feat(routing): upstream (northbound) declaration aggregation

DEFAULT_CONFIG.json5

@@ -349,6 +349,29 @@
   //   publishers: [
   //     // key_expression
   //   ],
+  //   /// Router upstream (northbound) declaration aggregation. Opt-in; effective only on a
+  //   /// router's north-bound forwarding (distinct from the session-local subscribers/publishers
+  //   /// above). A downstream subscriber/queryable whose key-expression is included by one of
+  //   /// these prefixes is folded into a single `${prefix}` declaration toward the upstream and
+  //   /// suppressed upstream (kept locally for downward routing), so an upstream router holds one
+  //   /// resource per configured prefix instead of one per forwarded child key-expression. The
+  //   /// folded queryable is advertised complete=false (presence, not authority); `target=AllComplete`
+  //   /// queries are still forwarded through the aggregate to the real child queryables.
+  //   /// Trade-offs when enabled — apply per-key policy/introspection at the router that performs the
+  //   /// fold (where child keys are still visible): upstream per-key declaration ACL/QoS/interceptors
+  //   /// and admin-space enumeration see only the `${prefix}` aggregate; the wildcard aggregate may
+  //   /// forward data toward the folding router for unsubscribed keys; liveliness tokens are NOT
+  //   /// folded. Each prefix should cover the key-expression sub-tree a single downstream branch owns;
+  //   /// prefixes should be disjoint, not a bare `**` root, and not under the `@/` admin key-space
+  //   /// (misconfigured prefixes are warned about at startup).
+  //   upstream: {
+  //     subscribers: [
+  //       // key_expression
+  //     ],
+  //     queryables: [
+  //       // key_expression
+  //     ],
+  //   },
   // },
 
   // /// Namespace prefix.

commons/zenoh-config/src/lib.rs

@@ -655,6 +655,44 @@ validated_struct::validator! {
             subscribers: Vec<OwnedKeyExpr>,
             /// A list of key-expressions for which all included publishers will be aggregated into.
             publishers: Vec<OwnedKeyExpr>,
+            /// Router upstream (northbound) declaration aggregation. **Opt-in** and effective only on
+            /// a router's north-bound forwarding — distinct from the session-local `subscribers`/
+            /// `publishers` above. When a north-bound router HAT forwards a downstream subscriber or
+            /// queryable whose key-expression is *included* by one of these prefixes, the per-key
+            /// children are folded into a single `${prefix}` declaration toward the upstream and the
+            /// children are suppressed upstream (they remain registered locally for downward
+            /// routing), so an upstream router holds one routing resource per configured prefix
+            /// instead of one per forwarded child key-expression. The folded queryable is advertised
+            /// `complete=false` (presence, not authority) so it never shadows a distinct complete
+            /// source; `target=AllComplete` queries are still forwarded through the aggregate to the
+            /// real child queryables. Empty by default → behaviour is identical to stock.
+            ///
+            /// **Trade-offs when enabled — apply per-key policy and introspection at the router that
+            /// performs the fold (downstream of the aggregation point), where the child key-expressions
+            /// are still visible:**
+            /// - Per-key declaration ACL / QoS-overwrite / interceptors at the *upstream* node act on
+            ///   the `${prefix}` aggregate, not the individual child keys (granularity degrades to the
+            ///   prefix).
+            /// - Upstream admin-space enumeration (`@/${zid}/router/subscriber|queryable/**`) shows the
+            ///   aggregate, not the per-key children.
+            /// - The aggregate is a wildcard, so data published to *unsubscribed* keys under the prefix
+            ///   is still forwarded toward the folding router (which drops it); keep prefixes as tight
+            ///   as the actually-subscribed sub-tree.
+            /// - Only subscribers and queryables are folded. Liveliness tokens are intentionally NOT
+            ///   folded: a liveliness sample's key-expression *is* the token's own key (there is no
+            ///   reply fan-out to recover per-key identity, unlike a wildcard queryable `get`), so a
+            ///   single `${prefix}/**` token could neither enumerate the live per-key set nor emit a
+            ///   per-key removal when one child token is undeclared.
+            /// - Each prefix should cover the key-expression sub-tree a single downstream branch owns;
+            ///   prefixes should be disjoint, not a bare `**` root, and not under the `@/` admin
+            ///   key-space (misconfigured prefixes are warned about at startup).
+            pub upstream: #[derive(Default)]
+            UpstreamAggregationConf {
+                /// Key-expression prefixes whose downstream subscribers are folded upstream.
+                subscribers: Vec<OwnedKeyExpr>,
+                /// Key-expression prefixes whose downstream queryables are folded upstream.
+                queryables: Vec<OwnedKeyExpr>,
+            },
         },
 
         /// Overwrite QoS options for Zenoh messages by key expression (ignores Zenoh API QoS config)

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

