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

@@ -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<uint32_t>()) {
         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<std::mutex> 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<uint32_t>(incoming.peerId);
             response["ttlSeconds"].set<uint32_t>(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<uint32_t>(incoming.peerId);
+            response["ttlSeconds"].set<uint32_t>(ttlSeconds);
+            return true;
+        }
+
+        if (incoming.patches.empty()) {
+            if (m_peerPatches.erase(incoming.peerId) == 0U) {
+                response = snapshotLocked();
+                response["acceptedPeerId"].set<uint32_t>(incoming.peerId);
+                response["ttlSeconds"].set<uint32_t>(ttlSeconds);
+                return true;
+            }
+        }
+        else {
             m_peerPatches[incoming.peerId] = incoming;
+        }
         bumpRevisionLocked();
+        if (changed != nullptr)
+            *changed = true;
 
         response = snapshotLocked();
         response["acceptedPeerId"].set<uint32_t>(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

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.

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);

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;
 

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
         }