fix(zenoh-ext): stop calling user code while holding subscriber state locks

[YuanYuYuan]

Summary

Two zenoh-ext subscribers call user code while holding a non-reentrant std::sync::Mutex. A callback that re-enters the subscriber therefore takes a mutex its own thread already holds, and deadlocks against itself. This PR fixes both.

commit	file	sites
1	advanced_subscriber.rs	the sample callback, the miss callbacks, the flush helpers, the drop callback
2	querying_subscriber.rs	the sample callback, RepliesHandler::drop
3	both	review follow-up: closes two windows the deferral opened, and documents what it does not do
4	both	review follow-up: stops a poisoned mutex defeating the unwind paths

Both use the same mechanism, described below. std::sync::Mutex is not reentrant, so neither deadlock is a race — they reproduce on the first sample, every time.

This is an established class in zenoh, and the rule already exists

The project has recognised and fixed "a lock is held while user code runs" more than once. zenoh-ext was never brought in line.

Four examples follow. Each row was checked against the tracker; the list is illustrative rather than exhaustive, since issue search is keyword-based.

issue	what deadlocked	resolution
#359	put inside a subscriber callback	closed 2022
#1092	tables locks held across liveliness and matching-status callbacks	closed 2024
#1560	publishing inside a matching_listener callback	#1583, merged 2024
#2634	dropping a Querier whose completion callback re-enters the Session	closed 2026

#1092 states the rule this PR applies, in its own words:

It would be good if the Zenoh tables lock where released before calling liveliness subscribers and matching status listeners user callbacks.

and names both failure modes it prevents — Zenoh operations performed inside the callback, and user locks taken inside the callback while another task holds them and calls into Zenoh. Both apply here unchanged.

#2634 is the same mechanism as this PR's second commit. A lock held while a callback is dropped, on a non-reentrant primitive, re-entered on the same thread.

So this PR adds no new rule. It applies the existing one to the last two places that do not follow it.

Zenoh core already follows it

Read at 8aaf85c60f378fc9136511c49f90c9194f91b0ad:

delivery path	before the callout
Session::execute_subscriber_callbacks	drop(state)
Session::send_push_consume	drop(state)

Why it deadlocks

sequenceDiagram
    autonumber
    participant App as Publishing thread
    participant Sub as zenoh-ext subscriber
    participant Core as Session

    App->>Sub: sample arrives
    activate Sub
    Sub->>Sub: zlock!(state)
    Sub->>App: user callback runs, guard held
    App->>Core: publish, fetch, or drop a handle
    Core->>Sub: re-enters on the same thread
    Sub--xSub: zlock!(state) — this thread holds it
    Note over Sub: the thread waits on itself
    deactivate Sub

Loading

Step 6 has several shapes. For AdvancedSubscriber, SampleMissListener::drop and SampleMissListenerBuilder::wait both take statesref. For QueryingSubscriber, FetchingSubscriber::fetch resolves to register_handler, which takes state. For both, a callback that publishes re-enters through core's synchronous local delivery.

The Drop shapes are the ones that reach a user unannounced. Dropping a value inside a callback does not read as re-entering the middleware.

What fails without this

Each fix was measured against a control that carries the tests and not the fix.

control	tests	result
main + the test file	AdvancedSubscriber scenarios	fail, Got Deadlocked.
commit 1 + the full test file, querying_subscriber.rs untouched	5 reentrancy tests	3 passed, 2 failed — exactly the two new ones, both Deadlocked
commit 3 reverted	5 reentrancy tests	all pass — no failing baseline, see below
commit 4 reverted	5 reentrancy tests	all pass — no failing baseline, see below

The second control is the narrow one: it keeps the first fix and removes only the second, so the two new tests pin the second commit specifically rather than re-proving the first.

Note

Commit 3 has no failing baseline, and a test cannot honestly give it one. It closes two windows that are races: wait_callbacks() returning while a staged call is still owed, and a staged sample discarded by an undeclare landing between the stage and the drain. Both need the undeclare to land inside a window bounded by one guard release and the next acquisition. A test that hit it would be timing-dependent, and a test that missed it would read as proof of absence.

The argument is the control flow instead, and it is stated in wait_for_delivery's doc comment: a staging thread releases the guard before it claims the delivering role, so delivering.is_some() is false while the outbox is non-empty. Anything that watched only delivering returned early in that window.