@@ -369,10 +369,12 @@ impl Face {
                 }
             }
             QueryTarget::AllComplete => {
-                for qabl in qabls
-                    .iter()
-                    .filter(|q| q.info.is_none_or(|info| info.complete) && filter(q))
-                {
+                for qabl in qabls.iter().filter(|q| {
+                    // B-A2: also forward to transparent forwarders (non-complete entries toward a
+                    // router whose resource covers the query, e.g. a northbound aggregate hiding
+                    // complete children) so the next router re-applies AllComplete against them.
+                    (q.info.is_none_or(|info| info.complete) || q.forwarder) && filter(q)
+                }) {
                     route.insert(qabl.dir.dst_face.id, || {
                         let mut dir = qabl.dir.clone();
                         let rid = insert_pending_query(&mut dir.dst_face, query.clone());

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

@@ -86,6 +86,14 @@ pub(crate) struct QueryTargetQabl {
     pub(crate) dir: Direction,
     pub(crate) info: Option<QueryableInfoType>,
     pub(crate) region: Region,
+    /// A *transparent forwarder*: a non-complete route entry toward a router whose resource
+    /// covers the query key-expr (e.g. a northbound `${prefix}/**` aggregate that suppresses its
+    /// per-key — possibly `complete=true` — children upstream). `QueryTarget::AllComplete` forwards
+    /// to such entries so the next router can re-apply the filter against its real children;
+    /// forwarding to a genuinely-incomplete remote is harmless (dropped one hop later). It never
+    /// affects `BestMatching` (which only short-circuits on `info.complete`), so it cannot shadow a
+    /// distinct complete source. Always `false` for terminal/face-local queryables.
+    pub(crate) forwarder: bool,
 }
 
 impl QueryTargetQabl {
@@ -109,6 +117,7 @@ impl QueryTargetQabl {
                 distance: if ctx.face.is_local { 0 } else { 1 },
             }),
             region: *region,
+            forwarder: false,
         })
     }
 }

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

@@ -25,7 +25,7 @@ use std::{
 
 use uhlc::HLC;
 use zenoh_config::{unwrap_or_default, Config};
-use zenoh_keyexpr::keyexpr;
+use zenoh_keyexpr::{keyexpr, OwnedKeyExpr};
 use zenoh_protocol::{
     core::{Bound, ExprId, Region, WireExpr, ZenohIdProto},
     network::Mapping,
@@ -145,6 +145,16 @@ pub(crate) struct TablesData {
 
     pub(crate) hats: RegionMap<HatTablesData>,
     pub(crate) routes_version: RoutesVersion,
+
+    /// Northbound declaration aggregation (`aggregation.upstream`). Prefixes whose downstream
+    /// subscribers/queryables a north-bound router HAT folds into one `${prefix}` declaration
+    /// upstream. `*_resources` are the corresponding aggregate `Resource`s, pre-created once at
+    /// gateway build (parallel to the prefix vecs) so the fold path never creates resources and the
+    /// aggregate's match-set is wired exactly once. Empty ⇒ behaviour identical to stock zenoh.
+    pub(crate) agg_sub_prefixes: Vec<OwnedKeyExpr>,
+    pub(crate) agg_sub_resources: Vec<Arc<Resource>>,
+    pub(crate) agg_qabl_prefixes: Vec<OwnedKeyExpr>,
+    pub(crate) agg_qabl_resources: Vec<Arc<Resource>>,
 }
 
 impl Debug for TablesData {
@@ -184,6 +194,8 @@ impl TablesData {
             Duration::from_millis(unwrap_or_default!(config.queries_default_timeout()));
         let interests_timeout =
             Duration::from_millis(unwrap_or_default!(config.routing().interests().timeout()));
+        let agg_sub_prefixes = config.aggregation().upstream().subscribers().clone();
+        let agg_qabl_prefixes = config.aggregation().upstream().queryables().clone();
         #[cfg(feature = "stats")]
         let mut stats_keys = zenoh_stats::StatsKeysTree::default();
         #[cfg(feature = "stats")]
@@ -210,6 +222,10 @@ impl TablesData {
             #[cfg(feature = "stats")]
             stats,
             routes_version: 0,
+            agg_sub_prefixes,
+            agg_sub_resources: Vec::new(),
+            agg_qabl_prefixes,
+            agg_qabl_resources: Vec::new(),
         })
     }
 
@@ -294,6 +310,130 @@ pub struct Tables {
 }
 
 impl Tables {
+    /// Pre-create the aggregate `Resource` for every configured `aggregation.upstream` prefix, once,
+    /// at gateway build (where the full `Tables` is available for `make_resource`). The northbound
+    /// fold path then only *looks these up* — it never creates resources — and because each
+    /// aggregate persists for the gateway's lifetime its match-set is wired exactly once, so
+    /// reconnect/churn cannot accumulate duplicate match cross-links. No-op when unconfigured.
+    pub(crate) fn precreate_upstream_aggregates(&mut self) {
+        self.warn_on_aggregation_misconfig();
+        // Create + wire each UNIQUE prefix's aggregate exactly once (a prefix may be listed in both
+        // `subscribers` and `queryables`), then fill the parallel resource vecs by lookup. This
+        // avoids re-running `match_resource` on a shared prefix (which would duplicate match links).
+        let mut by_ke: HashMap<OwnedKeyExpr, Arc<Resource>> = HashMap::new();
+        let prefixes: Vec<OwnedKeyExpr> = self
+            .data
+            .agg_sub_prefixes
+            .iter()
+            .chain(self.data.agg_qabl_prefixes.iter())
+            .cloned()
+            .collect();
+        for ke in &prefixes {
+            if !by_ke.contains_key(ke) {
+                let res = self.make_aggregate_resource(ke);
+                by_ke.insert(ke.clone(), res);
+            }
+        }
+        let sub_res: Vec<Arc<Resource>> = self
+            .data
+            .agg_sub_prefixes
+            .iter()
+            .map(|ke| by_ke[ke].clone())
+            .collect();
+        let qabl_res: Vec<Arc<Resource>> = self
+            .data
+            .agg_qabl_prefixes
+            .iter()
+            .map(|ke| by_ke[ke].clone())
+            .collect();
+        self.data.agg_sub_resources = sub_res;
+        self.data.agg_qabl_resources = qabl_res;
+    }
+
+    /// Emit operator-facing warnings for suspicious `aggregation.upstream` prefixes (root wildcards,
+    /// admin-space prefixes, duplicates, mutual inclusion) plus a one-time note about the trade-offs.
+    /// Soft: never rejects config — folding still applies to the prefixes exactly as given.
+    fn warn_on_aggregation_misconfig(&self) {
+        let all: Vec<(&str, &OwnedKeyExpr)> = self
+            .data
+            .agg_sub_prefixes
+            .iter()
+            .map(|k| ("subscribers", k))
+            .chain(
+                self.data
+                    .agg_qabl_prefixes
+                    .iter()
+                    .map(|k| ("queryables", k)),
+            )
+            .collect();
+        if all.is_empty() {
+            return;
+        }
+        tracing::info!(
+            "northbound aggregation enabled for {} prefix declaration(s): upstream per-key declaration \
+             ACL/QoS/interceptors and upstream admin-space enumeration see only the ${{prefix}} aggregate, \
+             and the wildcard aggregate may forward data toward the edge for unsubscribed keys — apply \
+             per-key policy and per-key introspection at the owning edge",
+            all.len()
+        );
+        for (kind, ke) in &all {
+            let s = ke.as_str();
+            if s == "**" || s == "*" {
+                tracing::warn!(
+                    "aggregation.upstream.{kind} prefix `{s}` is a root wildcard and would fold EVERY \
+                     matching declaration into one aggregate — almost certainly a misconfiguration"
+                );
+            }
+            if s == "@" || s.starts_with("@/") {
+                tracing::warn!(
+                    "aggregation.upstream.{kind} prefix `{s}` targets the `@/` admin key-space; folding \
+                     admin declarations breaks per-node introspection — almost certainly a misconfiguration"
+                );
+            }
+        }
+        // Duplicates / mutual inclusion: matching is first-match, so a narrower prefix listed after a
+        // broader one that includes it never receives children (its aggregate stays empty).
+        for i in 0..all.len() {
+            for j in (i + 1)..all.len() {
+                let (a, b) = (all[i].1, all[j].1);
+                if a == b {
+                    tracing::warn!(
+                        "aggregation.upstream prefix `{}` is listed more than once (redundant)",
+                        a.as_str()
+                    );
+                } else if a.includes(b) {
+                    tracing::warn!(
+                        "aggregation.upstream prefix `{}` includes `{}`: children of the narrower prefix \
+                         fold into the broader one (first match), so `{}` may never aggregate anything",
+                        a.as_str(),
+                        b.as_str(),
+                        b.as_str()
+                    );
+                } else if b.includes(a) {
+                    tracing::warn!(
+                        "aggregation.upstream prefix `{}` includes `{}`: children of the narrower prefix \
+                         fold into the broader one (first match), so `{}` may never aggregate anything",
+                        b.as_str(),
+                        a.as_str(),
+                        a.as_str()
+                    );
+                }
+            }
+        }
+    }
+
+    fn make_aggregate_resource(&mut self, ke: &OwnedKeyExpr) -> Arc<Resource> {
+        // Compute matches BEFORE creating the resource so it does not appear in its own match-set
+        // (mirrors `Face::with_mapped_expr`), then wire it in so a wildcard route resolving to the
+        // aggregate still reaches the (later-declared) children.
+        let mut matches = Resource::get_matches(&self.data, ke);
+        let mut root = self.data.root_res.clone();
+        let mut res = Resource::make_resource(self, &mut root, ke.as_str());
+        matches.push(Arc::downgrade(&res));
+        Resource::match_resource(&self.data, &mut res, matches);
+        res
+    }
+
     pub(crate) fn sourced_subscribers(&self) -> HashMap<Arc<Resource>, Sources> {
         self.hats
             .values()

zenoh/src/net/routing/gateway.rs

@@ -188,9 +188,14 @@ impl<'conf> GatewayBuilder<'conf> {
             stats,
         )?;
 
+        let mut tables = Tables { data, hats };
+        // Pre-create the northbound-aggregation aggregate resources (needs the fully-assembled
+        // `Tables`); no-op unless `aggregation.upstream` is configured.
+        tables.precreate_upstream_aggregates();
+
         Ok(Gateway {
             tables: Arc::new(TablesLock {
-                tables: RwLock::new(Tables { data, hats }),
+                tables: RwLock::new(tables),
                 ctrl_lock: Mutex::new(()),
                 queries_lock: RwLock::new(()),
             }),

zenoh/src/net/routing/hat/broker/queries.rs

@@ -190,18 +190,17 @@ impl Hat {
             return false;
         };
 
-        let is_matching_res = |qabl: &Resource| {
+        let is_matching = |qabl: &Resource, info: &QueryableInfoType| {
             let Some(ke) = qabl.keyexpr() else {
                 bug!("Queryable resource should not be root");
                 return false;
             };
-
-            ke.includes(key_expr)
-        };
-        let is_matching_info = |info: &QueryableInfoType| !complete || info.complete;
-
-        let is_matching = |qabl: &Resource, info: &QueryableInfoType| {
-            is_matching_res(qabl) && is_matching_info(info)
+            // Under AllComplete (`complete`), match a complete queryable OR a strict wildcard
+            // superset (`ke != key_expr`) — a transparent forwarder (e.g. a northbound aggregate
+            // that suppresses its complete children upstream). A same-key non-complete queryable
+            // stays non-matching. Mirrors the AllComplete route-path forwarder in
+            // dispatcher::compute_final_route (B-A2).
+            ke.includes(key_expr) && (!complete || info.complete || ke != key_expr)
         };
 
         let compute_other_matches = || {

zenoh/src/net/routing/hat/client/queries.rs

@@ -191,6 +191,7 @@ impl HatQueriesTrait for Hat {
                             node_id: DEFAULT_NODE_ID,
                         },
                         region: self.region,
+                        forwarder: false,
                     });
                 }
             }

zenoh/src/net/routing/hat/peer/queries.rs

@@ -317,6 +317,7 @@ impl HatQueriesTrait for Hat {
                         },
                         info: None,
                         region: self.region(),
+                        forwarder: false,
                     });
                 }
             }
@@ -334,6 +335,7 @@ impl HatQueriesTrait for Hat {
                     },
                     info: None,
                     region: self.region(),
+                    forwarder: false,
                 });
             }
         }

