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 }