Commit 4 has none either, for a different reason. It only changes what happens when the state mutex is poisoned, which requires a thread to panic while holding the state guard. No test in this suite does that, and one written to force it would be exercising std::sync::Mutex's poisoning contract rather than this code. The argument is the table below.

Commit 4 is the Copilot review's finding, extended to the one site it did not name. Four lock acquisitions can run inside a Drop; each mishandled poisoning, in one of two ways:

site	on a poisoned mutex, before	after
DeliveringRole::drop, both files	if let Ok(..) — silently skips, leaving the role claimed	into_inner(), marker cleared unconditionally
wait_for_delivery, initial acquisition	zlock! — unwraps and panics	returns, matching what its own wait loop already did
the subscriber's drop callback	zlock! — unwraps and panics	into_inner(); the handles still have to be released

The two zlock! sites are the sharper ones: both run inside CallbackDrop's destructor, and a panic there is a panic in a Drop, which aborts the process. The two if let Ok(..) sites cannot abort, but they defeat the point of an RAII guard whose contract is to release the marker however it is left.

test result: FAILED. 0 passed; 1 failed; 0 ignored; 2 filtered out; finished in 10.00s

thread 'dropping_a_miss_listener_inside_the_callback_is_safe' panicked at
zenoh-ext/tests/reentrancy.rs:238:5:
`SampleMissListener::drop` takes `statesref`. This deadlocked until the sample
callback stopped being invoked under that guard; a `Deadlocked` here means
collect · release · call has been undone somewhere. Got Deadlocked.

The fix

Delivery is interleaved with the state walk, so it cannot simply be hoisted out of the locked region. It is recorded inside the lock and replayed outside it.

Each subscriber's state gains a FIFO outbox and a delivering marker. Calls are staged under the guard. Whichever thread claims the delivering role drains the outbox with the guard released.

stage	before	after
state walk	under the guard	under the guard, unchanged
the callout	under the guard	staged, then called with the guard released
delivery order	the guard serialised it	the FIFO preserves it
two threads delivering	the guard excluded them	the delivering marker excludes them
a callback that re-enters	deadlocks	stages its work and returns

The shape becomes acquire · stage · release · call, and the re-entrant call lands on an unlocked mutex:

sequenceDiagram
    autonumber
    participant App as Delivering thread
    participant Sub as zenoh-ext subscriber
    participant Out as outbox FIFO

    App->>Sub: sample arrives
    activate Sub
    Sub->>Sub: zlock!(state)
    Sub->>Out: stage the call
    Sub->>Sub: release the guard
    deactivate Sub
    Sub->>Out: claim the delivering role
    Out->>App: user callback runs, no guard held
    App->>Sub: publish, fetch, or drop a handle
    Sub->>Sub: zlock!(state) — free, taken and released
    Sub->>Out: stage, then return
    Note over Out: the outer lap delivers it

Loading

Step 9 is the whole point. The re-entrant call now finds the mutex unlocked, does its work and returns, instead of waiting on a guard its own thread holds.

A staged call carries its own target

Staging is a commitment to deliver, so each Call carries the callback it is for rather than looking one up when it is drained.

	resolve at drain time	carry the target (what ships)
normal delivery	identical	identical
undeclare between the stage and the drain	State::callback is gone, the sample is discarded	delivered through the handle registered when it arrived
before delivery was deferred	could not happen — the sample was delivered under the same guard	matches that

Call::Miss already worked this way, because a miss listener can be unregistered independently. Call::Sample now does too. The cost is one handle clone per staged call.

The delivering role

One thread drains at a time. That is what keeps callbacks mutually excluded and delivery ordered now that the state guard no longer does it.

stateDiagram-v2
    direction LR
    Idle: delivering is None
    Draining: delivering is Some thread id

    [*] --> Idle
    Idle --> Draining: outbox non-empty, claim the role
    Draining --> Draining: batch done, outbox refilled, take it
    Draining --> Idle: batch done, outbox empty, release
    Draining --> Idle: callback unwound, the role guard clears it

Loading

Two edges are load-bearing.

