fix key-value fired duplicated events in case of multi subscribers

packages/integration_tests/src/key_value.rs

@@ -0,0 +1,203 @@
+#[cfg(test)]
+mod tests {
+    use async_std::prelude::FutureExt;
+    use async_std::task::JoinHandle;
+    use atm0s_sdn::SharedRouter;
+    use atm0s_sdn::{convert_enum, NetworkPlane, NetworkPlaneConfig};
+    use atm0s_sdn::{KeyValueBehavior, KeyValueBehaviorEvent, KeyValueHandlerEvent, KeyValueSdk, KeyValueSdkEvent};
+    use atm0s_sdn::{LayersSpreadRouterSyncBehavior, LayersSpreadRouterSyncBehaviorEvent, LayersSpreadRouterSyncHandlerEvent};
+    use atm0s_sdn::{ManualBehavior, ManualBehaviorConf, ManualBehaviorEvent, ManualHandlerEvent};
+    use atm0s_sdn::{NodeAddr, NodeAddrBuilder, NodeId};
+    use atm0s_sdn::{OptionUtils, SystemTimer};
+    use atm0s_sdn_transport_vnet::VnetEarth;
+    use std::{sync::Arc, time::Duration, vec};
+
+    #[derive(convert_enum::From, convert_enum::TryInto)]
+    enum ImplBehaviorEvent {
+        KeyValue(KeyValueBehaviorEvent),
+        RouterSync(LayersSpreadRouterSyncBehaviorEvent),
+        Manual(ManualBehaviorEvent),
+    }
+
+    #[derive(convert_enum::From, convert_enum::TryInto)]
+    enum ImplHandlerEvent {
+        KeyValue(KeyValueHandlerEvent),
+        RouterSync(LayersSpreadRouterSyncHandlerEvent),
+        Manual(ManualHandlerEvent),
+    }
+
+    #[derive(convert_enum::From, convert_enum::TryInto)]
+    enum ImplSdkEvent {
+        KeyValue(KeyValueSdkEvent),
+    }
+
+    async fn run_node(vnet: Arc<VnetEarth>, node_id: NodeId, seeds: Vec<NodeAddr>) -> (KeyValueSdk, NodeAddr, JoinHandle<()>) {
+        log::info!("Run node {} connect to {:?}", node_id, seeds);
+        let node_addr = Arc::new(NodeAddrBuilder::new(node_id));
+        let transport = Box::new(atm0s_sdn_transport_vnet::VnetTransport::new(vnet, node_addr.addr()));
+        let timer = Arc::new(SystemTimer());
+
+        let router = SharedRouter::new(node_id);
+        let manual = ManualBehavior::new(ManualBehaviorConf {
+            node_id,
+            node_addr: node_addr.addr(),
+            seeds,
+            local_tags: vec![],
+            connect_tags: vec![],
+        });
+
+        let router_sync_behaviour = LayersSpreadRouterSyncBehavior::new(router.clone());
+        let kv_sdk = KeyValueSdk::new();
+        let kv_behaviour = KeyValueBehavior::new(node_id, 3000, Some(Box::new(kv_sdk.clone())));
+
+        let mut plane = NetworkPlane::<ImplBehaviorEvent, ImplHandlerEvent, ImplSdkEvent>::new(NetworkPlaneConfig {
+            node_id,
+            tick_ms: 100,
+            behaviors: vec![Box::new(kv_behaviour), Box::new(router_sync_behaviour), Box::new(manual)],
+            transport,
+            timer,
+            router: Arc::new(router.clone()),
+        });
+
+        let join = async_std::task::spawn(async move {
+            plane.started();
+            while let Ok(_) = plane.recv().await {}
+            plane.stopped();
+        });
+
+        (kv_sdk, node_addr.addr(), join)
+    }
+
+    #[async_std::test]
+    async fn local_node_simple() {
+        let vnet = Arc::new(VnetEarth::default());
+        let (sdk, _addr, join) = run_node(vnet, 1, vec![]).await;
+
+        async_std::task::sleep(Duration::from_millis(300)).await;
+
+        //for simple key-value
+        const KEY_ID: u64 = 1000;
+        let mut event_rx = sdk.subscribe(KEY_ID, None);
+        sdk.set(KEY_ID, vec![1, 2, 3], None);
+
+        let (key, value, _, source) = event_rx.recv().timeout(Duration::from_millis(300)).await.expect("Should receive event").expect("Should has event");
+        assert_eq!(key, KEY_ID);
+        assert_eq!(value, Some(vec![1, 2, 3]));
+        assert_eq!(source, 1);
+
+        sdk.del(KEY_ID);
+        let (key, value, _, source) = event_rx.recv().timeout(Duration::from_millis(300)).await.expect("Should receive event").expect("Should has event");
+        assert_eq!(key, KEY_ID);
+        assert_eq!(value, None);
+        assert_eq!(source, 1);
+
+        join.cancel().await.print_none("Should cancel join");
+    }
+
+    #[async_std::test]
+    async fn local_node_hmap() {
+        let vnet = Arc::new(VnetEarth::default());
+        let (sdk, _addr, join) = run_node(vnet, 1, vec![]).await;
+
+        async_std::task::sleep(Duration::from_millis(300)).await;
+
+        //for simple key-value
+        const KEY_ID: u64 = 1000;
+        const SUB_KEY: u64 = 2000;
+        let mut event_rx = sdk.hsubscribe(KEY_ID, None);
+        sdk.hset(KEY_ID, SUB_KEY, vec![1, 2, 3], None);
+
+        let (key, sub_key, value, _, source) = event_rx.recv().timeout(Duration::from_millis(300)).await.expect("Should receive event").expect("Should has event");
+        assert_eq!(key, KEY_ID);
+        assert_eq!(sub_key, SUB_KEY);
+        assert_eq!(value, Some(vec![1, 2, 3]));
+        assert_eq!(source, 1);
+
+        sdk.hdel(KEY_ID, SUB_KEY);
+
+        let (key, sub_key, value, _, source) = event_rx.recv().timeout(Duration::from_millis(300)).await.expect("Should receive event").expect("Should has event");
+        assert_eq!(key, KEY_ID);
+        assert_eq!(sub_key, SUB_KEY);
+        assert_eq!(value, None);
+        assert_eq!(source, 1);
+
+        join.cancel().await.print_none("Should cancel join");
+    }
+
+    #[async_std::test]
+    async fn local_node_hmap_raw() {
+        let vnet = Arc::new(VnetEarth::default());
+        let (sdk, _addr, join) = run_node(vnet, 1, vec![]).await;
+
+        async_std::task::sleep(Duration::from_millis(300)).await;
+
+        //for simple key-value
+        const KEY_ID: u64 = 1000;
+        const SUB_KEY: u64 = 2000;
+        let (tx2, rx2) = async_std::channel::bounded(10);
+        sdk.hsubscribe_raw(KEY_ID, 2000, None, tx2);
+
+        sdk.hset(KEY_ID, SUB_KEY, vec![1, 2, 3], None);
+
+        let (key, sub_key, value, _, source) = rx2.recv().timeout(Duration::from_millis(300)).await.expect("Should receive event").expect("Should has event");
+        assert_eq!(key, KEY_ID);
+        assert_eq!(sub_key, SUB_KEY);
+        assert_eq!(value, Some(vec![1, 2, 3]));
+        assert_eq!(source, 1);
+
+        sdk.hdel(KEY_ID, SUB_KEY);
+
+        let (key, sub_key, value, _, source) = rx2.recv().timeout(Duration::from_millis(300)).await.expect("Should receive event").expect("Should has event");
+        assert_eq!(key, KEY_ID);
+        assert_eq!(sub_key, SUB_KEY);
+        assert_eq!(value, None);
+        assert_eq!(source, 1);
+
+        join.cancel().await.print_none("Should cancel join");
+    }
+
+    #[async_std::test]
+    async fn local_node_hmap_both() {
+        let vnet = Arc::new(VnetEarth::default());
+        let (sdk, _addr, join) = run_node(vnet, 1, vec![]).await;
+
+        async_std::task::sleep(Duration::from_millis(300)).await;
+
+        //for simple key-value
+        const KEY_ID: u64 = 1000;
+        const SUB_KEY: u64 = 2000;
+        let (tx2, rx2) = async_std::channel::bounded(10);
+        let mut event_rx = sdk.hsubscribe(KEY_ID, None);
+        sdk.hsubscribe_raw(KEY_ID, 20000, None, tx2);
+
+        sdk.hset(KEY_ID, SUB_KEY, vec![1, 2, 3], None);
+
+        let (key, sub_key, value, _, source) = event_rx.recv().timeout(Duration::from_millis(300)).await.expect("Should receive event").expect("Should has event");
+        assert_eq!(key, KEY_ID);
+        assert_eq!(sub_key, SUB_KEY);
+        assert_eq!(value, Some(vec![1, 2, 3]));
+        assert_eq!(source, 1);
+
+        let (key, sub_key, value, _, source) = rx2.recv().timeout(Duration::from_millis(300)).await.expect("Should receive event").expect("Should has event");
+        assert_eq!(key, KEY_ID);
+        assert_eq!(sub_key, SUB_KEY);
+        assert_eq!(value, Some(vec![1, 2, 3]));
+        assert_eq!(source, 1);
+
+        sdk.hdel(KEY_ID, SUB_KEY);
+
+        let (key, sub_key, value, _, source) = event_rx.recv().timeout(Duration::from_millis(300)).await.expect("Should receive event").expect("Should has event");
+        assert_eq!(key, KEY_ID);
+        assert_eq!(sub_key, SUB_KEY);
+        assert_eq!(value, None);
+        assert_eq!(source, 1);
+
+        let (key, sub_key, value, _, source) = rx2.recv().timeout(Duration::from_millis(300)).await.expect("Should receive event").expect("Should has event");
+        assert_eq!(key, KEY_ID);
+        assert_eq!(sub_key, SUB_KEY);
+        assert_eq!(value, None);
+        assert_eq!(source, 1);
+
+        join.cancel().await.print_none("Should cancel join");
+    }
+}

