[syncd]: Route FDB LEARN/AGED/MOVE events via ProducerStateTable for key-based dedup

lib/sairediscommon.h

@@ -97,6 +97,8 @@
 /**
  * @brief Table which will be used to send API response from syncd.
  */
+#define REDIS_TABLE_FDB_EVENT_STATE "FDB_EVENT_STATE"
+
 #define REDIS_TABLE_GETRESPONSE     "GETRESPONSE"
 
 // REDIS default database defines

syncd/NotificationProcessor.cpp

@@ -10,6 +10,8 @@
 #include "swss/logger.h"
 #include "swss/notificationproducer.h"
 
+#include "swss/producerstatetable.h"
+
 #include <inttypes.h>
 
 using namespace syncd;
@@ -19,7 +21,8 @@ NotificationProcessor::NotificationProcessor(
         _In_ std::shared_ptr<NotificationProducerBase> producer,
         _In_ std::shared_ptr<BaseRedisClient> client,
         _In_ std::function<void(const swss::KeyOpFieldsValuesTuple&)> synchronizer,
-        _In_ std::function<bool(sai_object_id_t, sai_port_oper_status_t)> linkEventDampingApplier):
+        _In_ std::function<bool(sai_object_id_t, sai_port_oper_status_t)> linkEventDampingApplier,
+        _In_ std::shared_ptr<swss::DBConnector> dbAsic):
     m_synchronizer(synchronizer),
     m_linkEventDampingApplier(linkEventDampingApplier),
     m_client(client),
@@ -30,6 +33,12 @@ NotificationProcessor::NotificationProcessor(
     m_runThread = false;
 
     m_notificationQueue = std::make_shared<NotificationQueue>();
+
+    if (dbAsic)
+    {
+        m_fdbEventStateProducer = std::make_shared<swss::ProducerStateTable>(
+                dbAsic.get(), REDIS_TABLE_FDB_EVENT_STATE);
+    }
 }
 
 NotificationProcessor::~NotificationProcessor()