Draining --> Idle on the empty outbox happens in a single lock acquisition. The "is the outbox empty" test and the release must not be split. If they were, a sample staged in between would find delivering still set, decline to deliver, and sit in the outbox until some later sample happened to flush it.

Draining --> Idle on unwind exists because a user callback can panic. DeliveringRole is an RAII guard whose Drop clears the marker. Without it the role would stay claimed forever and the subscriber would stop delivering for the rest of its life. It runs on the unwind path, so it takes the mutex without unwrapping — a panic in a destructor aborts the process.

Why not a per-call-site buffer

Callback is Send + Sync, so two threads can be inside a sample callback at once. With buffers local to each call site, one thread could advance the state to sn=5 and another to sn=6, then deliver 6 before 5. Holding the mutex across advance-and-deliver is exactly what used to prevent that.

Rejected, and why:

sequenceDiagram
    autonumber
    participant A as Thread A
    participant B as Thread B
    Note over A,B: per-call-site buffers
    A->>A: advance state to sn=5, buffer locally
    B->>B: advance state to sn=6, buffer locally
    B-->>B: deliver 6
    A-->>A: deliver 5
    Note over A,B: the subscriber observes 6 before 5

Loading

What shipped instead:

sequenceDiagram
    autonumber
    participant A as Thread A
    participant O as outbox FIFO
    participant B as Thread B
    A->>O: stage 5
    B->>O: stage 6
    A->>O: claim the delivering role
    B--xB: role already claimed, return
    O-->>A: deliver 5, then 6
    Note over A,B: the subscriber observes 5 then 6

Loading

Thread B does not block and does not lose its sample. It stages and returns; A picks the work up on its next lap.

Ordered delivery is the feature both types exist to provide — AdvancedSubscriber reorders by sequence number, and QueryingSubscriber merges fetched replies with live samples. One shared FIFO with one active deliverer restores it: calls leave in the order the state machine produced them, whichever thread drains them, and callbacks stay mutually excluded as before.

How the two commits align

The second commit reuses the first's mechanism rather than inventing one. The staging structure, the single-deliverer marker and the unwind-safe role guard are the same shape in both files.

They are deliberately not shared. QueryingSubscriber has no wait_callbacks-style API, so its copy needs no condition variable, and the type is deprecated — factoring the two together would tie a deprecated type's lifetime to the current one's. The querying_subscriber.rs copy says so at the type.

Two quieter call-outs

Dropping a user callback runs user Drop code. So unregister_miss_callback returns the callback for the caller to drop with the guard released, and the subscriber's drop callback takes its handles out before it releases them.

Behaviour changes

	change	who is affected	action
B1	wait_callbacks() on a sample-miss listener now genuinely blocks	callers who relied on it returning immediately	none, unless you depended on the no-op
B2	callbacks run with the state guard released	callbacks that observed the implicit exclusion	none — exclusion and ordering are preserved, now explicitly
B3	a sample may be delivered on a different thread from the one that received it	callbacks that read thread-local state, or assume the delivering thread	audit callbacks for thread-affinity or thread-local assumptions — see below
B4	the outbox is not backpressure	a producer faster than the callback	none available today — the queue grows where the producer used to block. See below
B5	a callback that republishes on every sample livelocks instead of deadlocking	callbacks that publish unconditionally to their own key	make the republish conditional. It was already a bug; the symptom changed

B1 needs explanation. wait_callbacks() was a no-op, and its comment said why:

// Note: no particular synchronization is required as of now since miss listener callbacks are always executed under state lock

Acquiring the state lock during undeclare was the wait. Removing callbacks-under-the-lock silently disarmed it: the method still returned Ok, just without the guarantee it documents. It now blocks on the delivering marker.

That marker is a thread id rather than a bool, and that is load-bearing. A callback that undeclares while it is itself being delivered would otherwise wait on its own delivery — the original defect wearing a different hat.

The thread id is what makes wait_callbacks safe to call from inside a callback:

sequenceDiagram
    autonumber
    participant T as Delivering thread
    participant S as State

    T->>T: draining, delivering = this thread
    T->>T: user callback runs
    T->>S: undeclare, with wait_callbacks
    alt if delivering were a bool
        S--xT: block until nobody is delivering
        Note over T: nobody is this very thread
    else delivering is a ThreadId
        S-->>T: owner is me, do not wait
    end

