diff --git a/CLAUDE.md b/CLAUDE.md index 8d2223c..e6e6d26 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -238,6 +238,7 @@ min_qos_depth: 1 # Minimum KEEP_LAST subscription depth after aggr max_qos_depth: 100 # Maximum KEEP_LAST subscription depth after aggregating publisher depths topic_poll_interval: 1.0 # Seconds between topics_changed notification polls; 0 disables polling client_backlog_size: 100 # Max frames queued per slow client before dropping the oldest (must be > 0) +heavy_frame_threshold_bytes: 262144 # Isolate messages >= this size into their own size-class frame; 0 disables tls: false # Enable TLS (wss://); requires certfile and keyfile certfile: "" # TLS server certificate file keyfile: "" # TLS private key file @@ -247,14 +248,14 @@ keyfile: "" # TLS private key file ```bash pj_bridge_rti --domains 0 1 --port 9090 --publish-rate 50 --session-timeout 10 \ --topic-whitelist ".*" --topic-poll-interval 1.0 --client-backlog-size 100 \ - --certfile cert.pem --keyfile key.pem + --heavy-frame-threshold-bytes 262144 --certfile cert.pem --keyfile key.pem ``` ### FastDDS (via CLI flags): ```bash pj_bridge_fastdds --domains 0 1 --port 9090 --publish-rate 50 --session-timeout 10 \ --topic-whitelist ".*" --topic-poll-interval 1.0 --client-backlog-size 100 \ - --certfile cert.pem --keyfile key.pem + --heavy-frame-threshold-bytes 262144 --certfile cert.pem --keyfile key.pem ``` See `docs/API.md` for full semantics of each option (topic whitelist matching rules, diff --git a/CMakeLists.txt b/CMakeLists.txt index 50bfbec..f941358 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -256,6 +256,7 @@ if(BUILD_TESTING AND ament_cmake_FOUND) ament_add_gtest(${PROJECT_NAME}_tests tests/unit/test_websocket_middleware.cpp + tests/unit/test_backpressure.cpp tests/unit/test_bounded_frame_queue.cpp tests/unit/test_message_buffer.cpp tests/unit/test_session_manager.cpp diff --git a/README.md b/README.md index b4fe2d7..91aa6fc 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,7 @@ independently. | `max_qos_depth` | int | 100 | ROS2 only: maximum KEEP_LAST subscription depth after aggregating publisher depths | | `topic_poll_interval` | double | 1.0 | Seconds between `topics_changed` notification polls; `0` disables polling | | `client_backlog_size` | int | 100 | Max binary frames queued per slow client before the oldest is dropped (must be `> 0`) | +| `heavy_frame_threshold_bytes` | int | 262144 | Isolate messages ≥ this size (bytes) into their own size-class frame so they don't starve small topics; `0` disables (must be `>= 0`) | | `tls` | bool | false | Enable TLS (`wss://`); requires `certfile` and `keyfile` | | `certfile` | string | `""` | TLS server certificate file | | `keyfile` | string | `""` | TLS private key file | @@ -71,6 +72,7 @@ independently. | `--topic-whitelist` | string list | `.*` | Full-match regex patterns (ECMAScript), repeatable | | `--topic-poll-interval` | double | 1.0 | Seconds between `topics_changed` notification polls; `0` disables polling | | `--client-backlog-size` | int | 100 | Max binary frames queued per slow client before the oldest is dropped (range `1`-`1000000`) | +| `--heavy-frame-threshold-bytes` | int | 262144 | Isolate messages ≥ this size (bytes) into their own size-class frame; `0` disables (range `0`-`1000000000`) | | `--certfile` | string | (none) | TLS server certificate file; enables `wss://`, requires `--keyfile` | | `--keyfile` | string | (none) | TLS private key file; enables `wss://`, requires `--certfile` | | `--qos-profile` | string | (none) | RTI only: QoS profile XML file path | diff --git a/app/include/pj_bridge/bridge_server.hpp b/app/include/pj_bridge/bridge_server.hpp index 40c31d9..7ce7f42 100644 --- a/app/include/pj_bridge/bridge_server.hpp +++ b/app/include/pj_bridge/bridge_server.hpp @@ -28,6 +28,7 @@ #include "pj_bridge/message_buffer.hpp" #include "pj_bridge/middleware/middleware_interface.hpp" +#include "pj_bridge/protocol_constants.hpp" #include "pj_bridge/session_manager.hpp" #include "pj_bridge/subscription_manager_interface.hpp" #include "pj_bridge/topic_source_interface.hpp" @@ -47,6 +48,21 @@ namespace pj_bridge { * Thread-safe for concurrent client connections. * Event loop is driven externally (no internal timers). */ + +/// Tunable configuration for BridgeServer (backend-agnostic). Bundled into one +/// struct so entry points construct the server with a single named aggregate +/// rather than a long positional argument list. +struct BridgeServerConfig { + int port = 9090; ///< WebSocket port + double session_timeout = 10.0; ///< client session timeout, seconds + double publish_rate = 50.0; ///< message aggregation/publish rate, Hz + WhitelistFilter whitelist = {}; ///< topic whitelist (default: matches everything) + /// Per-message byte size at or above which a topic's message is isolated into + /// its own size-class ("heavy") frame instead of being aggregated with light + /// topics. 0 disables splitting (single aggregated frame). Default: 256 KiB. + size_t heavy_frame_threshold_bytes = kDefaultHeavyFrameThresholdBytes; +}; + class BridgeServer { public: struct StatsSnapshot { @@ -61,16 +77,12 @@ class BridgeServer { * @param topic_source Backend-specific topic discovery and schema provider * @param subscription_manager Backend-specific subscription manager * @param middleware Middleware interface for network communication - * @param port Server port (default: 9090) - * @param session_timeout Session timeout in seconds (default: 10.0) - * @param publish_rate Message aggregation publish rate in Hz (default: 50.0) - * @param whitelist Topic whitelist filter (default: matches everything) + * @param config Tunable server configuration (see BridgeServerConfig) */ explicit BridgeServer( std::shared_ptr topic_source, std::shared_ptr subscription_manager, - std::shared_ptr middleware, int port = 9090, double session_timeout = 10.0, - double publish_rate = 50.0, WhitelistFilter whitelist = {}); + std::shared_ptr middleware, BridgeServerConfig config = {}); /// Shuts down middleware before members are destroyed, preventing /// disconnect callbacks from firing into a partially destroyed object. @@ -211,6 +223,9 @@ class BridgeServer { double session_timeout_; double publish_rate_; WhitelistFilter whitelist_; + // Per-message byte size at or above which a topic is isolated into its own + // size-class ("heavy") frame; 0 disables splitting. See publish_aggregated_messages(). + size_t heavy_frame_threshold_bytes_; // State std::atomic initialized_; diff --git a/app/include/pj_bridge/message_serializer.hpp b/app/include/pj_bridge/message_serializer.hpp index 1c12740..9df4185 100644 --- a/app/include/pj_bridge/message_serializer.hpp +++ b/app/include/pj_bridge/message_serializer.hpp @@ -89,12 +89,14 @@ class AggregatedMessageSerializer { * - Offset 0: magic (uint32_t "PJRB" = 0x42524A50, little-endian) * - Offset 4: message_count (uint32_t, little-endian) * - Offset 8: uncompressed_size (uint32_t, little-endian) - * - Offset 12: flags (uint32_t, reserved = 0) + * - Offset 12: flags (uint32_t; bit0 = heavy frame, else reserved = 0) * - Offset 16+: ZSTD-compressed payload * + * @param flags Header flag bits written at offset 12 (default 0). Use + * kFrameFlagHeavy to mark an isolated large/size-class frame. * @return Vector containing header + compressed payload */ - std::vector finalize(); + std::vector finalize(uint32_t flags = 0); /** * @brief Compress data using ZSTD (compression level 1) diff --git a/app/include/pj_bridge/middleware/backpressure.hpp b/app/include/pj_bridge/middleware/backpressure.hpp new file mode 100644 index 0000000..1c93679 --- /dev/null +++ b/app/include/pj_bridge/middleware/backpressure.hpp @@ -0,0 +1,105 @@ +/* + * Copyright (C) 2026 Davide Faconti + * + * This file is part of pj_bridge. + * + * pj_bridge is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * pj_bridge is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with pj_bridge. If not, see . + */ + +#pragma once + +#include +#include +#include + +#include "pj_bridge/middleware/middleware_interface.hpp" + +namespace pj_bridge { + +/// Outcome of run_backpressure() for one outgoing frame. The disposition is a +/// single discriminant (no contradictory flag combinations are representable); +/// the counters are companion data. +struct SendOutcome { + SendResult result = SendResult::kClientGone; ///< how the current frame was handled + size_t frames_flushed = 0; ///< backlog frames flushed to the socket this call + size_t dropped = 0; ///< backlog frames evicted on overflow (only when result == kQueued) +}; + +/// Socket-agnostic backpressure policy shared by the send path. The caller +/// injects the socket/queue primitives (as callables — templated to avoid +/// std::function type-erasure on the hot path) so the policy is unit-testable +/// without a real connection: +/// - @p buffered_amount : `size_t()` — current socket buffer bytes +/// - @p pop_pending : `std::optional>()` — pop oldest queued frame (nullopt if empty) +/// - @p send : `bool(const vector&)` — transmit a frame; false = client gone +/// - @p queue_pending : `std::optional(const vector&)` — enqueue (drop-oldest), +/// returns #dropped, or nullopt if the client is gone +/// +/// Policy: first flush the backlog while the socket has room (re-checking the +/// watermark each iteration so an already-congested socket is never fed +/// further, and never flushing more than @p max_flush frames so a concurrent +/// producer cannot make one call flush forever); then handle the current +/// frame — send it if there is room (kDelivered), else drop a kHeavy frame +/// before transmit (kShed) or enqueue a kNormal frame (kQueued). A vanished +/// client surfaces as kClientGone. +/// +/// @param max_flush upper bound on frames flushed this call — pass the backlog +/// size observed at call start. +template +SendOutcome run_backpressure( + FramePriority priority, const std::vector& frame, size_t watermark, size_t max_flush, + const BufferedAmount& buffered_amount, const PopPending& pop_pending, const Send& send, + const QueuePending& queue_pending) { + SendOutcome out; + + // Flush queued frames to the socket, re-checking the watermark each iteration + // so a socket that fills up mid-flush is never fed further (the flush-recheck + // fix: the old loop computed the flush count once and could dump a burst of + // stale frames onto an already-congested socket). Bounded by max_flush so a + // producer enqueueing concurrently cannot keep this single call flushing. + while (out.frames_flushed < max_flush && buffered_amount() < watermark) { + std::optional> queued = pop_pending(); + if (!queued) { + break; + } + if (!send(*queued)) { + out.result = SendResult::kClientGone; + return out; + } + out.frames_flushed++; + } + + // Handle the current frame against the live socket buffer. + if (buffered_amount() < watermark) { + out.result = send(frame) ? SendResult::kDelivered : SendResult::kClientGone; + return out; + } + + // Socket congested: shed heavy frames before transmit; queue normal frames. + if (priority == FramePriority::kHeavy) { + out.result = SendResult::kShed; + return out; + } + + std::optional dropped = queue_pending(frame); + if (!dropped) { + out.result = SendResult::kClientGone; + return out; + } + out.dropped = *dropped; + out.result = SendResult::kQueued; + return out; +} + +} // namespace pj_bridge diff --git a/app/include/pj_bridge/middleware/middleware_interface.hpp b/app/include/pj_bridge/middleware/middleware_interface.hpp index ec62b2e..d3422a1 100644 --- a/app/include/pj_bridge/middleware/middleware_interface.hpp +++ b/app/include/pj_bridge/middleware/middleware_interface.hpp @@ -28,6 +28,23 @@ namespace pj_bridge { +/// Delivery priority for a per-client binary frame. Under socket congestion a +/// `kHeavy` (large/size-class) frame is dropped before transmit rather than +/// queued, so one big frame cannot starve the small frames behind it; a +/// `kNormal` frame instead falls back to the queue-with-drop-oldest backlog. +/// (When the socket has room, both are sent immediately.) See docs/API.md. +enum class FramePriority { kNormal, kHeavy }; + +/// Outcome of send_binary(): how the transport handled the frame. Only +/// `kDelivered` and `kQueued` are (or will be) put on the wire; `kShed` and +/// `kClientGone` are never delivered, so callers must NOT count them as sent. +enum class SendResult { + kDelivered, ///< written to the socket now (or flushed from the backlog) + kQueued, ///< enqueued for later delivery (kNormal frame, socket congested) + kShed, ///< dropped before transmit (kHeavy frame, socket congested) + kClientGone, ///< the client disconnected; nothing was sent +}; + /// Abstract transport layer between BridgeServer and clients. /// /// Implementations handle connection management and bidirectional messaging. @@ -65,8 +82,14 @@ class MiddlewareInterface { virtual bool publish_data(const std::vector& data) = 0; /// Send binary data to a specific client (used for per-client aggregated frames). - /// @return true if the message was sent, false if the client is gone. - virtual bool send_binary(const std::string& client_identity, const std::vector& data) = 0; + /// @param priority kHeavy frames are shed before transmit under congestion + /// instead of queued (default kNormal preserves the legacy behavior). + /// @return how the frame was handled (see SendResult). Callers counting + /// forwarded bytes/messages must treat only kDelivered/kQueued as + /// forwarded — kShed and kClientGone never reach the client. + virtual SendResult send_binary( + const std::string& client_identity, const std::vector& data, + FramePriority priority = FramePriority::kNormal) = 0; /// Discard any queued outbound data for this client (e.g. when its session /// is destroyed server-side while the socket stays open). Default no-op for diff --git a/app/include/pj_bridge/middleware/websocket_middleware.hpp b/app/include/pj_bridge/middleware/websocket_middleware.hpp index f1c1da4..550f954 100644 --- a/app/include/pj_bridge/middleware/websocket_middleware.hpp +++ b/app/include/pj_bridge/middleware/websocket_middleware.hpp @@ -33,6 +33,7 @@ #include #include +#include "pj_bridge/middleware/backpressure.hpp" #include "pj_bridge/middleware/bounded_frame_queue.hpp" #include "pj_bridge/middleware/middleware_interface.hpp" @@ -48,7 +49,13 @@ struct TlsConfig { class WebSocketMiddleware : public MiddlewareInterface { public: - explicit WebSocketMiddleware(size_t client_backlog_size = 100, std::optional tls = std::nullopt); + /// @param socket_buffer_watermark bytes of outgoing socket buffer at/above + /// which a client is considered congested (frames queue or shed). + /// Defaults to kSocketBufferHighWatermark; primarily overridden in + /// tests to exercise the congested path deterministically. + explicit WebSocketMiddleware( + size_t client_backlog_size = 100, std::optional tls = std::nullopt, + size_t socket_buffer_watermark = kSocketBufferHighWatermark); ~WebSocketMiddleware() override; WebSocketMiddleware(const WebSocketMiddleware&) = delete; @@ -61,7 +68,11 @@ class WebSocketMiddleware : public MiddlewareInterface { bool receive_request(std::vector& data, std::string& client_identity) override; bool send_reply(const std::string& client_identity, const std::vector& data) override; bool publish_data(const std::vector& data) override; - bool send_binary(const std::string& client_identity, const std::vector& data) override; + // NOTE: the FramePriority default lives only on the base MiddlewareInterface + // declaration (defaults are bound statically, so repeating it here could + // silently diverge). + SendResult send_binary( + const std::string& client_identity, const std::vector& data, FramePriority priority) override; bool is_ready() const override; void set_on_connect(ConnectionCallback callback) override; void set_on_disconnect(ConnectionCallback callback) override; @@ -71,6 +82,17 @@ class WebSocketMiddleware : public MiddlewareInterface { /// across all clients (currently connected and already disconnected). uint64_t dropped_frame_count() const; + /// Total number of kHeavy frames shed before transmit under congestion + /// (dropped instead of queued), summed over the middleware's lifetime. + uint64_t heavy_shed_count() const; + + /// Lossy-send policy watermark adapted from foxglove_bridge (MIT License, + /// Copyright (c) Foxglove Technologies Inc): once a client's outgoing socket + /// buffer reaches this many bytes, further frames are queued (kNormal) or shed + /// (kHeavy) instead of blocking or disconnecting the client. Public so entry + /// points can sanity-check a configured heavy-frame threshold against it. + static constexpr size_t kSocketBufferHighWatermark = 1u << 20; // 1 MiB + private: struct IncomingRequest { std::string client_id; @@ -98,7 +120,13 @@ class WebSocketMiddleware : public MiddlewareInterface { // total). Guarded by clients_mutex_. uint64_t dropped_from_disconnected_{0}; + // Lifetime count of kHeavy frames shed before transmit under congestion + // (dropped rather than queued). Distinct from dropped_frame_count(), which + // counts backlog-overflow drops of kNormal frames. Guarded by clients_mutex_. + uint64_t heavy_shed_total_{0}; + size_t client_backlog_size_; + size_t socket_buffer_watermark_; std::optional tls_; ConnectionCallback on_connect_; @@ -111,12 +139,6 @@ class WebSocketMiddleware : public MiddlewareInterface { static constexpr int kShutdownTimeoutSeconds = 3; static constexpr size_t kMaxIncomingQueueSize = 1024; - - // Lossy-send policy adapted from foxglove_bridge (MIT License, Copyright - // (c) Foxglove Technologies Inc): once a client's outgoing socket buffer - // exceeds this watermark, new frames are queued (dropping the oldest on - // overflow) instead of blocking or disconnecting the client. - static constexpr size_t kSocketBufferHighWatermark = 1u << 20; // 1 MiB static constexpr int kDropWarnIntervalSeconds = 30; }; diff --git a/app/include/pj_bridge/protocol_constants.hpp b/app/include/pj_bridge/protocol_constants.hpp index 9f6d285..af57f3c 100644 --- a/app/include/pj_bridge/protocol_constants.hpp +++ b/app/include/pj_bridge/protocol_constants.hpp @@ -33,6 +33,21 @@ static constexpr uint32_t kBinaryFrameMagic = 0x42524A50; /// Size of the binary frame header in bytes static constexpr size_t kBinaryHeaderSize = 16; +/// Binary frame header flag bit (offset 12 of the 16-byte header) reserved for a +/// future "heavy" (isolated large/size-class message) marker. NOT currently +/// emitted: existing PlotJuggler plugins reject any frame with flags != 0, so +/// heavy frames ship unflagged (flags == 0) and heaviness is conveyed +/// server-side via FramePriority instead. Reserved here for a future +/// capability-negotiated rollout (see docs/API.md). +static constexpr uint32_t kFrameFlagHeavy = 0x1; + +/// Default per-message byte threshold at or above which a topic's message is +/// isolated into its own "heavy" size-class frame instead of being aggregated +/// with light topics (see docs/API.md). Chosen comfortably below the 1 MiB +/// socket high-watermark and well above typical scalar/odom/tf frames. A +/// threshold of 0 disables splitting (single aggregated frame, legacy behavior). +static constexpr size_t kDefaultHeavyFrameThresholdBytes = 256 * 1024; // 256 KiB + /// Schema encoding identifier for ROS2 message definitions inline constexpr const char* kSchemaEncodingRos2Msg = "ros2msg"; @@ -49,6 +64,7 @@ inline constexpr const char* kServerCapabilities[] = { "latched_replay", // retained samples replayed after subscribe/resume "topics_changed", // pushed topic advertisement (subscribe_topic_updates) "per_topic_rate_limit", // subscribe entries accept {name, max_rate_hz} + "size_class_frames", // large topics isolated into own frames (header flag bit0 = heavy) }; } // namespace pj_bridge diff --git a/app/src/bridge_server.cpp b/app/src/bridge_server.cpp index 3682674..fe3551e 100644 --- a/app/src/bridge_server.cpp +++ b/app/src/bridge_server.cpp @@ -60,14 +60,15 @@ double clamp_rate_hz(double rate_hz) { BridgeServer::BridgeServer( std::shared_ptr topic_source, std::shared_ptr subscription_manager, std::shared_ptr middleware, - int port, double session_timeout, double publish_rate, WhitelistFilter whitelist) + BridgeServerConfig config) : topic_source_(std::move(topic_source)), subscription_manager_(std::move(subscription_manager)), middleware_(std::move(middleware)), - port_(port), - session_timeout_(session_timeout), - publish_rate_(publish_rate), - whitelist_(std::move(whitelist)), + port_(config.port), + session_timeout_(config.session_timeout), + publish_rate_(config.publish_rate), + whitelist_(std::move(config.whitelist)), + heavy_frame_threshold_bytes_(config.heavy_frame_threshold_bytes), initialized_(false), total_messages_published_(0), total_bytes_published_(0), @@ -1057,6 +1058,11 @@ void BridgeServer::publish_aggregated_messages() { std::vector compressed_data; size_t msg_count; std::vector client_ids; + bool is_heavy = false; // isolated large/size-class frame (kFrameFlagHeavy) + // Per-topic message counts carried by THIS frame, folded into + // topic_forward_counts_ only when the frame is actually sent (so a + // partial send after the split doesn't over-count unsent topics). + std::unordered_map topic_counts; }; std::vector frames; @@ -1064,12 +1070,48 @@ void BridgeServer::publish_aggregated_messages() { std::lock_guard sent_lock(last_sent_mutex_); for (const auto& [group_key, client_ids] : subscription_groups) { - AggregatedMessageSerializer serializer; + AggregatedMessageSerializer light_serializer; + std::vector heavy_frames_for_group; size_t group_msg_count = 0; const auto& representative_subs = client_subs[client_ids.front()]; auto& rep_last_sent = last_sent_times_[client_ids.front()]; + // Per-topic counts for messages routed into THIS group's light frame; + // folded into the light frame's topic_counts when it is built. + std::unordered_map light_topic_counts; + + // Route one rate-admitted message into either the aggregated light + // frame or its own isolated heavy frame, based on the per-message CDR + // byte size. The rate gate (below) decides WHETHER a message is + // admitted; this decides WHICH frame it lands in — orthogonal. A + // threshold of 0 disables splitting (everything goes to the light frame). + auto route_message = [&](const std::string& topic, const BufferedMessage& msg) { + const bool heavy = heavy_frame_threshold_bytes_ != 0 && msg.data->size() >= heavy_frame_threshold_bytes_; + if (heavy) { + // One single-message frame per heavy message (same one-message-per-frame + // shape as collect_latched_replay's retained-sample frame). The frame is + // NOT wire-flagged (flags stay 0): existing PlotJuggler plugins reject + // any frame with flags != 0, and the isolation/shedding benefit comes + // entirely from the separate frame + the in-memory is_heavy priority + // below — not from a wire marker. kFrameFlagHeavy is reserved for a + // future capability-negotiated rollout (see docs/API.md). + AggregatedMessageSerializer heavy_serializer; + heavy_serializer.serialize_message(topic, msg.timestamp_ns, msg.data->data(), msg.data->size()); + GroupFrame heavy_frame; + heavy_frame.compressed_data = heavy_serializer.finalize(); + heavy_frame.msg_count = 1; + heavy_frame.client_ids = client_ids; + heavy_frame.is_heavy = true; + heavy_frame.topic_counts[topic] = 1; + heavy_frames_for_group.push_back(std::move(heavy_frame)); + } else { + light_serializer.serialize_message(topic, msg.timestamp_ns, msg.data->data(), msg.data->size()); + light_topic_counts[topic]++; + } + group_msg_count++; + }; + for (const auto& [topic, msgs] : messages) { auto sub_it = representative_subs.find(topic); if (sub_it == representative_subs.end()) { @@ -1081,9 +1123,7 @@ void BridgeServer::publish_aggregated_messages() { if (rate_mhz == 0) { for (const auto& msg : msgs) { - serializer.serialize_message(topic, msg.timestamp_ns, msg.data->data(), msg.data->size()); - group_msg_count++; - forward_counts[topic]++; + route_message(topic, msg); } } else { uint64_t min_interval_ns = static_cast(1'000'000'000'000) / static_cast(rate_mhz); @@ -1091,10 +1131,8 @@ void BridgeServer::publish_aggregated_messages() { for (const auto& msg : msgs) { if (msg.timestamp_ns >= last_sent + min_interval_ns) { - serializer.serialize_message(topic, msg.timestamp_ns, msg.data->data(), msg.data->size()); + route_message(topic, msg); last_sent = msg.timestamp_ns; - group_msg_count++; - forward_counts[topic]++; } } rep_last_sent[topic] = last_sent; @@ -1105,19 +1143,28 @@ void BridgeServer::publish_aggregated_messages() { continue; } - GroupFrame frame; - frame.compressed_data = serializer.finalize(); - frame.msg_count = group_msg_count; - frame.client_ids = client_ids; - - // Propagate rate-limiting state to other clients in the group + // Propagate rate-limiting state to other clients in the group. Done + // once per group; both lanes advanced the same rep_last_sent. for (const auto& client_id : client_ids) { if (client_id != client_ids.front()) { last_sent_times_[client_id] = rep_last_sent; } } - frames.push_back(std::move(frame)); + // Emit the aggregated light frame (only if small topics landed in it), + // then each isolated heavy frame. Every frame carries the same + // client_ids, so the send loop below is unchanged. + if (light_serializer.get_message_count() > 0) { + GroupFrame light_frame; + light_frame.compressed_data = light_serializer.finalize(); + light_frame.msg_count = light_serializer.get_message_count(); + light_frame.client_ids = client_ids; + light_frame.topic_counts = std::move(light_topic_counts); + frames.push_back(std::move(light_frame)); + } + for (auto& heavy_frame : heavy_frames_for_group) { + frames.push_back(std::move(heavy_frame)); + } } } // last_sent_mutex_ released here @@ -1129,9 +1176,15 @@ void BridgeServer::publish_aggregated_messages() { if (it != paused_state.end() && it->second) { continue; } - if (middleware_->send_binary(client_id, frame.compressed_data)) { + const FramePriority priority = frame.is_heavy ? FramePriority::kHeavy : FramePriority::kNormal; + const SendResult result = middleware_->send_binary(client_id, frame.compressed_data, priority); + // Count a frame as forwarded only if it was delivered now or queued for + // later delivery. A kShed heavy frame is intentionally dropped under + // congestion (surfaced via the middleware's heavy_shed_count(), not in + // publish stats); kClientGone never reaches the client. + if (result == SendResult::kDelivered || result == SendResult::kQueued) { any_sent = true; - } else { + } else if (result == SendResult::kClientGone) { spdlog::debug("Failed to send binary frame to client '{}'", client_id); } } @@ -1139,6 +1192,9 @@ void BridgeServer::publish_aggregated_messages() { if (any_sent) { total_msg_count += frame.msg_count; total_bytes += frame.compressed_data.size(); + for (const auto& [topic, count] : frame.topic_counts) { + forward_counts[topic] += count; + } } } diff --git a/app/src/message_serializer.cpp b/app/src/message_serializer.cpp index c9f0f93..7a4d645 100644 --- a/app/src/message_serializer.cpp +++ b/app/src/message_serializer.cpp @@ -67,7 +67,7 @@ size_t AggregatedMessageSerializer::get_message_count() const { return message_count_; } -std::vector AggregatedMessageSerializer::finalize() { +std::vector AggregatedMessageSerializer::finalize(uint32_t flags) { // Build 16-byte header (uncompressed) std::vector header(kBinaryHeaderSize); @@ -83,8 +83,7 @@ std::vector AggregatedMessageSerializer::finalize() { uint32_t uncompressed = static_cast(serialized_data_.size()); std::memcpy(header.data() + 8, &uncompressed, sizeof(uncompressed)); - // Flags (offset 12, 4 bytes, reserved = 0) - uint32_t flags = 0; + // Flags (offset 12, 4 bytes; bit0 = heavy frame, see kFrameFlagHeavy) std::memcpy(header.data() + 12, &flags, sizeof(flags)); // Handle empty payload case diff --git a/app/src/middleware/websocket_middleware.cpp b/app/src/middleware/websocket_middleware.cpp index 2e920dc..901ff55 100644 --- a/app/src/middleware/websocket_middleware.cpp +++ b/app/src/middleware/websocket_middleware.cpp @@ -28,8 +28,12 @@ namespace pj_bridge { -WebSocketMiddleware::WebSocketMiddleware(size_t client_backlog_size, std::optional tls) - : client_backlog_size_(client_backlog_size), tls_(std::move(tls)), initialized_(false) {} +WebSocketMiddleware::WebSocketMiddleware( + size_t client_backlog_size, std::optional tls, size_t socket_buffer_watermark) + : client_backlog_size_(client_backlog_size), + socket_buffer_watermark_(socket_buffer_watermark), + tls_(std::move(tls)), + initialized_(false) {} WebSocketMiddleware::~WebSocketMiddleware() { shutdown(); @@ -296,101 +300,90 @@ bool WebSocketMiddleware::send_reply(const std::string& client_identity, const s return send_info.success; } -bool WebSocketMiddleware::send_binary(const std::string& client_identity, const std::vector& data) { +SendResult WebSocketMiddleware::send_binary( + const std::string& client_identity, const std::vector& data, FramePriority priority) { // Lossy-send policy adapted from foxglove_bridge (MIT License, Copyright // (c) Foxglove Technologies Inc): a slow client never blocks the publisher - // and is never disconnected for falling behind. Instead, once its socket's - // outgoing buffer crosses kSocketBufferHighWatermark, new frames are queued - // per-client (dropping the oldest queued frame on overflow) and flushed - // once the buffer drains below the watermark again. + // and is never disconnected for falling behind. The admit/flush/queue/shed + // decision itself lives in the socket-agnostic run_backpressure() policy; + // this method binds it to the real socket and the per-client backlog. // - // IMPORTANT: clients_mutex_ is only ever held to look up state (the client - // handle, pending_frames_, last_drop_warn_) — never while calling into - // ix::WebSocket (bufferedAmount()/sendBinary()). Those calls take - // IXWebSocket's own internal locks, and the Open-connection handler above - // takes clients_mutex_ from an IXWebSocket-owned thread that already holds - // some of those internal locks; holding clients_mutex_ across a socket call - // here would form a lock-order cycle with that path (observed as a TSAN - // lock-order-inversion during development). + // IMPORTANT: clients_mutex_ is only ever held to look up/mutate per-client + // state (the client handle, pending_frames_, last_drop_warn_) — never while + // calling into ix::WebSocket (bufferedAmount()/sendBinary()). Those calls + // take IXWebSocket's own internal locks, and the Open-connection handler + // takes clients_mutex_ from an IXWebSocket-owned thread already holding some + // of those locks; holding clients_mutex_ across a socket call here would form + // a lock-order cycle (observed as a TSAN lock-order-inversion). Each policy + // callback below therefore takes the lock only for its own state access. std::shared_ptr ws; + size_t max_flush = 0; { std::lock_guard lock(clients_mutex_); auto it = clients_.find(client_identity); if (it == clients_.end() || !it->second) { - return false; + return SendResult::kClientGone; } ws = it->second; + // Snapshot the backlog size so the flush is bounded to frames present now, + // never chasing frames a concurrent producer enqueues mid-flush. + auto pending_it = pending_frames_.find(client_identity); + if (pending_it != pending_frames_.end()) { + max_flush = pending_it->second.size(); + } } - if (ws->bufferedAmount() < kSocketBufferHighWatermark) { - // Flush any backlog first, in FIFO order, bounded to the number of - // frames present when this call started so a fast producer can't spin - // forever. Each frame is popped under the lock and sent outside it. - size_t to_flush = 0; - { - std::lock_guard lock(clients_mutex_); - auto pending_it = pending_frames_.find(client_identity); - if (pending_it != pending_frames_.end()) { - to_flush = pending_it->second.size(); - } - } + auto buffered_amount = [&]() -> size_t { return ws->bufferedAmount(); }; - for (size_t i = 0; i < to_flush; ++i) { - std::optional> frame; - { - std::lock_guard lock(clients_mutex_); - auto pending_it = pending_frames_.find(client_identity); - if (pending_it == pending_frames_.end()) { - break; - } - frame = pending_it->second.pop_front(); - } - if (!frame) { - break; - } - std::string frame_data(reinterpret_cast(frame->data()), frame->size()); - auto send_info = ws->sendBinary(frame_data); - if (!send_info.success) { - return false; - } + auto pop_pending = [&]() -> std::optional> { + std::lock_guard lock(clients_mutex_); + auto pending_it = pending_frames_.find(client_identity); + if (pending_it == pending_frames_.end()) { + return std::nullopt; } - - std::string binary_data(reinterpret_cast(data.data()), data.size()); - auto send_info = ws->sendBinary(binary_data); - return send_info.success; - } - - // Socket buffer is over the high watermark: queue the frame instead of - // sending it now. - // - // Note the backlog is only ever flushed from within send_binary itself: if - // a client's subscribed topics go quiet, queued frames sit here until the - // next send_binary call for that client (or its disconnect). This is - // deliberate — the queue is bounded, delivering stale frames on the next - // traffic burst is acceptable, and plotting clients care about fresh data, - // so a dedicated flush timer isn't worth the extra machinery. - size_t dropped; - uint64_t dropped_total_now; - { + return pending_it->second.pop_front(); + }; + + auto send = [&](const std::vector& frame) -> bool { + std::string frame_data(reinterpret_cast(frame.data()), frame.size()); + return ws->sendBinary(frame_data).success; + }; + + // Enqueue under the lock (drop-oldest on overflow). Re-check the client still + // exists: between the ws lookup above and here the disconnect callback may + // have erased it AND its pending_frames_ entry; recreating a queue for a dead + // client would strand the frame and skew dropped_frame_count(). nullopt tells + // the policy the client is gone, so send_binary reports kClientGone. + auto queue_pending = [&](const std::vector& frame) -> std::optional { std::lock_guard lock(clients_mutex_); - // Re-check the client still exists: between the ws lookup above (lock - // released before the socket calls) and here, the disconnect callback may - // have erased the client AND its pending_frames_ entry. Blindly - // try_emplace-ing would recreate a queue for a disconnected client, - // stranding the frame until shutdown and skewing dropped_frame_count(). - // Treat it like the top-of-function lookup miss: the client is gone. if (clients_.find(client_identity) == clients_.end()) { - return false; + return std::nullopt; } auto [pending_it, inserted] = pending_frames_.try_emplace(client_identity, BoundedFrameQueue(client_backlog_size_)); (void)inserted; - dropped = pending_it->second.push(data); - dropped_total_now = pending_it->second.dropped_total(); + return pending_it->second.push(frame); + }; + + SendOutcome outcome = run_backpressure( + priority, data, socket_buffer_watermark_, max_flush, buffered_amount, pop_pending, send, queue_pending); + + if (outcome.result == SendResult::kShed) { + std::lock_guard lock(clients_mutex_); + // Re-check liveness, mirroring the queue path's disconnect-race handling: if + // the client vanished between the ws lookup and the shed decision, report + // it gone rather than counting a shed for a dead client. + if (clients_.find(client_identity) == clients_.end()) { + return SendResult::kClientGone; + } + heavy_shed_total_++; } - if (dropped > 0) { + // Throttled slow-client warning, shared by the queue-drop and heavy-shed + // paths (at most one line per client per kDropWarnIntervalSeconds). + const bool dropped_a_frame = outcome.result == SendResult::kQueued && outcome.dropped > 0; + if (dropped_a_frame || outcome.result == SendResult::kShed) { auto now = std::chrono::steady_clock::now(); - bool should_warn; + bool should_warn = false; { std::lock_guard lock(clients_mutex_); auto warn_it = last_drop_warn_.find(client_identity); @@ -401,13 +394,15 @@ bool WebSocketMiddleware::send_binary(const std::string& client_identity, const } } if (should_warn) { - spdlog::warn( - "Slow client '{}': dropping oldest queued frame(s) ({} dropped total)", client_identity, dropped_total_now); + if (outcome.result == SendResult::kShed) { + spdlog::warn("Slow client '{}': shedding heavy frame(s) ({} shed total)", client_identity, heavy_shed_count()); + } else { + spdlog::warn("Slow client '{}': dropping oldest queued frame(s)", client_identity); + } } } - // The frame was accepted for later delivery, not a failure. - return true; + return outcome.result; } void WebSocketMiddleware::drop_pending(const std::string& client_identity) { @@ -434,6 +429,11 @@ uint64_t WebSocketMiddleware::dropped_frame_count() const { return total; } +uint64_t WebSocketMiddleware::heavy_shed_count() const { + std::lock_guard lock(clients_mutex_); + return heavy_shed_total_; +} + bool WebSocketMiddleware::publish_data(const std::vector& data) { std::vector> clients_copy; { diff --git a/app/src/standalone_event_loop.cpp b/app/src/standalone_event_loop.cpp index 73d6d94..01cd830 100644 --- a/app/src/standalone_event_loop.cpp +++ b/app/src/standalone_event_loop.cpp @@ -104,6 +104,7 @@ void run_standalone_event_loop( stats_msg += fmt::format( "\n Sent: {:.2f} MB/s", static_cast(snapshot.total_bytes_published) / elapsed / (1024.0 * 1024.0)); stats_msg += fmt::format("\n Dropped frames (slow clients): {}", middleware->dropped_frame_count()); + stats_msg += fmt::format("\n Shed heavy frames (congestion): {}", middleware->heavy_shed_count()); spdlog::info(stats_msg); last_stats_print = now; diff --git a/docs/API.md b/docs/API.md index aae63bd..60ddb9a 100644 --- a/docs/API.md +++ b/docs/API.md @@ -100,7 +100,7 @@ Every `get_topics` response carries a `server` object: {"server": {"name": "pj_bridge", "version": "0.8.0", "capabilities": ["include_schemas", "latched_badge", "latched_replay", "topics_changed", - "per_topic_rate_limit"]}} + "per_topic_rate_limit", "size_class_frames"]}} ``` Compatibility policy for clients: @@ -547,6 +547,27 @@ configurable: - **FastDDS / RTI**: CLI flag `--client-backlog-size`, default `100`, valid range `1`-`1000000`. +Under congestion, heavy (size-class) frames from the aggregated publish stream +are shed before transmit rather than queued (see +[Size-class frames](#size-class-frames)), so a continuous stream of +large frames cannot fill the backlog and evict small-topic frames. (One-shot +latched-replay frames are an exception: they are sent at normal priority — +never shed like heavy frames — so they may briefly occupy the backlog and, like +any normal frame, can be dropped only if the backlog itself overflows under +sustained congestion; they are not a continuous stream.) + +The per-message size at or above which a message is isolated into its own heavy +frame is configurable: + +- **ROS2**: int parameter `heavy_frame_threshold_bytes`, default `262144` + (256 KiB). Must be `>= 0`; `0` disables splitting (single aggregated frame, + legacy behavior). +- **FastDDS / RTI**: CLI flag `--heavy-frame-threshold-bytes`, default `262144`, + valid range `0`-`1000000000`. + +Keep the threshold below the 1 MiB socket watermark so a single heavy message +does not fill the socket buffer on its own. + ## TLS / wss:// The bridge can optionally serve the WebSocket endpoint over TLS (`wss://`) @@ -601,7 +622,7 @@ Binary frames consist of a fixed 16-byte header followed by ZSTD-compressed payl | 0 | 4 | magic | `0x42524A50` ("PJRB") | | 4 | 4 | message_count | Number of messages in frame | | 8 | 4 | uncompressed_size | Payload size before compression | -| 12 | 4 | flags | Reserved (must be 0) | +| 12 | 4 | flags | Reserved — currently always `0` (bit 0 is reserved for a future heavy-frame marker; see [Size-class frames](#size-class-frames)) | ### Payload (ZSTD-compressed) @@ -617,3 +638,27 @@ For each message: ``` The magic bytes allow clients to validate frame integrity before decompression. + +### Size-class frames + +Advertised by the `size_class_frames` capability. To keep a large topic (e.g. a +`PointCloud2`) from starving small topics under slow-link backpressure, the server +does **not** weld small and large messages into one frame. Instead, each message +whose serialized size is at or above a server-configured threshold +(`heavy_frame_threshold_bytes`, default 256 KiB; `0` disables splitting) is sent as +its own frame, while smaller messages stay aggregated in a single frame. Server-side +these large frames are treated as "heavy" and shed before transmit under congestion +(see [Slow clients / backpressure](#slow-clients--backpressure)). + +This is purely a framing change — the payload format is identical, and a publish +cycle may now emit several binary frames (at most one aggregated light frame, plus +one per heavy message — and no light frame at all when every admitted message in a +group is heavy) instead of one. Each frame is self-describing via its +`message_count`, so a client decodes them all the same way and needs no changes. + +**Wire compatibility:** heavy frames are **not** marked on the wire — the `flags` +field stays `0` on every frame, because existing PlotJuggler plugins reject any +frame with `flags != 0`. Heaviness is a purely server-side scheduling property. The +`kFrameFlagHeavy` bit (0x1) is reserved for a future capability-negotiated rollout +if clients ever need to distinguish heavy frames (e.g. to surface drop indicators). +`protocol_version` is unchanged. diff --git a/fastdds/src/main.cpp b/fastdds/src/main.cpp index 3e04920..bbd6b20 100644 --- a/fastdds/src/main.cpp +++ b/fastdds/src/main.cpp @@ -42,6 +42,7 @@ int main(int argc, char* argv[]) { std::vector topic_whitelist{".*"}; double topic_poll_interval = 1.0; int client_backlog_size = 100; + int heavy_frame_threshold_bytes = 262144; std::string certfile; std::string keyfile; @@ -60,6 +61,12 @@ int main(int argc, char* argv[]) { "Max frames queued per slow client before dropping the oldest (backpressure)") ->default_val(100) ->check(CLI::Range(1, 1000000)); + app.add_option( + "--heavy-frame-threshold-bytes", heavy_frame_threshold_bytes, + "Isolate messages this size (bytes) or larger into their own size-class frames; " + "0 disables (keep below the 1 MiB socket watermark)") + ->default_val(262144) + ->check(CLI::Range(0, 1000000000)); // Bound variable is already initialized to {".*"} (match everything); CLI11 // leaves it untouched if the flag is not passed, so no default_val() is // needed (and default_val() on a vector would round-trip through a @@ -86,6 +93,7 @@ int main(int argc, char* argv[]) { spdlog::info(" Topic whitelist: {}", fmt::join(topic_whitelist, ", ")); spdlog::info(" Topic poll interval: {:.1f} s", topic_poll_interval); spdlog::info(" Client backlog size: {}", client_backlog_size); + spdlog::info(" Heavy frame threshold: {} bytes", heavy_frame_threshold_bytes); spdlog::info(" TLS: {}", tls_enabled ? "enabled" : "disabled"); auto whitelist_result = pj_bridge::WhitelistFilter::create(topic_whitelist); @@ -99,6 +107,14 @@ int main(int argc, char* argv[]) { return 1; } + if (heavy_frame_threshold_bytes > 0 && + static_cast(heavy_frame_threshold_bytes) >= pj_bridge::WebSocketMiddleware::kSocketBufferHighWatermark) { + spdlog::warn( + "--heavy-frame-threshold-bytes ({}) >= socket watermark ({}): messages between the two sizes stay 'light' and " + "queue instead of shedding under congestion — keep the threshold below the watermark", + heavy_frame_threshold_bytes, pj_bridge::WebSocketMiddleware::kSocketBufferHighWatermark); + } + try { auto topic_source = std::make_shared(domain_ids); auto sub_manager = std::make_shared(*topic_source); @@ -110,8 +126,9 @@ int main(int argc, char* argv[]) { std::make_shared(static_cast(client_backlog_size), tls_config); pj_bridge::BridgeServer server( - topic_source, sub_manager, middleware, port, session_timeout, publish_rate, - std::move(whitelist_result.value())); + topic_source, sub_manager, middleware, + {port, session_timeout, publish_rate, std::move(whitelist_result.value()), + static_cast(heavy_frame_threshold_bytes)}); pj_bridge::run_standalone_event_loop( server, sub_manager, middleware, {port, publish_rate, session_timeout, stats_enabled, topic_poll_interval}); diff --git a/ros2/src/main.cpp b/ros2/src/main.cpp index 4ae111d..2f5b71b 100644 --- a/ros2/src/main.cpp +++ b/ros2/src/main.cpp @@ -49,6 +49,7 @@ int main(int argc, char** argv) { node->declare_parameter("max_qos_depth", 100); node->declare_parameter("topic_poll_interval", 1.0); node->declare_parameter("client_backlog_size", 100); + node->declare_parameter("heavy_frame_threshold_bytes", 262144); node->declare_parameter("tls", false); node->declare_parameter("certfile", ""); node->declare_parameter("keyfile", ""); @@ -62,6 +63,7 @@ int main(int argc, char** argv) { int64_t max_qos_depth = node->get_parameter("max_qos_depth").as_int(); double topic_poll_interval = node->get_parameter("topic_poll_interval").as_double(); int64_t client_backlog_size = node->get_parameter("client_backlog_size").as_int(); + int64_t heavy_frame_threshold_bytes = node->get_parameter("heavy_frame_threshold_bytes").as_int(); bool tls_enabled = node->get_parameter("tls").as_bool(); std::string certfile = node->get_parameter("certfile").as_string(); std::string keyfile = node->get_parameter("keyfile").as_string(); @@ -93,6 +95,23 @@ int main(int argc, char** argv) { return 1; } + if (heavy_frame_threshold_bytes < 0) { + RCLCPP_ERROR( + node->get_logger(), "Invalid heavy_frame_threshold_bytes: %ld (must be >= 0; 0 disables splitting)", + heavy_frame_threshold_bytes); + rclcpp::shutdown(); + return 1; + } + if (heavy_frame_threshold_bytes > 0 && + static_cast(heavy_frame_threshold_bytes) >= pj_bridge::WebSocketMiddleware::kSocketBufferHighWatermark) { + RCLCPP_WARN( + node->get_logger(), + "heavy_frame_threshold_bytes (%ld) >= socket watermark (%zu): messages between the two sizes stay 'light' and " + "queue instead of shedding under congestion — keep the threshold below the watermark", + heavy_frame_threshold_bytes, pj_bridge::WebSocketMiddleware::kSocketBufferHighWatermark); + } + RCLCPP_INFO(node->get_logger(), "heavy_frame_threshold_bytes=%ld", heavy_frame_threshold_bytes); + auto whitelist_result = pj_bridge::WhitelistFilter::create(topic_whitelist); if (!whitelist_result) { RCLCPP_ERROR(node->get_logger(), "Invalid topic_whitelist: %s", whitelist_result.error().c_str()); @@ -124,8 +143,9 @@ int main(int argc, char** argv) { // Create bridge server pj_bridge::BridgeServer server( - topic_source, sub_manager, middleware, port, session_timeout, publish_rate, - std::move(whitelist_result.value())); + topic_source, sub_manager, middleware, + {port, session_timeout, publish_rate, std::move(whitelist_result.value()), + static_cast(heavy_frame_threshold_bytes)}); if (!server.initialize()) { RCLCPP_ERROR(node->get_logger(), "Failed to initialize bridge server"); @@ -186,8 +206,10 @@ int main(int argc, char** argv) { auto [total_messages, total_bytes] = server.get_publish_stats(); RCLCPP_INFO( - node->get_logger(), "Final statistics: %lu messages published, %lu bytes transmitted", total_messages, - total_bytes); + node->get_logger(), + "Final statistics: %lu messages published, %lu bytes transmitted, %lu frames dropped (slow clients), " + "%lu heavy frames shed (congestion)", + total_messages, total_bytes, middleware->dropped_frame_count(), middleware->heavy_shed_count()); RCLCPP_INFO(node->get_logger(), "Bridge server shutdown complete"); diff --git a/rti/src/main.cpp b/rti/src/main.cpp index 396fdd8..f419961 100644 --- a/rti/src/main.cpp +++ b/rti/src/main.cpp @@ -43,6 +43,7 @@ int main(int argc, char* argv[]) { std::vector topic_whitelist{".*"}; double topic_poll_interval = 1.0; int client_backlog_size = 100; + int heavy_frame_threshold_bytes = 262144; std::string certfile; std::string keyfile; @@ -62,6 +63,12 @@ int main(int argc, char* argv[]) { "Max frames queued per slow client before dropping the oldest (backpressure)") ->default_val(100) ->check(CLI::Range(1, 1000000)); + app.add_option( + "--heavy-frame-threshold-bytes", heavy_frame_threshold_bytes, + "Isolate messages this size (bytes) or larger into their own size-class frames; " + "0 disables (keep below the 1 MiB socket watermark)") + ->default_val(262144) + ->check(CLI::Range(0, 1000000000)); // Bound variable is already initialized to {".*"} (match everything); CLI11 // leaves it untouched if the flag is not passed, so no default_val() is // needed (and default_val() on a vector would round-trip through a @@ -91,6 +98,7 @@ int main(int argc, char* argv[]) { spdlog::info(" Topic whitelist: {}", fmt::join(topic_whitelist, ", ")); spdlog::info(" Topic poll interval: {:.1f} s", topic_poll_interval); spdlog::info(" Client backlog size: {}", client_backlog_size); + spdlog::info(" Heavy frame threshold: {} bytes", heavy_frame_threshold_bytes); spdlog::info(" TLS: {}", tls_enabled ? "enabled" : "disabled"); auto whitelist_result = pj_bridge::WhitelistFilter::create(topic_whitelist); @@ -104,6 +112,14 @@ int main(int argc, char* argv[]) { return 1; } + if (heavy_frame_threshold_bytes > 0 && + static_cast(heavy_frame_threshold_bytes) >= pj_bridge::WebSocketMiddleware::kSocketBufferHighWatermark) { + spdlog::warn( + "--heavy-frame-threshold-bytes ({}) >= socket watermark ({}): messages between the two sizes stay 'light' and " + "queue instead of shedding under congestion — keep the threshold below the watermark", + heavy_frame_threshold_bytes, pj_bridge::WebSocketMiddleware::kSocketBufferHighWatermark); + } + try { auto topic_source = std::make_shared(domain_ids, qos_profile); auto sub_manager = std::make_shared(*topic_source); @@ -115,8 +131,9 @@ int main(int argc, char* argv[]) { std::make_shared(static_cast(client_backlog_size), tls_config); pj_bridge::BridgeServer server( - topic_source, sub_manager, middleware, port, session_timeout, publish_rate, - std::move(whitelist_result.value())); + topic_source, sub_manager, middleware, + {port, session_timeout, publish_rate, std::move(whitelist_result.value()), + static_cast(heavy_frame_threshold_bytes)}); pj_bridge::run_standalone_event_loop( server, sub_manager, middleware, {port, publish_rate, session_timeout, stats_enabled, topic_poll_interval}); diff --git a/tests/unit/test_backpressure.cpp b/tests/unit/test_backpressure.cpp new file mode 100644 index 0000000..2928938 --- /dev/null +++ b/tests/unit/test_backpressure.cpp @@ -0,0 +1,194 @@ +/* + * Copyright (C) 2026 Davide Faconti + * + * This file is part of pj_bridge. + * + * pj_bridge is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * pj_bridge is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with pj_bridge. If not, see . + */ + +#include + +#include +#include +#include + +#include "pj_bridge/middleware/backpressure.hpp" + +using namespace pj_bridge; + +namespace { +std::vector frame_of(size_t n) { + return std::vector(n, 0x7F); +} +} // namespace + +// Drives run_backpressure() against an in-memory fake socket whose buffered +// amount grows by each frame's size as it is "sent" (sockets drain over time, +// but within one synchronous call the buffer only grows). No real connection. +class BackpressureTest : public ::testing::Test { + protected: + size_t watermark_ = 1000; + size_t buffered_ = 0; + std::deque> backlog_; + std::vector> sent_; + size_t backlog_cap_ = 100; + + std::function buffered_amount() { + return [this] { return buffered_; }; + } + std::function>()> pop_pending() { + return [this]() -> std::optional> { + if (backlog_.empty()) { + return std::nullopt; + } + std::vector f = std::move(backlog_.front()); + backlog_.pop_front(); + return f; + }; + } + std::function&)> send() { + return [this](const std::vector& f) { + sent_.push_back(f); + buffered_ += f.size(); + return true; + }; + } + std::function(const std::vector&)> queue_pending() { + return [this](const std::vector& f) -> std::optional { + size_t dropped = 0; + if (backlog_.size() >= backlog_cap_) { + backlog_.pop_front(); + dropped = 1; + } + backlog_.push_back(f); + return dropped; + }; + } + SendOutcome run(FramePriority prio, const std::vector& frame) { + return run_backpressure( + prio, frame, watermark_, backlog_.size(), buffered_amount(), pop_pending(), send(), queue_pending()); + } +}; + +TEST_F(BackpressureTest, SendsImmediatelyWhenSocketHasRoom) { + auto out = run(FramePriority::kNormal, frame_of(100)); + EXPECT_EQ(out.result, SendResult::kDelivered); + EXPECT_EQ(out.frames_flushed, 0u); + ASSERT_EQ(sent_.size(), 1u); + EXPECT_EQ(sent_[0].size(), 100u); +} + +TEST_F(BackpressureTest, FlushStopsWhenWatermarkReachedMidFlush) { + // Five queued 600-byte frames, watermark 1000. Each send adds 600 to the + // socket buffer, so only two may flush before the socket is congested again. + for (int i = 0; i < 5; ++i) { + backlog_.push_back(frame_of(600)); + } + auto out = run(FramePriority::kNormal, frame_of(600)); + EXPECT_EQ(out.frames_flushed, 2u) << "flush must recheck the watermark and stop early"; + EXPECT_EQ(out.result, SendResult::kQueued); + EXPECT_EQ(sent_.size(), 2u); + // Three unflushed backlog frames remain, plus the current frame gets queued. + EXPECT_EQ(backlog_.size(), 4u); +} + +TEST_F(BackpressureTest, HeavyFrameShedWhenCongested) { + buffered_ = watermark_; // already congested + auto out = run(FramePriority::kHeavy, frame_of(5000)); + EXPECT_EQ(out.result, SendResult::kShed); + EXPECT_TRUE(sent_.empty()); + EXPECT_TRUE(backlog_.empty()) << "a heavy frame must be shed, never queued"; +} + +TEST_F(BackpressureTest, NormalFrameQueuedWhenCongested) { + buffered_ = watermark_; // congested + auto out = run(FramePriority::kNormal, frame_of(100)); + EXPECT_EQ(out.result, SendResult::kQueued); + EXPECT_TRUE(sent_.empty()); + ASSERT_EQ(backlog_.size(), 1u); +} + +TEST_F(BackpressureTest, SendFailureDuringFlushIsNotAccepted) { + backlog_.push_back(frame_of(100)); + auto out = run_backpressure( + FramePriority::kNormal, frame_of(100), watermark_, /*max_flush=*/1, buffered_amount(), pop_pending(), + [](const std::vector&) { return false; }, // send fails (client gone) + queue_pending()); + EXPECT_EQ(out.result, SendResult::kClientGone); +} + +TEST_F(BackpressureTest, FlushDrainsFullyThenSendsCurrentWhenRoom) { + // Backlog fits comfortably under the watermark: all of it flushes, then the + // current frame is sent too. + for (int i = 0; i < 3; ++i) { + backlog_.push_back(frame_of(10)); + } + auto out = run(FramePriority::kNormal, frame_of(10)); + EXPECT_EQ(out.result, SendResult::kDelivered); + EXPECT_EQ(out.frames_flushed, 3u); + EXPECT_TRUE(backlog_.empty()); + EXPECT_EQ(sent_.size(), 4u); // 3 flushed + current +} + +TEST_F(BackpressureTest, HeavyFrameSentWhenRoomAvailable) { + // A heavy frame is only shed under congestion; with room it is sent normally. + auto out = run(FramePriority::kHeavy, frame_of(5000)); + EXPECT_EQ(out.result, SendResult::kDelivered); + ASSERT_EQ(sent_.size(), 1u); + EXPECT_EQ(sent_[0].size(), 5000u); +} + +TEST_F(BackpressureTest, QueueClientGoneReturnsNotAccepted) { + // queue_pending() reporting the client vanished (nullopt) must surface as a + // failed send (matches the middleware disconnect-race handling). + buffered_ = watermark_; // congested -> normal frame takes the queue path + auto out = run_backpressure( + FramePriority::kNormal, frame_of(100), watermark_, /*max_flush=*/0, buffered_amount(), pop_pending(), send(), + [](const std::vector&) -> std::optional { return std::nullopt; }); + EXPECT_EQ(out.result, SendResult::kClientGone); +} + +TEST_F(BackpressureTest, FlushBoundedByMaxFlush) { + // Even with room for all, at most max_flush queued frames flush per call, so a + // concurrent producer cannot make one call flush forever. + for (int i = 0; i < 5; ++i) { + backlog_.push_back(frame_of(10)); + } + auto out = run_backpressure( + FramePriority::kNormal, frame_of(10), watermark_, /*max_flush=*/3, buffered_amount(), pop_pending(), send(), + queue_pending()); + EXPECT_EQ(out.frames_flushed, 3u); + EXPECT_EQ(backlog_.size(), 2u); +} + +TEST_F(BackpressureTest, FlushTerminatesDespiteConcurrentEnqueue) { + // A producer that enqueues a fresh frame on every pop (socket has endless + // room). Without the max_flush bound this would flush forever; with it, + // exactly max_flush frames flush and the call terminates. This is the + // concurrency property max_flush exists to guarantee. + backlog_.push_back(frame_of(1)); + auto refilling_pop = [this]() -> std::optional> { + if (backlog_.empty()) { + return std::nullopt; + } + std::vector f = std::move(backlog_.front()); + backlog_.pop_front(); + backlog_.push_back(frame_of(1)); // producer keeps the backlog non-empty + return f; + }; + auto out = run_backpressure( + FramePriority::kNormal, frame_of(1), watermark_, /*max_flush=*/3, buffered_amount(), refilling_pop, send(), + queue_pending()); + EXPECT_EQ(out.frames_flushed, 3u) << "max_flush must bound the flush despite continual refills"; +} diff --git a/tests/unit/test_bridge_server.cpp b/tests/unit/test_bridge_server.cpp index 8b2692b..7d0b62e 100644 --- a/tests/unit/test_bridge_server.cpp +++ b/tests/unit/test_bridge_server.cpp @@ -25,8 +25,8 @@ #include #include #include -#include #include +#include #include #include #include @@ -37,6 +37,7 @@ #include "pj_bridge/bridge_server.hpp" #include "pj_bridge/message_serializer.hpp" #include "pj_bridge/middleware/middleware_interface.hpp" +#include "pj_bridge/protocol_constants.hpp" #include "pj_bridge/subscription_manager_interface.hpp" #include "pj_bridge/topic_source_interface.hpp" #include "pj_bridge/whitelist_filter.hpp" @@ -84,11 +85,29 @@ class MockMiddleware : public MiddlewareInterface { return true; } - bool send_binary(const std::string& client_identity, const std::vector& data) override { + SendResult send_binary( + const std::string& client_identity, const std::vector& data, FramePriority priority) override { + // Heaviness is carried by the priority argument, not a wire flag (heavy + // frames are NOT wire-flagged so existing plugins accept them). + const bool is_heavy_frame = (priority == FramePriority::kHeavy); + // Test seam: simulate a client vanishing right before a heavy frame is + // delivered, so partial-send stats can be exercised. + if (fail_heavy_sends_ && is_heavy_frame) { + return SendResult::kClientGone; + } + // Test seam: simulate the middleware shedding a heavy frame under congestion + // (accepted-but-dropped, never delivered) so the bridge's "shed frames aren't + // counted as forwarded" rule can be verified without a real socket. + if (shed_heavy_sends_ && is_heavy_frame) { + return SendResult::kShed; + } + // Only delivered frames are recorded; binary_sends_ and binary_priorities_ + // stay index-parallel. log_send(SendKind::kBinary, client_identity); std::lock_guard lock(binary_mutex_); binary_sends_.emplace_back(client_identity, data); - return true; + binary_priorities_.push_back(priority); + return SendResult::kDelivered; } bool is_ready() const override { @@ -172,6 +191,27 @@ class MockMiddleware : public MiddlewareInterface { void clear_binary_sends() { std::lock_guard lock(binary_mutex_); binary_sends_.clear(); + binary_priorities_.clear(); + } + + /// Priorities passed to send_binary(), in call order (parallel to get_binary_sends()). + std::vector get_binary_priorities() { + std::lock_guard lock(binary_mutex_); + return binary_priorities_; + } + + /// When enabled, send_binary() returns kClientGone for frames marked + /// kFrameFlagHeavy (simulating the client disconnecting before the heavy + /// frame is delivered). + void set_fail_heavy_sends(bool fail) { + fail_heavy_sends_ = fail; + } + + /// When enabled, send_binary() returns kShed for frames marked kFrameFlagHeavy + /// (simulating the middleware shedding a heavy frame under congestion — the + /// frame is accepted by the policy but never delivered). + void set_shed_heavy_sends(bool shed) { + shed_heavy_sends_ = shed; } /// Return all replies sent to a given client, in send order (parsed JSON). @@ -225,6 +265,9 @@ class MockMiddleware : public MiddlewareInterface { std::mutex binary_mutex_; std::vector>> binary_sends_; + std::vector binary_priorities_; + bool fail_heavy_sends_{false}; + bool shed_heavy_sends_{false}; std::mutex send_log_mutex_; std::vector> send_log_; @@ -476,7 +519,8 @@ class BridgeServerTest : public ::testing::Test { mock_sub_manager_ = std::make_shared(); // Use a high port that the mock will never actually listen on. // session_timeout = 10s, publish_rate = 50 Hz - server_ = std::make_unique(mock_topic_source_, mock_sub_manager_, mock_, 19999, 10.0, 50.0); + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, 50.0}); } void TearDown() override { @@ -1967,14 +2011,14 @@ TEST_F(BridgeServerTest, UnsubscribeAcceptsObjectFormat) { // BridgeServer::initialize() should reject publish_rate <= 0. // --------------------------------------------------------------------------- TEST_F(BridgeServerTest, InitializeRejectsZeroPublishRate) { - auto zero_rate_server = - std::make_unique(mock_topic_source_, mock_sub_manager_, mock_, 19999, 10.0, 0.0); + auto zero_rate_server = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, 0.0}); EXPECT_FALSE(zero_rate_server->initialize()); } TEST_F(BridgeServerTest, InitializeRejectsNegativePublishRate) { - auto neg_rate_server = - std::make_unique(mock_topic_source_, mock_sub_manager_, mock_, 19999, 10.0, -1.0); + auto neg_rate_server = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, -1.0}); EXPECT_FALSE(neg_rate_server->initialize()); } @@ -2333,7 +2377,7 @@ TEST_F(BridgeServerTest, GetTopicsOmitsNonWhitelistedTopics) { auto whitelist_result = WhitelistFilter::create({"/allowed.*"}); ASSERT_TRUE(whitelist_result.has_value()); server_ = std::make_unique( - mock_topic_source_, mock_sub_manager_, mock_, 19999, 10.0, 50.0, whitelist_result.value()); + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, 50.0, whitelist_result.value()}); ASSERT_TRUE(server_->initialize()); mock_topic_source_->set_topics({{"/allowed/foo", "std_msgs/msg/String"}, {"/blocked", "std_msgs/msg/String"}}); @@ -2356,7 +2400,7 @@ TEST_F(BridgeServerTest, SubscribeToNonWhitelistedTopicFailsAllSubscriptionsFail auto whitelist_result = WhitelistFilter::create({"/allowed.*"}); ASSERT_TRUE(whitelist_result.has_value()); server_ = std::make_unique( - mock_topic_source_, mock_sub_manager_, mock_, 19999, 10.0, 50.0, whitelist_result.value()); + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, 50.0, whitelist_result.value()}); ASSERT_TRUE(server_->initialize()); mock_topic_source_->set_topics({{"/blocked", "std_msgs/msg/String"}}); @@ -2384,7 +2428,7 @@ TEST_F(BridgeServerTest, SubscribeMixedWhitelistedAndNonWhitelistedTopicsPartial auto whitelist_result = WhitelistFilter::create({"/allowed.*"}); ASSERT_TRUE(whitelist_result.has_value()); server_ = std::make_unique( - mock_topic_source_, mock_sub_manager_, mock_, 19999, 10.0, 50.0, whitelist_result.value()); + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, 50.0, whitelist_result.value()}); ASSERT_TRUE(server_->initialize()); mock_topic_source_->set_topics({{"/allowed/foo", "std_msgs/msg/String"}, {"/blocked", "std_msgs/msg/String"}}); @@ -2571,7 +2615,7 @@ TEST_F(BridgeServerTest, CheckTopicChangesIgnoresNonWhitelistedTopics) { auto whitelist_result = WhitelistFilter::create({"/allowed.*"}); ASSERT_TRUE(whitelist_result.has_value()); server_ = std::make_unique( - mock_topic_source_, mock_sub_manager_, mock_, 19999, 10.0, 50.0, whitelist_result.value()); + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, 50.0, whitelist_result.value()}); ASSERT_TRUE(server_->initialize()); mock_topic_source_->set_topics({{"/allowed/foo", "std_msgs/msg/String"}}); @@ -2676,7 +2720,8 @@ TEST_F(BridgeServerTest, GetTopicsCarriesServerInfo) { EXPECT_FALSE(info["version"].get().empty()); ASSERT_TRUE(info["capabilities"].is_array()); const auto& caps = info["capabilities"]; - for (const char* expected : {"include_schemas", "latched_badge", "latched_replay", "topics_changed"}) { + for (const char* expected : + {"include_schemas", "latched_badge", "latched_replay", "topics_changed", "size_class_frames"}) { EXPECT_TRUE(std::find(caps.begin(), caps.end(), json(expected)) != caps.end()) << "missing " << expected; } } @@ -3095,3 +3140,453 @@ TEST_F(BridgeServerTest, ResumeReplaysLatchedSampleForTopicSubscribedWhilePaused EXPECT_EQ(p_sends[0], MockMiddleware::SendKind::kText); EXPECT_EQ(p_sends[1], MockMiddleware::SendKind::kBinary); } + +// --------------------------------------------------------------------------- +// Size-class frame splitting +// +// A "heavy" (large) topic is isolated into its own binary frame (flagged +// kFrameFlagHeavy) instead of being welded into the aggregated frame with +// small topics. This lets slow-link backpressure shed the heavy frame without +// starving the light topics — fixing the priority-inversion described in +// docs/API.md. +// --------------------------------------------------------------------------- + +namespace { +// Build a shared CDR-style byte buffer of `n` bytes all set to `fill`. +std::shared_ptr> make_bytes(size_t n, uint8_t fill) { + return std::make_shared>(n, static_cast(fill)); +} +// Read the uint32 flags field at binary-header offset 12. +uint32_t frame_flags(const std::vector& frame) { + uint32_t flags = 0; + std::memcpy(&flags, frame.data() + 12, sizeof(flags)); + return flags; +} +} // namespace + +TEST_F(BridgeServerTest, PublishSplitsHeavyTopicIntoOwnFrame) { + // Rebuild the server with a small heavy-frame threshold so a modest payload + // qualifies as "heavy" without allocating hundreds of KiB. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, + BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, /*heavy_frame_threshold_bytes=*/64}); + ASSERT_TRUE(server_->initialize()); + + mock_topic_source_->set_topics({{"/small", "std_msgs/msg/String"}, {"/big", "sensor_msgs/msg/PointCloud2"}}); + mock_sub_manager_->add_known_topic("/small"); + mock_sub_manager_->add_known_topic("/big"); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({"/small", "/big"}); + mock_->push_request("client_a", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_a"); + mock_->clear_binary_sends(); + + mock_sub_manager_->deliver_message("/small", make_bytes(8, 0xAA), 111); // below threshold + mock_sub_manager_->deliver_message("/big", make_bytes(200, 0xBB), 222); // >= threshold + + server_->publish_aggregated_messages(); + + auto sends = mock_->get_binary_sends(); + auto priorities = mock_->get_binary_priorities(); + ASSERT_EQ(sends.size(), 2u) << "expected one light frame + one heavy frame"; + ASSERT_EQ(priorities.size(), 2u); + + // Classify by send priority (heavy frames are NOT wire-flagged so existing + // plugins accept them). Every frame must therefore carry flags == 0. + int light_idx = -1; + int heavy_idx = -1; + for (int i = 0; i < 2; ++i) { + EXPECT_EQ(frame_flags(sends[i].second), 0u) << "heavy frames must not set a wire flag (plugin compatibility)"; + if (priorities[i] == FramePriority::kHeavy) { + heavy_idx = i; + } else { + light_idx = i; + } + } + ASSERT_GE(light_idx, 0) << "no light (kNormal) frame found"; + ASSERT_GE(heavy_idx, 0) << "no heavy (kHeavy) frame found"; + + // Both frames go to the subscribing client. + EXPECT_EQ(sends[light_idx].first, "client_a"); + EXPECT_EQ(sends[heavy_idx].first, "client_a"); + + // Light frame carries only /small; heavy frame carries only /big. + auto light_msgs = decode_frame(sends[light_idx].second); + ASSERT_EQ(light_msgs.size(), 1u); + EXPECT_EQ(light_msgs[0].topic, "/small"); + + auto heavy_msgs = decode_frame(sends[heavy_idx].second); + ASSERT_EQ(heavy_msgs.size(), 1u); + EXPECT_EQ(heavy_msgs[0].topic, "/big"); +} + +TEST_F(BridgeServerTest, HeavyFrameThresholdZeroKeepsSingleFrame) { + // Threshold 0 disables splitting: small + large ride one aggregated frame + // (legacy behavior), and no heavy flag is set. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, + BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, /*heavy_frame_threshold_bytes=*/0}); + ASSERT_TRUE(server_->initialize()); + + mock_topic_source_->set_topics({{"/small", "std_msgs/msg/String"}, {"/big", "sensor_msgs/msg/PointCloud2"}}); + mock_sub_manager_->add_known_topic("/small"); + mock_sub_manager_->add_known_topic("/big"); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({"/small", "/big"}); + mock_->push_request("client_a", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_a"); + mock_->clear_binary_sends(); + + mock_sub_manager_->deliver_message("/small", make_bytes(8, 0xAA), 111); + mock_sub_manager_->deliver_message("/big", make_bytes(200, 0xBB), 222); + + server_->publish_aggregated_messages(); + + auto sends = mock_->get_binary_sends(); + ASSERT_EQ(sends.size(), 1u) << "threshold 0 must not split"; + EXPECT_EQ(frame_flags(sends[0].second), 0u); + EXPECT_EQ(decode_frame(sends[0].second).size(), 2u); +} + +TEST_F(BridgeServerTest, PartialSendDoesNotOvercountForwardStats) { + // A split group emits a light frame (/small) then a heavy frame (/big). If + // the heavy frame's send fails (client vanished mid-cycle), forward stats + // must count only the topic that was actually delivered — not the unsent one. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, + BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, /*heavy_frame_threshold_bytes=*/64}); + ASSERT_TRUE(server_->initialize()); + + mock_topic_source_->set_topics({{"/small", "std_msgs/msg/String"}, {"/big", "sensor_msgs/msg/PointCloud2"}}); + mock_sub_manager_->add_known_topic("/small"); + mock_sub_manager_->add_known_topic("/big"); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({"/small", "/big"}); + mock_->push_request("client_a", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_a"); + mock_->clear_binary_sends(); + + mock_->set_fail_heavy_sends(true); // heavy frame delivery fails + + mock_sub_manager_->deliver_message("/small", make_bytes(8, 0xAA), 111); + mock_sub_manager_->deliver_message("/big", make_bytes(200, 0xBB), 222); + server_->publish_aggregated_messages(); + + auto stats = server_->snapshot_and_reset_stats(); + EXPECT_EQ(stats.topic_forward_counts["/small"], 1u); + EXPECT_EQ(stats.topic_forward_counts["/big"], 0u) << "unsent heavy topic must not be counted"; +} + +TEST_F(BridgeServerTest, HeavyFrameThresholdBoundaryIsInclusive) { + // The classifier is `size >= threshold`: a message exactly at the threshold + // is heavy; one byte under is light. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, + BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, /*heavy_frame_threshold_bytes=*/64}); + ASSERT_TRUE(server_->initialize()); + + mock_topic_source_->set_topics({{"/at", "std_msgs/msg/String"}, {"/below", "std_msgs/msg/String"}}); + mock_sub_manager_->add_known_topic("/at"); + mock_sub_manager_->add_known_topic("/below"); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({"/at", "/below"}); + mock_->push_request("client_a", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_a"); + mock_->clear_binary_sends(); + + mock_sub_manager_->deliver_message("/at", make_bytes(64, 0x11), 111); // == threshold -> heavy + mock_sub_manager_->deliver_message("/below", make_bytes(63, 0x22), 222); // < threshold -> light + server_->publish_aggregated_messages(); + + auto sends = mock_->get_binary_sends(); + auto priorities = mock_->get_binary_priorities(); + ASSERT_EQ(sends.size(), 2u); + bool heavy_is_at = false; + bool light_is_below = false; + for (size_t i = 0; i < sends.size(); ++i) { + auto msgs = decode_frame(sends[i].second); + ASSERT_EQ(msgs.size(), 1u); + if (priorities[i] == FramePriority::kHeavy) { + heavy_is_at = (msgs[0].topic == "/at"); + } else { + light_is_below = (msgs[0].topic == "/below"); + } + } + EXPECT_TRUE(heavy_is_at) << "message == threshold must be heavy"; + EXPECT_TRUE(light_is_below) << "message < threshold must be light"; +} + +TEST_F(BridgeServerTest, OnlyHeavyTopicEmitsNoLightFrame) { + // A group whose only admitted message is heavy emits just the heavy frame — + // no empty light frame. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, + BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, /*heavy_frame_threshold_bytes=*/64}); + ASSERT_TRUE(server_->initialize()); + + mock_topic_source_->set_topics({{"/big", "sensor_msgs/msg/PointCloud2"}}); + mock_sub_manager_->add_known_topic("/big"); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({"/big"}); + mock_->push_request("client_a", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_a"); + mock_->clear_binary_sends(); + + mock_sub_manager_->deliver_message("/big", make_bytes(200, 0xBB), 222); + server_->publish_aggregated_messages(); + + auto sends = mock_->get_binary_sends(); + auto priorities = mock_->get_binary_priorities(); + ASSERT_EQ(sends.size(), 1u) << "no light frame when every topic is heavy"; + EXPECT_EQ(priorities[0], FramePriority::kHeavy); + auto msgs = decode_frame(sends[0].second); + ASSERT_EQ(msgs.size(), 1u); + EXPECT_EQ(msgs[0].topic, "/big"); +} + +TEST_F(BridgeServerTest, LargeLatchedSampleReplayedUnflagged) { + // A latched replay frame is built by collect_latched_replay via the plain + // finalize() (no flags) and must NOT be marked heavy even when the retained + // sample is large: it is a one-shot replay, not part of the size-class stream. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, + BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, /*heavy_frame_threshold_bytes=*/64}); + ASSERT_TRUE(server_->initialize()); + + mock_topic_source_->set_topics({{"/latched", "sensor_msgs/msg/PointCloud2"}}); + mock_sub_manager_->add_known_topic("/latched"); + mock_sub_manager_->set_transient_local("/latched", true); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({"/latched"}); + + mock_->push_request("client_a", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_a"); + mock_sub_manager_->deliver_message("/latched", make_bytes(200, 0xCC), 777); // large retained sample + server_->publish_aggregated_messages(); // drains the normal buffer + mock_->clear_binary_sends(); + + // Late subscriber gets the retained sample replayed. + mock_->push_request("client_b", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_b"); + + auto sends = mock_->get_binary_sends(); + ASSERT_EQ(sends.size(), 1u); + EXPECT_EQ(sends[0].first, "client_b"); + EXPECT_EQ(frame_flags(sends[0].second), 0u) << "latched replay must be unflagged even when large"; + auto msgs = decode_frame(sends[0].second); + ASSERT_EQ(msgs.size(), 1u); + EXPECT_EQ(msgs[0].topic, "/latched"); +} + +TEST_F(BridgeServerTest, HeavyFramesSentWithHeavyPriority) { + // The bridge must tag heavy frames kHeavy so the middleware can shed them + // under congestion; light frames stay kNormal. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, + BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, /*heavy_frame_threshold_bytes=*/64}); + ASSERT_TRUE(server_->initialize()); + + mock_topic_source_->set_topics({{"/small", "std_msgs/msg/String"}, {"/big", "sensor_msgs/msg/PointCloud2"}}); + mock_sub_manager_->add_known_topic("/small"); + mock_sub_manager_->add_known_topic("/big"); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({"/small", "/big"}); + mock_->push_request("client_a", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_a"); + mock_->clear_binary_sends(); + + mock_sub_manager_->deliver_message("/small", make_bytes(8, 0xAA), 111); + mock_sub_manager_->deliver_message("/big", make_bytes(200, 0xBB), 222); + server_->publish_aggregated_messages(); + + auto sends = mock_->get_binary_sends(); + auto priorities = mock_->get_binary_priorities(); + ASSERT_EQ(sends.size(), 2u); + ASSERT_EQ(priorities.size(), 2u); + // The kHeavy-priority frame must carry /big; the kNormal frame /small. (Frames + // are not wire-flagged, so priority is the only heaviness signal.) + for (size_t i = 0; i < sends.size(); ++i) { + EXPECT_EQ(frame_flags(sends[i].second), 0u); + auto msgs = decode_frame(sends[i].second); + ASSERT_EQ(msgs.size(), 1u); + EXPECT_EQ(msgs[0].topic, priorities[i] == FramePriority::kHeavy ? "/big" : "/small"); + } +} + +TEST_F(BridgeServerTest, RateLimitedHeavyTopicRespectsRateGate) { + // The per-topic rate gate is applied BEFORE size classification, so a + // rate-limited heavy topic emits one heavy frame only per admitted message. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, 64}); + ASSERT_TRUE(server_->initialize()); + mock_topic_source_->set_topics({{"/big", "sensor_msgs/msg/PointCloud2"}}); + mock_sub_manager_->add_known_topic("/big"); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({json{{"name", "/big"}, {"max_rate_hz", 10.0}}}); // 100 ms min interval + mock_->push_request("client_a", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_a"); + mock_->clear_binary_sends(); + + // Interval = 1e8 ns; last_sent starts at 0. Admitted: 1e8 and 2e8; dropped: 1.5e8. + mock_sub_manager_->deliver_message("/big", make_bytes(200, 0xB1), 100'000'000); + mock_sub_manager_->deliver_message("/big", make_bytes(200, 0xB2), 150'000'000); + mock_sub_manager_->deliver_message("/big", make_bytes(200, 0xB3), 200'000'000); + server_->publish_aggregated_messages(); + + auto sends = mock_->get_binary_sends(); + auto priorities = mock_->get_binary_priorities(); + ASSERT_EQ(sends.size(), 2u) << "rate gate must admit exactly 2 of the 3 heavy messages"; + for (size_t i = 0; i < sends.size(); ++i) { + EXPECT_EQ(priorities[i], FramePriority::kHeavy) << "each admitted heavy message is its own frame"; + auto msgs = decode_frame(sends[i].second); + ASSERT_EQ(msgs.size(), 1u); + EXPECT_EQ(msgs[0].topic, "/big"); + } + auto stats = server_->snapshot_and_reset_stats(); + EXPECT_EQ(stats.topic_forward_counts["/big"], 2u); +} + +TEST_F(BridgeServerTest, MultiClientGroupSplitFansOutButCountsOnce) { + // Two clients with the identical subscription share one group: the light and + // heavy frames are each fanned out to BOTH, but forward stats count each topic + // once per group, not once per client. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, 64}); + ASSERT_TRUE(server_->initialize()); + mock_topic_source_->set_topics({{"/small", "std_msgs/msg/String"}, {"/big", "sensor_msgs/msg/PointCloud2"}}); + mock_sub_manager_->add_known_topic("/small"); + mock_sub_manager_->add_known_topic("/big"); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({"/small", "/big"}); + for (const char* cid : {"client_a", "client_b"}) { + mock_->push_request(cid, sub.dump()); + server_->process_requests(); + mock_->pop_reply(cid); + } + mock_->clear_binary_sends(); + + mock_sub_manager_->deliver_message("/small", make_bytes(8, 0xAA), 111); + mock_sub_manager_->deliver_message("/big", make_bytes(200, 0xBB), 222); + server_->publish_aggregated_messages(); + + auto sends = mock_->get_binary_sends(); + ASSERT_EQ(sends.size(), 4u) << "2 clients x (1 light + 1 heavy)"; + std::unordered_map per_client; + for (const auto& [cid, frame] : sends) { + per_client[cid]++; + } + EXPECT_EQ(per_client["client_a"], 2); + EXPECT_EQ(per_client["client_b"], 2); + + auto stats = server_->snapshot_and_reset_stats(); + EXPECT_EQ(stats.topic_forward_counts["/small"], 1u) << "counted once per group, not per client"; + EXPECT_EQ(stats.topic_forward_counts["/big"], 1u); +} + +TEST_F(BridgeServerTest, MultipleHeavyTopicsEachGetOwnFrame) { + // Two heavy topics in one group must each get their own isolated heavy frame, + // alongside a single aggregated light frame. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, 64}); + ASSERT_TRUE(server_->initialize()); + mock_topic_source_->set_topics( + {{"/small", "std_msgs/msg/String"}, + {"/big1", "sensor_msgs/msg/PointCloud2"}, + {"/big2", "sensor_msgs/msg/PointCloud2"}}); + mock_sub_manager_->add_known_topic("/small"); + mock_sub_manager_->add_known_topic("/big1"); + mock_sub_manager_->add_known_topic("/big2"); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({"/small", "/big1", "/big2"}); + mock_->push_request("client_a", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_a"); + mock_->clear_binary_sends(); + + mock_sub_manager_->deliver_message("/small", make_bytes(8, 0xAA), 111); + mock_sub_manager_->deliver_message("/big1", make_bytes(200, 0xB1), 222); + mock_sub_manager_->deliver_message("/big2", make_bytes(200, 0xB2), 333); + server_->publish_aggregated_messages(); + + auto sends = mock_->get_binary_sends(); + auto priorities = mock_->get_binary_priorities(); + ASSERT_EQ(sends.size(), 3u) << "1 light + 2 heavy"; + std::set heavy_topics; + std::set light_topics; + for (size_t i = 0; i < sends.size(); ++i) { + auto msgs = decode_frame(sends[i].second); + if (priorities[i] == FramePriority::kHeavy) { + ASSERT_EQ(msgs.size(), 1u) << "a heavy frame carries a single message"; + heavy_topics.insert(msgs[0].topic); + } else { + for (const auto& m : msgs) { + light_topics.insert(m.topic); + } + } + } + EXPECT_EQ(heavy_topics, (std::set{"/big1", "/big2"})); + EXPECT_EQ(light_topics, (std::set{"/small"})); +} + +TEST_F(BridgeServerTest, ShedHeavyFrameNotCountedInForwardStats) { + // When the middleware sheds a heavy frame under congestion (accepted-but- + // dropped), the bridge must NOT count it as forwarded — otherwise "Sent" + // throughput over-reports and a starved heavy topic looks healthy. + server_ = std::make_unique( + mock_topic_source_, mock_sub_manager_, mock_, BridgeServerConfig{19999, 10.0, 50.0, WhitelistFilter{}, 64}); + ASSERT_TRUE(server_->initialize()); + mock_topic_source_->set_topics({{"/small", "std_msgs/msg/String"}, {"/big", "sensor_msgs/msg/PointCloud2"}}); + mock_sub_manager_->add_known_topic("/small"); + mock_sub_manager_->add_known_topic("/big"); + + json sub; + sub["command"] = "subscribe"; + sub["topics"] = json::array({"/small", "/big"}); + mock_->push_request("client_a", sub.dump()); + server_->process_requests(); + mock_->pop_reply("client_a"); + mock_->clear_binary_sends(); + + mock_->set_shed_heavy_sends(true); // middleware sheds the heavy frame + mock_sub_manager_->deliver_message("/small", make_bytes(8, 0xAA), 111); + mock_sub_manager_->deliver_message("/big", make_bytes(200, 0xBB), 222); + server_->publish_aggregated_messages(); + + auto stats = server_->snapshot_and_reset_stats(); + EXPECT_EQ(stats.topic_forward_counts["/small"], 1u); + EXPECT_EQ(stats.topic_forward_counts["/big"], 0u) << "a shed heavy frame is not forwarded"; + + // Only the light frame was actually delivered. + auto sends = mock_->get_binary_sends(); + ASSERT_EQ(sends.size(), 1u); + EXPECT_EQ(frame_flags(sends[0].second), 0u); +} diff --git a/tests/unit/test_message_serializer.cpp b/tests/unit/test_message_serializer.cpp index 378f866..12a02a3 100644 --- a/tests/unit/test_message_serializer.cpp +++ b/tests/unit/test_message_serializer.cpp @@ -22,6 +22,7 @@ #include #include "pj_bridge/message_serializer.hpp" +#include "pj_bridge/protocol_constants.hpp" using namespace pj_bridge; @@ -498,3 +499,34 @@ TEST_F(MessageSerializerTest, UncompressedSizeMatchesPayload) { std::memcpy(&uncompressed_size, result.data() + 8, sizeof(uncompressed_size)); EXPECT_EQ(uncompressed_size, 18u); } + +// ============================================================================ +// Frame flags — heavy/size-class frame marking (bit 0 of header flags field) +// ============================================================================ + +TEST_F(MessageSerializerTest, FinalizeWithHeavyFlagSetsHeaderBit) { + auto data = create_test_data({1, 2, 3, 4}); + serializer_.serialize_message("/big", 1000, data.data(), data.size()); + + auto result = serializer_.finalize(kFrameFlagHeavy); + + // Flags field at offset 12 carries the heavy bit. + uint32_t flags; + std::memcpy(&flags, result.data() + 12, sizeof(flags)); + EXPECT_EQ(flags, kFrameFlagHeavy); +} + +TEST_F(MessageSerializerTest, FinalizeFlagsDoNotAlterPayload) { + auto data = create_test_data({10, 20, 30, 40, 50}); + serializer_.serialize_message("/big", 12345, data.data(), data.size()); + + auto plain = serializer_.finalize(0); + auto heavy = serializer_.finalize(kFrameFlagHeavy); + + // Only the flags field differs; the compressed payload (offset 16+) is + // byte-identical regardless of flags. + ASSERT_EQ(plain.size(), heavy.size()); + std::vector plain_payload(plain.begin() + 16, plain.end()); + std::vector heavy_payload(heavy.begin() + 16, heavy.end()); + EXPECT_EQ(plain_payload, heavy_payload); +} diff --git a/tests/unit/test_websocket_middleware.cpp b/tests/unit/test_websocket_middleware.cpp index 573a73a..0dd9029 100644 --- a/tests/unit/test_websocket_middleware.cpp +++ b/tests/unit/test_websocket_middleware.cpp @@ -116,7 +116,7 @@ TEST_F(WebSocketMiddlewareTest, SendBinaryToUnknownClient) { ASSERT_TRUE(result.has_value()); std::vector data = {1, 2, 3}; - EXPECT_FALSE(middleware_->send_binary("nonexistent_client", data)); + EXPECT_EQ(middleware_->send_binary("nonexistent_client", data, FramePriority::kNormal), SendResult::kClientGone); } TEST_F(WebSocketMiddlewareTest, ReceiveRequestNotInitializedReturnsFast) { @@ -334,7 +334,7 @@ TEST_F(WebSocketMiddlewareTest, SendBinaryToConnectedClient) { // Send binary data from the server to this client std::vector binary_payload = {0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02, 0x03, 0x04}; - EXPECT_TRUE(middleware_->send_binary(client_id, binary_payload)); + EXPECT_EQ(middleware_->send_binary(client_id, binary_payload, FramePriority::kNormal), SendResult::kDelivered); // Wait for the client to receive the binary data { @@ -350,6 +350,48 @@ TEST_F(WebSocketMiddlewareTest, SendBinaryToConnectedClient) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); } +// With the socket watermark forced to 0, every connected client is treated as +// congested, which makes the otherwise-unreachable over-watermark path testable +// with a real connection: a kHeavy frame is shed (and counted) before transmit, +// while a kNormal frame is queued. This covers the shed-counter wiring that +// lives outside the pure run_backpressure() policy. +TEST(WebSocketMiddlewareShedTest, HeavyFrameShedUnderForcedCongestionIncrementsCounter) { + WebSocketMiddleware middleware(/*client_backlog_size=*/100, std::nullopt, /*socket_buffer_watermark=*/0); + ASSERT_TRUE(middleware.initialize(18110).has_value()); + + ix::WebSocket client; + client.setUrl("ws://127.0.0.1:18110"); + client.setOnMessageCallback([](const ix::WebSocketMessagePtr& /*msg*/) {}); + client.start(); + ASSERT_TRUE(wait_for_client_open(client)) << "Client failed to connect"; + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + client.send("register"); + std::vector data; + std::string client_id; + ASSERT_TRUE(poll_receive_request(middleware, data, client_id)); + ASSERT_FALSE(client_id.empty()); + + // The middleware is content-agnostic; it acts on the priority argument, not + // the frame bytes. A >=16-byte buffer is a realistic frame size. + std::vector frame(32, 0x7F); + + EXPECT_EQ(middleware.heavy_shed_count(), 0u); + + EXPECT_EQ(middleware.send_binary(client_id, frame, FramePriority::kHeavy), SendResult::kShed); + EXPECT_EQ(middleware.heavy_shed_count(), 1u); + + EXPECT_EQ(middleware.send_binary(client_id, frame, FramePriority::kHeavy), SendResult::kShed); + EXPECT_EQ(middleware.heavy_shed_count(), 2u); + + // A normal frame under the same congestion is queued, not shed. + EXPECT_EQ(middleware.send_binary(client_id, frame, FramePriority::kNormal), SendResult::kQueued); + EXPECT_EQ(middleware.heavy_shed_count(), 2u); + + client.stop(); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); +} + TEST_F(WebSocketMiddlewareTest, ShutdownWithConnectedClientDoesNotDeadlock) { auto result = middleware_->initialize(18094); ASSERT_TRUE(result.has_value()); @@ -474,7 +516,7 @@ TEST_F(WebSocketMiddlewareTest, DroppedFrameCountZeroAfterNormalTraffic) { std::vector payload = {1, 2, 3, 4}; for (int i = 0; i < 10; ++i) { - EXPECT_TRUE(middleware_->send_binary(client_id, payload)); + EXPECT_EQ(middleware_->send_binary(client_id, payload, FramePriority::kNormal), SendResult::kDelivered); } EXPECT_EQ(middleware_->dropped_frame_count(), 0u); @@ -499,7 +541,7 @@ TEST_F(WebSocketMiddlewareTest, SendBinaryToAbsentClientCreatesNoPendingState) { std::vector data = {1, 2, 3}; for (int i = 0; i < 5; ++i) { - EXPECT_FALSE(middleware_->send_binary("never-connected", data)); + EXPECT_EQ(middleware_->send_binary("never-connected", data, FramePriority::kNormal), SendResult::kClientGone); } EXPECT_EQ(middleware_->dropped_frame_count(), 0u); @@ -545,7 +587,7 @@ TEST_F(WebSocketMiddlewareTest, DropPendingIsSafeNoOpAndKeepsSocketUsable) { EXPECT_EQ(middleware_->dropped_frame_count(), 0u); std::vector payload = {1, 2, 3, 4}; - EXPECT_TRUE(middleware_->send_binary(client_id, payload)); + EXPECT_EQ(middleware_->send_binary(client_id, payload, FramePriority::kNormal), SendResult::kDelivered); EXPECT_EQ(middleware_->dropped_frame_count(), 0u); client.stop();