Respect liveliness subscriber history when processing token declarations (#2160)

* Consider liveliness sub history when processing token declarations

* Add regression test

* Remove unnecessary Option wrappers

* Address review comments

---------

Co-authored-by: OlivierHecart <olivier.hecart@adlinktech.com>

Author: oteffahi

zenoh/src/api/admin.rs

@@ -221,6 +221,7 @@ impl Handler {
                             ..Put::default()
                         }))
                     },
+                    false,
                     #[cfg(feature = "unstable")]
                     Reliability::Reliable,
                 );
@@ -280,6 +281,7 @@ impl PeerHandler {
             &self.expr.clone().with_suffix(suffix),
             Default::default(),
             msg,
+            false,
             #[cfg(feature = "unstable")]
             Reliability::Reliable,
         );

zenoh/src/api/session.rs

@@ -338,6 +338,7 @@ impl SessionState {
             key_expr: key_expr.clone().into_owned(),
             origin,
             callback,
+            history: false,
         };
 
         let declared_sub = origin != Locality::SessionLocal;
@@ -1709,6 +1710,7 @@ impl SessionInner {
             key_expr: key_expr.clone().into_owned(),
             origin,
             callback: callback.clone(),
+            history,
         };
 
         let sub_state = Arc::new(sub_state);
