Fix reader-thread wedge: decouple republish onto a bounded publish-worker queue by rolker · Pull Request #28 · rolker/udp_bridge

rolker · 2026-05-25T22:30:49Z

Closes #10

Problem

The bridge's UDP reader wedges — reader thread blocked, kernel Recv-Q backs up, forwarded topics go silent — when a local subscriber dies during active forwarding. Observed 4× in one field day (2026-04-27); one wedge masked an RTK/NTRIP alert by freezing /diagnostics stale on the operator side.

Root cause: decodeData's publisher->publish() runs in the MutuallyExclusive socket_drain_group_ (spin_once → decode → decodeData). The destination publisher is deliberately RELIABLE (an rmw_zenoh_cpp 0.2.9 graph-visibility workaround — see qos_design.md). When a local reliable subscriber dies uncleanly, DDS keeps it matched and the publish blocks on a full history slot, which stalls spin_once → recvfrom → Recv-Q backup. The first-arrival create_generic_publisher() and sendBridgeInfo() are on the same thread and can block on rmw discovery too.

QoS can't be the fix (qos_design.md rules out all three knobs: BEST_EFFORT breaks CAMP's reliable subscribers, BEST_AVAILABLE is blocked by the rmw bug and still behaves reliable on the local hop). So the fix is transport-level and rmw-independent.

Fix

Decouple the rmw-touching tail of decodeData onto a dedicated single-worker PublishQueue (include/udp_bridge/publish_queue.h):

