AllComplete fails when two same-keyExpr queryables on a single router have mixed complete flags

[leon-somana]

Describe the bug

When a single router (or any session) declares two queryables on the exact same key_expr — one with complete=true and one with complete=false — a client query with QueryTarget::AllComplete against a matching key receives no replies at all. Neither queryable's callback is invoked. The query times out client-side after ResponseFinal arrives with an empty result set.

Other combinations on the same key_expr work as expected:

Config (one router, one client session)	target=AllComplete	target=All
1 queryable, complete=true	✓	✓
2 queryables, both complete=true, same key_expr	✓ both reply	✓ both reply
2 queryables, both complete=false, same key_expr	n/a (no complete)	✓ both reply
2 queryables, complete=true + complete=false, same key_expr	✗ neither replies	✓ both reply
2 queryables, complete=true + complete=false, different key_exprs	✓	✓

The trigger is specifically the combination of same key_expr + mixed complete flags + target=AllComplete. Either separating the key_exprs or using target=All works around it.

To reproduce

Rust binary to run with cargo run --release:

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::Mutex;
use zenoh::query::{ConsolidationMode, QueryTarget};
use zenoh::sample::Sample;
use zenoh::Config;

type AnyErr = Box<dyn std::error::Error + Send + Sync>;
type Store = Arc<Mutex<HashMap<String, Sample>>>;

