fix(pubsub): stop running subscriber callbacks on the publishing thread

[YuanYuYuan]

Part of #282 — that issue states the defect class, the shared fix shape and the merge order.

Summary

Publishing from inside a subscriber callback on the same session deadlocked deterministically. This PR moves session-local delivery off the publishing thread onto a bounded per-subscriber drain queue. It also stops declaring zenoh-ext's AdvancedSubscriber for QoS profiles that do not need it.

Role in #282: instance fix. It targets main and uses no tracked lock types, so it can merge in any order relative to the keystone (#255). Its locks are candidates for conversion in the follow-up that closes #282.

Issue

Fixes #249. zenoh-ext's AdvancedSubscriber invokes the user callback while holding its own state mutex. That mutex is a std::sync::Mutex, which is not re-entrant. zenoh delivers a session-local sample inline on the publishing thread. A callback that publishes to its own topic therefore blocks on a lock its own thread already holds.

After this PR the callout runs on a separate drain thread, holding nothing — the acq · rel · callout shape #282 describes:

sequenceDiagram
    autonumber
    participant App as User thread
    participant Q as Bounded drain queue
    participant Drain as hiroz-sub-drain thread
    participant Cb as User callback

    App->>Q: publish(topic, m1) → enqueue, evicting if full
    Note over App: publish() returns immediately
    Q->>Drain: dequeue (state lock released before the callout)
    Drain->>Cb: invoke callback — holding nothing
    activate Cb
    Cb->>Q: publish(topic, m2) → enqueue
    Note over Q,Cb: no lock is held, so this returns.<br/>The loop iterates instead of recursing.
    deactivate Cb

Loading

Before / after

What a user observes, whatever the implementation:

	Before	After
Publish from inside a subscriber callback	deadlocks, deterministically	returns; the loop iterates
Volatile subscriber	declares zenoh-ext's AdvancedSubscriber, which holds a non-reentrant mutex across the callout	declares a plain subscriber
Session-local delivery	runs the callback inline on the publishing thread	enqueues; a drain thread runs the callback holding no lock
publish() returning	means the subscriber has already run	means the sample is queued
Undelivered same-session samples	none exist — delivery is inline, so loss is structurally impossible	bounded at the history depth, drop-oldest, with an escalating warn!
drop(subscriber)	returns at once; there is no queue and no thread to join	discards the backlog, then blocks until any in-flight callback returns

The last three rows are breaking changes. The Breaking changes section states each one in full.

What this PR does

Three stacked mechanisms produced the hang. This PR removes all three.

#	Change	Why
1	Declare AdvancedSubscriber only when QoS needs it (qos_needs_advanced)	It is the component holding the non-reentrant lock across the callout. For Volatile — the ROS 2 default — the wrapper added no liveliness subscriber, heartbeat or detection token, so it was pure overhead plus that lock. TransientLocal keeps it, for history replay and miss recovery.
2	Move session-local delivery onto a bounded drain queue (CallbackDispatcher::spawn)	The publishing thread no longer runs the callback, so it cannot re-enter its own lock.
3	Teardown discards the backlog instead of running it inline (Drop, dequeue checks closed before pending)	Draining first made drop(subscriber) run a callback per queued sample — backlog × callback_duration, unbounded.

All four CallbackDispatcher::spawn sites pass dispatch_capacity — the plain and advanced arms of both the typed builder (pubsub.rs) and the FFI raw subscriber (node.rs). DISPATCH_UNBOUNDED survives only as the KeepAll arm of dispatch_capacity itself.

Breaking changes

Seven. BC1–BC6 follow from moving session-local delivery off the publishing thread. BC7 is API surface.

tag	What changes	Who is affected	Before → after	Action
BC1	Session-local delivery is asynchronous	Anyone publishing and then asserting on a same-session side effect	publish() returning meant the subscriber had run → it does not	Synchronise explicitly
BC2	Same-session samples can be dropped	Callback subscribers with KeepLast(depth)	Inline delivery made loss structurally impossible → the queue is bounded at the history depth and drops oldest, with an escalating warn!	Use KeepAll if losslessness is required
BC3	Subscriber callbacks are no longer mutually excluded	Callbacks doing a non-atomic read-modify-write	zenoh-ext invoked the callback under its Mutex<State>, so callbacks were serialised by construction → they are not now	Add your own synchronisation
BC4	Local and remote publications on one topic are no longer ordered relative to each other	Plain (non-TransientLocal) subscribers	Single interleaved order → two independent paths	Do not rely on cross-source ordering
BC5	Dropping a subscriber discards its undelivered backlog, and blocks on any in-flight callback without bound	Anyone dropping a subscriber; anyone relying on teardown draining	main has no queue and no Drop impl; every accepted sample was delivered inline before publish returned, and drop(sub) joined nothing → the backlog is discarded (as destroying an rclcpp subscription does), but Drop now joins the drain thread	Never drop a subscriber while holding a lock, GIL or channel its callback also touches — see D1. Drain before dropping if the backlog matters
BC6	New public types	ffi consumers	SubscriberHandle and CallbackDispatcher are new public items; RawSubscriber::inner changes from AdvancedSubscriber<()> to SubscriberHandle	Update any direct use of inner
BC7	Synchronous local delivery is gone as a guarantee	Downstream code only	Nothing in this repo depended on it — rmw-zenoh-rs uses build_with_notifier, and wait_for_subscription reads the graph rather than local delivery	—