packages/integration_tests/src/lib.rs

@@ -1,2 +1,3 @@
+mod key_value;
 mod pubsub;
 mod rpc;

packages/services/key_value/src/behavior.rs

@@ -52,11 +52,8 @@ where
         }
     }
 
-    fn pop_all_events<BE>(&mut self, _ctx: &BehaviorContext)
-    where
-        BE: Send + Sync + 'static,
-    {
-        while let Some(action) = self.simple_remote.pop_action() {
+    fn pop_all_events(&mut self, _ctx: &BehaviorContext, now_ms: u64) {
+        while let Some(action) = self.simple_remote.pop_action(now_ms) {
             log::debug!("[KeyValueBehavior {}] pop_all_events simple remote: {:?}", self.node_id, action);
             let header = MsgHeader::build(KEY_VALUE_SERVICE_ID, KEY_VALUE_SERVICE_ID, action.1).set_from_node(Some(self.node_id));
             self.outputs
@@ -95,7 +92,7 @@ where
             }
         }
 
-        while let Some(action) = self.hashmap_remote.pop_action() {
+        while let Some(action) = self.hashmap_remote.pop_action(now_ms) {
             log::debug!("[KeyValueBehavior {}] pop_all_events hashmap remote: {:?}", self.node_id, action);
             let header = MsgHeader::build(KEY_VALUE_SERVICE_ID, KEY_VALUE_SERVICE_ID, action.1).set_from_node(Some(self.node_id));
             self.outputs
@@ -143,22 +140,22 @@ where
             KeyValueMsg::SimpleRemote(msg) => {
                 log::debug!("[KeyValueBehavior {}] process_key_value_msg simple remote: {:?} from {}", self.node_id, msg, from);
                 self.simple_remote.on_event(now_ms, from, msg);
-                self.pop_all_events::<BE>(ctx);
+                self.pop_all_events(ctx, now_ms);
             }
             KeyValueMsg::SimpleLocal(msg) => {
                 log::debug!("[KeyValueBehavior {}] process_key_value_msg simple local: {:?} from {}", self.node_id, msg, from);
                 self.simple_local.on_event(from, msg);
-                self.pop_all_events::<BE>(ctx);
+                self.pop_all_events(ctx, now_ms);
             }
             KeyValueMsg::HashmapRemote(msg) => {
                 log::debug!("[KeyValueBehavior {}] process_key_value_msg hashmap remote: {:?} from {}", self.node_id, msg, from);
                 self.hashmap_remote.on_event(now_ms, from, msg);
-                self.pop_all_events::<BE>(ctx);
+                self.pop_all_events(ctx, now_ms);
             }
             KeyValueMsg::HashmapLocal(msg) => {
                 log::debug!("[KeyValueBehavior {}] process_key_value_msg hashmap local: {:?} from {}", self.node_id, msg, from);
                 self.hashmap_local.on_event(from, msg);
-                self.pop_all_events::<BE>(ctx);
+                self.pop_all_events(ctx, now_ms);
             }
         }
     }
@@ -217,7 +214,7 @@ where
         self.simple_local.tick(now_ms);
         self.hashmap_remote.tick(now_ms);
         self.hashmap_local.tick(now_ms);
-        self.pop_all_events::<BE>(ctx);
+        self.pop_all_events(ctx, now_ms);
     }
 
     fn on_awake(&mut self, ctx: &BehaviorContext, now_ms: u64) {
@@ -229,7 +226,7 @@ where
                 break;
             }
         }