async fn attach(sess: &zenoh::Session, name: &'static str, key_expr: &'static str, complete: bool) -> Result<(), AnyErr> {
    let store: Store = Arc::new(Mutex::new(HashMap::new()));
    let sub_store = store.clone();
    let _sub = sess.declare_subscriber(key_expr).callback(move |sample| {
        let s = sub_store.clone();
        tokio::spawn(async move {
            eprintln!("  [{}] sub got put for {}", name, sample.key_expr());
            s.lock().await.insert(sample.key_expr().to_string(), sample);
        });
    }).await?;
    std::mem::forget(_sub);
    let q_store = store.clone();
    let _qabl = sess.declare_queryable(key_expr).complete(complete).callback(move |query| {
        let s = q_store.clone();
        tokio::spawn(async move {
            let ke = query.key_expr().clone();
            if let Some(sample) = s.lock().await.get(&ke.to_string()) {
                eprintln!("  [{}] qabl replying for {}", name, ke);
                let _ = query.reply(ke.clone(), sample.payload().clone()).timestamp(*sample.timestamp().unwrap()).await;
            } else {
                eprintln!("  [{}] qabl: {} not present", name, ke);
            }
        });
    }).await?;
    std::mem::forget(_qabl);
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), AnyErr> {
    let mut router_cfg = Config::default();
    router_cfg.insert_json5("mode", r#""router""#).unwrap();
    router_cfg.insert_json5("listen/endpoints", r#"["tcp/127.0.0.1:17447"]"#).unwrap();
    let router = zenoh::open(router_cfg).await?;

    // Two queryables on the SAME key_expr, mixed `complete`.
    attach(&router, "primary", "test/state/**", true).await?;
    attach(&router, "backup",  "test/state/**", false).await?;

    let mut client_cfg = Config::default();
    client_cfg.insert_json5("mode", r#""client""#).unwrap();
    client_cfg.insert_json5("connect/endpoints", r#"["tcp/127.0.0.1:17447"]"#).unwrap();
    let client = zenoh::open(client_cfg).await?;
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;

    let key = "test/state/probe-key";
    client.put(key, "hello".as_bytes()).await?;

    let start = Instant::now();
    let replies = client.get(key)
        .target(QueryTarget::AllComplete)
        .consolidation(ConsolidationMode::Monotonic)
        .await?;
    let mut n = 0usize;
    while let Ok(r) = replies.recv_async().await {
        if r.result().is_ok() { n += 1; }
    }
    eprintln!("AllComplete: {} reply(ies) in {}ms", n, start.elapsed().as_millis());
    assert!(n > 0, "AllComplete returned no replies — bug reproduced");
    Ok(())
}

System info

zenoh 1.9.0

[leon-somana]

The bug might be:

  hat/broker/queries.rs:343-354 in register_queryable:

  match res.face_ctxs.get_mut(&ctx.src_face.id) {
      Some(ctx) => {
          get_mut_unchecked(ctx).qabl = Some(*info);   // ← OVERWRITES, doesn't merge
      }
      None => {
          let ctx = res.face_ctxs.entry(ctx.src_face.id)
              .or_insert_with(|| Arc::new(FaceContext::new(ctx.src_face.clone())));
          get_mut_unchecked(ctx).qabl = Some(*info);
      }
  }

[yellowhatter]

Tagging @OlivierHecart

[fuzzypixelz]

This behavior is expected and was intentionally maintained during the recent routing layer rewrite in #2096. Queryable information is only reduced in gateways that need to represent multiple south-bound notes. The protocol treats each queryable (re-)declaration on the same keyexpr as an info—i.e. completeness, distance—update. This behavior is the same for session-to-runtime and on-the-wire queryable declarations.

@leon-somana: does this behavior break a specific use case for you?

[leon-somana]

Hey @fuzzypixelz ,

My use case is that I have two storages (update: I incorrectly wrote "two storage plugins" earlier) that both subscribe to the same key expression - but one is complete and the other is an incomplete fallback (more specifically, one is a local rocksdb and the other is an eventually consistent S3 replica). It sounds like what you're saying is that the second storage plugin declaration (complete = false) overrides the first storage plugin declaration (complete = true) and so there isn't a way to use target to filter?

[fuzzypixelz]

My use case is that I have two storage plugins that both subscribe to the same key expression - but one is complete and the other is an incomplete fallback (more specifically, one is a local rocksdb and the other is an eventually consistent S3 replica). It sounds like what you're saying is that the second storage plugin declaration (complete = false) overrides the first storage plugin declaration (complete = true) and so there isn't a way to use target to filter?

Plugins don't share the same session, so my reasoning above doesn't apply. To simulate a multi-session scenario you can write:

#!/usr/bin/env nuze -X0

zenoh runtime open r {mode:router}
zenoh session open --runtime r -s scomplete
zenoh session open --runtime r -s sincomplete

let qcomplete = job spawn {
  zenoh queryable -s scomplete k --complete true { |s| "complete" }
}

sleep 500ms

let qincomplete = job spawn {
  zenoh queryable -s sincomplete k --complete false { |s| "incomplete" }
}

(run the above with nuze -X0e 'source <file.nu>')

$ nuze -X0e 'zenoh open {mode:client}'
> zenoh get k --target all --consolidation none | get payload
╭───┬────────────╮
│ 0 │ incomplete │
│ 1 │ complete   │
╰───┴────────────╯
> zenoh get k --target all-complete --consolidation none | get payload
╭───┬──────────╮
│ 0 │ complete │
╰───┴──────────╯

Thus I cannot reproduce the issue. If I were to use one session only:

#!/usr/bin/env nuze -X0

zenoh open {mode:"router"}

let qcomplete = job spawn {
  zenoh queryable k --complete true { |s| "complete" }
}

sleep 500ms

let qincomplete = job spawn {
  zenoh queryable k --complete false { |s| "incomplete" }
}

$ nuze -X0e 'zenoh open {mode:client}'
> zenoh get k --target all --consolidation none | get payload
╭───┬────────────╮
│ 0 │ incomplete │
│ 1 │ complete   │
╰───┴────────────╯
> zenoh get k --target all-complete --consolidation none | get payload
╭────────────╮
│ empty list │
╰────────────╯

Now this corresponds to what you observe, yet it doesn't line up with the scenario you describe.

To conclude, I believe the "same-session completeness overriding" behavior is not the root cause of your issue, something else is going on.

[leon-somana]

Sorry, my terminology was wrong when I said "two storage plugins". I meant two storages set-up against the standard storage manager plugin instantiation.

Plugins don't share the same session, but multiple storages do share the same session within the storage plugin.

- plugins/zenoh-plugin-storage-manager/src/lib.rs:132 — one session per plugin: let session = Arc::new(zenoh::session::init(runtime.clone()).wait()?);
- spawn_storage (lib.rs:289) passes self.session.clone() into create_and_start_storage for every storage.
- Each storage then declares its queryable on that shared session with its own flag: src/storages_mgt/service.rs:153-154 → self.session.declare_queryable(...).complete(self.configuration.complete).

So this is specifically replicable with zenohd + 2 storages.

  {
    plugins: {
      storage_manager: {
        volumes: {
          rocksdb: { /* ... */ },
          s3:      { backend: "s3", /* ... */ },
        },
        storages: {
          primary: {
            key_expr: "state/**",
            volume:   "rocksdb",
            complete: true,        // authoritative — answers AllComplete
          },
          fallback: {
            key_expr: "state/**",   // identical key_expr
            volume:   "s3",
            complete: false,       // best-effort — should NOT answer AllComplete
          },
        },
      },
    },
  }

[fuzzypixelz]

I see two possible solution here:

1. Update the storage manager plugin to use one session per storage.
2. Update the session to merge queryable info for multiple queryables on the same keyexpr.

I would lean toward (2) as the least intrusive of the two.