Loading

The regression was confirmed before it was fixed, not assumed. test_callback_drop_on_undeclare_advanced_sample_miss_listener catches it, because it counts callback drops as well as calls.

What the wait covers. Watching the delivering marker alone is not sufficient, and commit 3 fixes that. A staging thread releases the guard before it claims the role, so there is a window in which delivering is None while the outbox holds calls the subscriber has already committed to:

moment	delivering	outbox	a wait on delivering alone
thread A stages a miss, releases the guard	None	1 call	returns — wrong
A claims the role	Some(A)	0	waits
A runs the callback	Some(A)	0	waits
A releases the role	None	0	returns — correct

In row 1 the listener could still fire after wait_callbacks() had returned, because each staged Call carries its own handle and unregistering does not reach it. A non-empty outbox now counts as outstanding work. That is safe to wait for: every site that stages calls dispatch_outbox on the way out, so a non-empty outbox always has a thread on its way to drain it.

The wait is subscriber-wide, not per-listener: a sample callback in flight is waited out too. The staged calls share one queue, so a listener's own cannot be separated from them.

B3 is new with the outbox. A thread that finds another already draining stages its sample and returns, so the active deliverer runs that callback. Zenoh already invoked subscriber callbacks from arbitrary RX workers, so no thread identity was ever promised — but one thread delivering another thread's sample is new, and a callback keyed on thread-local state would notice.

B4 — the outbox is not backpressure. The state mutex used to throttle producers: a thread that arrived while a callback was running blocked on the guard until it finished. Staging returns immediately, so a producer faster than the callback grows the outbox instead of being slowed by it. Nothing bounds that.

Bounding it means either dropping samples or reintroducing a block, both larger semantic changes than removing the deadlock. This PR states the property rather than deciding it. If a bound is wanted, it deserves its own change and its own argument.

B5 — a re-entrant publish is safe only if it is conditional. A callback that publishes to its own subscriber stages and returns, and the outer lap delivers it; that is the fix working. A callback that republishes on every sample refills the outbox as fast as the drain empties it, so dispatch_outbox never leaves its refill loop. It livelocks where it used to deadlock.

Both are the callback's own unbounded recursion, and the old code caught it only by deadlocking. The tests keep it in check with once.swap(true, ..), which is the same discipline stated explicitly.

The only public signature edit is self → mut self on wait_callbacks. That is a binding mode, not part of the callable API, so it is not source-breaking.

The internal surface does change: State::unregister_miss_callback now returns Option<Callback<Miss>> and is #[must_use], so the caller drops the callback with the guard released. It is a private method, so no downstream code sees it.

Testing

The tests avoid the trap that a timeout alone proves nothing: "the callback ran and wedged" and "the callback never ran" look identical from outside.

• Each scenario asserts it entered the callback.
• run_scenario returns a three-way Completed / Deadlocked / Panicked rather than a bool, so an assertion failure inside a scenario cannot be reported as a hang.
• the_callback_guard_fires_when_the_callback_never_runs is the control. It publishes where the subscriber does not match and asserts the guard fires, which proves the guard executes at all.
• The scenarios open with scouting, gossip and listeners disabled. They are single-process and need no network. On defaults they peer with siblings running in parallel, which falsifies the control's premise.

Full zenoh-ext suite, measured on 46934fc7c (the current tip, with commits 3 and 4): 30 passed, 0 failed — 5 reentrancy, 12 advanced, 4 liveliness, 9 serialization and utils. cargo fmt --check and cargo clippy --all-targets --features unstable -- -D warnings are clean on the same commit.

The advanced and liveliness suites are the ones that would catch a delivery-ordering regression, and the liveliness suite exercises QueryingSubscriber and FetchingSubscriber directly. Commit 3 changes what a staged call carries and what the wait covers, so those two suites passing on the current tip is the evidence that neither change reordered or dropped anything. Commit 4 changes only poisoned-mutex behaviour, so it is invisible to them by construction — that is stated rather than implied.

One #[allow] is added, on the Call enum. clippy::large_enum_variant wants Sample boxed; boxing it would put a heap allocation on every delivered sample to shrink a queue that is empty whenever delivery keeps up. The reason is stated at the attribute.