decodeData now only deserializes the MessageInternal, builds the inner SerializedMessage, and enqueues a PublishItem.
The worker (publishItem) does find-or-create publisher + first-arrival sendBridgeInfo + publish(). A stalled publish blocks only the worker; recvfrom keeps draining.
The queue is byte-bounded (publish_queue_max_bytes, declareIfMissing), drop-oldest (KEEP_LAST-consistent), with an atomic drop counter surfaced via a publish queue diagnostic (WARN on recent drops). Queue drops are unrecoverable loss after wire delivery — documented in qos_design.md as the local-hop expression of the bridge's best-effort-with-loss-reduction model.
Worker runs only while ACTIVE (start in on_activate, stop+join in on_deactivate, backstop in on_cleanup + destructor). publishItem catches+logs (mirroring decode()'s catch-all) and run() has a last-resort barrier, so a bad packet can't std::terminate the node.

Test plan

colcon build clean (only pre-existing -Wpedantic/sign-compare warnings).
Full suite: 94 tests, 0 failures.
New test_publish_queue.cpp (8 cases): non-stall under a blocking sink (the bug, via fake sink), FIFO, byte-budget drop-oldest, oversize-item enqueue-then-evict, push-after-stop / push-before-start no-op, stop-while-blocked join, throwing-sink survival.
Deferred to Bench-test harness for field failure modes (resend amp, wedge, rate-limit accuracy, saturation) #18 bench harness: live-DDS kill-a-subscriber-mid-forward reproduction (acceptance criterion Install Udp bridge node/ui #1). Bench-test harness for field failure modes (resend amp, wedge, rate-limit accuracy, saturation) #18 (PR Bench-test harness for field failure modes (netns + tc netem) #27) already targets the "wedge" failure mode.

Notes

Single worker fixes the reported reader-thread wedge (all 4 occurrences). Per-publisher isolation (so a stalled subscriber on one topic can't head-of-line-block others) is a deferred follow-up to file when this lands — the drop counter makes the residual observable.
Reviewed locally via /review-code (Deep; Claude + Copilot adversarial). Findings (missing worker exception barrier; ACTIVE-only lifecycle) addressed before this description. Plan + review timeline in .agent/work-plans/issue-10/.

Authored-By: Claude Code Agent
Model: Claude Opus 4.7 (1M context)

Decouple destination publish from the socket-drain thread via a bounded publish-worker queue, so a RELIABLE publish() stalling on an uncleanly-dead subscriber can no longer wedge recvfrom / back up the kernel Recv-Q. QoS knobs are ruled out by qos_design.md (BEST_EFFORT breaks CAMP, BEST_AVAILABLE blocked by rmw_zenoh_cpp 0.2.9); the fix is transport-level and rmw-independent.

Move all first-arrival rmw work (create_generic_publisher + sendBridgeInfo + publish) into the worker sink, not just publish(); byte-bounded queue with injected sink for testability; construct/join worker in on_configure/on_cleanup (non-lifecycle publishers — join gates teardown); update udp_bridge_node.cpp invariant comment + qos_design unrecoverable-drop note.

Header-only single-worker queue that decouples republishing from the caller. push() never blocks; bound is in bytes with drop-oldest (KEEP_LAST-consistent) and an atomic drop counter; the publish work is an injected sink so the queue unit-tests with a fake sink. test_publish_queue covers the non-stall property (blocking sink), FIFO, byte-budget drop-oldest, oversize-item handling, push-after-stop/before-start no-op, and stop-while-blocked join.

decodeData now only deserializes the MessageInternal and enqueues a PublishItem; the rmw-touching tail — find-or-create the destination publisher, first-arrival sendBridgeInfo, and publish() — runs on the publish worker (publishItem). A RELIABLE publish stalling on an uncleanly-dead subscriber, or a first-arrival create_generic_publisher blocking on rmw discovery, now blocks only the worker; recvfrom keeps draining and the kernel Recv-Q no longer backs up. The fix is transport-level and independent of the destination publisher's reliability (RELIABLE workaround / BEST_AVAILABLE intent) — see qos_design.md. Worker starts in on_configure, stops+joins in on_cleanup (dtor backstop); publish_queue_max_bytes is a declareIfMissing parameter; a publish-queue diagnostic WARNs on recent drops. Updates the socket_drain_group_ invariant comment that predicted this change.

Local /review-code (Deep; Claude + Copilot adversarial agreed): - (must-fix) The publish worker had no exception barrier. The sink can throw (create_generic_publisher on unknown/cross-version datatype, publish, and sendBridgeInfo's ConnectionException) — and before #10 this ran inside decode()'s catch-all on the executor thread. On a plain std::thread an uncaught throw is std::terminate: a single bad packet would crash the bridge. publishItem now try/catches + logs (RCLCPP_ERROR, mirroring decode's catch-all); PublishQueue::run adds a last-resort catch(...) backstop. New test ThrowingSinkDoesNotKillWorker covers it. - Run the worker only while ACTIVE: start() in on_activate, stop()+join in on_deactivate (was start in on_configure). Stops it publishing onto the lifecycle-gated bridge_info/topic_statistics publishers while INACTIVE and matches the state-guard discipline of the other publish paths. Backstop stop() retained in on_cleanup + the destructor. - Document the new concurrent outbound-send caller in the callback-group audit comment, and the shutdown-join bound (one in-flight publish max_blocking_time).

Copilot

Pull request overview

This PR addresses issue #10 by preventing the UDP socket-drain (reader) thread from wedging when destination-side republishing blocks (e.g., a RELIABLE publisher stalled by a dead-but-still-matched subscriber). It does this by moving all rmw-touching republish work off the drain thread onto a bounded, single-worker publish queue, with observability via diagnostics and unit tests that exercise the non-stall contract.

Changes:

Introduce udp_bridge::PublishQueue (byte-bounded, drop-oldest, single worker) and wire decodeData() to enqueue PublishItems instead of publishing inline.
Add lifecycle-managed worker start/stop, a publish-queue diagnostic, and a configurable publish_queue_max_bytes parameter.
Add focused unit tests for queue behavior (non-blocking push under blocking sink, FIFO, dropping semantics, stop/join behavior, exception containment) and document the design in qos_design.md.

Reviewed changes

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

Show a summary per file

File	Description
`udp_bridge/include/udp_bridge/publish_queue.h`	New bounded worker queue abstraction with injected sink and drop accounting.
`udp_bridge/src/udp_bridge.cpp`	Rewire `decodeData()` → enqueue; add `publishItem()` worker sink; lifecycle start/stop; parameter + diagnostic.
`udp_bridge/include/udp_bridge/udp_bridge.h`	Declare publish-queue members, parameter storage, and diagnostics helper.
`udp_bridge/src/udp_bridge_node.cpp`	Update callback-group/threading architecture comments to reflect the new worker.
`udp_bridge/test/test_publish_queue.cpp`	New unit tests covering non-stall, ordering, drops, shutdown, and exception survival.
`udp_bridge/doc/qos_design.md`	Document transport-level decoupling rationale and unrecoverable post-wire-drop semantics.
`udp_bridge/CMakeLists.txt`	Register the new gtest target for the publish-queue tests.
`.agent/work-plans/issue-10/plan.md`	Planning document for the change.
`.agent/work-plans/issue-10/progress.md`	Progress/status tracking for the plan.

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

+  // Spin until the worker is parked inside the sink (has popped an item).
+  void wait_until_blocked()
+  {
+    while(calls.load() == 0)
+      std::this_thread::yield();
+  }


+  declareIfMissing("publish_queue_max_bytes",
+    static_cast<int64_t>(kDefaultPublishQueueMaxBytes));
+  {
+    int64_t configured = get_parameter("publish_queue_max_bytes").as_int();
+    if(configured < static_cast<int64_t>(kMinPublishQueueMaxBytes))
+    {
+      RCLCPP_WARN_STREAM(get_logger(),
+        "publish_queue_max_bytes " << configured << " is below "
+        << kMinPublishQueueMaxBytes << "; clamping to "
+        << kMinPublishQueueMaxBytes);
+      configured = static_cast<int64_t>(kMinPublishQueueMaxBytes);
+    }
+    publish_queue_max_bytes_ = static_cast<size_t>(configured);
+  }


- test_publish_queue: BlockingSink::wait_until_blocked() now waits against a 2s deadline with ASSERT_LT instead of an unbounded spin, so a worker that never starts fails fast with a clear message rather than wedging until the colcon test timeout. - on_configure: publish_queue_max_bytes now clamps an upper bound (2 GiB) with a warning too, matching the both-bounds discipline of port / maximum_packet_size / history_depth and removing the int64->size_t truncation edge on 32-bit size_t platforms.

Copilot

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 1 comment.

+    stop_requested_ = false;
+    running_ = true;
+    worker_ = std::thread(&PublishQueue::run, this);


…ueue::start() Set running_=true only after the std::thread is constructed. If the thread ctor throws (std::system_error on resource exhaustion), running_ stays false so the queue is cleanly not-running (push() drops, start() retryable) instead of stuck running_=true with no worker to drain. run() never reads running_, so the reorder is behaviour-neutral on the success path.

Copilot

Pull request overview

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

Claude Code Agent added 9 commits May 25, 2026 18:29

progress: plan authored for #10

36579bf

plan: record single-worker scope decision for #10

e7a2b6e

progress: plan review for #10 (verdict revise, revisions applied)

3cc9527

progress: implementation + local review for #10

e37e733

rolker changed the title ~~[PLAN] Bridge wedges (reader thread blocked, Recv-Q backup) when remote subscriber dies~~ Fix reader-thread wedge: decouple republish onto a bounded publish-worker queue May 25, 2026

rolker marked this pull request as ready for review May 25, 2026 23:58

Copilot AI review requested due to automatic review settings May 25, 2026 23:58

Copilot started reviewing on behalf of rolker May 25, 2026 23:58 View session

Copilot AI reviewed May 26, 2026

View reviewed changes

This was referenced May 26, 2026

Cross-topic isolation for the publish worker — open design question (not committed) #29

Open

Bench-test harness for field failure modes (resend amp, wedge, rate-limit accuracy, saturation) #18

Open

Claude Code Agent added 3 commits May 25, 2026 20:16

progress: external review for #10

73f455d

progress: address Copilot review + pre-push re-review for #10

c655350

Copilot AI review requested due to automatic review settings May 26, 2026 01:11

Copilot started reviewing on behalf of rolker May 26, 2026 01:12 View session

Copilot AI reviewed May 26, 2026

View reviewed changes

Comment thread udp_bridge/include/udp_bridge/publish_queue.h

Comment on lines +83 to +85

stop_requested_ = false;

running_ = true;

worker_ = std::thread(&PublishQueue::run, this);

Claude Code Agent added 3 commits May 25, 2026 21:36

progress: external review (re-review) for #10

da30538

progress: resolve start() exception-safety re-review finding for #10

ce13281

Copilot AI review requested due to automatic review settings May 26, 2026 01:57

Copilot started reviewing on behalf of rolker May 26, 2026 01:57 View session

Copilot AI reviewed May 26, 2026

View reviewed changes

rolker mentioned this pull request May 26, 2026

Bench-test harness for field failure modes (netns + tc netem) #27

Open

5 tasks

progress: external review for #10 (current-HEAD re-review clean)

bbddeb7

rolker merged commit 09d43ed into jazzy May 26, 2026

rolker deleted the feature/issue-10 branch May 26, 2026 04:36

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

Fix reader-thread wedge: decouple republish onto a bounded publish-worker queue#28
rolker merged 16 commits into
jazzyfrom
feature/issue-10

rolker commented May 25, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

rolker commented May 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Problem

Fix

Test plan

Notes

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

rolker commented May 25, 2026 •

edited

Loading