aggregate peer affiliations and unit registrations upstream;

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<uint32_t>& registrations)
+{
+    std::vector<uint32_t> replacement = registrations;
+    std::sort(replacement.begin(), replacement.end());
+    replacement.erase(std::unique(replacement.begin(), replacement.end()), replacement.end());
+
+    const std::unordered_map<uint32_t, Timer> currentTimers = m_unitRegTimers.get();
+    std::unordered_map<uint32_t, Timer> 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<uint32_t, uint32_t>& affiliations)
+{
+    const std::unordered_map<uint32_t, uint32_t> currentAffiliations = m_grpAffTable.get();
+    const std::unordered_map<uint32_t, Timer> currentTimers = m_grpAffTimers.get();
+    std::unordered_map<uint32_t, Timer> 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<uint32_t> AffiliationLookup::clearGroupAff(uint32_t dstId, bool releaseAll)

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<uint32_t>& 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<uint32_t, uint32_t>& affiliations);
         /**
          * @brief Helper to release group affiliations.
          * @param dstId Talkgroup ID.

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<uint32_t, uint32_t>& registrations)
+{
+    std::vector<uint32_t> 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()

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<uint32_t, uint32_t>& registrations);
         /**
          * @brief Helper to release unit registrations.
          */

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<std::mutex> lock(m_peerAffiliationsMutex);
-
-    auto it = m_peerAffiliations.find(peerId);
-    if (it != m_peerAffiliations.end()) {
-        m_peerAffiliations.erase(peerId);
-    }
+    bool replaced = false;
+    std::unordered_map<uint32_t, uint32_t> aggregate;
+    std::vector<uint32_t> aggregateRegistrations;
+    {
+        std::lock_guard<std::mutex> updateLock(m_affiliationUpdateMutex);
+        {
+            std::lock_guard<std::mutex> peerLock(m_peerAffiliationsMutex);
 
-    lookups::ChannelLookup* chLookup = new lookups::ChannelLookup();
-    std::shared_ptr<fne_lookups::AffiliationLookup> 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<fne_lookups::AffiliationLookup> 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<std::mutex> lock(m_peerAffiliationsMutex);
+    bool erased = false;
+    std::unordered_map<uint32_t, uint32_t> aggregate;
+    std::vector<uint32_t> aggregateRegistrations;
+    {
+        std::lock_guard<std::mutex> updateLock(m_affiliationUpdateMutex);
+        {
+            std::lock_guard<std::mutex> 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::PeerAffiliationMapPair> 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<uint32_t, uint32_t> aggregate;
+    {
+        std::lock_guard<std::mutex> updateLock(m_affiliationUpdateMutex);
+        std::shared_ptr<fne_lookups::AffiliationLookup> 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<uint32_t, uint32_t> aggregate;
+    {
+        std::lock_guard<std::mutex> updateLock(m_affiliationUpdateMutex);
+        std::shared_ptr<fne_lookups::AffiliationLookup> 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<uint32_t, uint32_t>& affiliations)
+{
+    std::unordered_map<uint32_t, uint32_t> aggregate;
+    {
+        std::lock_guard<std::mutex> updateLock(m_affiliationUpdateMutex);
+        std::shared_ptr<fne_lookups::AffiliationLookup> 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<uint32_t, uint32_t> TrafficNetwork::buildAggregatedAffiliations()
+{
+    std::vector<PeerAffiliationMapPair> peerSnapshot = peerAffiliationsSnapshot();
+    std::sort(peerSnapshot.begin(), peerSnapshot.end(), [](const PeerAffiliationMapPair& lhs, const PeerAffiliationMapPair& rhs) {
+        return lhs.first < rhs.first;
+    });
+
+    std::unordered_map<uint32_t, uint32_t> aggregate;
+    for (const auto& peer : peerSnapshot) {
+        if (peer.second == nullptr)
+            continue;
+
+        const std::unordered_map<uint32_t, uint32_t> 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<uint32_t, uint32_t>& 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<uint32_t> aggregate;
+    {
+        std::lock_guard<std::mutex> updateLock(m_affiliationUpdateMutex);
+        std::shared_ptr<fne_lookups::AffiliationLookup> 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<uint32_t> aggregate;
+    std::unordered_map<uint32_t, uint32_t> aggregateAffiliations;
+    {
+        std::lock_guard<std::mutex> updateLock(m_affiliationUpdateMutex);
+        std::shared_ptr<fne_lookups::AffiliationLookup> 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<uint32_t>& registrations, uint32_t ssrc)
+{
+    std::vector<uint32_t> aggregate;
+    {
+        std::lock_guard<std::mutex> updateLock(m_affiliationUpdateMutex);
+        std::shared_ptr<fne_lookups::AffiliationLookup> peerRegistrations = getPeerAffiliations(peerId);
+        if (peerRegistrations == nullptr)
+            return false;
+
+        std::unordered_map<uint32_t, uint32_t> 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<uint32_t> TrafficNetwork::buildAggregatedUnitRegistrations()
+{
+    std::vector<PeerAffiliationMapPair> peerSnapshot = peerAffiliationsSnapshot();
+    std::sort(peerSnapshot.begin(), peerSnapshot.end(), [](const PeerAffiliationMapPair& lhs, const PeerAffiliationMapPair& rhs) {
+        return lhs.first < rhs.first;
+    });
+
+    std::unordered_map<uint32_t, uint32_t> aggregateOrigins;
+    for (const auto& peer : peerSnapshot) {
+        if (peer.second == nullptr)
+            continue;
+
+        const std::vector<uint32_t> 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<uint32_t> 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<uint32_t>& 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)

src/fne/network/TrafficNetwork.h

@@ -404,9 +404,10 @@ namespace network
         mutable std::shared_timed_mutex m_peersMutex;
         std::unordered_map<uint32_t, json::array> m_peerReplicaPeers;
         mutable std::mutex m_peerReplicaPeersMutex;
-        typedef std::pair<const uint32_t, std::shared_ptr<fne_lookups::AffiliationLookup>> PeerAffiliationMapPair;
+        typedef std::pair<uint32_t, std::shared_ptr<fne_lookups::AffiliationLookup>> PeerAffiliationMapPair;
         concurrent::unordered_map<uint32_t, std::shared_ptr<fne_lookups::AffiliationLookup>> m_peerAffiliations;
         mutable std::mutex m_peerAffiliationsMutex;
+        mutable std::mutex m_affiliationUpdateMutex;
         concurrent::shared_unordered_map<uint32_t, std::vector<uint32_t>> m_ccPeerMap;
         static std::timed_mutex s_keyQueueMutex;
         std::unordered_map<uint32_t, uint16_t> m_peerReplicaKeyQueue;
@@ -783,6 +784,73 @@ namespace network
          * @returns std::vector<PeerAffiliationMapPair> Snapshot of peer affiliation entries.
          */
         std::vector<PeerAffiliationMapPair> 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<uint32_t, uint32_t>& affiliations);
+        /**
+         * @brief Rebuilds the aggregate downstream affiliation snapshot.
+         * @note The caller must hold m_affiliationUpdateMutex.
+         * @returns std::unordered_map<uint32_t, uint32_t> Aggregate affiliation snapshot.
+         */
+        std::unordered_map<uint32_t, uint32_t> buildAggregatedAffiliations();
+        /**
+         * @brief Publishes an aggregate downstream affiliation snapshot upstream.
+         * @param affiliations Complete aggregate source-to-talkgroup affiliation snapshot.
+         */
+        void publishAggregatedAffiliations(const std::unordered_map<uint32_t, uint32_t>& 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<uint32_t>& registrations, uint32_t ssrc);
+        /**
+         * @brief Rebuilds the aggregate downstream unit registration snapshot.
+         * @note The caller must hold m_affiliationUpdateMutex.
+         * @returns std::vector<uint32_t> Aggregate unit registration snapshot.
+         */
+        std::vector<uint32_t> buildAggregatedUnitRegistrations();
+        /**
+         * @brief Publishes an aggregate downstream unit registration snapshot upstream.
+         * @param registrations Complete aggregate source ID registration snapshot.
+         */
+        void publishAggregatedUnitRegistrations(const std::vector<uint32_t>& registrations);
         /**
          * @brief Helper to disconnect a downstream peer.
          * @param peerId Peer ID.

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<int>(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<int>(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<int>(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<int>(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<uint32_t>(req->length) - 4U) / 8U;
+                        if (len != availableEntries || static_cast<uint32_t>(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<uint32_t, uint32_t> 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<uint32_t>(req->length) - 4U) / 3U;
+                        if (len != availableEntries || static_cast<uint32_t>(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<uint32_t> 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 {

tests/fne/FNENetwork_Tests.cpp

@@ -12,8 +12,10 @@
 
 #include "fne/FNETestHooks.h"
 #include "fne/HostFNE.h"
+#include "common/lookups/AffiliationLookup.h"
 
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 // ---------------------------------------------------------------------------
@@ -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<uint32_t, uint32_t> 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<uint32_t>{ 1002U, 1003U });
+
+    registrations.replaceUnitRegistrations({});
+    REQUIRE(registrations.unitRegTable().empty());
+}
+
 TEST_CASE("FNE test hooks expose peer connection state", "[fne][hooks]")
 {
     FNEHarness harness;