@@ -316,6 +325,15 @@ void NotificationProcessor::process_on_fdb_event(
 
     bool sendntf = true;
 
+    /*
+     * Separate FLUSH events from LEARN/AGED/MOVE events.
+     * FLUSH events are sent via PUBLISH (NotificationProducer) as before.
+     * LEARN/AGED/MOVE events use ProducerStateTable so rapid events for the
+     * same entry are merged before orchagent processes them.
+     */
+    std::vector<sai_fdb_event_notification_data_t> flushEvents;
+    std::vector<sai_fdb_event_notification_data_t> nonFlushEvents;
+
     for (uint32_t i = 0; i < count; i++)
     {
         sai_fdb_event_notification_data_t *fdb = &data[i];
@@ -336,24 +354,76 @@ void NotificationProcessor::process_on_fdb_event(
 
         m_translator->translateRidToVid(SAI_OBJECT_TYPE_FDB_ENTRY, fdb->fdb_entry.switch_id, fdb->attr_count, fdb->attr, true);
 
-        /*
-         * Currently because of brcm bug, we need to install fdb entries in
-         * asic view and currently this event don't have fdb type which is
-         * required on creation.
-         */
-
         redisPutFdbEntryToAsicView(fdb);
+
+        if (fdb->event_type == SAI_FDB_EVENT_FLUSHED)
+        {
+            flushEvents.push_back(*fdb);
+        }
+        else
+        {
+            nonFlushEvents.push_back(*fdb);
+        }
     }
 
-    if (sendntf)
+    if (!sendntf)
+    {
+        SWSS_LOG_ERROR("FDB notification was not sent since it contain invalid OIDs, bug?");
+        return;
+    }
+
+    /* Send FLUSH events via PUBLISH (preserves existing behavior) */
+    if (!flushEvents.empty())
     {
-        std::string s = sai_serialize_fdb_event_ntf(count, data);
+        std::string s = sai_serialize_fdb_event_ntf(
+                (uint32_t)flushEvents.size(), flushEvents.data());
 
         sendNotification(SAI_SWITCH_NOTIFICATION_NAME_FDB_EVENT, s);
     }
-    else
+
+    /* Send LEARN/AGED/MOVE events via ProducerStateTable (merged by key) */
+    if (m_fdbEventStateProducer && !nonFlushEvents.empty())
     {
-        SWSS_LOG_ERROR("FDB notification was not sent since it contain invalid OIDs, bug?");
+        for (auto &fdb : nonFlushEvents)
+        {
+            std::string key = sai_serialize_fdb_entry(fdb.fdb_entry);
+
+            if (fdb.event_type == SAI_FDB_EVENT_AGED)
+            {
+                m_fdbEventStateProducer->del(key);
+            }
+            else
+            {
+                std::vector<swss::FieldValueTuple> fvs;
+
+                fvs.emplace_back("event_type", sai_serialize_fdb_event(fdb.event_type));
+
+                for (uint32_t j = 0; j < fdb.attr_count; j++)
+                {
+                    if (fdb.attr[j].id == SAI_FDB_ENTRY_ATTR_BRIDGE_PORT_ID)
+                    {
+                        fvs.emplace_back("bridge_port_id",
+                                sai_serialize_object_id(fdb.attr[j].value.oid));
+                    }
+                    else if (fdb.attr[j].id == SAI_FDB_ENTRY_ATTR_TYPE)
+                    {
+                        fvs.emplace_back("sai_fdb_type",
+                                sai_serialize_enum(fdb.attr[j].value.s32,
+                                    &sai_metadata_enum_sai_fdb_entry_type_t));
+                    }
+                }
+
+                m_fdbEventStateProducer->set(key, fvs);
+            }
+        }
+    }
+    else if (!nonFlushEvents.empty())
+    {
+        /* Fallback: m_fdbEventStateProducer not initialized, use PUBLISH */
+        std::string s = sai_serialize_fdb_event_ntf(
+                (uint32_t)nonFlushEvents.size(), nonFlushEvents.data());
+
+        sendNotification(SAI_SWITCH_NOTIFICATION_NAME_FDB_EVENT, s);
     }
 }
 

syncd/NotificationProcessor.h

@@ -7,6 +7,7 @@
 #include "FlowDump.h"
 
 #include "swss/notificationproducer.h"
+#include "swss/producerstatetable.h"
 
 #include <thread>
 #include <memory>
@@ -23,7 +24,8 @@ namespace syncd
                     _In_ std::shared_ptr<NotificationProducerBase> producer,
                     _In_ std::shared_ptr<BaseRedisClient> client,
                     _In_ std::function<void(const swss::KeyOpFieldsValuesTuple&)> synchronizer,
-                    _In_ std::function<bool(sai_object_id_t, sai_port_oper_status_t)> linkEventDampingApplier = nullptr);
+                    _In_ std::function<bool(sai_object_id_t, sai_port_oper_status_t)> linkEventDampingApplier = nullptr,
+                    _In_ std::shared_ptr<swss::DBConnector> dbAsic = nullptr);
 
             virtual ~NotificationProcessor();
 
@@ -221,5 +223,7 @@ namespace syncd
             std::shared_ptr<BaseRedisClient> m_client;
 
             std::shared_ptr<NotificationProducerBase> m_notifications;
+
+            std::shared_ptr<swss::ProducerStateTable> m_fdbEventStateProducer;
     };
 }

syncd/Syncd.cpp

@@ -187,7 +187,8 @@ Syncd::Syncd(
             m_notifications,
             m_client,
             std::bind(&Syncd::syncProcessNotification, this, _1),
-            std::bind(&Syncd::applyLinkEventDamping, this, _1, _2));
+            std::bind(&Syncd::applyLinkEventDamping, this, _1, _2),
+            m_dbAsic);
     m_handler = std::make_shared<NotificationHandler>(m_processor);
 
     m_sn.onFdbEvent = std::bind(&NotificationHandler::onFdbEvent, m_handler.get(), _1, _2);

unittest/syncd/TestNotificationProcessor.cpp

@@ -6,6 +6,9 @@
 #include "lib/sairediscommon.h"
 #include "vslib/Sai.h"
 
+#include "swss/table.h"
+#include "swss/logger.h"
+
 #include <gtest/gtest.h>
 
 using namespace syncd;
@@ -144,3 +147,170 @@ TEST(NotificationProcessor, NotificationProcessorTest)
     notificationProcessor->syncProcessNotification(flowBulkGetSessionEventItem);
     translator->eraseRidAndVid(0x123456789abcdef, 0x123456789abcdef);
 }
