P0 correctness & shutdown safety (Epics #2–#5) #10

Merged

p13marc merged 4 commits from fix/p0-correctness-0.5.0 into master

2026-06-29 14:07:38 +00:00

p13marc commented

2026-06-29 13:56:58 +00:00

(Migrated from github.com)

Implements the four P0 bug epics from the 0.5.0 milestone — the report's coherent "correctness" group. Self-contained fixes; no dependency or toolchain changes. One commit per epic.

Closes #2. Closes #3. Closes #4. Closes #5.

Epic #2 — Concurrency & shutdown safety

The state mutex was held across blocking Zenoh I/O, serializing stat/state reads against publish/receive and defeating the source's lock-free flush design.

Sink: publishing moved to a dedicated worker thread; render()/render_list() take a short lock to clone Arc handles, release it, then hand the payload to the worker and wait — bounded by a new publish-timeout-ms and cancellable via an unlocked atomic (unlock()/unlock_stop() + flush events). The NULL/flush transition no longer hangs under default congestion-control=block with no subscriber.
Source: flushing moved onto self (set lock-free); subscriber Arc-wrapped; recv_timeout loop runs with state released.
Bounded zenoh::open(...); poison-tolerant hot-path locks.

Epic #3 — Error handling & session lifecycle

Replaced string-matched error classification (format!("{:?}", e).contains("Timeout"), dead contains("timeout"|...)) with typed handling — recv_timeout already maps timeouts to Ok(None), so Err is a true disconnect.
Session registry now refcounts and closes on last release (was leaked forever), opens outside the global lock (double-checked insert), and warns on config mismatch. Wired into teardown via SessionWrapper::SharedGroup Drop.

Epic #4 — zenohdemux correctness

Removed the unused second subscriber (doubled inbound traffic).
Pads map keyed by full key expression (not derived pad name) with unique-name disambiguation; stable FNV-1a hash replaces DefaultHasher.
No more unwrap() panics in the receiver thread (surfaces a GStreamer error).
EOS pushed to dynamic pads on stop; no_more_pads() emitted via a quiescence window (no-more-pads-timeout-ms); errors stat exposed; demux session leak fixed.

Epic #5 — Compression & batched-render correctness

Breaking vs 0.4.x compression wire format (0.4.0 unpublished — intentional).

Self-describing payload frame (magic + algo + version) → a no-decoder receiver errors clearly instead of forwarding garbage.
LZ4 length-prefix removes the hardcoded 16 MB ceiling; HIGHCOMPRESSION fixes the inverted level semantics.
Compression promoted to a first-class gst.compression metadata field (no more user.* smuggling).
render_list parity: shared encode_and_publish() so batched buffers get identical compression + buffer-timing metadata (previously sent raw, dropping both).
Removed the dead dropped statistic.

Verification

Full test suite green for default and --features compression; each compression-* feature builds individually; release build green.
cargo fmt --check clean; cargo clippy --all-targets --all-features clean for the new code.
New tests: shutdown-no-hang, re-enabled stats integration, session refcount lifecycle, pad-collision, EOS-on-stop, >16 MB LZ4 round-trip, render_list compression+meta parity.
Manual: sender with no subscriber under default block mode shut down ~52 ms after SIGINT (no hang).

🤖 Generated with Claude Code

Implements the four **P0** bug epics from the 0.5.0 milestone — the report's coherent "correctness" group. Self-contained fixes; no dependency or toolchain changes. One commit per epic. Closes #2. Closes #3. Closes #4. Closes #5. ## Epic #2 — Concurrency & shutdown safety The `state` mutex was held across blocking Zenoh I/O, serializing stat/state reads against publish/receive and defeating the source's lock-free flush design. - Sink: publishing moved to a dedicated worker thread; `render()`/`render_list()` take a short lock to clone `Arc` handles, release it, then hand the payload to the worker and wait — bounded by a new **`publish-timeout-ms`** and cancellable via an `unlocked` atomic (`unlock()`/`unlock_stop()` + flush events). The NULL/flush transition no longer hangs under default `congestion-control=block` with no subscriber. - Source: `flushing` moved onto `self` (set lock-free); subscriber `Arc`-wrapped; `recv_timeout` loop runs with `state` released. - Bounded `zenoh::open(...)`; poison-tolerant hot-path locks. ## Epic #3 — Error handling & session lifecycle - Replaced string-matched error classification (`format!("{:?}", e).contains("Timeout")`, dead `contains("timeout"|...)`) with typed handling — `recv_timeout` already maps timeouts to `Ok(None)`, so `Err` is a true disconnect. - Session registry now **refcounts** and closes on last release (was leaked forever), opens **outside** the global lock (double-checked insert), and warns on config mismatch. Wired into teardown via `SessionWrapper::SharedGroup` `Drop`. ## Epic #4 — zenohdemux correctness - Removed the unused second subscriber (doubled inbound traffic). - Pads map keyed by **full key expression** (not derived pad name) with unique-name disambiguation; stable **FNV-1a** hash replaces `DefaultHasher`. - No more `unwrap()` panics in the receiver thread (surfaces a GStreamer error). - **EOS** pushed to dynamic pads on stop; **`no_more_pads()`** emitted via a quiescence window (`no-more-pads-timeout-ms`); `errors` stat exposed; demux session leak fixed. ## Epic #5 — Compression & batched-render correctness **Breaking vs 0.4.x compression wire format (0.4.0 unpublished — intentional).** - Self-describing payload frame (magic + algo + version) → a no-decoder receiver errors clearly instead of forwarding garbage. - LZ4 length-prefix removes the hardcoded 16 MB ceiling; `HIGHCOMPRESSION` fixes the inverted level semantics. - Compression promoted to a first-class `gst.compression` metadata field (no more `user.*` smuggling). - **`render_list` parity**: shared `encode_and_publish()` so batched buffers get identical compression + buffer-timing metadata (previously sent raw, dropping both). - Removed the dead `dropped` statistic. ## Verification - Full test suite green for default **and** `--features compression`; each `compression-*` feature builds individually; release build green. - `cargo fmt --check` clean; `cargo clippy --all-targets --all-features` clean for the new code. - New tests: shutdown-no-hang, re-enabled stats integration, session refcount lifecycle, pad-collision, EOS-on-stop, >16 MB LZ4 round-trip, render_list compression+meta parity. - Manual: sender with no subscriber under default block mode shut down ~52 ms after SIGINT (no hang). 🤖 Generated with [Claude Code](https://claude.com/claude-code)