@@ -1975,6 +1977,7 @@ impl SessionInner {
         wire_expr: &WireExpr,
         qos: push::ext::QoSType,
         msg: impl FnOnce() -> &'a mut PushBody,
+        historical: bool,
         #[cfg(feature = "unstable")] reliability: Reliability,
     ) {
         let mut callbacks = SingleOrVec::default();
@@ -1985,9 +1988,10 @@ impl SessionInner {
         if wire_expr.suffix.is_empty() {
             match state.get_res(&wire_expr.scope, wire_expr.mapping, local) {
                 Some(Resource::Node(res)) => {
-                    for sub in res.subscribers(kind) {
-                        if sub.origin == Locality::Any
-                            || (local == (sub.origin == Locality::SessionLocal))
+                    for sub in res.subscribers(kind).iter() {
+                        if (sub.origin == Locality::Any
+                            || (local == (sub.origin == Locality::SessionLocal)))
+                            && (sub.history || (!historical))
                         {
                             callbacks.push((sub.callback.clone(), res.key_expr.clone().into()));
                         }
@@ -2011,6 +2015,7 @@ impl SessionInner {
                     for sub in state.subscribers(kind).values() {
                         if (sub.origin == Locality::Any
                             || (local == (sub.origin == Locality::SessionLocal)))
+                            && (sub.history || (!historical))
                             && key_expr.intersects(&sub.key_expr)
                         {
                             callbacks.push((sub.callback.clone(), key_expr.clone().into_owned()));
@@ -2115,6 +2120,7 @@ impl SessionInner {
                     }
                     &mut push.payload
                 },
+                false,
                 #[cfg(feature = "unstable")]
                 reliability,
             );
@@ -2550,6 +2556,9 @@ impl Primitives for WeakSession {
                                 &m.wire_expr,
                                 Default::default(),
                                 || body.insert(Put::default().into()),
+                                // interest_id is set if the Token is an Interest::Current.
+                                // This is used to decide if subs with history=false should be called or not
+                                msg.interest_id.is_some(),
                                 #[cfg(feature = "unstable")]
                                 Reliability::Reliable,
                             );
@@ -2567,6 +2576,10 @@ impl Primitives for WeakSession {
                     if state.primitives.is_none() {
                         return; // Session closing or closed
                     }
+                    // interest_id is set if the Token is an Interest::Current.
+                    // This is used to decide if liveliness subs with history=false should be called or not
+                    // NOTE: an UndeclareToken is most likely not an Interest::Current
+                    let interest_current = msg.interest_id.is_some();
                     if let Some(key_expr) = state.remote_tokens.remove(&m.id) {
                         drop(state);
                         let mut body = None;
@@ -2576,6 +2589,7 @@ impl Primitives for WeakSession {
                             &key_expr.to_wire(self),
                             Default::default(),
                             || body.insert(Del::default().into()),
+                            interest_current,
                             #[cfg(feature = "unstable")]
                             Reliability::Reliable,
                         );
@@ -2593,6 +2607,7 @@ impl Primitives for WeakSession {
                                     &key_expr.to_wire(self),
                                     Default::default(),
                                     || body.insert(Del::default().into()),
+                                    interest_current,
                                     #[cfg(feature = "unstable")]
                                     Reliability::Reliable,
                                 );
@@ -2625,6 +2640,7 @@ impl Primitives for WeakSession {
             &msg.wire_expr,
             msg.ext_qos,
             || &mut msg.payload,
+            false,
             #[cfg(feature = "unstable")]
             _reliability,
         );

zenoh/src/api/subscriber.rs

@@ -37,6 +37,7 @@ pub(crate) struct SubscriberState {
     pub(crate) key_expr: KeyExpr<'static>,
     pub(crate) origin: Locality,
     pub(crate) callback: Callback<Sample>,
+    pub(crate) history: bool,
 }
 
 impl fmt::Debug for SubscriberState {

zenoh/tests/liveliness.rs

@@ -4746,3 +4746,100 @@ async fn test_liveliness_double_undeclare_clique() {
     peer1.close().await.unwrap();
     peer2.close().await.unwrap();
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_liveliness_sub_history_conflict() {
+    // https://github.com/eclipse-zenoh/zenoh/issues/2071
+    use std::time::Duration;
+
+    use zenoh::sample::SampleKind;
+    use zenoh_config::WhatAmI;
+    use zenoh_link::EndPoint;
+
+    const TIMEOUT: Duration = Duration::from_secs(60);
+    const SLEEP: Duration = Duration::from_secs(1);
+    const ROUTER_ENDPOINT: &str = "tcp/localhost:30516";
+    const LIVELINESS_KEYEXPR: &str = "test/liveliness/history/conflict";
+
+    zenoh_util::init_log_from_env_or("error");
+
+    let router = {
+        let mut c = zenoh::Config::default();
+        c.listen
+            .endpoints
+            .set(vec![ROUTER_ENDPOINT.parse::<EndPoint>().unwrap()])
+            .unwrap();
+        c.scouting.multicast.set_enabled(Some(false)).unwrap();
+        let _ = c.set_mode(Some(WhatAmI::Router));
+        let s = ztimeout!(zenoh::open(c)).unwrap();
+        tracing::info!("Router ZID: {}", s.zid());
+        s
+    };
+
+    let client_tok = {
+        let mut c = zenoh::Config::default();
+        c.connect
+            .endpoints
+            .set(vec![ROUTER_ENDPOINT.parse::<EndPoint>().unwrap()])
+            .unwrap();
+        c.scouting.multicast.set_enabled(Some(false)).unwrap();
+        let _ = c.set_mode(Some(WhatAmI::Peer));
+        let s = ztimeout!(zenoh::open(c)).unwrap();
+        tracing::info!("Client (token) ZID: {}", s.zid());
+        s
+    };
+
+    let token = ztimeout!(client_tok.liveliness().declare_token(LIVELINESS_KEYEXPR)).unwrap();
+    tokio::time::sleep(SLEEP).await;
+
+    let client_subs = {
+        let mut c = zenoh::Config::default();
+        c.connect
+            .endpoints
+            .set(vec![ROUTER_ENDPOINT.parse::<EndPoint>().unwrap()])
+            .unwrap();
+        c.scouting.multicast.set_enabled(Some(false)).unwrap();
+        let _ = c.set_mode(Some(WhatAmI::Client));
+        let s = ztimeout!(zenoh::open(c)).unwrap();
+        tracing::info!("Client (subs) ZID: {}", s.zid());
+        s
+    };
+
+    let sub_no_history = ztimeout!(client_subs
+        .liveliness()
+        .declare_subscriber(LIVELINESS_KEYEXPR)
+        .history(false))
+    .unwrap();
+    tokio::time::sleep(SLEEP).await;
+
+    assert!(sub_no_history.try_recv().unwrap().is_none());
+
+    let sub_history = ztimeout!(client_subs
+        .liveliness()
+        .declare_subscriber(LIVELINESS_KEYEXPR)
+        .history(true))
+    .unwrap();
+    tokio::time::sleep(SLEEP).await;
+
+    let sample = ztimeout!(sub_history.recv_async()).unwrap();
+    assert!(sample.kind() == SampleKind::Put);
+    assert!(sample.key_expr().as_str() == LIVELINESS_KEYEXPR);
+    assert!(sub_history.try_recv().unwrap().is_none());
+
+    assert!(sub_no_history.try_recv().unwrap().is_none());
+
+    token.undeclare().await.unwrap();
+    tokio::time::sleep(SLEEP).await;
+
+    let sample = ztimeout!(sub_history.recv_async()).unwrap();
+    assert!(sample.kind() == SampleKind::Delete);
+    assert!(sample.key_expr().as_str() == LIVELINESS_KEYEXPR);
+
+    let sample = ztimeout!(sub_no_history.recv_async()).unwrap();
+    assert!(sample.kind() == SampleKind::Delete);
+    assert!(sample.key_expr().as_str() == LIVELINESS_KEYEXPR);
+
+    client_tok.close().await.unwrap();
+    client_subs.close().await.unwrap();
+    router.close().await.unwrap();
+}