From 8d56e503b405a01bc573e4e4ef376bf0d56e4dc9 Mon Sep 17 00:00:00 2001 From: BYGX-wcr Date: Mon, 27 Jul 2026 21:25:54 +0000 Subject: [PATCH] swss-common-bridge: replay unchanged producer updates Signed-off-by: BYGX-wcr --- crates/swss-common-bridge/src/lib.rs | 26 +---------- crates/swss-common-bridge/src/producer.rs | 53 ++++++++++++----------- 2 files changed, 29 insertions(+), 50 deletions(-) diff --git a/crates/swss-common-bridge/src/lib.rs b/crates/swss-common-bridge/src/lib.rs index 913d0f54..06aa5ff2 100644 --- a/crates/swss-common-bridge/src/lib.rs +++ b/crates/swss-common-bridge/src/lib.rs @@ -5,8 +5,8 @@ use std::collections::HashMap; use swss_common::{FieldValues, KeyOpFieldValues, KeyOperation}; /// An in-memory copy of a table. -/// We keep a copy so that we can skip update if there is no change. The cache is established from the previous -/// request sent to the Producer. If the Producer is restarted, the cache will be empty. +/// The consumer bridge uses this to merge incremental updates into complete entries and skip duplicate notifications. +/// The cache is established from updates read from the consumer table and is empty after the bridge restarts. #[derive(Default)] pub(crate) struct TableCache(HashMap); @@ -39,26 +39,4 @@ impl TableCache { } } } - - /// Replace the cached entry with the provided kfv. - /// Returns false if operation is SET and field values are the same as cached ones. - /// For Del it always return true because the local cache is cleared after restart. - fn replace_kfv(&mut self, kfv: KeyOpFieldValues) -> bool { - match kfv.operation { - KeyOperation::Set => { - let field_values = self.0.entry(kfv.key.clone()).or_default(); - - if kfv.field_values == *field_values { - return false; - } - - *field_values = kfv.field_values; - true - } - KeyOperation::Del => { - self.0.remove(&kfv.key); - true - } - } - } } diff --git a/crates/swss-common-bridge/src/producer.rs b/crates/swss-common-bridge/src/producer.rs index ae83fb93..5bb423c5 100644 --- a/crates/swss-common-bridge/src/producer.rs +++ b/crates/swss-common-bridge/src/producer.rs @@ -1,4 +1,3 @@ -use crate::TableCache; use std::{future::Future, sync::Arc}; use swbus_actor::ActorMessage; use swbus_edge::{ @@ -32,7 +31,6 @@ where { let swbus = SimpleSwbusEdgeClient::new(rt, addr, false, false); tokio::task::spawn(async move { - let mut table_cache = TableCache::default(); loop { let Some(msg) = swbus.recv().await else { // Swbus shut down, we might as well quit. @@ -47,12 +45,8 @@ where let (error_code, error_message) = match ActorMessage::deserialize(&payload) { Ok(actor_msg) => match actor_msg.deserialize_data::() { Ok(kfv) => { - if table_cache.replace_kfv(kfv.clone()) { - table.apply_kfv(kfv).await; - (SwbusErrorCode::Ok, String::new()) - } else { - (SwbusErrorCode::Ok, "No change in data".to_string()) - } + table.apply_kfv(kfv).await; + (SwbusErrorCode::Ok, String::new()) } Err(e) => ( SwbusErrorCode::InvalidPayload, @@ -256,8 +250,14 @@ mod test { // Spawn the bridge (keeps the producer alive across consumer recreations) let _bridge = ProducerBridge::spawn(rt, sp("mytable-bridge"), zpst); + let kfvs = vec![KeyOpFieldValues { + key: random_string(), + operation: KeyOperation::Set, + field_values: random_fvs(), + }]; + // First run - timeout(Duration::from_secs(5), run_test_with_swbus(&swbus, zcst)) + timeout(Duration::from_secs(5), run_test_with_swbus(&swbus, zcst, kfvs.clone())) .await .unwrap(); @@ -270,15 +270,18 @@ mod test { let mut zmqs2 = ZmqServer::new(&zmq_endpoint).unwrap(); let zcst2 = ZmqConsumerStateTable::new(redis.db_connector(), "mytable", &mut zmqs2, None, None).unwrap(); - // Second run with recreated consumer - timeout(Duration::from_secs(5), run_test_with_swbus(&swbus, zcst2)) + // Replay the same data after recreating the consumer. + timeout(Duration::from_secs(5), run_test_with_swbus(&swbus, zcst2, kfvs)) .await .unwrap(); } - async fn run_test_with_swbus(swbus: &SimpleSwbusEdgeClient, mut consumer_table: C) { - // Send some updates to the bridge - let mut kfvs = random_kfvs(); + async fn run_test_with_swbus( + swbus: &SimpleSwbusEdgeClient, + mut consumer_table: C, + mut kfvs: Vec, + ) { + // Send updates to the bridge for kfv in &kfvs { let msg = OutgoingMessage { destination: sp("mytable-bridge"), @@ -388,25 +391,23 @@ mod test { // Resend the same updates to the producer bridge for kfv in &kfvs { let msg = OutgoingMessage { - destination: sp("dpu-bridge"), + destination: sp("mytable-bridge"), body: MessageBody::Request { payload: encode_kfv(kfv), }, }; swbus.send(msg).await.unwrap(); - println!("Sent: {}", kfv.key); } - let result = timeout(Duration::from_secs(3), consumer_table.read_data()).await; - if result.is_ok() { - // If we got here, it means the bridge did not skip the updates - let received = consumer_table.pops().await; - if !received.is_empty() { - for kfv in received { - println!("Received: {}", kfv.key); - } - panic!("Expected bridge to skip duplicate updates, but it processed them"); - } + + // The bridge must replay identical updates because the destination may + // have lost its state without the bridge restarting. + let mut replayed_kfvs = Vec::new(); + while replayed_kfvs.len() < kfvs.len() { + consumer_table.read_data().await; + replayed_kfvs.extend(consumer_table.pops().await); } + replayed_kfvs.sort_unstable(); + assert_eq!(kfvs, replayed_kfvs); } fn encode_kfv(kfv: &KeyOpFieldValues) -> Vec {