From 9c62b8222257f518a4e608c61e4b3ebec917beb0 Mon Sep 17 00:00:00 2001 From: "Lorenzo L. Romero" Date: Fri, 26 Jun 2026 21:03:02 -0400 Subject: [PATCH] Route patch status replication over metadata network Move console patch status fanout and downstream FNE replication off the traffic network path and onto the metadata connection. Add registry change detection so duplicate or stale replicated snapshots do not advance revisions or get re-broadcast. --- src/fne/PatchStatusRegistry.cpp | 75 +++++++++++++++++++++++++++-- src/fne/PatchStatusRegistry.h | 5 +- src/fne/network/MetadataNetwork.cpp | 34 +++++++++++-- src/fne/network/MetadataNetwork.h | 3 ++ src/fne/network/TrafficNetwork.cpp | 19 +++++--- 5 files changed, 120 insertions(+), 16 deletions(-) diff --git a/src/fne/PatchStatusRegistry.cpp b/src/fne/PatchStatusRegistry.cpp index eb5870dc..afa75c9c 100644 --- a/src/fne/PatchStatusRegistry.cpp +++ b/src/fne/PatchStatusRegistry.cpp @@ -53,8 +53,11 @@ void PatchStatusRegistry::configure(uint32_t defaultTtlSeconds, uint32_t minTtlS /* Publishes a complete patch snapshot for one console peer. */ -bool PatchStatusRegistry::publish(json::object& request, json::object& response, std::string& errorMessage) +bool PatchStatusRegistry::publish(json::object& request, json::object& response, std::string& errorMessage, bool* changed) { + if (changed != nullptr) + *changed = false; + if (!request["peerId"].is()) { errorMessage = "peerId was not a valid integer"; return false; @@ -109,18 +112,35 @@ bool PatchStatusRegistry::publish(json::object& request, json::object& response, { std::lock_guard guard(m_mutex); auto existing = m_peerPatches.find(incoming.peerId); - if (existing != m_peerPatches.end() && incoming.sequence > 0U && existing->second.sequence > incoming.sequence) { + if (existing != m_peerPatches.end() && incoming.sequence > 0U && existing->second.sequence >= incoming.sequence && + existing->second.originFnePeerId == incoming.originFnePeerId) { response = snapshotLocked(); response["acceptedPeerId"].set(incoming.peerId); response["ttlSeconds"].set(ttlSeconds); return true; } - if (incoming.patches.empty()) - m_peerPatches.erase(incoming.peerId); - else + if (existing != m_peerPatches.end() && peerSnapshotsEqual(existing->second, incoming)) { + response = snapshotLocked(); + response["acceptedPeerId"].set(incoming.peerId); + response["ttlSeconds"].set(ttlSeconds); + return true; + } + + if (incoming.patches.empty()) { + if (m_peerPatches.erase(incoming.peerId) == 0U) { + response = snapshotLocked(); + response["acceptedPeerId"].set(incoming.peerId); + response["ttlSeconds"].set(ttlSeconds); + return true; + } + } + else { m_peerPatches[incoming.peerId] = incoming; + } bumpRevisionLocked(); + if (changed != nullptr) + *changed = true; response = snapshotLocked(); response["acceptedPeerId"].set(incoming.peerId); @@ -320,6 +340,51 @@ json::object PatchStatusRegistry::peerSnapshotToJson(const PeerPatchSnapshot& pe return obj; } +/* Compares two patch members for logical equality. */ + +bool PatchStatusRegistry::patchMembersEqual(const PatchMember& lhs, const PatchMember& rhs) +{ + return lhs.system == rhs.system && + lhs.mode == rhs.mode && + lhs.tgid == rhs.tgid && + lhs.slot == rhs.slot; +} + +/* Compares two patch records for logical equality. */ + +bool PatchStatusRegistry::patchRecordsEqual(const PatchRecord& lhs, const PatchRecord& rhs) +{ + if (lhs.patchId != rhs.patchId || lhs.active != rhs.active || lhs.oneWay != rhs.oneWay || lhs.members.size() != rhs.members.size()) + return false; + + for (size_t i = 0U; i < lhs.members.size(); i++) { + if (!patchMembersEqual(lhs.members[i], rhs.members[i])) + return false; + } + + return true; +} + +/* Compares two peer snapshots for logical equality. */ + +bool PatchStatusRegistry::peerSnapshotsEqual(const PeerPatchSnapshot& lhs, const PeerPatchSnapshot& rhs) +{ + if (lhs.peerId != rhs.peerId || + lhs.originFnePeerId != rhs.originFnePeerId || + lhs.peerName != rhs.peerName || + lhs.sequence != rhs.sequence || + lhs.patches.size() != rhs.patches.size()) { + return false; + } + + for (size_t i = 0U; i < lhs.patches.size(); i++) { + if (!patchRecordsEqual(lhs.patches[i], rhs.patches[i])) + return false; + } + + return true; +} + /* Parses one patch record from JSON. */ bool PatchStatusRegistry::parsePatch(json::object& obj, PatchRecord& patch, std::string& errorMessage) const diff --git a/src/fne/PatchStatusRegistry.h b/src/fne/PatchStatusRegistry.h index 3c8ee02c..4d647247 100644 --- a/src/fne/PatchStatusRegistry.h +++ b/src/fne/PatchStatusRegistry.h @@ -77,7 +77,7 @@ public: * @param errorMessage Validation error text populated when the request is invalid. * @returns bool True, if the publish request was valid and applied, otherwise false. */ - bool publish(json::object& request, json::object& response, std::string& errorMessage); + bool publish(json::object& request, json::object& response, std::string& errorMessage, bool* changed = nullptr); /** * @brief Removes all patch records associated with a console peer. * @param peerId Console peer ID whose records should be removed. @@ -193,6 +193,9 @@ private: * @returns json::object JSON peer patch snapshot. */ static json::object peerSnapshotToJson(const PeerPatchSnapshot& peer); + static bool patchMembersEqual(const PatchMember& lhs, const PatchMember& rhs); + static bool patchRecordsEqual(const PatchRecord& lhs, const PatchRecord& rhs); + static bool peerSnapshotsEqual(const PeerPatchSnapshot& lhs, const PeerPatchSnapshot& rhs); /** * @brief Parses one patch record from JSON. diff --git a/src/fne/network/MetadataNetwork.cpp b/src/fne/network/MetadataNetwork.cpp index b0356b98..606d0dec 100644 --- a/src/fne/network/MetadataNetwork.cpp +++ b/src/fne/network/MetadataNetwork.cpp @@ -164,6 +164,31 @@ void MetadataNetwork::close() m_status = NET_STAT_INVALID; } +/* Helper to send a metadata message to a peer's metadata port. */ + +bool MetadataNetwork::writePeerMetadata(FNEPeerConnection* connection, uint32_t ssrc, FrameQueue::OpcodePair opcode, const uint8_t* data, + uint32_t length, uint16_t pktSeq, uint32_t streamId) const +{ + if (connection == nullptr) + return false; + if (m_status != NET_STAT_MST_RUNNING) + return false; + if (m_frameQueue == nullptr) + return false; + + sockaddr_storage addr; + uint32_t addrLen = 0U; + uint16_t port = connection->port() + 1U; + + if (udp::Socket::lookup(connection->address(), port, addr, addrLen) != 0) { + LogWarning(LOG_NET, "PEER %u (%s) failed to resolve metadata endpoint %s:%u", connection->id(), + connection->identWithQualifier().c_str(), connection->address().c_str(), port); + return false; + } + + return m_frameQueue->write(data, length, streamId, connection->id(), ssrc, opcode, pktSeq, addr, addrLen); +} + // --------------------------------------------------------------------------- // Private Class Members // --------------------------------------------------------------------------- @@ -433,14 +458,17 @@ void MetadataNetwork::taskNetworkRx(NetPacketRequest* req) json::object response = json::object(); std::string errorMessage; - if (!network->patchStatusRegistry().publish(reqObj, response, errorMessage)) { + bool changed = false; + if (!network->patchStatusRegistry().publish(reqObj, response, errorMessage, &changed)) { LogWarning(LOG_MASTER, "PEER %u (%s) invalid patch status payload, %s", pktPeerId, connection->identWithQualifier().c_str(), errorMessage.c_str()); network->writePeerNAK(pktPeerId, network->createStreamId(), TAG_TRANSFER_PATCH_STATUS, NET_CONN_NAK_ILLEGAL_PACKET); break; } - network->writePatchStatusToConsoles(response); - network->replicatePatchStatus(reqObj); + if (changed) { + network->writePatchStatusToConsoles(response); + network->replicatePatchStatus(reqObj); + } } else { network->writePeerNAK(pktPeerId, network->createStreamId(), TAG_TRANSFER_PATCH_STATUS, NET_CONN_NAK_FNE_UNAUTHORIZED); diff --git a/src/fne/network/MetadataNetwork.h b/src/fne/network/MetadataNetwork.h index 4d516075..b0864c75 100644 --- a/src/fne/network/MetadataNetwork.h +++ b/src/fne/network/MetadataNetwork.h @@ -91,6 +91,9 @@ namespace network private: friend class TrafficNetwork; + bool writePeerMetadata(FNEPeerConnection* connection, uint32_t ssrc, FrameQueue::OpcodePair opcode, const uint8_t* data, + uint32_t length, uint16_t pktSeq, uint32_t streamId) const; + TrafficNetwork* m_trafficNetwork; HostFNE* m_host; diff --git a/src/fne/network/TrafficNetwork.cpp b/src/fne/network/TrafficNetwork.cpp index a358770f..4468d1b8 100644 --- a/src/fne/network/TrafficNetwork.cpp +++ b/src/fne/network/TrafficNetwork.cpp @@ -553,11 +553,15 @@ void TrafficNetwork::processReplicatedPatchStatus(uint32_t peerId, json::object json::object response = json::object(); std::string errorMessage; - if (!m_patchStatusRegistry.publish(obj, response, errorMessage)) { + bool changed = false; + if (!m_patchStatusRegistry.publish(obj, response, errorMessage, &changed)) { LogWarning(LOG_MASTER, "PEER %u invalid replicated patch status payload, %s", peerId, errorMessage.c_str()); return; } + if (!changed) + return; + writePatchStatusToConsoles(response); replicatePatchStatus(obj, peerId); } @@ -2525,6 +2529,8 @@ bool TrafficNetwork::writePatchStatusPayload(FNEPeerConnection* connection, json return false; if (!m_patchStatusEnabled) return false; + if (m_host->m_mdNetwork == nullptr) + return false; if (!connection->connected()) return false; if (connection->peerClass() != PEER_CONN_CLASS_CONSOLE) @@ -2543,15 +2549,12 @@ bool TrafficNetwork::writePatchStatusPayload(FNEPeerConnection* connection, json ::memset(buffer, 0x00U, DATA_PACKET_LENGTH); ::memcpy(buffer + 11U, payload.c_str(), len); - sockaddr_storage addr = connection->socketStorage(); - uint32_t addrLen = connection->sockStorageLen(); - if (m_debug) { LogDebug(LOG_MASTER, "PEER %u (%s) sending patch status registry, len = %u", connection->id(), connection->identWithQualifier().c_str(), len); } - return m_frameQueue->write(buffer, len + 11U, createStreamId(), connection->id(), m_peerId, - { NET_FUNC::TRANSFER, NET_SUBFUNC::TRANSFER_SUBFUNC_PATCH_STATUS }, RTP_END_OF_CALL_SEQ, addr, addrLen); + return m_host->m_mdNetwork->writePeerMetadata(connection, m_peerId, + { NET_FUNC::TRANSFER, NET_SUBFUNC::TRANSFER_SUBFUNC_PATCH_STATUS }, buffer, len + 11U, RTP_END_OF_CALL_SEQ, createStreamId()); } /* Helper to serialize and queue a patch status replication payload. */ @@ -2562,6 +2565,8 @@ bool TrafficNetwork::writePatchStatusReplicationPayload(FNEPeerConnection* conne return false; if (!m_patchStatusEnabled) return false; + if (m_host->m_mdNetwork == nullptr) + return false; if (!connection->connected()) return false; if (connection->peerClass() != PEER_CONN_CLASS_NEIGHBOR || !connection->isReplica()) @@ -2585,7 +2590,7 @@ bool TrafficNetwork::writePatchStatusReplicationPayload(FNEPeerConnection* conne connection->identWithQualifier().c_str(), pkt.fragments.size(), streamId); if (pkt.fragments.size() > 0U) { for (auto frag : pkt.fragments) { - writePeer(connection->id(), m_peerId, { NET_FUNC::REPL, NET_SUBFUNC::REPL_PATCH_STATUS }, + m_host->m_mdNetwork->writePeerMetadata(connection, m_peerId, { NET_FUNC::REPL, NET_SUBFUNC::REPL_PATCH_STATUS }, frag.second->data, FRAG_SIZE, RTP_END_OF_CALL_SEQ, streamId); Thread::sleep(60U); // pace block transmission }