+
+
+/*
+ * Helper: build a processor with a DB connection so m_fdbEventStateProducer
+ * is initialized.
+ */
+static std::shared_ptr<NotificationProcessor> makeProcessorWithDb(
+    std::shared_ptr<swss::DBConnector> dbAsic,
+    std::shared_ptr<VirtualOidTranslator> &outTranslator)
+{
+    SWSS_LOG_ENTER();
+    auto sai      = std::make_shared<saivs::Sai>();
+    auto client   = std::make_shared<RedisClient>(dbAsic);
+    auto producer = std::make_shared<syncd::RedisNotificationProducer>("ASIC_DB");
+
+    auto np = std::make_shared<NotificationProcessor>(producer, client,
+                                                      [](const swss::KeyOpFieldsValuesTuple&){},
+                                                      nullptr, dbAsic);
+
+    auto switchConfigContainer  = std::make_shared<sairedis::SwitchConfigContainer>();
+    auto ridxGen = std::make_shared<sairedis::RedisVidIndexGenerator>(dbAsic, REDIS_KEY_VIDCOUNTER);
+    auto voim = std::make_shared<sairedis::VirtualObjectIdManager>(0, switchConfigContainer, ridxGen);
+
+    outTranslator = std::make_shared<VirtualOidTranslator>(client, voim, sai);
+    np->m_translator = outTranslator;
+
+    return np;
+}
+
+/*
+ * ProducerStateTable writes field data to "_FDB_EVENT_STATE:{key}" in Redis.
+ * Use this helper to verify what was written.
+ */
+static std::string fdbStateKey(const std::string &fdbKey)
+{
+    // SWSS_LOG_ENTER omitted (simple helper)
+    return "_FDB_EVENT_STATE:" + fdbKey;
+}
+
+/*
+ * Test: LEARN event is written to FDB_EVENT_STATE via ProducerStateTable.
+ */
+TEST(NotificationProcessor, FdbLearnGoesToProducerStateTable)
+{
+    auto dbAsic = std::make_shared<swss::DBConnector>("ASIC_DB", 0);
+    std::shared_ptr<VirtualOidTranslator> translator;
+    auto np = makeProcessorWithDb(dbAsic, translator);
+
+    translator->insertRidAndVid(0x21000000000000, 0x210000000000);
+    translator->insertRidAndVid(0x1003a0000004c,  0x3a000000000c00);
+    translator->insertRidAndVid(0x2600000010,     0x26000000000010);
+
+    std::string fdbKey =
+        "{\"bvid\":\"oid:0x26000000000010\","
+        "\"mac\":\"00:00:00:00:00:10\","
+        "\"switch_id\":\"oid:0x210000000000\"}";
+
+    dbAsic->del(fdbStateKey(fdbKey));
+
+    std::string learnData =
+        "[{\"fdb_entry\":\"{\\\"bvid\\\":\\\"oid:0x2600000010\\\","
+        "\\\"mac\\\":\\\"00:00:00:00:00:10\\\","
+        "\\\"switch_id\\\":\\\"oid:0x21000000000000\\\"}\","
+        "\"fdb_event\":\"SAI_FDB_EVENT_LEARNED\","
+        "\"list\":["
+            "{\"id\":\"SAI_FDB_ENTRY_ATTR_BRIDGE_PORT_ID\","
+             "\"value\":\"oid:0x1003a0000004c\"},"
+            "{\"id\":\"SAI_FDB_ENTRY_ATTR_TYPE\","
+             "\"value\":\"SAI_FDB_ENTRY_TYPE_DYNAMIC\"}"
+        "]}]";
+
+    std::vector<swss::FieldValueTuple> empty;
+    swss::KeyOpFieldsValuesTuple item(SAI_SWITCH_NOTIFICATION_NAME_FDB_EVENT, learnData, empty);
+    np->syncProcessNotification(item);
+
+    auto val = dbAsic->hget(fdbStateKey(fdbKey), "event_type");
+    EXPECT_NE(val, nullptr) << "LEARN event not written to FDB_EVENT_STATE";
+    if (val)
+    {
+        EXPECT_EQ(*val, "SAI_FDB_EVENT_LEARNED");
+    }
+
+    dbAsic->flushdb();
+    translator->eraseRidAndVid(0x21000000000000, 0x210000000000);
+    translator->eraseRidAndVid(0x1003a0000004c,  0x3a000000000c00);
+    translator->eraseRidAndVid(0x2600000010,     0x26000000000010);
+}
+
+/*
+ * Test: AGED event issues DEL on FDB_EVENT_STATE (deletes the key).
+ */
+TEST(NotificationProcessor, FdbAgedDeletesFromProducerStateTable)
+{
+    auto dbAsic = std::make_shared<swss::DBConnector>("ASIC_DB", 0);
+    std::shared_ptr<VirtualOidTranslator> translator;
+    auto np = makeProcessorWithDb(dbAsic, translator);
+
+    translator->insertRidAndVid(0x21000000000000, 0x210000000000);
+    translator->insertRidAndVid(0x2600000011,     0x26000000000011);
+
+    std::string fdbKey =
+        "{\"bvid\":\"oid:0x26000000000011\","
+        "\"mac\":\"00:00:00:00:00:11\","
+        "\"switch_id\":\"oid:0x210000000000\"}";
+
+    /* Pre-populate so we can verify deletion */
+    dbAsic->hset(fdbStateKey(fdbKey), "event_type", "SAI_FDB_EVENT_LEARNED");
+
+    std::string agedData =
+        "[{\"fdb_entry\":\"{\\\"bvid\\\":\\\"oid:0x2600000011\\\","
+        "\\\"mac\\\":\\\"00:00:00:00:00:11\\\","
+        "\\\"switch_id\\\":\\\"oid:0x21000000000000\\\"}\","
+        "\"fdb_event\":\"SAI_FDB_EVENT_AGED\","
+        "\"list\":[]}]";
+
+    std::vector<swss::FieldValueTuple> empty;
+    swss::KeyOpFieldsValuesTuple item(SAI_SWITCH_NOTIFICATION_NAME_FDB_EVENT, agedData, empty);
+    np->syncProcessNotification(item);
+
+    /* ProducerStateTable::del does not remove the key immediately in Redis;
+     * it publishes a DEL notification. The _G_ scratch key should be deleted. */
+    auto val = dbAsic->hget(fdbStateKey(fdbKey), "event_type");
+    EXPECT_EQ(val, nullptr) << "AGED event should delete entry from FDB_EVENT_STATE scratch";
+
+    dbAsic->flushdb();
+    translator->eraseRidAndVid(0x21000000000000, 0x210000000000);
+    translator->eraseRidAndVid(0x2600000011,     0x26000000000011);
+}
+
+/*
+ * Test: FLUSH event goes via PUBLISH, not to FDB_EVENT_STATE.
+ */
+TEST(NotificationProcessor, FdbFlushDoesNotWriteToProducerStateTable)
+{
+    auto dbAsic = std::make_shared<swss::DBConnector>("ASIC_DB", 0);
+    std::shared_ptr<VirtualOidTranslator> translator;
+    auto np = makeProcessorWithDb(dbAsic, translator);
+
+    translator->insertRidAndVid(0x21000000000000, 0x210000000000);
+    translator->insertRidAndVid(0x2600000012,     0x26000000000012);
+
+    std::string fdbKey =
+        "{\"bvid\":\"oid:0x26000000000012\","
+        "\"mac\":\"00:00:00:00:00:00\","
+        "\"switch_id\":\"oid:0x210000000000\"}";
+
+    dbAsic->del(fdbStateKey(fdbKey));
+
+    std::string flushData =
+        "[{\"fdb_entry\":\"{\\\"bvid\\\":\\\"oid:0x2600000012\\\","
+        "\\\"mac\\\":\\\"00:00:00:00:00:00\\\","
+        "\\\"switch_id\\\":\\\"oid:0x21000000000000\\\"}\","
+        "\"fdb_event\":\"SAI_FDB_EVENT_FLUSHED\","
+        "\"list\":[]}]";
+
+    std::vector<swss::FieldValueTuple> empty;
+    swss::KeyOpFieldsValuesTuple item(SAI_SWITCH_NOTIFICATION_NAME_FDB_EVENT, flushData, empty);
+    np->syncProcessNotification(item);
+
+    /* FLUSH must NOT appear in the scratch table */
+    auto val = dbAsic->hget(fdbStateKey(fdbKey), "event_type");
+    EXPECT_EQ(val, nullptr) << "FLUSH event must not be written to FDB_EVENT_STATE";
+
+    dbAsic->flushdb();
+    translator->eraseRidAndVid(0x21000000000000, 0x210000000000);
+    translator->eraseRidAndVid(0x2600000012,     0x26000000000012);
+}