zenoh/src/net/routing/hat/router/mod.rs

@@ -105,6 +105,14 @@ pub(crate) struct Hat {
     router_qabls: HashSet<Arc<Resource>>,
     routers_net: Option<Network>,
     routers_trees_worker: TreesComputationWorker,
+    /// Northbound declaration aggregation: the per-key children folded into each configured
+    /// `aggregation.upstream` prefix, keyed by index into `TablesData::agg_sub_prefixes` /
+    /// `agg_qabl_prefixes`. The aggregate `Resource` itself is pre-created in `TablesData`; here we
+    /// only track folded children so the aggregate is declared once (first fold) and withdrawn on
+    /// the last. Queryable children additionally carry their `QueryableInfoType` so the aggregate's
+    /// advertised info is the merge of its children. Empty ⇒ stock behaviour.
+    north_agg_subs: HashMap<usize, HashSet<Arc<Resource>>>,
+    north_agg_qabls: HashMap<usize, HashMap<Arc<Resource>, QueryableInfoType>>,
     #[cfg(test)]
     disable_async_tree_computation: bool,
 }
@@ -124,6 +132,8 @@ impl Hat {
             router_tokens: HashSet::new(),
             routers_net: None,
             routers_trees_worker: TreesComputationWorker::new(region),
+            north_agg_subs: HashMap::new(),
+            north_agg_qabls: HashMap::new(),
             #[cfg(test)]
             disable_async_tree_computation: false,
         }