-        self.pop_all_events::<BE>(ctx);
+        self.pop_all_events(ctx, now_ms);
     }
 
     fn check_incoming_connection(&mut self, ctx: &BehaviorContext, now_ms: u64, node: NodeId, conn_id: ConnId) -> Result<(), ConnectionRejectReason> {
@@ -256,7 +253,7 @@ where
     fn on_sdk_msg(&mut self, ctx: &BehaviorContext, now_ms: u64, from_service: u8, event: SE) {
         if let Ok(event) = event.try_into() {
             self.process_sdk_event(ctx, now_ms, from_service, event);
-            self.pop_all_events::<BE>(ctx);
+            self.pop_all_events(ctx, now_ms);
         } else {
             debug_assert!(false, "Invalid event")
         }

packages/services/key_value/src/behavior/event_acks.rs

@@ -2,8 +2,10 @@ use std::collections::{HashMap, VecDeque};
 
 use crate::ReqId;
 
+pub const RESEND_AFTER_MS: u64 = 500;
+
 pub struct EventAckManager<T: Clone> {
-    events: HashMap<ReqId, (T, u8)>,
+    events: HashMap<ReqId, (u64, T, u8)>,
     actions: VecDeque<T>,
 }
 
@@ -19,20 +21,20 @@ where
     }
 
     // add to queue and auto retry each tick
-    pub fn add_event(&mut self, req_id: ReqId, event: T, retry_count: u8) {
+    pub fn add_event(&mut self, now_ms: u64, req_id: ReqId, event: T, retry_count: u8) {
         log::debug!("[EventAckManager] Add event with req_id {}, retry_count {}", req_id, retry_count);
         self.actions.push_back(event.clone());
-        self.events.insert(req_id, (event, retry_count));
+        self.events.insert(req_id, (now_ms, event, retry_count));
     }
 
     pub fn on_ack(&mut self, req_id: ReqId) {
         log::debug!("[EventAckManager] On ack with req_id {}", req_id);
         self.events.remove(&req_id);
     }
 
-    pub fn tick(&mut self) {
-        for (_req_id, (event, retry_count)) in self.events.iter_mut() {
-            if *retry_count > 0 {
+    pub fn tick(&mut self, now_ms: u64) {
+        for (_req_id, (pushed_ms, event, retry_count)) in self.events.iter_mut() {
+            if *retry_count > 0 && now_ms >= *pushed_ms + RESEND_AFTER_MS {
                 log::debug!("[EventAckManager] Retry event with req_id {}, retry_count {}", _req_id, *retry_count);
                 *retry_count -= 1;
                 self.actions.push_back(event.clone());