The three red Lints and doc tests jobs are not from this PR

Note

They fail on zenoh/src/net/runtime/adminspace.rs, which this PR does not touch. Rust 1.98.0 added clippy::useless_borrows_in_formatting, and the offending line is on main unchanged. #2756 fixes it repo-wide; its own lint jobs are green. This branch needs a rebase once that merges, and no change of its own.

evidence
what the branch changes	zenoh-ext/src/advanced_subscriber.rs, zenoh-ext/src/querying_subscriber.rs, zenoh-ext/tests/reentrancy.rs — nothing else
the failing line	version = &*LONG_VERSION, in metrics, identical on main
not specific to this branch	the same three jobs fail on #2751 and #2755
before the toolchain moved	this branch was 28/28 green on 2026-08-20

---

🏷️ Label-Based Checklist

Based on the labels applied to this PR, please complete these additional requirements:

Labels: bug

🐛 Bug Fix Requirements

Since this PR is labeled as a bug fix, please ensure:

• Root cause documented - Explain what caused the bug in the PR description
• Reproduction test added - Test that fails on main branch without the fix
• Test passes with fix - The reproduction test passes with your changes
• Regression prevention - Test will catch if this bug reoccurs in the future
• Fix is minimal - Changes are focused only on fixing the bug
• Related bugs checked - Verified no similar bugs exist in related code

Why this matters: Bugs without tests often reoccur.

Instructions:

1. Check off items as you complete them (change - [ ] to - [x])
2. The PR checklist CI will verify these are completed

This checklist updates automatically when labels change, but preserves your checked boxes.

[codecov]

Codecov Report

❌ Patch coverage is 84.00000% with 48 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.86%. Comparing base (8aaf85c) to head (e2d4dc6).
⚠️ Report is 2 commits behind head on main.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
zenoh-ext/src/advanced_subscriber.rs	85.89%	34 Missing ⚠️
zenoh-ext/src/querying_subscriber.rs	76.27%	14 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2744      +/-   ##
==========================================
+ Coverage   74.67%   74.86%   +0.19%     
==========================================
  Files         425      419       -6     
  Lines       63930    63960      +30     
==========================================
+ Hits        47739    47883     +144     
+ Misses      16191    16077     -114     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

[Copilot]

Pull request overview

This PR fixes deterministic self-deadlocks in zenoh-ext subscribers caused by invoking (or dropping) user-provided callbacks while holding a non-reentrant std::sync::Mutex, by deferring callback execution until after the state lock is released. The approach introduces an outbox FIFO plus a single “delivering role” marker to preserve ordering and mutual exclusion while preventing re-entrant lock acquisition.

Changes:

• Defers AdvancedSubscriber callback/miss-callback delivery by staging Calls into a FIFO outbox under the lock and draining it after releasing the lock (including updated wait_callbacks semantics).
• Applies the same “stage then drain” mechanism to FetchingSubscriber / QueryingSubscriber, including avoiding dropping user callbacks while holding the state lock.
• Adds zenoh-ext/tests/reentrancy.rs to reproduce and prevent regressions for the re-entrancy deadlock class.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
zenoh-ext/src/advanced_subscriber.rs	Introduces outbox-based deferred delivery + delivering role + wait_callbacks support to avoid calling/dropping user code under the subscriber state mutex.
zenoh-ext/src/querying_subscriber.rs	Defers sample delivery (and callback drops) outside the InnerState mutex via an outbox and delivering role to prevent re-entrant deadlocks.
zenoh-ext/tests/reentrancy.rs	Adds deterministic re-entrancy tests ensuring callbacks are not invoked while holding subscriber state locks.

Suppressed comments (1)

zenoh-ext/src/advanced_subscriber.rs:810

• wait_for_delivery uses zlock!(statesref) to acquire the mutex, which unwraps poisoning. If this is ever called from a destructor/unwind path while the mutex is poisoned, it can trigger a panic-in-drop abort. Since this helper already treats poisoning as "nothing useful to do" later in the loop, it should also avoid unwrapping on the initial lock acquisition.

fn wait_for_delivery(statesref: &Arc<Mutex<State>>) {
    let me = std::thread::current().id();
    let mut guard = zlock!(statesref);
    let condvar = guard.delivery_done.clone();
    loop {

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.