From 067b622fe9f7989521c98c3dcd393c5f6b407ade Mon Sep 17 00:00:00 2001 From: Bryan Biedenkapp Date: Sat, 17 Oct 2026 16:23:22 -0400 Subject: [PATCH] aggregate peer affiliations and unit registrations upstream; --- src/common/lookups/AffiliationLookup.cpp | 56 +++++ src/common/lookups/AffiliationLookup.h | 14 ++ src/fne/lookups/AffiliationLookup.cpp | 14 ++ src/fne/lookups/AffiliationLookup.h | 5 + src/fne/network/TrafficNetwork.cpp | 297 +++++++++++++++++++++-- src/fne/network/TrafficNetwork.h | 70 +++++- src/fne/network/packet/Announce.cpp | 176 ++++++-------- tests/fne/FNENetwork_Tests.cpp | 34 +++ 8 files changed, 539 insertions(+), 127 deletions(-) diff --git a/src/common/lookups/AffiliationLookup.cpp b/src/common/lookups/AffiliationLookup.cpp index a441c4ad..b9fc640c 100644 --- a/src/common/lookups/AffiliationLookup.cpp +++ b/src/common/lookups/AffiliationLookup.cpp @@ -202,6 +202,34 @@ bool AffiliationLookup::isUnitReg(uint32_t srcId) const } } +/* Atomically replaces the unit registration table with a peer snapshot. */ + +void AffiliationLookup::replaceUnitRegistrations(const std::vector& registrations) +{ + std::vector replacement = registrations; + std::sort(replacement.begin(), replacement.end()); + replacement.erase(std::unique(replacement.begin(), replacement.end()), replacement.end()); + + const std::unordered_map currentTimers = m_unitRegTimers.get(); + std::unordered_map replacementTimers; + replacementTimers.reserve(replacement.size()); + + for (const uint32_t srcId : replacement) { + const auto currentTimer = currentTimers.find(srcId); + if (currentTimer != currentTimers.end()) { + replacementTimers.emplace(srcId, currentTimer->second); + } + else { + Timer timer(1000U, UNIT_REG_TIMEOUT); + timer.start(); + replacementTimers.emplace(srcId, timer); + } + } + + m_unitRegTable = replacement; + m_unitRegTimers = replacementTimers; +} + /* Helper to release unit registrations. */ void AffiliationLookup::clearUnitReg() @@ -299,6 +327,34 @@ bool AffiliationLookup::isGroupAff(uint32_t srcId, uint32_t dstId) const return false; } +/* Atomically replaces the group affiliation table with a peer snapshot. */ + +void AffiliationLookup::replaceGroupAffiliations(const std::unordered_map& affiliations) +{ + const std::unordered_map currentAffiliations = m_grpAffTable.get(); + const std::unordered_map currentTimers = m_grpAffTimers.get(); + std::unordered_map replacementTimers; + replacementTimers.reserve(affiliations.size()); + + for (const auto& affiliation : affiliations) { + const auto currentAffiliation = currentAffiliations.find(affiliation.first); + const auto currentTimer = currentTimers.find(affiliation.first); + if (currentAffiliation != currentAffiliations.end() && currentAffiliation->second == affiliation.second && + currentTimer != currentTimers.end()) { + replacementTimers.emplace(affiliation.first, currentTimer->second); + } + else { + Timer timer(1000U, GRP_AFF_TIMEOUT); + timer.start(); + replacementTimers.emplace(affiliation.first, timer); + } + } + + // Routing readers observe either the old or complete replacement table. + m_grpAffTable = affiliations; + m_grpAffTimers = replacementTimers; +} + /* Helper to release group affiliations. */ std::vector AffiliationLookup::clearGroupAff(uint32_t dstId, bool releaseAll) diff --git a/src/common/lookups/AffiliationLookup.h b/src/common/lookups/AffiliationLookup.h index 206092c0..325a4c35 100644 --- a/src/common/lookups/AffiliationLookup.h +++ b/src/common/lookups/AffiliationLookup.h @@ -103,6 +103,13 @@ namespace lookups * @returns bool True, if source ID has registered, otherwise false. */ virtual bool isUnitReg(uint32_t srcId) const; + /** + * @brief Atomically replaces the unit registration table with a peer snapshot. + * @param[in] registrations Complete source ID registration snapshot. + * @note Timers for unchanged registrations are preserved; new registrations + * receive new timers. + */ + virtual void replaceUnitRegistrations(const std::vector& registrations); /** * @brief Helper to release unit registrations. */ @@ -145,6 +152,13 @@ namespace lookups * @returns bool True, if the source ID is affiliated to the talkgroup ID, otherwise false. */ virtual bool isGroupAff(uint32_t srcId, uint32_t dstId) const; + /** + * @brief Atomically replaces the group affiliation table with a peer snapshot. + * @param[in] affiliations Complete source-to-talkgroup affiliation snapshot. + * @note Timers for unchanged affiliations are preserved; new and changed + * affiliations receive new timers. + */ + virtual void replaceGroupAffiliations(const std::unordered_map& affiliations); /** * @brief Helper to release group affiliations. * @param dstId Talkgroup ID. diff --git a/src/fne/lookups/AffiliationLookup.cpp b/src/fne/lookups/AffiliationLookup.cpp index 6a7a32e8..03ad7bff 100644 --- a/src/fne/lookups/AffiliationLookup.cpp +++ b/src/fne/lookups/AffiliationLookup.cpp @@ -34,6 +34,7 @@ AffiliationLookup::~AffiliationLookup() = default; void AffiliationLookup::unitReg(uint32_t srcId, uint32_t ssrc) { if (isUnitReg(srcId)) { + m_unitRegPeerTable[srcId] = ssrc; return; } @@ -92,6 +93,19 @@ uint32_t AffiliationLookup::getSSRCByUnitReg(uint32_t srcId) return 0U; } +/* Atomically replaces unit registrations and their originating peers. */ + +void AffiliationLookup::replaceUnitRegistrations(const std::unordered_map& registrations) +{ + std::vector sourceIds; + sourceIds.reserve(registrations.size()); + for (const auto& registration : registrations) + sourceIds.push_back(registration.first); + + ::lookups::AffiliationLookup::replaceUnitRegistrations(sourceIds); + m_unitRegPeerTable = registrations; +} + /* Helper to release unit registrations. */ void AffiliationLookup::clearUnitReg() diff --git a/src/fne/lookups/AffiliationLookup.h b/src/fne/lookups/AffiliationLookup.h index 6c9f7a00..c793d8ab 100644 --- a/src/fne/lookups/AffiliationLookup.h +++ b/src/fne/lookups/AffiliationLookup.h @@ -75,6 +75,11 @@ namespace fne_lookups * @returns uint32_t Originating Peer ID. */ uint32_t getSSRCByUnitReg(uint32_t srcId); + /** + * @brief Atomically replaces unit registrations and their originating peers. + * @param[in] registrations Map of source IDs to originating peer IDs. + */ + void replaceUnitRegistrations(const std::unordered_map& registrations); /** * @brief Helper to release unit registrations. */ diff --git a/src/fne/network/TrafficNetwork.cpp b/src/fne/network/TrafficNetwork.cpp index 60e60639..36a4d900 100644 --- a/src/fne/network/TrafficNetwork.cpp +++ b/src/fne/network/TrafficNetwork.cpp @@ -109,6 +109,8 @@ TrafficNetwork::TrafficNetwork(HostFNE* host, const std::string& address, uint16 m_peerReplicaPeers(), m_peerReplicaPeersMutex(), m_peerAffiliations(), + m_peerAffiliationsMutex(), + m_affiliationUpdateMutex(), m_ccPeerMap(), m_peerReplicaKeyQueue(), m_peerReplicaKEKKeyQueue(), @@ -1584,44 +1586,81 @@ void TrafficNetwork::eraseStreamPktSeq(uint32_t peerId, uint32_t streamId) void TrafficNetwork::createPeerAffiliations(uint32_t peerId, std::string peerName) { - std::lock_guard lock(m_peerAffiliationsMutex); - - auto it = m_peerAffiliations.find(peerId); - if (it != m_peerAffiliations.end()) { - m_peerAffiliations.erase(peerId); - } + bool replaced = false; + std::unordered_map aggregate; + std::vector aggregateRegistrations; + { + std::lock_guard updateLock(m_affiliationUpdateMutex); + { + std::lock_guard peerLock(m_peerAffiliationsMutex); - lookups::ChannelLookup* chLookup = new lookups::ChannelLookup(); - std::shared_ptr aff( - new fne_lookups::AffiliationLookup(peerName, chLookup, m_verbose), - [](fne_lookups::AffiliationLookup* p) { - if (p != nullptr) { - lookups::ChannelLookup* rfCh = p->rfCh(); - if (rfCh != nullptr) { - delete rfCh; - } - delete p; + auto it = m_peerAffiliations.find(peerId); + if (it != m_peerAffiliations.end()) { + m_peerAffiliations.erase(peerId); + replaced = true; } - }); - aff->setDisableUnitRegTimeout(true); // FNE doesn't allow unit registration timeouts (notification must come from the peers) - aff->setDisableGrpAffTimeout(true); // FNE doesn't allow group affiliation timeouts (notification must come from the peers) - m_peerAffiliations.insert(peerId, aff); + lookups::ChannelLookup* chLookup = new lookups::ChannelLookup(); + std::shared_ptr aff( + new fne_lookups::AffiliationLookup(peerName, chLookup, m_verbose), + [](fne_lookups::AffiliationLookup* p) { + if (p != nullptr) { + lookups::ChannelLookup* rfCh = p->rfCh(); + if (rfCh != nullptr) { + delete rfCh; + } + delete p; + } + }); + + aff->setDisableUnitRegTimeout(true); // FNE doesn't allow unit registration timeouts (notification must come from the peers) + aff->setDisableGrpAffTimeout(true); // FNE doesn't allow group affiliation timeouts (notification must come from the peers) + m_peerAffiliations.insert(peerId, aff); + } + + if (replaced) { + aggregate = buildAggregatedAffiliations(); + aggregateRegistrations = buildAggregatedUnitRegistrations(); + } + } + + if (replaced) { + publishAggregatedAffiliations(aggregate); + publishAggregatedUnitRegistrations(aggregateRegistrations); + } } /* Helper to erase the peer from the peers affiliations list. */ bool TrafficNetwork::erasePeerAffiliations(uint32_t peerId) { - std::lock_guard lock(m_peerAffiliationsMutex); + bool erased = false; + std::unordered_map aggregate; + std::vector aggregateRegistrations; + { + std::lock_guard updateLock(m_affiliationUpdateMutex); + { + std::lock_guard peerLock(m_peerAffiliationsMutex); - auto it = m_peerAffiliations.find(peerId); - if (it != m_peerAffiliations.end()) { - m_peerAffiliations.erase(peerId); - return true; + auto it = m_peerAffiliations.find(peerId); + if (it != m_peerAffiliations.end()) { + m_peerAffiliations.erase(peerId); + erased = true; + } + } + + if (erased) { + aggregate = buildAggregatedAffiliations(); + aggregateRegistrations = buildAggregatedUnitRegistrations(); + } } - return false; + if (erased) { + publishAggregatedAffiliations(aggregate); + publishAggregatedUnitRegistrations(aggregateRegistrations); + } + + return erased; } /* Helper to get the peer affiliations entry for a peer. */ @@ -1653,6 +1692,212 @@ std::vector TrafficNetwork::peerAffiliat return snapshot; } +/* Adds or updates one peer affiliation and publishes the aggregate upstream. */ + +bool TrafficNetwork::updatePeerAffiliation(uint32_t peerId, uint32_t srcId, uint32_t dstId) +{ + std::unordered_map aggregate; + { + std::lock_guard updateLock(m_affiliationUpdateMutex); + std::shared_ptr affiliations = getPeerAffiliations(peerId); + if (affiliations == nullptr) + return false; + + affiliations->groupUnaff(srcId); + affiliations->groupAff(srcId, dstId); + aggregate = buildAggregatedAffiliations(); + } + + publishAggregatedAffiliations(aggregate); + return true; +} + +/* Removes one peer affiliation and publishes the aggregate upstream. */ + +bool TrafficNetwork::removePeerAffiliation(uint32_t peerId, uint32_t srcId) +{ + std::unordered_map aggregate; + { + std::lock_guard updateLock(m_affiliationUpdateMutex); + std::shared_ptr affiliations = getPeerAffiliations(peerId); + if (affiliations == nullptr) + return false; + + affiliations->groupUnaff(srcId); + aggregate = buildAggregatedAffiliations(); + } + + publishAggregatedAffiliations(aggregate); + return true; +} + +/* Replaces one peer's affiliations and publishes the aggregate upstream. */ + +bool TrafficNetwork::replacePeerAffiliations(uint32_t peerId, const std::unordered_map& affiliations) +{ + std::unordered_map aggregate; + { + std::lock_guard updateLock(m_affiliationUpdateMutex); + std::shared_ptr peerAffiliations = getPeerAffiliations(peerId); + if (peerAffiliations == nullptr) + return false; + + peerAffiliations->replaceGroupAffiliations(affiliations); + aggregate = buildAggregatedAffiliations(); + } + + publishAggregatedAffiliations(aggregate); + return true; +} + +/* Rebuilds the aggregate downstream affiliation snapshot. */ + +std::unordered_map TrafficNetwork::buildAggregatedAffiliations() +{ + std::vector peerSnapshot = peerAffiliationsSnapshot(); + std::sort(peerSnapshot.begin(), peerSnapshot.end(), [](const PeerAffiliationMapPair& lhs, const PeerAffiliationMapPair& rhs) { + return lhs.first < rhs.first; + }); + + std::unordered_map aggregate; + for (const auto& peer : peerSnapshot) { + if (peer.second == nullptr) + continue; + + const std::unordered_map affiliations = peer.second->grpAffTable(); + for (const auto& affiliation : affiliations) { + const auto result = aggregate.emplace(affiliation.first, affiliation.second); + if (!result.second && result.first->second != affiliation.second) { + LogWarning(LOG_MASTER, "PEER %u affiliation conflict, srcId = %u, dstId = %u already represented by dstId = %u from a lower peer ID", + peer.first, affiliation.first, affiliation.second, result.first->second); + } + } + } + + m_globalAff->replaceGroupAffiliations(aggregate); + return aggregate; +} + +/* Publishes an aggregate downstream affiliation snapshot upstream. */ + +void TrafficNetwork::publishAggregatedAffiliations(const std::unordered_map& affiliations) +{ + for (auto& peer : m_host->m_peerNetworks) { + if (peer.second != nullptr && peer.second->isEnabled() && peer.second->isReplica()) + peer.second->announceAffiliationUpdate(affiliations); + } +} + +/* Adds one peer unit registration and publishes the aggregate upstream. */ + +bool TrafficNetwork::updatePeerUnitRegistration(uint32_t peerId, uint32_t srcId, uint32_t ssrc) +{ + std::vector aggregate; + { + std::lock_guard updateLock(m_affiliationUpdateMutex); + std::shared_ptr registrations = getPeerAffiliations(peerId); + if (registrations == nullptr) + return false; + + registrations->unitReg(srcId, ssrc); + aggregate = buildAggregatedUnitRegistrations(); + } + + publishAggregatedUnitRegistrations(aggregate); + return true; +} + +/* Removes one peer unit registration and publishes the aggregate upstream. */ + +bool TrafficNetwork::removePeerUnitRegistration(uint32_t peerId, uint32_t srcId) +{ + std::vector aggregate; + std::unordered_map aggregateAffiliations; + { + std::lock_guard updateLock(m_affiliationUpdateMutex); + std::shared_ptr registrations = getPeerAffiliations(peerId); + if (registrations == nullptr) + return false; + + registrations->unitDereg(srcId); + aggregate = buildAggregatedUnitRegistrations(); + aggregateAffiliations = buildAggregatedAffiliations(); + } + + publishAggregatedUnitRegistrations(aggregate); + publishAggregatedAffiliations(aggregateAffiliations); + return true; +} + +/* Replaces one peer's unit registrations and publishes the aggregate upstream. */ + +bool TrafficNetwork::replacePeerUnitRegistrations(uint32_t peerId, const std::vector& registrations, uint32_t ssrc) +{ + std::vector aggregate; + { + std::lock_guard updateLock(m_affiliationUpdateMutex); + std::shared_ptr peerRegistrations = getPeerAffiliations(peerId); + if (peerRegistrations == nullptr) + return false; + + std::unordered_map registrationOrigins; + registrationOrigins.reserve(registrations.size()); + for (const uint32_t srcId : registrations) + registrationOrigins[srcId] = ssrc; + + peerRegistrations->replaceUnitRegistrations(registrationOrigins); + aggregate = buildAggregatedUnitRegistrations(); + } + + publishAggregatedUnitRegistrations(aggregate); + return true; +} + +/* Rebuilds the aggregate downstream unit registration snapshot. */ + +std::vector TrafficNetwork::buildAggregatedUnitRegistrations() +{ + std::vector peerSnapshot = peerAffiliationsSnapshot(); + std::sort(peerSnapshot.begin(), peerSnapshot.end(), [](const PeerAffiliationMapPair& lhs, const PeerAffiliationMapPair& rhs) { + return lhs.first < rhs.first; + }); + + std::unordered_map aggregateOrigins; + for (const auto& peer : peerSnapshot) { + if (peer.second == nullptr) + continue; + + const std::vector registrations = peer.second->unitRegTable(); + for (const uint32_t srcId : registrations) { + const uint32_t ssrc = peer.second->getSSRCByUnitReg(srcId); + const auto result = aggregateOrigins.emplace(srcId, ssrc); + if (!result.second && result.first->second != ssrc) { + LogWarning(LOG_MASTER, "PEER %u unit registration conflict, srcId = %u already represented by lower peer ID", + peer.first, srcId); + } + } + } + + m_globalAff->replaceUnitRegistrations(aggregateOrigins); + + std::vector aggregate; + aggregate.reserve(aggregateOrigins.size()); + for (const auto& registration : aggregateOrigins) + aggregate.push_back(registration.first); + std::sort(aggregate.begin(), aggregate.end()); + return aggregate; +} + +/* Publishes an aggregate downstream unit registration snapshot upstream. */ + +void TrafficNetwork::publishAggregatedUnitRegistrations(const std::vector& registrations) +{ + for (auto& peer : m_host->m_peerNetworks) { + if (peer.second != nullptr && peer.second->isEnabled() && peer.second->isReplica()) + peer.second->announceUnitRegUpdate(registrations); + } +} + /* Helper to disconnect a downstream peer. */ void TrafficNetwork::disconnectPeer(uint32_t peerId, const PeerConnectionPtr& connection) diff --git a/src/fne/network/TrafficNetwork.h b/src/fne/network/TrafficNetwork.h index f0f1c722..b625e4a8 100644 --- a/src/fne/network/TrafficNetwork.h +++ b/src/fne/network/TrafficNetwork.h @@ -404,9 +404,10 @@ namespace network mutable std::shared_timed_mutex m_peersMutex; std::unordered_map m_peerReplicaPeers; mutable std::mutex m_peerReplicaPeersMutex; - typedef std::pair> PeerAffiliationMapPair; + typedef std::pair> PeerAffiliationMapPair; concurrent::unordered_map> m_peerAffiliations; mutable std::mutex m_peerAffiliationsMutex; + mutable std::mutex m_affiliationUpdateMutex; concurrent::shared_unordered_map> m_ccPeerMap; static std::timed_mutex s_keyQueueMutex; std::unordered_map m_peerReplicaKeyQueue; @@ -783,6 +784,73 @@ namespace network * @returns std::vector Snapshot of peer affiliation entries. */ std::vector peerAffiliationsSnapshot() const; + /** + * @brief Adds or updates one peer affiliation and publishes the aggregate upstream. + * @param peerId Peer ID. + * @param srcId Source Radio ID. + * @param dstId Talkgroup ID. + * @returns bool True, if the peer affiliation lookup exists, otherwise false. + */ + bool updatePeerAffiliation(uint32_t peerId, uint32_t srcId, uint32_t dstId); + /** + * @brief Removes one peer affiliation and publishes the aggregate upstream. + * @param peerId Peer ID. + * @param srcId Source Radio ID. + * @returns bool True, if the peer affiliation lookup exists, otherwise false. + */ + bool removePeerAffiliation(uint32_t peerId, uint32_t srcId); + /** + * @brief Replaces one peer's affiliations and publishes the aggregate upstream. + * @param peerId Peer ID. + * @param affiliations Complete source-to-talkgroup affiliation snapshot. + * @returns bool True, if the peer affiliation lookup exists, otherwise false. + */ + bool replacePeerAffiliations(uint32_t peerId, const std::unordered_map& affiliations); + /** + * @brief Rebuilds the aggregate downstream affiliation snapshot. + * @note The caller must hold m_affiliationUpdateMutex. + * @returns std::unordered_map Aggregate affiliation snapshot. + */ + std::unordered_map buildAggregatedAffiliations(); + /** + * @brief Publishes an aggregate downstream affiliation snapshot upstream. + * @param affiliations Complete aggregate source-to-talkgroup affiliation snapshot. + */ + void publishAggregatedAffiliations(const std::unordered_map& affiliations); + /** + * @brief Adds one peer unit registration and publishes the aggregate upstream. + * @param peerId Peer ID. + * @param srcId Source Radio ID. + * @param ssrc Originating peer ID. + * @returns bool True, if the peer affiliation lookup exists, otherwise false. + */ + bool updatePeerUnitRegistration(uint32_t peerId, uint32_t srcId, uint32_t ssrc); + /** + * @brief Removes one peer unit registration and publishes the aggregate upstream. + * @param peerId Peer ID. + * @param srcId Source Radio ID. + * @returns bool True, if the peer affiliation lookup exists, otherwise false. + */ + bool removePeerUnitRegistration(uint32_t peerId, uint32_t srcId); + /** + * @brief Replaces one peer's unit registrations and publishes the aggregate upstream. + * @param peerId Peer ID. + * @param registrations Complete source ID registration snapshot. + * @param ssrc Originating peer ID. + * @returns bool True, if the peer affiliation lookup exists, otherwise false. + */ + bool replacePeerUnitRegistrations(uint32_t peerId, const std::vector& registrations, uint32_t ssrc); + /** + * @brief Rebuilds the aggregate downstream unit registration snapshot. + * @note The caller must hold m_affiliationUpdateMutex. + * @returns std::vector Aggregate unit registration snapshot. + */ + std::vector buildAggregatedUnitRegistrations(); + /** + * @brief Publishes an aggregate downstream unit registration snapshot upstream. + * @param registrations Complete aggregate source ID registration snapshot. + */ + void publishAggregatedUnitRegistrations(const std::vector& registrations); /** * @brief Helper to disconnect a downstream peer. * @param peerId Peer ID. diff --git a/src/fne/network/packet/Announce.cpp b/src/fne/network/packet/Announce.cpp index f383c348..7b96b5cf 100644 --- a/src/fne/network/packet/Announce.cpp +++ b/src/fne/network/packet/Announce.cpp @@ -40,21 +40,15 @@ void MetadataNetwork::PacketHandler::announce(TrafficNetwork* network, MetadataN // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && aff != nullptr) { - uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address - uint32_t dstId = GET_UINT24(req->buffer, 3U); // Destination Address - aff->groupUnaff(srcId); - aff->groupAff(srcId, dstId); - - // attempt to repeat traffic to replica masters - if (network->m_host->m_peerNetworks.size() > 0) { - for (auto& peer : network->m_host->m_peerNetworks) { - if (peer.second != nullptr) { - if (peer.second->isEnabled() && peer.second->isReplica()) { - peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_GRP_AFFIL }, - req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); - } - } - } + if (req->length >= static_cast(MSG_ANNC_GRP_AFFIL)) { + uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address + uint32_t dstId = GET_UINT24(req->buffer, 3U); // Destination Address + network->updatePeerAffiliation(peerId, srcId, dstId); + } + else { + LogWarning(LOG_MASTER, "PEER %u (%s) announced malformed group affiliation, length = %u", + peerId, connection->identWithQualifier().c_str(), req->length); + network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } } else { @@ -77,20 +71,14 @@ void MetadataNetwork::PacketHandler::announce(TrafficNetwork* network, MetadataN // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && aff != nullptr) { - uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address - aff->unitReg(srcId, ssrc); - network->m_globalAff->unitReg(srcId, ssrc); - - // attempt to repeat traffic to replica masters - if (network->m_host->m_peerNetworks.size() > 0) { - for (auto& peer : network->m_host->m_peerNetworks) { - if (peer.second != nullptr) { - if (peer.second->isEnabled() && peer.second->isReplica()) { - peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_UNIT_REG }, - req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true, 0U, ssrc); - } - } - } + if (req->length >= static_cast(MSG_ANNC_UNIT_REG)) { + uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address + network->updatePeerUnitRegistration(peerId, srcId, ssrc); + } + else { + LogWarning(LOG_MASTER, "PEER %u (%s) announced malformed unit registration, length = %u", + peerId, connection->identWithQualifier().c_str(), req->length); + network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } } else { @@ -112,20 +100,14 @@ void MetadataNetwork::PacketHandler::announce(TrafficNetwork* network, MetadataN // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && aff != nullptr) { - uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address - aff->unitDereg(srcId); - network->m_globalAff->unitDereg(srcId); - - // attempt to repeat traffic to replica masters - if (network->m_host->m_peerNetworks.size() > 0) { - for (auto& peer : network->m_host->m_peerNetworks) { - if (peer.second != nullptr) { - if (peer.second->isEnabled() && peer.second->isReplica()) { - peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_UNIT_DEREG }, - req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); - } - } - } + if (req->length >= static_cast(MSG_ANNC_UNIT_REG)) { + uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address + network->removePeerUnitRegistration(peerId, srcId); + } + else { + LogWarning(LOG_MASTER, "PEER %u (%s) announced malformed unit deregistration, length = %u", + peerId, connection->identWithQualifier().c_str(), req->length); + network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } } else { @@ -148,20 +130,14 @@ void MetadataNetwork::PacketHandler::announce(TrafficNetwork* network, MetadataN // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && aff != nullptr) { - uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address - aff->groupUnaff(srcId); - network->m_globalAff->groupUnaff(srcId); - - // attempt to repeat traffic to replica masters - if (network->m_host->m_peerNetworks.size() > 0) { - for (auto& peer : network->m_host->m_peerNetworks) { - if (peer.second != nullptr) { - if (peer.second->isEnabled() && peer.second->isReplica()) { - peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_GRP_UNAFFIL }, - req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); - } - } - } + if (req->length >= static_cast(MSG_ANNC_GRP_UNAFFIL)) { + uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address + network->removePeerAffiliation(peerId, srcId); + } + else { + LogWarning(LOG_MASTER, "PEER %u (%s) announced malformed group unaffiliation, length = %u", + peerId, connection->identWithQualifier().c_str(), req->length); + network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } } else { @@ -185,33 +161,33 @@ void MetadataNetwork::PacketHandler::announce(TrafficNetwork* network, MetadataN network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } - if (aff != nullptr) { - aff->clearGroupAff(0U, true); + if (aff != nullptr && req->length >= 4) { + const uint32_t len = GET_UINT32(req->buffer, 0U); + const uint32_t availableEntries = (static_cast(req->length) - 4U) / 8U; + if (len != availableEntries || static_cast(req->length) != 4U + (len * 8U)) { + LogWarning(LOG_MASTER, "PEER %u (%s) announced malformed affiliations, count = %u, length = %u", + peerId, connection->identWithQualifier().c_str(), len, req->length); + network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); + break; + } - // update TGID lists - uint32_t len = GET_UINT32(req->buffer, 0U); + std::unordered_map affiliations; + affiliations.reserve(len); uint32_t offs = 4U; - for (uint32_t i = 0; i < len; i++) { - uint32_t srcId = GET_UINT24(req->buffer, offs); - uint32_t dstId = GET_UINT24(req->buffer, offs + 4U); - - aff->groupAff(srcId, dstId); - network->m_globalAff->groupAff(srcId, dstId); + for (uint32_t i = 0U; i < len; i++) { + const uint32_t srcId = GET_UINT24(req->buffer, offs); + const uint32_t dstId = GET_UINT24(req->buffer, offs + 4U); + affiliations[srcId] = dstId; offs += 8U; } - LogInfoEx(LOG_MASTER, "PEER %u (%s) announced %u affiliations", peerId, connection->identWithQualifier().c_str(), len); - // attempt to repeat traffic to replica masters - if (network->m_host->m_peerNetworks.size() > 0) { - for (auto& peer : network->m_host->m_peerNetworks) { - if (peer.second != nullptr) { - if (peer.second->isEnabled() && peer.second->isReplica()) { - peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_AFFILS }, - req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); - } - } - } - } + network->replacePeerAffiliations(peerId, affiliations); + LogInfoEx(LOG_MASTER, "PEER %u (%s) announced %u affiliations", peerId, connection->identWithQualifier().c_str(), len); + } + else if (aff != nullptr) { + LogWarning(LOG_MASTER, "PEER %u (%s) announced malformed affiliations, length = %u", + peerId, connection->identWithQualifier().c_str(), req->length); + network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } } else { @@ -235,32 +211,32 @@ void MetadataNetwork::PacketHandler::announce(TrafficNetwork* network, MetadataN network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } - if (aff != nullptr) { - aff->clearUnitReg(); + if (aff != nullptr && req->length >= 4) { + const uint32_t len = GET_UINT32(req->buffer, 0U); + const uint32_t availableEntries = (static_cast(req->length) - 4U) / 3U; + if (len != availableEntries || static_cast(req->length) != 4U + (len * 3U)) { + LogWarning(LOG_MASTER, "PEER %u (%s) announced malformed unit registrations, count = %u, length = %u", + peerId, connection->identWithQualifier().c_str(), len, req->length); + network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); + break; + } - // update unit registration lists - uint32_t len = GET_UINT32(req->buffer, 0U); + std::vector registrations; + registrations.reserve(len); uint32_t offs = 4U; - for (uint32_t i = 0; i < len; i++) { - uint32_t srcId = GET_UINT24(req->buffer, offs); - - aff->unitReg(srcId, ssrc); - network->m_globalAff->unitReg(srcId, ssrc); + for (uint32_t i = 0U; i < len; i++) { + const uint32_t srcId = GET_UINT24(req->buffer, offs); + registrations.push_back(srcId); offs += 3U; } - LogInfoEx(LOG_MASTER, "PEER %u (%s) announced %u unit registrations", peerId, connection->identWithQualifier().c_str(), len); - // attempt to repeat traffic to replica masters - if (network->m_host->m_peerNetworks.size() > 0) { - for (auto& peer : network->m_host->m_peerNetworks) { - if (peer.second != nullptr) { - if (peer.second->isEnabled() && peer.second->isReplica()) { - peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_UNIT_REGS }, - req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); - } - } - } - } + network->replacePeerUnitRegistrations(peerId, registrations, ssrc); + LogInfoEx(LOG_MASTER, "PEER %u (%s) announced %u unit registrations", peerId, connection->identWithQualifier().c_str(), len); + } + else if (aff != nullptr) { + LogWarning(LOG_MASTER, "PEER %u (%s) announced malformed unit registrations, length = %u", + peerId, connection->identWithQualifier().c_str(), req->length); + network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } } else { diff --git a/tests/fne/FNENetwork_Tests.cpp b/tests/fne/FNENetwork_Tests.cpp index f9e85eb2..84dc4d50 100644 --- a/tests/fne/FNENetwork_Tests.cpp +++ b/tests/fne/FNENetwork_Tests.cpp @@ -12,8 +12,10 @@ #include "fne/FNETestHooks.h" #include "fne/HostFNE.h" +#include "common/lookups/AffiliationLookup.h" #include +#include #include // --------------------------------------------------------------------------- @@ -52,6 +54,38 @@ public: network::MetadataNetwork metadata; }; +TEST_CASE("Affiliation snapshots replace prior state", "[fne][affiliations]") +{ + lookups::AffiliationLookup affiliations("test-affiliations", nullptr, false); + affiliations.groupAff(1001U, 3100U); + affiliations.groupAff(1002U, 3200U); + + std::unordered_map replacement = { + { 1002U, 3300U }, + { 1003U, 3400U } + }; + affiliations.replaceGroupAffiliations(replacement); + + REQUIRE(affiliations.grpAffTable() == replacement); + + affiliations.replaceGroupAffiliations({}); + REQUIRE(affiliations.grpAffTable().empty()); +} + +TEST_CASE("Unit registration snapshots replace prior state", "[fne][registrations]") +{ + lookups::AffiliationLookup registrations("test-registrations", nullptr, false); + registrations.unitReg(1001U); + registrations.unitReg(1002U); + + registrations.replaceUnitRegistrations({ 1002U, 1003U, 1003U }); + + REQUIRE(registrations.unitRegTable() == std::vector{ 1002U, 1003U }); + + registrations.replaceUnitRegistrations({}); + REQUIRE(registrations.unitRegTable().empty()); +} + TEST_CASE("FNE test hooks expose peer connection state", "[fne][hooks]") { FNEHarness harness;