Important

Scope qualifier on BC2. It applies to every sample only on the advanced (TransientLocal) path, where always_shim enqueues unconditionally. On the plain path only locally published samples pass through the queue; a remote sample runs the callback inline on an RX worker and is neither queued nor dropped. The capacity constant is the same on both paths; which samples pass through it is not — see D2.

Queue bound versus rmw_zenoh_cpp

Read from source at rmw_zenoh_cpp/src. Upstream's zenoh sample callback only calls SubscriptionData::add_new_message, which locks, bounds, enqueues and notifies. User code runs later on the rclcpp executor thread, never on a zenoh delivery thread, on any profile.

Property	rmw_zenoh_cpp	hiroz after this PR	Verdict
bound	adapted_qos_profile.depth	dispatch_capacity: KeepLast(depth).max(1), KeepAll → unbounded	aligned
drop policy	drop oldest, pop_front before emplace_back	pop_front before push_back at len >= capacity	aligned
KEEP_ALL	unbounded	DISPATCH_UNBOUNDED	aligned
TransientLocal exemption	none — the check reads history policy only	none	aligned
applies to	every arriving sample, remote included	advanced path yes; plain path local samples only	divergent — see D2
advanced-subscriber cache	adv_sub_opts.history->max_samples = qos_.depth	cache_depth_from_history	aligned
teardown	is_shutdown_ set, undeclare, queue never drained, takes short-circuit	closed set, drain loop exits, backlog discarded	aligned
notifier on the delivery thread	trigger_callback() and the wait-set notify run under mutex_	DataHandler::handle calls notifier() after push returns, holding nothing	hiroz-better — upstream has the #282 shape here
loss reporting	MESSAGE_LOST from arrival sequence-number gaps; depth drops are debug-log only	escalating warn!; no MESSAGE_LOST at all	aligned on the narrow claim; hiroz-better on visibility; the absent MESSAGE_LOST is pre-existing (#292)
publisher backpressure	BLOCK only for RELIABLE && KEEP_ALL, else DROP	BLOCK for every Reliable	hiroz-worse, pre-existing — see D4
advanced subscriber gating	declared unconditionally; the durability if only adds options	plain subscriber for Volatile	divergent by design — the point of change 1

Note

The "zero depth" divergence is not reachable and is not a behavioural difference. Upstream replaces a zero depth with 42; dispatch_capacity floors it at 1. Every constructor of the value it receives already normalises zero away: QosHistory::KeepLast holds a NonZeroUsize, from_depth maps 0 → 10, the FFI conversion maps <= 0 → 10, and the default is KeepLast(10). All four call sites pass a locally declared entity's QoS, so depth.max(1) cannot observe a zero.

Two caveats stand: hiroz answers the same question three ways (10, 42, 1), of which only the 42 matches upstream; and upstream's in-tree comment claiming a floor of 1 in rmw_create_subscription has no matching code.

This PR changes no existing test. Both test files it touches are new. transient_local_delivery_preserves_order asserts a strictly increasing subsequence of what the publisher sent. It does not assert that every sample arrives, because KeepLast(10) promises no such thing. keep_all_delivers_every_local_sample covers losslessness on the profile that does promise it.

Evidence

New tests: 15 Rust integration tests across reentrant_publish.rs and dispatch_backpressure.rs, 2 qos_needs_advanced unit tests in pubsub.rs, and 9 Python cases in test_reentrant_publish.py.

Measurement	Taken on	How
28 check runs SUCCESS or SKIPPED, 0 failed, 0 pending; license/cla SUCCESS	5aa7e383	check rollup, read after the last run completed
Forcing qos_needs_advanced to return true — the shape of #249 — fails volatile_does_not_need_an_advanced_subscriber and passes 1 of 2. The unmodified code passes 2 of 2	reverse on f724a0a5, forward re-run on 5aa7e383 (292 lib tests, both detectors)	both directions run, each asserting a non-zero test count. The two commits differ only in comments
Reverting the advanced-path queue bound to DISPATCH_UNBOUNDED fails exactly one test: transient_local_keep_last_drops_the_oldest_local_samples	2376a01a	both directions run
dispatch_backpressure passes 4/4 with the bound in place	2376a01a	test run

keep_last_drops_the_oldest_local_samples detects unbounded growth and drop-newest, because it asserts values ([0, 46, 47, 48, 49]) rather than counts. Its determinism is structural, not slept-on: the callback sets parked before it waits on the latch, and the publisher spins on parked with a deadline. The drain thread therefore cannot pop during the burst.

What the tests actually pin

Property	Test
A callback publishing to its own topic iterates instead of recursing, on both paths	self_feeding_callback_loop_iterates_without_a_depth_cap, transient_local_self_feeding_callback_loop_iterates, intra_closed_loop_runs_iteratively
Dropping a subscriber from inside its own callback returns	transient_local_subscriber_dropped_inside_its_own_callback_does_not_deadlock
Teardown shuts the delivery thread down and leaks none	transient_local_subscriber_drop_shuts_down_delivery_thread, test_transient_local_dispatcher_threads_do_not_leak
A slow callback does not block the publisher	indirect: the burst runs under run_with_deadline, so a blocked publisher fails the deadline rather than being asserted against directly
Publishing from a callback does not hold the GIL	test_interpreter_stays_alive_during_reentrant_publish — a watchdog thread must keep ticking. On an unfixed build it hangs the process rather than failing, so it detects only under an external wall-clock timeout

Properties no test pins

These are argued in doc comments at their call sites. They are not evidence. A reviewer should read them as design intent that a future regression would not trip:

no lost wakeup between enqueue and dequeue; Drop not holding the queue lock across join(); no thread leaked when declare_subscriber fails; field order undeclaring before it joins; the plain path holding no zenoh lock inline; no lock-order cycle with zenoh-ext's statesref; the panic guard; KeepLast(usize::MAX); counter saturation; action-client teardown.

They are coverage gaps of the same kind #296 records, not claims this PR demonstrates.

Known defects in this change, disclosed

D1 is the one that changes what a reviewer must decide.

Drop for CallbackDispatcher joins a thread that is running user code. Concretely:

• Dropping a subscriber is an unbounded callout to user code, made while the dropping thread holds whatever it holds.
• It is reachable synchronously — ZSub owns SubscriberHandle by value, and both variants own dispatcher by value.
• If a callback takes a mutex the main thread also takes, drop(sub) never returns. No timeout, no log, no panic.
• On main there is no Drop impl and no thread, so drop(sub) always returned. This is Re-entrancy: user callbacks invoked while a lock is held #282's own shape relocated to teardown.
• It is a new hazard on a public API. build_with_callback's # Ownership section discloses it: dropping a subscriber blocks for as long as the callback runs, with no timeout. The hazard stands; only the surprise is removed.
• hiroz-py already patches the two instances it hit, both via py.allow_threads — direct evidence the hazard is real. Every Rust caller remains exposed.

The rest are pre-existing or enforcement gaps:

tag	Defect	Consequence
D2	The plain path bounds locally published samples only. A remote sample runs inline on a zenoh RX worker and never enters the queue	This is the qualifier on BC2. Upstream bounds every arriving sample. Over-warning, not under-warning
D3	LocalPublishGuard is pub(crate), so its doc's demand that "a fifth publish path must do the same" is unenforceable outside hiroz. One out-of-crate session.put exists, in the WASM plugin host transport	No reachable failure: that host's subscribers are raw zenoh handles, not ZSubs. Enforcement gap, not a live defect
D4	Pre-existing, new reach: Reliable maps to CongestionControl::Block on every history policy, where upstream uses Block only for RELIABLE && KEEP_ALL	A blocked put inside a callback now stalls that subscriber's drain loop, and the queue then drop-oldests silently. Before this PR the same publish stalled the caller's own thread, which was visible
D5	Pre-existing: destroy_subscriber resolves by a per-node id carrying no node identity	Two nodes in one interpreter both mint owned_id == 0. n1.destroy_subscriber(s2) tears down n1's subscription and returns Ok(())
D6	Each callback subscriber gets its own drain thread, spawned eagerly at build time. Measured on Linux x86-64, release: thread count is exactly baseline + N, and per subscriber this costs +12 KB RSS and +66 MB of virtual address space (a 2 MB stack plus a 64 MB glibc per-thread malloc arena)	Bounded and non-leaking — threads return to baseline on drop. The rmw path is unaffected: rmw-zenoh-rs builds subscriptions with build_with_notifier, so runs_user_code is false and no dispatcher is created. The cost falls on native Rust and hiroz-py callback subscribers only. Address space, not RSS, is the binding limit; MALLOC_ARENA_MAX=2 removes 64 of the 66 MB

What green CI does not prove

#296 records reverts that should fail the suite and do not — places where a future regression would pass unnoticed. None is a defect in this PR.

Note

The gating decision has a detector; the wiring does not. volatile_does_not_need_an_advanced_subscriber and transient_local_needs_an_advanced_subscriber pin qos_needs_advanced in both directions. Forcing it to return true — the shape of #249 — fails the first of the two.

No test asserts matches!(handle, SubscriberHandle::Plain { .. }) for a Volatile subscriber, because ZSub holds its handle privately. A rewiring that ignored qos_needs_advanced would therefore still pass. That assertion stays #296's first item.

Filed rather than fixed here:

• Notifier subscribers are exempt from the dispatcher on an unenforced claim #290 — notifier subscribers skip the dispatcher on an unenforced claim.
• The ffi feature is built by no crate on the PR gate, and its re-entrancy detector was deleted #291 — the gate builds ffi but never lints or tests it, and it lost its re-entrancy detector.
• RMW_EVENT_MESSAGE_LOST is plumbed but never raised #292 — hiroz never raises MESSAGE_LOST.

Important

D1 is disclosed, not fixed. build_with_callback documents that dropping a subscriber blocks for as long as its callback runs. A bounded teardown is the remaining work, and it is what a Rust caller needs before relying on drop(sub) returning.

[Copilot]

Pull request overview

Fixes subscriber callback deadlocks by moving hazardous delivery off publishing threads and updating Python bindings accordingly.

Changes:

• Adds QoS-aware subscriber selection and callback dispatch queues.
• Releases Python’s GIL during publishing and removes a receive-side payload copy.
• Adds Rust and Python regression and backpressure tests.

Reviewed changes

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

Show a summary per file

File	Description
crates/hiroz/src/pubsub.rs	Implements dispatching and QoS-aware subscribers.
crates/hiroz/src/node.rs	Applies dispatching to raw subscribers.
crates/hiroz/src/ffi/subscriber.rs	Stores generalized subscriber handles.
crates/hiroz/src/common.rs	Identifies handlers that execute user code.
crates/hiroz-tests/tests/reentrant_publish.rs	Tests reentrant publishing and teardown.
crates/hiroz-tests/tests/dispatch_backpressure.rs	Tests dispatcher queue bounds.
crates/hiroz-py/tests/test_reentrant_publish.py	Tests Python reentrancy and thread cleanup.
crates/hiroz-py/src/pubsub.rs	Releases the GIL while publishing.
crates/hiroz-py/src/node.rs	Uses borrowed sample payloads in callbacks.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (4)

crates/hiroz/src/pubsub.rs:1325

• This branch runs before runs_user_code is checked, so every TransientLocal queue/notifier subscriber (including rmw's build_with_notifier path) gets an unbounded dispatcher even though its handler only enqueues. That delays the wait-set notification and permits an unbounded backlog ahead of the already bounded BoundedQueue, contradicting the queue-mode contract below. Split the advanced path on runs_user_code and wire queue handlers directly to the advanced subscriber callback.

        let inner = if qos_needs_advanced(&self.entity.qos) {

crates/hiroz/src/pubsub.rs:1526

• build_internal already performs this encoding validation for every DataHandler at lines 1300–1320. Wrapping the new callback again parses each encoding twice and emits duplicate mismatch/unknown-format logs. Pass the callback directly to build_internal.

        let expected_encoding = self.expected_encoding.clone();
        let callback = Arc::new(move |sample: Sample| {

crates/hiroz-py/tests/test_reentrant_publish.py:132

• This makes the Python suite fail unconditionally on macOS, although macOS Python wheels are supported (docs/bindings/python.md:13-14). /proc/self/task is Linux-only and is needed solely by the drain-thread leak detector; skip that one test on unsupported platforms (a skip is not a false pass) while still running the portable re-entrancy tests.

    # The leak test can only see the Rust drain thread through procfs.
    assert os.path.isdir("/proc/self/task"), (
        "/proc/self/task is unavailable - the drain-thread leak test cannot run "
        "on this platform and must not be reported as passing"
    )

crates/hiroz-py/src/pubsub.rs:38

• The watchdog test does not independently exercise this GIL release: with dispatcher delivery enabled, the seed publish() returns quickly, and the test then sleeps for 300 ms before measuring progress, so reverting only allow_threads still passes. Add a detector that keeps the zenoh publish blocked while another Python thread must make progress; otherwise this regression can return unnoticed.

        py.allow_threads(|| self.inner.publish(zbuf.into()))
            .map_err(|e| e.into_pyerr())