Skip to content

Six foxglove_bridge-parity features: whitelist, QoS depths, topic push, backpressure, latched replay, TLS#7

Open
facontidavide wants to merge 16 commits into
mainfrom
feature/foxglove-parity
Open

Six foxglove_bridge-parity features: whitelist, QoS depths, topic push, backpressure, latched replay, TLS#7
facontidavide wants to merge 16 commits into
mainfrom
feature/foxglove-parity

Conversation

@facontidavide

@facontidavide facontidavide commented Jul 6, 2026

Copy link
Copy Markdown
Contributor

Summary

Closes the feature gap with foxglove_bridge on the topic-subscription feature set, per the side-by-side comparison. All protocol changes are additive — protocol_version stays 1; existing clients are unaffected.

Feature Config
Topic whitelist — full-match ECMAScript regex filtering in get_topics, subscribe, and topic-change detection topic_whitelist (ROS2) / --topic-whitelist (CLI)
QoS depth aggregation — subscription depth = sum of publisher depths, clamped; unknown/KEEP_ALL depths count as 100 (FastRTPS doesn't propagate depths); wrap-proof headroom arithmetic min_qos_depth, max_qos_depth
Pushed topic advertisement (opt-in) — subscribe_topic_updates command + topics_changed notification; eager baseline snapshot at startup topic_poll_interval (0 disables)
Slow-client backpressure — bounded per-client drop-oldest frame queues gated on bufferedAmount(); never disconnects; queues cleared on disconnect and session-timeout client_backlog_size
transient_local latched replay (ROS2) — late subscribers to shared latched subscriptions get the retained sample right after the subscribe/resume response; teardown clears retained state; duplicate-gated against the normal publish path automatic
TLS / wss:// — OpenSSL server certificate via ix::SocketTLSOptions; PJ_BRIDGE_TLS CMake option for the FetchContent path tls, certfile, keyfile / --certfile, --keyfile

Logic adapted from foxglove_bridge (QoS depth rule, lossy-send policy) carries MIT attribution comments; MIT is compatible with this repo's AGPL-3.0.

Design spec and implementation plan are included under docs/superpowers/.

Review process

Every feature commit passed a dual review (Claude + Codex); the six fix: commits are the review findings, including a FastRTPS KEEP_LAST(1) regression, a TSAN lock-order inversion, an orphan-queue disconnect race, replay-before-schema ordering, stale latched samples across teardown, and middleware queue leakage across session timeouts.

Test plan

  • pixi run -e humble build && pixi run -e humble test — 233 tests, 0 failures (baseline 177)
  • TSAN — no reports in pj_bridge code (only pre-existing suppressed IXWebSocket/OpenSSL/CycloneDDS internals)
  • ASAN — clean (pre-existing suppressions only)
  • pre-commit run -a clean
  • TLS round-trip test runs for real against the conda IXWebSocket (TLS-enabled)
  • FastDDS/RTI entry-point wiring is hand-verified only (backends not buildable in this env); CI legs with those backends should confirm

Note for reviewers: the depth tests force RMW_IMPLEMENTATION=rmw_cyclonedds_cpp (with a probe-and-skip fallback) because FastRTPS reports depth 0 through discovery; package.xml gained <test_depend>rmw_cyclonedds_cpp</test_depend> so the apt CI legs install it.

🤖 Generated with Claude Code

facontidavide and others added 16 commits July 6, 2026 13:58
@facontidavide @claude
docs: add foxglove-parity feature design spec
7e2ad49 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
docs: add foxglove-parity implementation plan
0d0384e 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
feat: add topic whitelist regex filtering
bd879e4 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
feat: aggregate publisher QoS depths with min/max clamping
cf493e4 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
fix: fall back to depth 100 when publisher history depth is unknown
72b2708 
Under RMWs that don't propagate publisher history depth through
discovery (e.g. rmw_fastrtps_cpp, the default), every publisher reports
depth 0, so the aggregated total clamped to min_qos_depth and produced
depth-1 subscription queues that drop messages on high-rate topics.
Treat a total of 0 as "unknown" and keep the historical default of
min(100, max_qos_depth) instead.

Also declare rmw_cyclonedds_cpp as a test_depend (the depth aggregation
tests force it, and apt-based CI containers don't ship it), and fix
include ordering in test_whitelist_filter.cpp so pre-commit is clean
repo-wide.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
feat: opt-in pushed topic advertisement (topics_changed notification)
866371d 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
refactor: unify QoS depth rule with per-publisher fallback and satura…
c659d32 
…ting sum

Replace the total==0 special case with a uniform per-publisher rule:
a publisher reporting depth 0 (RMW didn't propagate depth, or KEEP_ALL)
contributes the fallback of 100 instead of 0, so unknown depths never
shrink the queue and mixed known/unknown topics aggregate sensibly.
Make the sum saturating (each contribution capped at max_qos_depth,
running total clamped to max_qos_depth) so it can never wrap size_t.

Tests: add a mixed KEEP_ALL + depth-10 publisher case, and a runtime
RMW probe that skips the depth-assertion tests when the active RMW
doesn't propagate publisher history depths through discovery (protects
against a harness overriding the forced CycloneDDS selection).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
fix: take topic baseline snapshot eagerly at startup
e26c1ec 
Topics appearing between server initialization and the first
topic_poll_interval tick were folded into the silent baseline and never
notified. Snapshot immediately after initialize() instead, before any
client can connect. (Codex review finding on 866371d.)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
feat: bounded per-client send queue with drop-oldest backpressure
a3d90c5 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
feat: replay latest transient_local message to late subscribers (ROS2)
37007b8 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
fix: address backpressure review findings (orphan queue race, wrap-pr…
569a0f3 
…oof depth sum)

- send_binary: re-check client existence under clients_mutex_ before
  enqueueing an over-watermark frame; a concurrent disconnect between the
  socket-handle lookup and the enqueue could otherwise recreate a pending
  queue for a gone client, stranding the frame until shutdown and skewing
  dropped_frame_count(). Return false (same contract as the initial lookup
  miss) instead.
- Document that backlog flushing only happens inside send_binary by design
  (bounded queue, fresh-data-first plotting clients).
- generic_subscription_manager: replace add-then-clamp saturating depth sum
  with headroom arithmetic so size_t wrap is structurally impossible even
  for extreme max_qos_depth_ values via the constructor API.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
fix: latched replay ordering, stale-sample teardown, and duplicate gate
2a33289 
- Send the replay binary frame strictly AFTER the subscribe response:
  handle_subscribe now queues serialized frames in pending_replays_ (new
  leaf replays_mutex_); process_single_request flushes them right after
  send_reply succeeds, and drops them when the reply fails or the session
  is cleaned up. The client always has the schema before the frame.
- Clear latched state when the last subscription ref is released (new
  release_subscription_ref helper used by unsubscribe/pause/cleanup/
  rollback, backed by SubscriptionManagerInterface::is_subscribed with a
  default-false implementation and a ROS2 forwarding override): a stale
  retained sample is never replayed ahead of the fresh DDS transient_local
  redelivery on the next subscription.
- Skip the replay while the retained sample is still in the normal
  aggregation buffer (MessageBuffer::get_latched_for_replay): the regular
  publish cycle delivers it to the new subscriber anyway, so replaying
  would duplicate it.
- docs/API.md: document teardown, duplicate gate, and that replay frames
  bypass rate limiting and publish statistics.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
feat: optional TLS (wss://) via OpenSSL server certificate
9f6f467 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
fix: replay latched sample on resume for topics subscribed while paused
c270966 
- A client that subscribes to a latched (transient_local) topic while
  paused acquires no middleware ref, so it got neither the replay frame
  nor DDS redelivery when resume joined an already-shared subscription.
  handle_resume now runs the same latched bookkeeping as handle_subscribe
  after each successful re-subscribe; the frame is flushed by
  process_single_request right after the resume response, preserving the
  response-before-frame ordering.
- Factor the shared logic into BridgeServer::collect_latched_replay
  (set_latched + get_latched_for_replay + serialize + queue under the
  leaf replays_mutex_), replacing handle_subscribe's two-phase
  replay_candidates machinery; document the benign queue-vs-cleanup race
  there.
- docs/API.md (TLS): note that only cert/key readability is validated at
  startup — a mismatched pair only surfaces as per-connection handshake
  failures in the server log (IXWebSocket defers TLS setup to accept
  time).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
docs: document foxglove-parity features
3a0ba1d 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@facontidavide @claude
fix: discard middleware backpressure backlog on session cleanup
894851f 
When a session died server-side (heartbeat timeout) while the socket
stayed open, WebSocketMiddleware's pending_frames_/last_drop_warn_
entries for that client survived: a timed-out slow client parked up to
client_backlog_size frames indefinitely, and stale frames could flush
into a new logical session on the same connection.

- MiddlewareInterface: new drop_pending(client) hook, default no-op.
- WebSocketMiddleware: override folds the client's dropped_total() into
  the lifetime counter and erases its queue + warn-throttle entry (the
  socket itself is untouched).
- BridgeServer::cleanup_session: call drop_pending after the
  cleanup_mutex_ scope (middleware calls must not run under that lock).
- docs/API.md: correct backlog flush wording (delivered on next send
  attempt, no background flush), note frames in flight may arrive after
  an unsubscribe response, document latched replay on resume and that
  replay frames share the backpressure queue.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

1 participant

@facontidavide