Fix reader-thread wedge: decouple republish onto a bounded publish-worker queue

.agent/work-plans/issue-10/plan.md

@@ -0,0 +1,195 @@
+# Plan: Bridge wedges (reader thread blocked, Recv-Q backup) when remote subscriber dies
+
+## Issue
+
+https://github.com/rolker/udp_bridge/issues/10
+
+## Context
+
+The receive path runs entirely in `socket_drain_group_` (MutuallyExclusive):
+`spin_once` (10 ms timer) → `recvfrom` loop → `decode` → `decodeData` →
+`publisher->publish()` at `src/udp_bridge.cpp:776`. The destination publisher
+is created **RELIABLE** + `KEEP_LAST(1)` (`qos_resolution.h` — a deliberate
+`rmw_zenoh_cpp` 0.2.9 graph-visibility workaround; design intent is
+`BEST_AVAILABLE`). When a local subscriber (e.g. CAMP) dies *uncleanly*, DDS
+keeps it matched until liveliness times out, and a RELIABLE `publish()` blocks
+waiting for a history slot. Because that publish is on the **MutuallyExclusive**
+drain group, the blocked call stops `spin_once` from re-running → `recvfrom`
+stalls → kernel `Recv-Q` backs up (361 KB observed) → forwarded topics go
+silent. Four occurrences on 2026-04-27; one masked an RTK alert by freezing
+`/diagnostics` stale on the operator side.
+
+`qos_design.md` (lines 36-39, 89-96, 305-309) already documents this mechanism
+and rules out the QoS knobs as the fix: `BEST_EFFORT` would break CAMP's
+RELIABLE subscribers (operator goes dark), `BEST_AVAILABLE` is blocked by the
+rmw bug, and even BEST_AVAILABLE behaves RELIABLE for the local hop when a
+RELIABLE subscriber matches — so it can still block. The fix must be
+**transport-level, independent of QoS and rmw**: the drain thread must never
+make a call that can block on a single subscriber.
+
+`publish()` is not the *only* rmw call on the drain path. `decodeData` also
+runs, on the same thread: `create_generic_publisher()` (first-arrival, line
+759 — the existing comment at 733-737 calls this "exactly the wedge mechanism")
+and `sendBridgeInfo()` (first-arrival, line 773 — a graph query + two-mutex
+snapshot + UDP send). Both can block on rmw discovery. So the decoupling must
+move *all* of first-arrival create + bridge-info + publish off the drain
+thread, not just the `publish()` call — otherwise the first message on any new
+topic can still wedge the reader. The drain thread's only remaining
+`decodeData` work is the (CPU-bound) `MessageInternal` deserialize + building
+the inner `SerializedMessage`.
+
+## Approach
+
+Decouple the whole rmw-touching tail of `decodeData` from the socket-drain
+thread via a dedicated **publish-worker thread** fed by a bounded queue. The
+drain thread deserializes the `MessageInternal`, builds the inner
+`SerializedMessage`, and enqueues a self-contained item; the worker does
+lookup-or-create publisher + first-arrival bridge-info + `publish()`. A stalled
+publish (or a blocking first-arrival create) then blocks only the worker —
+`spin_once` keeps draining the socket. This realizes the package's documented
+"best-effort with loss reduction" model on the local hop: when the local
+publish path stalls, we drop republished messages rather than backing up the
+wire reader.
+
+1. **`PublishQueue` component** (`include/udp_bridge/publish_queue.h`,
+   header-only). A bounded MPSC queue of `PublishItem`
+   `{topic, datatype, reliability, durability, history_depth, SerializedMessage}`
+   plus one worker thread. The worker is parameterized by an injected **sink**
+   `std::function<void(PublishItem&&)>` — keeps all rmw/`UDPBridge` coupling out
+   of the component so it unit-tests with a fake sink (a deliberately-blocking
+   sink proves the non-stall property; a recording sink proves FIFO). `push()`
+   never blocks; the bound is in **bytes** (`PublishItem::byte_size()`, since a
+   reassembled image is large) and on overflow it drops the **oldest** item(s)
+   and increments an atomic dropped counter. `start()` launches the worker,
+   `stop()` signals + joins; `push()` after `stop()` is a no-op (drop), so it is
+   safe to call from any thread at any lifecycle phase. `run()` wraps the sink
+   call in a last-resort `catch(...)` barrier so a throwing sink can never
+   `std::terminate` the process (review finding — see Implementation Notes).
+2. **Wire into `UDPBridge`** — own a `PublishQueue` member. The injected sink
+   is a member lambda (`publishItem`) that runs the existing three-phase
+   lookup-or-create, the first-arrival `sendBridgeInfo()`, and
+   `publisher->publish()` — that block moves out of `decodeData` into the sink,
+   wrapped in a `try/catch` that logs via `RCLCPP_ERROR` (mirroring `decode()`'s
+   catch-all; the work previously ran under that barrier on the executor
+   thread). `decodeData` keeps only the deserialize + inner-`SerializedMessage`
+   build, then `publish_queue_.push(std::move(item))`. Lifecycle: `configure()`
+   the queue in `on_configure`, `start()` in `on_activate`, `stop()`+join in
+   `on_deactivate` (so the worker runs **only while ACTIVE** — matching the
+   state-guard discipline of the other publish paths, and never publishing on
+   the non-lifecycle destination publishers while INACTIVE), with a backstop
+   `stop()` in `on_cleanup` and the destructor. The join gates teardown:
+   destination publishers are **non-lifecycle** `GenericPublisher`s, so the
+   worker must be joined before the node/`publishers_` are torn down (the member
+   is declared last for the destructor-ordering backstop). Re-entrancy: an
+   activate→deactivate→activate cycle stops+restarts a fresh worker.
+3. **Bound + observability** — byte budget as a ROS parameter
+   (`publish_queue_max_bytes`, declared via `declareIfMissing`); register a
+   single global drop diagnostic via the existing `syncDiagnosticTasks()` /
+   `diagnostic_task_names_` pattern so a stalled local subscriber surfaces as a
+   WARN instead of silently. (Drop count is a global tally, not per-remote.)
+4. **Unit test** (`test/test_publish_queue.cpp`) — the non-stall property at the
+   component level, which is also the bug: with a sink that blocks on a latch
+   (simulating a stuck `publish()`), assert `push()` still returns promptly and
+   the producer is never blocked; FIFO order via a recording sink; drop-oldest
+   + counter when over the byte budget; oversize-item enqueue-then-evict;
+   push-after-stop / push-before-start no-op; clean `stop()`/join while the sink
+   is blocked; and a throwing sink does not kill the worker. Register via
+   `add_udp_bridge_gtest` in `CMakeLists.txt`.
+5. **End-to-end reproduction** — the kill-a-subscriber-mid-forward repro
+   (acceptance #1) belongs in the issue #18 bench harness (PR #27 draft, which
+   already targets the "wedge" failure mode). Coordinate rather than duplicate:
+   this PR lands the component + the blocking-sink non-stall unit test;
+   cross-link #18 for the live-DDS integration repro. (Residual: no test in
+   *this* PR exercises a real dying DDS subscriber — the blocking-sink test is
+   the proxy; see Open Questions.)
+6. **Docs** — add a "Receive-path publish decoupling" subsection to
+   `qos_design.md`: why the wedge is fixed at the transport layer (not QoS) and
+   holds regardless of the RELIABLE/BEST_AVAILABLE default, and that queue drops
+   are **unrecoverable** loss *after* successful wire delivery — distinct from,
+   and not recoverable by, the resend layer (which operates pre-`decodeData`).
+   Also update the `socket_drain_group_` invariant comment in
+   `udp_bridge_node.cpp` (which currently predicts this exact "move publish work
+   to a separate group via an internal queue" change).
+
+## Files to Change
+
+| File | Change |
+|------|--------|
+| `udp_bridge/include/udp_bridge/publish_queue.h` | New header-only bounded publish-worker queue (`PublishItem`, byte-bounded, injected sink, atomic drop counter) |
+| `udp_bridge/src/udp_bridge.cpp` | Move first-arrival create + `sendBridgeInfo` + `publish` into the queue sink lambda (`publishItem`, wrapped in try/catch+log); `decodeData` deserializes + enqueues; `configure` queue in `on_configure`, `start` in `on_activate`, `stop`+join in `on_deactivate` (backstop in `on_cleanup`); `declareIfMissing("publish_queue_max_bytes")`; drop diagnostic via `syncDiagnosticTasks` |
+| `udp_bridge/include/udp_bridge/udp_bridge.h` | `PublishQueue` member + `publish_queue_max_bytes_`; the sink helper decl |
+| `udp_bridge/src/udp_bridge_node.cpp` | Update the `socket_drain_group_` invariant comment (it predicts this change) |
+| `udp_bridge/test/test_publish_queue.cpp` | New unit test (blocking-sink non-stall, FIFO, byte-budget drop-oldest, join-while-blocked) |
+| `udp_bridge/CMakeLists.txt` | Register `test_publish_queue` via `add_udp_bridge_gtest` |
+| `udp_bridge/doc/qos_design.md` | Document transport-level decoupling as the #10 fix + unrecoverable-drop note |
+
+## Principles Self-Check
+
+| Principle | Consideration |
+|---|---|
+| Test what breaks | Unit-test the non-stall/drop invariants — the exact failure seen 4× in the field — not framework glue |
+| A change includes its consequences | `qos_design.md` updated; diagnostic surfaces drops; #18 coordination noted, not deferred silently |
+| Only what's needed | One worker + bounded queue; watchdog (issue fix-shape #2) deferred — decoupling removes the blocking site, making Recv-Q polling redundant |
+| Capture decisions | Rationale (why transport-level, not QoS) recorded in `qos_design.md`, not just the diff |
+| Human control & transparency | Drop count is observable via diagnostics; queue bound is a tunable parameter |
+
+## ADR Compliance
+
+| ADR | Triggered | How addressed |
+|---|---|---|
+| ADR-0008 (ROS 2 conventions) | Yes | Worker thread is internal; lifecycle start/stop follows LifecycleNode activate/deactivate contract; QoS contract unchanged |
+| ADR-0013 (progress.md vocabulary) | Yes | `progress.md` gets a `## Plan Authored` entry |
+
+## Consequences
+
+| If we change... | Also update... | Included? |
+|---|---|---|
+| publish path moves off drain thread | `qos_design.md` receive-path description | Yes |
+| add `publish_queue_max` param | parameter docs / README if it lists params | Yes (verify during impl) |
+| reader-thread non-stall property | issue #18 bench harness adds the e2e wedge repro | No — cross-linked follow-up in #18 |
+
+## Decisions (resolved)
+
+- **Cross-topic isolation scope → single worker** (user, 2026-05-25). One worker
+  thread + bounded queue this PR; it fixes the reported reader-thread wedge (all
+  four field occurrences). Per-publisher isolation (so a stalled subscriber on
+  one topic can't head-of-line-block others) is deferred to a follow-up issue,
+  to be filed when this PR lands. The bounded-queue drop counter + diagnostic
+  make the head-of-line residual observable in the meantime.
+
+## Open Questions
+
+- **E2E repro home.** Default: the kill-subscriber integration repro lands in
+  the #18 bench harness (PR #27), which already targets the "wedge" failure
+  mode — not a standalone launch_test here. Flag if you'd rather it live in
+  this repo's tests.
+
+## Estimated Scope
+
+Single PR for the component + unit test + wiring + docs. The end-to-end
+reproduction (acceptance #1) is coordinated into issue #18's harness.
+
+## Implementation Notes
+
+Rationale-bearing pivots from the local `/review-code` pass (Deep tier;
+Claude + Copilot adversarial agreed), beyond what's visible in the diff:
+
+- **Worker exception barrier.** The sink (`publishItem`) can throw —
+  `create_generic_publisher` on an unknown/cross-version `datatype`,
+  `publish`, and `sendBridgeInfo`'s `ConnectionException` on socket errors.
+  Before #10 this work ran inside `decode()`'s catch-all on the executor
+  thread (logged + dropped). A plain `std::thread` has no such backstop, so an
+  uncaught throw would `std::terminate` the bridge — a regression. Fixed in two
+  layers: `publishItem` catches + logs via `RCLCPP_ERROR` (informative, has the
+  node logger); `PublishQueue::run` has a last-resort `catch(...)` so the
+  worker can never terminate the process even if a future sink forgets.
+- **Worker runs only while ACTIVE.** Moved `start()` to `on_activate` and added
+  `stop()` to `on_deactivate` (was: start in `on_configure`). Otherwise the
+  worker could publish queued items after deactivation onto the
+  lifecycle-gated `bridge_info`/`topic_statistics` publishers (warn + drop) and
+  diverge from the state-guard discipline every other publish path follows.
+- **Shutdown-join bound.** `stop()` discards the backlog and waits at most for
+  one in-flight publish to return (the rmw `max_blocking_time`), not the whole
+  queue — so leaving ACTIVE / cleanup can't hang on the backlog. A single
+  genuinely-infinite `publish()` is the inherent limit of joining a thread
+  blocked in a syscall; documented, not engineered around.

.agent/work-plans/issue-10/progress.md

@@ -0,0 +1,111 @@
+---
+issue: 10
+---
+
+# Issue #10 — Bridge wedges (reader thread blocked, Recv-Q backup) when remote subscriber dies
+
+## Plan Authored
+**Status**: complete
+**When**: 2026-05-25 18:30 -0400
+**By**: Claude Code Agent (Claude Opus 4.7 (1M context))
+
+**Plan**: `.agent/work-plans/issue-10/plan.md` at `44a81ad`
+**PR**: https://github.com/rolker/udp_bridge/pull/28 (`[PLAN]` prefix)
+**Phases**: single
+
+### Open questions
+- [x] Cross-topic isolation scope — **decided: single publish-worker this PR; per-publisher isolation deferred to a follow-up issue** (user, 2026-05-25).
+- [ ] Confirm the kill-subscriber end-to-end reproduction (acceptance #1) lands in the issue #18 bench harness (PR #27), not a standalone launch_test here.
+
+## Plan Review
+**Status**: complete — verdict **revise**, revisions applied
+**When**: 2026-05-25 18:30 -0400
+**By**: Claude Code Agent (Claude Opus 4.7 (1M context)) — fresh-context internal review (pre-Copilot)
+
+**Plan**: `.agent/work-plans/issue-10/plan.md` at `078f794`
+
+Findings, all incorporated into the revised plan:
+- [x] **B1** (blocking): `create_generic_publisher()` (first-arrival) is also on the drain thread and can block on rmw → move the whole first-arrival create + bridge-info + publish into the worker sink, not just `publish()`.
+- [x] **B2**: first-arrival `sendBridgeInfo()` also inline on drain → folded into B1's sink move.
+- [x] **B3/B4** (lifecycle): construct/start queue in `on_configure`, `stop()`+join in `on_cleanup`; `push()` no-op after stop; join gates post-deactivation publishes (destination publishers are non-lifecycle); cleanup→configure re-creates a fresh worker.
+- [x] **S2**: bound the queue by **bytes**, not message count (reassembled images are large).
+- [x] **N1/N2**: injected `std::function` sink seam → unit-test non-stall with a blocking fake sink (the proxy for a dying DDS subscriber).
+- [x] **S1/S3/S4/N3**: doc unrecoverable drop; diagnostic via `syncDiagnosticTasks`; `declareIfMissing`; update `udp_bridge_node.cpp` invariant comment.
+
+## Implementation
+**Status**: complete
+**When**: 2026-05-25 19:48 -0400
+**By**: Claude Code Agent (Claude Opus 4.7 (1M context))
+
+**Branch**: feature/issue-10 at `558bebf`
+Implemented per plan: `PublishQueue` component + `decodeData`/`publishItem` rewiring + lifecycle wiring + diagnostic + `qos_design.md` + `udp_bridge_node.cpp` comment. Build clean; full suite **94 tests, 0 failures** (8 new `PublishQueue` tests).
+
+## Local Review (Pre-Push)
+**Status**: complete
+**When**: 2026-05-25 19:48 -0400
+**By**: Claude Code Agent (Claude Opus 4.7 (1M context))
+**Verdict**: changes-requested (all resolved)
+
+**Branch**: feature/issue-10 at `558bebf`
+**Mode**: pre-push
+**Depth**: Deep (reason: concurrency + lifecycle in a core comms package)
+**Must-fix**: 1 | **Suggestions**: 2
+
+### Findings
+- [x] (must-fix) Worker `run()` had no exception barrier — a throwing sink (`create_generic_publisher`/`publish`/`sendBridgeInfo`) → `std::terminate`, a regression vs `decode()`'s old catch-all. Fixed: `publishItem` try/catch+log + `run()` `catch(...)` backstop + `ThrowingSinkDoesNotKillWorker` test — `publish_queue.h`, `udp_bridge.cpp` `publishItem`
+- [x] (suggestion) Worker published while INACTIVE — moved `start`→`on_activate`, `stop`→`on_deactivate` — `udp_bridge.cpp` `on_activate`/`on_deactivate`
+- [x] (suggestion) New concurrent outbound-send caller + shutdown-join bound documented — `udp_bridge_node.cpp` callback-group comment
+- [x] (static) cppcheck: all findings on pre-existing/context lines (not changed lines) — dropped per silence filter
+
+## External Review
+**Status**: complete
+**When**: 2026-05-25 20:14 -0400
+**By**: Claude Code Agent (Claude Opus 4.7 (1M context))
+
+**PR**: #28 at `e37e733`
+**Reviews**: 1 review (Copilot bot), 2 valid, 0 false positives
+**CI**: copilot-pull-request-reviewer success (repo has no build/test CI; local build + 94-test suite is the gate)
+
+### Actions
+- [x] Fix: `test_publish_queue.cpp` `wait_until_blocked()` — bounded 2s deadline + `ASSERT_LT` (commit `241f2e5`).
+- [x] Fix: `udp_bridge.cpp` on_configure — upper-bound clamp (2 GiB) + warning for `publish_queue_max_bytes` (commit `241f2e5`).
+
+## Local Review (Pre-Push)
+**Status**: complete
+**When**: 2026-05-25 20:14 -0400
+**By**: Claude Code Agent (Claude Opus 4.7 (1M context))
+**Verdict**: approved
+
+**Branch**: feature/issue-10 at `241f2e5`
+**Mode**: pre-push
+**Depth**: Light (reason: two small low-risk follow-up fixes; bulk already Deep-reviewed)
+**Must-fix**: 0 | **Suggestions**: 0
+
+### Findings
+- [ ] No issues found. LGTM. (cppcheck: only a cross-TU `unusedStructMember` false positive on `publish_queue_max_bytes_`; Copilot: "No issues." Build clean, 94 tests pass.)
+
+## External Review
+**Status**: complete
+**When**: 2026-05-25 21:31 -0400
+**By**: Claude Code Agent (Claude Opus 4.7 (1M context))
+
+**PR**: #28 at `c655350`
+**Reviews**: 2 reviews (Copilot bot) — fresh re-review at HEAD; 1 valid, 0 false positives; 2 prior comments now addressed/stale
+**CI**: copilot-pull-request-reviewer success
+
+### Actions
+- [x] Fix: `publish_queue.h` `start()` — `running_ = true` now set after the `std::thread` is constructed (commit `329d308`; re-reviewed clean — cppcheck + Copilot "No issues"). Build clean, 94 tests pass.
+- [x] (addressed `241f2e5`) `wait_until_blocked()` bounded wait — confirmed in re-review as stale.
+- [x] (addressed `241f2e5`) `publish_queue_max_bytes` upper clamp — confirmed in re-review as stale.
+
+## External Review
+**Status**: complete
+**When**: 2026-05-26 00:20 -0400
+**By**: Claude Code Agent (Claude Opus 4.7 (1M context))
+
+**PR**: #28 at `ce13281`
+**Reviews**: 3 (Copilot bot) — fresh re-review at current HEAD found 0 comments; all prior findings stale/resolved
+**CI**: copilot-pull-request-reviewer success
+
+### Actions
+- [x] No outstanding review comments. All three rounds resolved (`241f2e5`, `329d308`); current-HEAD re-review clean. PR ready to merge.

udp_bridge/CMakeLists.txt

@@ -143,6 +143,9 @@ if(BUILD_TESTING)
   # through the same helper so it gets UDP_BRIDGE_BUILD_TESTING too —
   # required for in-package ABI alignment per the helper's contract.
   add_udp_bridge_gtest(test_giveup_diagnostic test/test_giveup_diagnostic.cpp)
+  # test_publish_queue (issue #10) exercises the bounded publish-worker
+  # queue in isolation via an injected sink — no UDPBridge / node needed.
+  add_udp_bridge_gtest(test_publish_queue test/test_publish_queue.cpp)
 endif()
 
 ament_package()