fix: Route invalidation is not hat-local

Also merges 8d059ff9

Author: fuzzypixelz

commons/zenoh-protocol/src/zenoh/query.rs

@@ -80,7 +80,7 @@ pub mod flag {
     pub const Z: u8 = 1 << 7; // 0x80 Extensions    if Z==1 then an extension will follow
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
 pub struct Query {
     pub consolidation: ConsolidationMode,
     pub parameters: String,

examples/examples/z_info.rs

@@ -60,13 +60,15 @@ async fn main() {
         let transport_events = info
             .transport_events_listener()
             .history(false) // Don't repeat transports we already printed
-            .await;
+            .await
+            .expect("Failed to declare transport events listener");
 
         // Set up link events listener (using default handler)
         let link_events = info
             .link_events_listener()
             .history(false) // Don't repeat links we already printed
-            .await;
+            .await
+            .expect("Failed to declare link events listener");
 
         // Listen for events until CTRL-C
         loop {

zenoh/src/api/builders/info_links.rs

@@ -148,7 +148,8 @@ impl std::fmt::Debug for LinkEventsListenerInner {
 ///     .link_events_listener()
 ///     .history(true)
 ///     .with(flume::bounded(32))
-///     .await;
+///     .await
+///     .expect("Failed to declare link events listener");
 ///
 /// while let Ok(event) = listener.recv_async().await {
 ///     match event.kind() {
@@ -181,7 +182,8 @@ impl<Handler> LinkEventsListener<Handler> {
     /// let listener = session.info()
     ///     .link_events_listener()
     ///     .with(flume::bounded(32))
-    ///     .await;
+    ///     .await
+    ///     .expect("Failed to declare link events listener");
     /// listener.undeclare().await.unwrap();
     /// # }
     /// ```
@@ -300,7 +302,8 @@ impl<Handler> IntoFuture for LinkEventsListenerUndeclaration<Handler> {
 ///     .link_events_listener()
 ///     .history(true)
 ///     .with(flume::bounded(32))
-///     .await;
+///     .await
+///     .expect("Failed to declare link events listener");
 ///
 /// while let Ok(event) = listener.recv_async().await {
 ///     match event.kind() {
@@ -437,7 +440,7 @@ where
     Handler: IntoHandler<LinkEvent> + Send,
     Handler::Handler: Send,
 {
-    type To = LinkEventsListener<Handler::Handler>;
+    type To = ZResult<LinkEventsListener<Handler::Handler>>;
 }
 
 #[zenoh_macros::unstable]
@@ -448,19 +451,20 @@ where
 {
     fn wait(self) -> Self::To {
         let (callback, handler) = self.handler.into_handler();
-        let state = self
-            .session
-            .declare_transport_links_listener_inner(callback, self.history, self.transport)
-            .expect("Failed to declare link events listener");
+        let state = self.session.declare_transport_links_listener_inner(
+            callback,
+            self.history,
+            self.transport,
+        )?;
 
-        LinkEventsListener {
+        Ok(LinkEventsListener {
             inner: LinkEventsListenerInner {
                 session: self.session.clone(),
                 id: state.id,
                 undeclare_on_drop: true,
             },
             handler,
-        }
+        })
     }
 }
 

zenoh/src/api/builders/info_transport.rs

@@ -116,7 +116,8 @@ impl std::fmt::Debug for TransportEventsListenerInner {
 ///     .transport_events_listener()
 ///     .history(true)
 ///     .with(flume::bounded(32))
-///     .await;
+///     .await
+///     .expect("Failed to declare transport events listener");
 ///
 /// while let Ok(event) = listener.recv_async().await {
 ///     match event.kind() {
@@ -147,7 +148,8 @@ impl<Handler> TransportEventsListener<Handler> {
     /// let listener = session.info()
     ///     .transport_events_listener()
     ///     .with(flume::bounded(32))
-    ///     .await;
+    ///     .await
+    ///     .expect("Failed to declare transport events listener");
     /// listener.undeclare().await.unwrap();
     /// # }
     /// ```
@@ -263,7 +265,8 @@ impl<Handler> IntoFuture for TransportEventsListenerUndeclaration<Handler> {
 ///     .transport_events_listener()
 ///     .history(true)
 ///     .with(flume::bounded(32))
-///     .await;
+///     .await
+///     .expect("Failed to declare transport events listener");
 ///
 /// while let Ok(event) = events.recv_async().await {
 ///     match event.kind() {
@@ -378,7 +381,7 @@ where
     Handler: IntoHandler<TransportEvent> + Send,
     Handler::Handler: Send,
 {
-    type To = TransportEventsListener<Handler::Handler>;
+    type To = ZResult<TransportEventsListener<Handler::Handler>>;
 }
 
 #[zenoh_macros::unstable]
@@ -391,17 +394,16 @@ where
         let (callback, handler) = self.handler.into_handler();
         let state = self
             .session
-            .declare_transport_events_listener_inner(callback, self.history)
-            .expect("Failed to declare transport events listener");
+            .declare_transport_events_listener_inner(callback, self.history)?;
 
-        TransportEventsListener {
+        Ok(TransportEventsListener {
             inner: TransportEventsListenerInner {
                 session: self.session.clone(),
                 id: state.id,
                 undeclare_on_drop: true,
             },
             handler,
-        }
+        })
     }
 }
 

zenoh/src/api/info.rs

@@ -167,7 +167,8 @@ impl SessionInfo {
     ///     .transport_events_listener()
     ///     .history(true)
     ///     .with(flume::bounded(32))
-    ///     .await;
+    ///     .await
+    ///     .expect("Failed to declare transport events listener");
     ///
     /// while let Ok(event) = events.recv_async().await {
     ///     match event.kind() {
@@ -195,7 +196,8 @@ impl SessionInfo {
     ///     .link_events_listener()
     ///     .history(true)
     ///     .with(flume::bounded(32))
-    ///     .await;
+    ///     .await
+    ///     .expect("Failed to declare link events listener");
     ///
     /// while let Ok(event) = listener.recv_async().await {
     ///     match event.kind() {

zenoh/src/net/routing/dispatcher/face.rs

@@ -566,7 +566,7 @@ impl Primitives for Face {
         let src_fid = ctx.src_face.id;
 
         for mut res in hats[region].unregister_face_subscriptions(ctx.reborrow()) {
-            disable_matches_data_routes(&mut res, &region);
+            disable_matches_data_routes(ctx.tables, &mut res);
 
             let mut remaining = hats
                 .values_mut()
@@ -585,7 +585,7 @@ impl Primitives for Face {
         }
 
         for mut res in hats[region].unregister_face_queryables(ctx.reborrow()) {
-            disable_matches_query_routes(&mut res, &region);
+            disable_matches_query_routes(ctx.tables, &mut res);
 
             let remaining = hats
                 .iter()

zenoh/src/net/routing/dispatcher/pubsub.rs

@@ -32,6 +32,7 @@ use crate::net::routing::{
     dispatcher::{
         face::Face,
         local_resources::{LocalResourceInfoTrait, LocalResources},
+        tables::TablesData,
     },
     hat::{BaseContext, SendDeclare},
     router::{get_or_set_route, Direction, RouteBuilder},
@@ -113,7 +114,8 @@ impl Face {
                         .reduce(|_, _| SubscriberInfo);
 
                     hats[region].propagate_subscription(ctx.reborrow(), res.clone(), other_info);
-                    disable_matches_data_routes(&mut res, &region);
+
+                    disable_matches_data_routes(ctx.tables, &mut res);
                 }
 
                 drop(wtables);
@@ -188,7 +190,7 @@ impl Face {
         if let Some(mut res) =
             hats[region].unregister_subscription(ctx.reborrow(), id, res.clone(), node_id)
         {
-            disable_matches_data_routes(&mut res, &region);
+            disable_matches_data_routes(ctx.tables, &mut res);
 
             let mut remaining = tables
                 .hats
@@ -208,16 +210,29 @@ impl Face {
     }
 }
 
-/// Disables data routes of the given regions's hat.
+/// Disables data routes for the given [`Resource`].
+///
+/// ## Note
 ///
-/// A subscription declaration or undeclaration should in theory only invalidate data routes of its owner hat.
-pub(crate) fn disable_matches_data_routes(res: &mut Arc<Resource>, region: &Region) {
+/// **Changes in data/query routes are not hat-local**. For example, a north peer hat has routes for data
+/// that originate from south-bound remotes but has no routes for data that originate in its north
+/// region, thus a change in a broker's data routes affects the routes of the north peer hat.
+pub(crate) fn disable_matches_data_routes(_tables: &mut TablesData, res: &mut Arc<Resource>) {
     if res.ctx.is_some() {
-        get_mut_unchecked(res).context_mut().hats[region].disable_data_routes();
+        for hat in get_mut_unchecked(res).context_mut().hats.values_mut() {
+            hat.disable_data_routes();
+        }
+
         for match_ in &res.context().matches {
             let mut match_ = match_.upgrade().unwrap();
             if !Arc::ptr_eq(&match_, res) {
-                get_mut_unchecked(&mut match_).context_mut().hats[region].disable_data_routes();
+                for hat in get_mut_unchecked(&mut match_)
+                    .context_mut()
+                    .hats
+                    .values_mut()
+                {
+                    hat.disable_data_routes();
+                }
             }
         }
     }