From c10d25e0478e380807e778304a3630b0b1aa061a Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Wed, 8 Jul 2026 12:22:14 +0200 Subject: [PATCH 1/8] feat(bridge): isolate heavy topics into their own size-class frames MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of the backpressure priority-inversion fix. publish_aggregated_messages() previously welded every subscribed topic — small (/odom, /tf) and large (a PointCloud2) — into one ZSTD frame, so slow-link backpressure could only drop whole frames and one heavy topic starved all the light ones. Now each message whose CDR size is >= heavy_frame_threshold_bytes (default 256 KiB, 0 disables) is serialized into its own frame marked with a new header flag bit (kFrameFlagHeavy at offset 12), while small topics stay aggregated. Rate-limiting state and the subscription-group sharing optimization are unchanged — each group just emits one light frame plus N heavy frames, all to the same clients. Wire-compatible (payload format unchanged; old clients ignore the flag); advertised via the new "size_class_frames" capability, no protocol_version bump. - AggregatedMessageSerializer::finalize(flags): stamp header flags - kFrameFlagHeavy, kDefaultHeavyFrameThresholdBytes constants - BridgeServer heavy_frame_threshold_bytes ctor param + split logic - tests: serializer flag round-trip, heavy/light split, threshold=0 disable, capability advertisement (248 tests pass) Middleware shed-before-transmit (so heavy frames are dropped rather than queued under congestion) follows in the next milestone. Co-Authored-By: Claude Opus 4.8 --- app/include/pj_bridge/bridge_server.hpp | 11 +- app/include/pj_bridge/message_serializer.hpp | 6 +- app/include/pj_bridge/protocol_constants.hpp | 14 +++ app/src/bridge_server.cpp | 62 +++++++--- app/src/message_serializer.cpp | 5 +- tests/unit/test_bridge_server.cpp | 115 ++++++++++++++++++- tests/unit/test_message_serializer.cpp | 32 ++++++ 7 files changed, 222 insertions(+), 23 deletions(-) diff --git a/app/include/pj_bridge/bridge_server.hpp b/app/include/pj_bridge/bridge_server.hpp index 40c31d9..5dd3168 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" @@ -65,12 +66,17 @@ class BridgeServer { * @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 heavy_frame_threshold_bytes 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. */ 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 = {}); + double publish_rate = 50.0, WhitelistFilter whitelist = {}, + size_t heavy_frame_threshold_bytes = kDefaultHeavyFrameThresholdBytes); /// Shuts down middleware before members are destroyed, preventing /// disconnect callbacks from firing into a partially destroyed object. @@ -211,6 +217,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/protocol_constants.hpp b/app/include/pj_bridge/protocol_constants.hpp index 9f6d285..7c0191d 100644 --- a/app/include/pj_bridge/protocol_constants.hpp +++ b/app/include/pj_bridge/protocol_constants.hpp @@ -33,6 +33,19 @@ static constexpr uint32_t kBinaryFrameMagic = 0x42524A50; /// Size of the binary frame header in bytes static constexpr size_t kBinaryHeaderSize = 16; +/// Binary frame header flag bits (offset 12 of the 16-byte header). Bit 0 +/// marks a "heavy" frame carrying an isolated large/size-class message +/// (see docs/API.md); remaining bits are reserved = 0. Old clients ignore +/// this field, so the flag is purely additive. +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 +62,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..6e123b3 100644 --- a/app/src/bridge_server.cpp +++ b/app/src/bridge_server.cpp @@ -60,7 +60,8 @@ 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) + int port, double session_timeout, double publish_rate, WhitelistFilter whitelist, + size_t heavy_frame_threshold_bytes) : topic_source_(std::move(topic_source)), subscription_manager_(std::move(subscription_manager)), middleware_(std::move(middleware)), @@ -68,6 +69,7 @@ BridgeServer::BridgeServer( session_timeout_(session_timeout), publish_rate_(publish_rate), whitelist_(std::move(whitelist)), + heavy_frame_threshold_bytes_(heavy_frame_threshold_bytes), initialized_(false), total_messages_published_(0), total_bytes_published_(0), @@ -1057,6 +1059,7 @@ 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) }; std::vector frames; @@ -1064,12 +1067,37 @@ 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()]; + // 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 frame per heavy message (mirrors collect_latched_replay). + 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(kFrameFlagHeavy); + heavy_frame.msg_count = 1; + heavy_frame.client_ids = client_ids; + heavy_frame.is_heavy = true; + 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()); + } + group_msg_count++; + forward_counts[topic]++; + }; + for (const auto& [topic, msgs] : messages) { auto sub_it = representative_subs.find(topic); if (sub_it == representative_subs.end()) { @@ -1081,9 +1109,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 +1117,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 +1129,27 @@ 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; + 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 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/tests/unit/test_bridge_server.cpp b/tests/unit/test_bridge_server.cpp index 8b2692b..87cf713 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" @@ -2676,7 +2677,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 +3097,112 @@ 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_, 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(); + ASSERT_EQ(sends.size(), 2u) << "expected one light frame + one heavy frame"; + + // Classify by the header heavy flag, not by emission order. + int light_idx = -1; + int heavy_idx = -1; + for (int i = 0; i < 2; ++i) { + if ((frame_flags(sends[i].second) & kFrameFlagHeavy) != 0) { + heavy_idx = i; + } else { + light_idx = i; + } + } + ASSERT_GE(light_idx, 0) << "no light frame (flags == 0) found"; + ASSERT_GE(heavy_idx, 0) << "no heavy frame (kFrameFlagHeavy) 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_, 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); +} 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); +} From b54ebf307d8077b8527077cd71632922120df805 Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Wed, 8 Jul 2026 12:34:23 +0200 Subject: [PATCH 2/8] fix(bridge): per-frame forward stats + document heavy flag (Codex review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the Codex review of the size-class split: - forward stats accuracy: topic_forward_counts_ was folded from a group-wide map gated only by the global "any frame sent" flag, so after the split a light frame sending while a heavy frame failed (client vanished mid-cycle) over-counted the unsent heavy topic. Each GroupFrame now carries its own topic_counts, folded in only when that frame is actually delivered. - docs/API.md: the flags field said "Reserved (must be 0)" while heavy frames now set bit 0 — update the header table, add a "Size-class frames" section, and list the size_class_frames capability. Tests: partial-send stats (red-first), threshold boundary inclusivity, only-heavy group emits no light frame, large latched replay stays unflagged (252 tests pass). Co-Authored-By: Claude Opus 4.8 --- app/src/bridge_server.cpp | 15 ++- docs/API.md | 22 ++++- tests/unit/test_bridge_server.cpp | 157 ++++++++++++++++++++++++++++++ 3 files changed, 191 insertions(+), 3 deletions(-) diff --git a/app/src/bridge_server.cpp b/app/src/bridge_server.cpp index 6e123b3..c9c3e59 100644 --- a/app/src/bridge_server.cpp +++ b/app/src/bridge_server.cpp @@ -1060,6 +1060,10 @@ void BridgeServer::publish_aggregated_messages() { 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; @@ -1074,6 +1078,10 @@ void BridgeServer::publish_aggregated_messages() { 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 @@ -1090,12 +1098,13 @@ void BridgeServer::publish_aggregated_messages() { 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++; - forward_counts[topic]++; }; for (const auto& [topic, msgs] : messages) { @@ -1145,6 +1154,7 @@ void BridgeServer::publish_aggregated_messages() { 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) { @@ -1171,6 +1181,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/docs/API.md b/docs/API.md index aae63bd..bc505e6 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: @@ -601,7 +601,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 | Bit 0 (`0x1`) = heavy frame (see [Size-class frames](#size-class-frames-heavy-flag)); other bits reserved = 0 | ### Payload (ZSTD-compressed) @@ -617,3 +617,21 @@ For each message: ``` The magic bytes allow clients to validate frame integrity before decompression. + +### Size-class frames (heavy flag) + +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 **heavy** frame with header flag bit 0 (`0x1`) set, while smaller messages +stay aggregated in a single unflagged frame. + +This is purely a framing change — the payload format is identical, and a publish +cycle may now emit several binary frames (one light frame plus one per heavy +message) instead of one. Clients that do not care about the distinction can ignore +the flag and decode every frame the same way (each frame is self-describing via its +`message_count`). Clients that do care may use bit 0 to, for example, surface +heavy-topic drop indicators. The flag never affects decoding, so old clients remain +compatible and `protocol_version` is unchanged. diff --git a/tests/unit/test_bridge_server.cpp b/tests/unit/test_bridge_server.cpp index 87cf713..c624d68 100644 --- a/tests/unit/test_bridge_server.cpp +++ b/tests/unit/test_bridge_server.cpp @@ -86,6 +86,15 @@ class MockMiddleware : public MiddlewareInterface { } bool send_binary(const std::string& client_identity, const std::vector& data) override { + // Test seam: simulate a client vanishing right before a heavy frame is + // delivered (send returns false), so partial-send stats can be exercised. + if (fail_heavy_sends_ && data.size() >= 16) { + uint32_t flags = 0; + std::memcpy(&flags, data.data() + 12, sizeof(flags)); + if ((flags & kFrameFlagHeavy) != 0) { + return false; + } + } log_send(SendKind::kBinary, client_identity); std::lock_guard lock(binary_mutex_); binary_sends_.emplace_back(client_identity, data); @@ -175,6 +184,12 @@ class MockMiddleware : public MiddlewareInterface { binary_sends_.clear(); } + /// When enabled, send_binary() returns false 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; + } + /// Return all replies sent to a given client, in send order (parsed JSON). std::vector get_replies(const std::string& client_id) { std::lock_guard lock(reply_mutex_); @@ -226,6 +241,7 @@ class MockMiddleware : public MiddlewareInterface { std::mutex binary_mutex_; std::vector>> binary_sends_; + bool fail_heavy_sends_{false}; std::mutex send_log_mutex_; std::vector> send_log_; @@ -3206,3 +3222,144 @@ TEST_F(BridgeServerTest, HeavyFrameThresholdZeroKeepsSingleFrame) { 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_, 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_, 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(); + ASSERT_EQ(sends.size(), 2u); + bool heavy_is_at = false; + bool light_is_below = false; + for (const auto& [cid, frame] : sends) { + auto msgs = decode_frame(frame); + ASSERT_EQ(msgs.size(), 1u); + if ((frame_flags(frame) & kFrameFlagHeavy) != 0) { + 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_, 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(); + ASSERT_EQ(sends.size(), 1u) << "no light frame when every topic is heavy"; + EXPECT_NE(frame_flags(sends[0].second) & kFrameFlagHeavy, 0u); + 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_, 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"); +} From 5b5f5ca748b4ab959e6e58c4084c7f4c8a516e4f Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Wed, 8 Jul 2026 12:46:12 +0200 Subject: [PATCH 3/8] feat(middleware): shed heavy frames before transmit + fix flush-recheck bug MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 (part 2) of the backpressure fix. The send path now distinguishes frame priority so a large frame can't starve small frames sharing the socket. Extracts the admit/flush/queue/shed decision into a socket-agnostic policy (run_backpressure, backpressure.hpp) that the caller drives with injected buffered_amount/send/pop/queue primitives — making it unit-testable without a real connection, which the existing middleware tests could not do. Behavior changes: - FramePriority{kNormal,kHeavy} added to MiddlewareInterface::send_binary (defaulted, so all existing callers are unchanged); BridgeServer tags heavy size-class frames kHeavy. - kHeavy frames are SHED (dropped before transmit) when the socket is over the watermark instead of queued — a queued 5 MB cloud is stale by flush time and just refills the buffer. Because heavy frames are never queued, the per-client backlog now only holds small frames, dissolving the old ~500 MB worst case. - Flush-recheck bug fixed: the flush loop re-reads bufferedAmount() each iteration and stops once congested, instead of dumping a burst of stale frames onto an already-full socket (computed the count once before). - New heavy_shed_count() observability counter, distinct from dropped_frame_count(). Tests: test_backpressure.cpp drives the policy with a fake socket — flush stops at the watermark mid-flush (red-first, reproduced the bug), heavy shed vs normal queue when congested, send-now with room, send-failure. Plus a bridge-level check that heavy frames are sent with kHeavy priority (258 tests pass). Co-Authored-By: Claude Opus 4.8 --- CMakeLists.txt | 1 + .../pj_bridge/middleware/backpressure.hpp | 103 ++++++++++++++ .../middleware/middleware_interface.hpp | 15 +- .../middleware/websocket_middleware.hpp | 14 +- app/src/bridge_server.cpp | 3 +- app/src/middleware/websocket_middleware.cpp | 134 ++++++++---------- tests/unit/test_backpressure.cpp | 130 +++++++++++++++++ tests/unit/test_bridge_server.cpp | 47 +++++- 8 files changed, 366 insertions(+), 81 deletions(-) create mode 100644 app/include/pj_bridge/middleware/backpressure.hpp create mode 100644 tests/unit/test_backpressure.cpp 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/app/include/pj_bridge/middleware/backpressure.hpp b/app/include/pj_bridge/middleware/backpressure.hpp new file mode 100644 index 0000000..7fa016a --- /dev/null +++ b/app/include/pj_bridge/middleware/backpressure.hpp @@ -0,0 +1,103 @@ +/* + * 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 + +#include "pj_bridge/middleware/middleware_interface.hpp" + +namespace pj_bridge { + +/// Outcome of run_backpressure() for one outgoing frame. +struct SendOutcome { + bool accepted = false; ///< value send_binary() should return to its caller + size_t frames_flushed = 0; ///< queued frames flushed to the socket this call + bool shed_heavy = false; ///< a kHeavy frame was dropped before transmit + size_t dropped = 0; ///< frames evicted from the backlog on queue overflow + bool send_failed = false; ///< a send() returned false (client gone) +}; + +/// Socket-agnostic backpressure policy shared by the send path. The caller +/// injects the socket/queue primitives so the policy is unit-testable without a +/// real connection: +/// - @p buffered_amount : current outgoing socket buffer size, in bytes +/// - @p pop_pending : remove + return the oldest queued frame (nullopt if empty) +/// - @p send : transmit a frame; returns success (false = client gone) +/// - @p queue_pending : enqueue a frame with drop-oldest overflow; returns the +/// number 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); then handle the current frame — send it if there is room, else drop +/// a kHeavy frame before transmit (shed) or enqueue a kNormal frame. +inline SendOutcome run_backpressure( + FramePriority priority, const std::vector& frame, size_t watermark, + const std::function& buffered_amount, + const std::function>()>& pop_pending, + const std::function&)>& send, + const std::function(const std::vector&)>& 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). + while (buffered_amount() < watermark) { + std::optional> queued = pop_pending(); + if (!queued) { + break; + } + if (!send(*queued)) { + out.send_failed = true; + out.accepted = false; + return out; + } + out.frames_flushed++; + } + + // Handle the current frame against the live socket buffer. + if (buffered_amount() < watermark) { + out.accepted = send(frame); + out.send_failed = !out.accepted; + return out; + } + + // Socket congested: shed heavy frames before transmit; queue normal frames. + if (priority == FramePriority::kHeavy) { + out.shed_heavy = true; + out.accepted = true; // accepted-for-later semantics, same as a queued normal frame + return out; + } + + std::optional dropped = queue_pending(frame); + if (!dropped) { + out.accepted = false; + out.send_failed = true; + return out; + } + out.dropped = *dropped; + out.accepted = true; + 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..3c42c8a 100644 --- a/app/include/pj_bridge/middleware/middleware_interface.hpp +++ b/app/include/pj_bridge/middleware/middleware_interface.hpp @@ -28,6 +28,12 @@ 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; `kNormal` +/// frames use the queue-with-drop-oldest backlog. See docs/API.md. +enum class FramePriority { kNormal, kHeavy }; + /// Abstract transport layer between BridgeServer and clients. /// /// Implementations handle connection management and bidirectional messaging. @@ -65,8 +71,13 @@ 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 true if the message was sent or accepted for later delivery, + /// false if the client is gone. + virtual bool 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..83037d6 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" @@ -61,7 +62,9 @@ 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; + bool send_binary( + const std::string& client_identity, const std::vector& data, + FramePriority priority = FramePriority::kNormal) override; bool is_ready() const override; void set_on_connect(ConnectionCallback callback) override; void set_on_disconnect(ConnectionCallback callback) override; @@ -71,6 +74,10 @@ 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; + private: struct IncomingRequest { std::string client_id; @@ -98,6 +105,11 @@ 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_; std::optional tls_; diff --git a/app/src/bridge_server.cpp b/app/src/bridge_server.cpp index c9c3e59..b0b5aa6 100644 --- a/app/src/bridge_server.cpp +++ b/app/src/bridge_server.cpp @@ -1171,7 +1171,8 @@ 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; + if (middleware_->send_binary(client_id, frame.compressed_data, priority)) { any_sent = true; } else { spdlog::debug("Failed to send binary frame to client '{}'", client_id); diff --git a/app/src/middleware/websocket_middleware.cpp b/app/src/middleware/websocket_middleware.cpp index 2e920dc..a039de8 100644 --- a/app/src/middleware/websocket_middleware.cpp +++ b/app/src/middleware/websocket_middleware.cpp @@ -296,22 +296,22 @@ 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) { +bool 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; { std::lock_guard lock(clients_mutex_); @@ -322,75 +322,50 @@ bool WebSocketMiddleware::send_binary(const std::string& client_identity, const ws = it->second; } - 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 returns false. + 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, kSocketBufferHighWatermark, buffered_amount, pop_pending, send, queue_pending); + + if (outcome.shed_heavy) { + std::lock_guard lock(clients_mutex_); + 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). + if (outcome.dropped > 0 || outcome.shed_heavy) { 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 +376,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.shed_heavy) { + 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.accepted; } void WebSocketMiddleware::drop_pending(const std::string& client_identity) { @@ -434,6 +411,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/tests/unit/test_backpressure.cpp b/tests/unit/test_backpressure.cpp new file mode 100644 index 0000000..f3e5655 --- /dev/null +++ b/tests/unit/test_backpressure.cpp @@ -0,0 +1,130 @@ +/* + * 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_, buffered_amount(), pop_pending(), send(), queue_pending()); + } +}; + +TEST_F(BackpressureTest, SendsImmediatelyWhenSocketHasRoom) { + auto out = run(FramePriority::kNormal, frame_of(100)); + EXPECT_TRUE(out.accepted); + 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(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_TRUE(out.shed_heavy); + EXPECT_TRUE(out.accepted); + 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_TRUE(out.accepted); + EXPECT_FALSE(out.shed_heavy); + 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_, buffered_amount(), pop_pending(), + [](const std::vector&) { return false; }, // send fails (client gone) + queue_pending()); + EXPECT_FALSE(out.accepted); + EXPECT_TRUE(out.send_failed); +} diff --git a/tests/unit/test_bridge_server.cpp b/tests/unit/test_bridge_server.cpp index c624d68..c476b64 100644 --- a/tests/unit/test_bridge_server.cpp +++ b/tests/unit/test_bridge_server.cpp @@ -85,7 +85,9 @@ class MockMiddleware : public MiddlewareInterface { return true; } - bool send_binary(const std::string& client_identity, const std::vector& data) override { + bool send_binary( + const std::string& client_identity, const std::vector& data, + FramePriority priority = FramePriority::kNormal) override { // Test seam: simulate a client vanishing right before a heavy frame is // delivered (send returns false), so partial-send stats can be exercised. if (fail_heavy_sends_ && data.size() >= 16) { @@ -98,6 +100,7 @@ class MockMiddleware : public MiddlewareInterface { log_send(SendKind::kBinary, client_identity); std::lock_guard lock(binary_mutex_); binary_sends_.emplace_back(client_identity, data); + binary_priorities_.push_back(priority); return true; } @@ -182,6 +185,13 @@ 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 false for frames marked kFrameFlagHeavy @@ -241,6 +251,7 @@ class MockMiddleware : public MiddlewareInterface { std::mutex binary_mutex_; std::vector>> binary_sends_; + std::vector binary_priorities_; bool fail_heavy_sends_{false}; std::mutex send_log_mutex_; @@ -3363,3 +3374,37 @@ TEST_F(BridgeServerTest, LargeLatchedSampleReplayedUnflagged) { 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_, 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); + for (size_t i = 0; i < sends.size(); ++i) { + const bool heavy_frame = (frame_flags(sends[i].second) & kFrameFlagHeavy) != 0; + EXPECT_EQ(priorities[i], heavy_frame ? FramePriority::kHeavy : FramePriority::kNormal); + } +} From 5c9dee8b76d155a503cbca32ca62c119d2f64e5c Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Wed, 8 Jul 2026 12:52:20 +0200 Subject: [PATCH 4/8] fix(middleware): bound flush + recheck liveness on heavy shed (Codex review) Addresses the Codex review of the shed-before-transmit change: - Unbounded-flush regression: the new flush loop dropped the original guard that bounded flushing to the backlog size at call start, so under concurrent producers a single send_binary() could keep flushing newly-enqueued frames. run_backpressure() now takes max_flush (the backlog size snapshotted under the lock) and stops after that many, in addition to the watermark recheck. - Heavy-shed disconnect race: the shed path returned accepted=true without re-checking client liveness, unlike the queue path (which returns false via queue_pending -> nullopt). send_binary() now re-checks clients_ under the lock before counting a shed and returns false if the client vanished. Tests: flush bounded by max_flush, full-drain-then-send-current, heavy frame sent when there is room (not shed), queue-client-gone returns not-accepted (262 tests pass). Co-Authored-By: Claude Opus 4.8 --- .../pj_bridge/middleware/backpressure.hpp | 16 ++++-- app/src/middleware/websocket_middleware.cpp | 17 ++++++- tests/unit/test_backpressure.cpp | 51 ++++++++++++++++++- 3 files changed, 75 insertions(+), 9 deletions(-) diff --git a/app/include/pj_bridge/middleware/backpressure.hpp b/app/include/pj_bridge/middleware/backpressure.hpp index 7fa016a..9c2a331 100644 --- a/app/include/pj_bridge/middleware/backpressure.hpp +++ b/app/include/pj_bridge/middleware/backpressure.hpp @@ -48,10 +48,15 @@ struct SendOutcome { /// /// 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); then handle the current frame — send it if there is room, else drop -/// a kHeavy frame before transmit (shed) or enqueue a kNormal frame. +/// 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, else drop a kHeavy frame before transmit +/// (shed) or enqueue a kNormal frame. +/// +/// @param max_flush upper bound on frames flushed this call — pass the backlog +/// size observed at call start. inline SendOutcome run_backpressure( - FramePriority priority, const std::vector& frame, size_t watermark, + FramePriority priority, const std::vector& frame, size_t watermark, size_t max_flush, const std::function& buffered_amount, const std::function>()>& pop_pending, const std::function&)>& send, @@ -61,8 +66,9 @@ inline SendOutcome run_backpressure( // 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). - while (buffered_amount() < watermark) { + // 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; diff --git a/app/src/middleware/websocket_middleware.cpp b/app/src/middleware/websocket_middleware.cpp index a039de8..c867b1c 100644 --- a/app/src/middleware/websocket_middleware.cpp +++ b/app/src/middleware/websocket_middleware.cpp @@ -313,6 +313,7 @@ bool WebSocketMiddleware::send_binary( // 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); @@ -320,6 +321,12 @@ bool WebSocketMiddleware::send_binary( return false; } 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(); + } } auto buffered_amount = [&]() -> size_t { return ws->bufferedAmount(); }; @@ -353,11 +360,17 @@ bool WebSocketMiddleware::send_binary( return pending_it->second.push(frame); }; - SendOutcome outcome = - run_backpressure(priority, data, kSocketBufferHighWatermark, buffered_amount, pop_pending, send, queue_pending); + SendOutcome outcome = run_backpressure( + priority, data, kSocketBufferHighWatermark, max_flush, buffered_amount, pop_pending, send, queue_pending); if (outcome.shed_heavy) { 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 + // the client gone (false) rather than counting a shed for a dead client. + if (clients_.find(client_identity) == clients_.end()) { + return false; + } heavy_shed_total_++; } diff --git a/tests/unit/test_backpressure.cpp b/tests/unit/test_backpressure.cpp index f3e5655..ac8116d 100644 --- a/tests/unit/test_backpressure.cpp +++ b/tests/unit/test_backpressure.cpp @@ -76,7 +76,8 @@ class BackpressureTest : public ::testing::Test { }; } SendOutcome run(FramePriority prio, const std::vector& frame) { - return run_backpressure(prio, frame, watermark_, buffered_amount(), pop_pending(), send(), queue_pending()); + return run_backpressure( + prio, frame, watermark_, backlog_.size(), buffered_amount(), pop_pending(), send(), queue_pending()); } }; @@ -122,9 +123,55 @@ TEST_F(BackpressureTest, NormalFrameQueuedWhenCongested) { TEST_F(BackpressureTest, SendFailureDuringFlushIsNotAccepted) { backlog_.push_back(frame_of(100)); auto out = run_backpressure( - FramePriority::kNormal, frame_of(100), watermark_, buffered_amount(), pop_pending(), + 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_FALSE(out.accepted); EXPECT_TRUE(out.send_failed); } + +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_TRUE(out.accepted); + 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_TRUE(out.accepted); + EXPECT_FALSE(out.shed_heavy); + 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_FALSE(out.accepted); + EXPECT_TRUE(out.send_failed); +} + +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); +} From a529e1018b27ec57dcb4442f4ab0ff3a1f4f6e0e Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Wed, 8 Jul 2026 12:56:49 +0200 Subject: [PATCH 5/8] feat(config): expose heavy_frame_threshold_bytes on all three backends Completes Phase 1 by making the size-class split threshold configurable per backend, following the existing client_backlog_size plumbing pattern: - ROS2: `heavy_frame_threshold_bytes` int parameter (default 262144, must be >= 0; validated at startup). - FastDDS / RTI: `--heavy-frame-threshold-bytes` CLI11 option (default 262144, range 0..1e9). Threaded into the BridgeServer constructor's existing parameter. 0 disables splitting (legacy single-frame behavior). Docs updated: docs/API.md gains a config subsection under backpressure, and CLAUDE.md's config listings include the new option. Only the ROS2 backend is built here (FastDDS/RTI are disabled), so their mains mirror the ROS2/client_backlog_size pattern exactly. 262 tests pass. Co-Authored-By: Claude Opus 4.8 --- CLAUDE.md | 5 +++-- docs/API.md | 17 +++++++++++++++++ fastdds/src/main.cpp | 12 ++++++++++-- ros2/src/main.cpp | 15 +++++++++++++-- rti/src/main.cpp | 12 ++++++++++-- 5 files changed, 53 insertions(+), 8 deletions(-) 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/docs/API.md b/docs/API.md index bc505e6..b32cd77 100644 --- a/docs/API.md +++ b/docs/API.md @@ -547,6 +547,23 @@ configurable: - **FastDDS / RTI**: CLI flag `--client-backlog-size`, default `100`, valid range `1`-`1000000`. +Under congestion this backlog only ever holds small frames: heavy +(size-class) frames are shed before transmit rather than queued (see +[Size-class frames](#size-class-frames-heavy-flag)), so a single large frame +cannot fill the backlog and evict small-topic frames. + +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://`) diff --git a/fastdds/src/main.cpp b/fastdds/src/main.cpp index 3e04920..a9365ea 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); @@ -110,8 +118,8 @@ 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..9ffbce9 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,15 @@ 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; + } + 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 +135,8 @@ 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"); diff --git a/rti/src/main.cpp b/rti/src/main.cpp index 396fdd8..a17d5ab 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); @@ -115,8 +123,8 @@ 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}); From 641f80902492ff91b2ecf9bdec5ce56f2f5293df Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Wed, 8 Jul 2026 13:00:09 +0200 Subject: [PATCH 6/8] docs: qualify backlog claim re: latched replay (Codex review) The backpressure section claimed the backlog "only ever holds small frames". That is true for the aggregated publish stream (heavy frames are shed, not queued) but not for one-shot latched-replay frames, which are delivered at normal priority so a late subscriber reliably receives the retained sample and can briefly occupy the backlog. Reworded to say so. Co-Authored-By: Claude Opus 4.8 --- docs/API.md | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/docs/API.md b/docs/API.md index b32cd77..88523c9 100644 --- a/docs/API.md +++ b/docs/API.md @@ -547,10 +547,13 @@ configurable: - **FastDDS / RTI**: CLI flag `--client-backlog-size`, default `100`, valid range `1`-`1000000`. -Under congestion this backlog only ever holds small frames: heavy -(size-class) frames are shed before transmit rather than queued (see -[Size-class frames](#size-class-frames-heavy-flag)), so a single large frame -cannot fill the backlog and evict small-topic frames. +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-heavy-flag)), so a continuous stream of +large frames cannot fill the backlog and evict small-topic frames. (One-shot +latched-replay frames are the exception: they are delivered reliably at normal +priority so a late subscriber always receives the retained sample, and may +therefore briefly occupy the backlog — but 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: From 91078e943d4da04e6fb739106e458b31596f8c1d Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Wed, 8 Jul 2026 19:13:07 +0200 Subject: [PATCH 7/8] refactor: address full PR review (SendResult, config struct, stats, tests) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the findings from the multi-agent + Codex review of PR #9. Correctness (the one real bug — shed frames counted as delivered throughput): - send_binary() now returns SendResult{kDelivered,kQueued,kShed,kClientGone} instead of bool. BridgeServer counts a frame as forwarded only when delivered or queued, so a shed heavy frame no longer inflates total_bytes_published_ / topic_forward_counts_ ("Sent MB/s" no longer over-reports under congestion). - heavy_shed_count() is now surfaced in the periodic (FastDDS/RTI) and final (ROS2) operator stats lines, next to dropped_frame_count(). Type/design refactors: - SendOutcome's flag struct -> a SendResult discriminant + counters, removing the dead/redundant send_failed field and making the contradictory flag combinations unrepresentable. - run_backpressure() is templated on its callbacks (drops std::function type-erasure on the per-frame send hot path). - BridgeServer's 8-arg constructor -> a BridgeServerConfig aggregate; all 17 call sites (3 mains + 14 tests) migrated. - FramePriority default argument now lives only on the base interface declaration (not the override), avoiding silent static-default divergence. Test coverage (closes the reviewer-identified gaps): - rate-limited heavy topic (rate gate x split), multi-client group fan-out with once-per-group stats, multiple heavy topics each isolated, shed-heavy frame not counted in stats, and flush-bound-survives-concurrent-refill. - A socket_buffer_watermark constructor seam (default kSocketBufferHighWatermark) makes the over-watermark path testable: a new test drives a real client with watermark=0 and asserts heavy_shed_count() increments on shed. Docs/robustness: - Reconcile the latched-replay backpressure wording (normal priority, never shed, but can be dropped only on sustained backlog overflow). - README config tables gain heavy_frame_threshold_bytes / --heavy-frame-threshold-bytes. - All three entry points warn when the threshold is set >= the socket watermark (kSocketBufferHighWatermark is now public for that check). - docs/API.md: "at most one light frame" (none when a group is all-heavy). 268 tests pass in Release, TSAN, and ASAN. Co-Authored-By: Claude Opus 4.8 --- README.md | 2 + app/include/pj_bridge/bridge_server.hpp | 28 +- .../pj_bridge/middleware/backpressure.hpp | 56 ++-- .../middleware/middleware_interface.hpp | 22 +- .../middleware/websocket_middleware.hpp | 30 ++- app/src/bridge_server.cpp | 25 +- app/src/middleware/websocket_middleware.cpp | 29 ++- app/src/standalone_event_loop.cpp | 1 + docs/API.md | 12 +- fastdds/src/main.cpp | 13 +- ros2/src/main.cpp | 19 +- rti/src/main.cpp | 13 +- tests/unit/test_backpressure.cpp | 41 ++- tests/unit/test_bridge_server.cpp | 243 +++++++++++++++--- tests/unit/test_websocket_middleware.cpp | 52 +++- 15 files changed, 443 insertions(+), 143 deletions(-) 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 5dd3168..7ce7f42 100644 --- a/app/include/pj_bridge/bridge_server.hpp +++ b/app/include/pj_bridge/bridge_server.hpp @@ -48,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 { @@ -62,21 +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 heavy_frame_threshold_bytes 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. + * @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 = {}, - size_t heavy_frame_threshold_bytes = kDefaultHeavyFrameThresholdBytes); + std::shared_ptr middleware, BridgeServerConfig config = {}); /// Shuts down middleware before members are destroyed, preventing /// disconnect callbacks from firing into a partially destroyed object. diff --git a/app/include/pj_bridge/middleware/backpressure.hpp b/app/include/pj_bridge/middleware/backpressure.hpp index 9c2a331..1c93679 100644 --- a/app/include/pj_bridge/middleware/backpressure.hpp +++ b/app/include/pj_bridge/middleware/backpressure.hpp @@ -20,7 +20,6 @@ #pragma once #include -#include #include #include @@ -28,39 +27,40 @@ namespace pj_bridge { -/// Outcome of run_backpressure() for one outgoing frame. +/// 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 { - bool accepted = false; ///< value send_binary() should return to its caller - size_t frames_flushed = 0; ///< queued frames flushed to the socket this call - bool shed_heavy = false; ///< a kHeavy frame was dropped before transmit - size_t dropped = 0; ///< frames evicted from the backlog on queue overflow - bool send_failed = false; ///< a send() returned false (client gone) + 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 so the policy is unit-testable without a -/// real connection: -/// - @p buffered_amount : current outgoing socket buffer size, in bytes -/// - @p pop_pending : remove + return the oldest queued frame (nullopt if empty) -/// - @p send : transmit a frame; returns success (false = client gone) -/// - @p queue_pending : enqueue a frame with drop-oldest overflow; returns the -/// number dropped, or nullopt if the client is gone +/// 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, else drop a kHeavy frame before transmit -/// (shed) or enqueue a kNormal frame. +/// 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. -inline SendOutcome run_backpressure( +template +SendOutcome run_backpressure( FramePriority priority, const std::vector& frame, size_t watermark, size_t max_flush, - const std::function& buffered_amount, - const std::function>()>& pop_pending, - const std::function&)>& send, - const std::function(const std::vector&)>& queue_pending) { + 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 @@ -74,8 +74,7 @@ inline SendOutcome run_backpressure( break; } if (!send(*queued)) { - out.send_failed = true; - out.accepted = false; + out.result = SendResult::kClientGone; return out; } out.frames_flushed++; @@ -83,26 +82,23 @@ inline SendOutcome run_backpressure( // Handle the current frame against the live socket buffer. if (buffered_amount() < watermark) { - out.accepted = send(frame); - out.send_failed = !out.accepted; + 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.shed_heavy = true; - out.accepted = true; // accepted-for-later semantics, same as a queued normal frame + out.result = SendResult::kShed; return out; } std::optional dropped = queue_pending(frame); if (!dropped) { - out.accepted = false; - out.send_failed = true; + out.result = SendResult::kClientGone; return out; } out.dropped = *dropped; - out.accepted = true; + out.result = SendResult::kQueued; return out; } diff --git a/app/include/pj_bridge/middleware/middleware_interface.hpp b/app/include/pj_bridge/middleware/middleware_interface.hpp index 3c42c8a..d3422a1 100644 --- a/app/include/pj_bridge/middleware/middleware_interface.hpp +++ b/app/include/pj_bridge/middleware/middleware_interface.hpp @@ -30,10 +30,21 @@ 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; `kNormal` -/// frames use the queue-with-drop-oldest backlog. See docs/API.md. +/// 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. @@ -73,9 +84,10 @@ class MiddlewareInterface { /// Send binary data to a specific client (used for per-client aggregated frames). /// @param priority kHeavy frames are shed before transmit under congestion /// instead of queued (default kNormal preserves the legacy behavior). - /// @return true if the message was sent or accepted for later delivery, - /// false if the client is gone. - virtual bool send_binary( + /// @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; diff --git a/app/include/pj_bridge/middleware/websocket_middleware.hpp b/app/include/pj_bridge/middleware/websocket_middleware.hpp index 83037d6..550f954 100644 --- a/app/include/pj_bridge/middleware/websocket_middleware.hpp +++ b/app/include/pj_bridge/middleware/websocket_middleware.hpp @@ -49,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; @@ -62,9 +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, - FramePriority priority = FramePriority::kNormal) 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; @@ -78,6 +86,13 @@ class WebSocketMiddleware : public MiddlewareInterface { /// (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; @@ -111,6 +126,7 @@ class WebSocketMiddleware : public MiddlewareInterface { uint64_t heavy_shed_total_{0}; size_t client_backlog_size_; + size_t socket_buffer_watermark_; std::optional tls_; ConnectionCallback on_connect_; @@ -123,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/src/bridge_server.cpp b/app/src/bridge_server.cpp index b0b5aa6..bda2ff4 100644 --- a/app/src/bridge_server.cpp +++ b/app/src/bridge_server.cpp @@ -60,16 +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, - size_t heavy_frame_threshold_bytes) + 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)), - heavy_frame_threshold_bytes_(heavy_frame_threshold_bytes), + 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), @@ -1090,7 +1089,8 @@ void BridgeServer::publish_aggregated_messages() { 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 frame per heavy message (mirrors collect_latched_replay). + // One single-message frame per heavy message (same one-message-per-frame + // shape as collect_latched_replay's retained-sample frame). AggregatedMessageSerializer heavy_serializer; heavy_serializer.serialize_message(topic, msg.timestamp_ns, msg.data->data(), msg.data->size()); GroupFrame heavy_frame; @@ -1172,9 +1172,14 @@ void BridgeServer::publish_aggregated_messages() { continue; } const FramePriority priority = frame.is_heavy ? FramePriority::kHeavy : FramePriority::kNormal; - if (middleware_->send_binary(client_id, frame.compressed_data, priority)) { + 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); } } diff --git a/app/src/middleware/websocket_middleware.cpp b/app/src/middleware/websocket_middleware.cpp index c867b1c..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,7 +300,7 @@ bool WebSocketMiddleware::send_reply(const std::string& client_identity, const s return send_info.success; } -bool WebSocketMiddleware::send_binary( +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 @@ -318,7 +322,7 @@ bool WebSocketMiddleware::send_binary( 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, @@ -349,7 +353,7 @@ bool WebSocketMiddleware::send_binary( // 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 returns false. + // 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_); if (clients_.find(client_identity) == clients_.end()) { @@ -361,22 +365,23 @@ bool WebSocketMiddleware::send_binary( }; SendOutcome outcome = run_backpressure( - priority, data, kSocketBufferHighWatermark, max_flush, buffered_amount, pop_pending, send, queue_pending); + priority, data, socket_buffer_watermark_, max_flush, buffered_amount, pop_pending, send, queue_pending); - if (outcome.shed_heavy) { + 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 - // the client gone (false) rather than counting a shed for a dead client. + // it gone rather than counting a shed for a dead client. if (clients_.find(client_identity) == clients_.end()) { - return false; + return SendResult::kClientGone; } heavy_shed_total_++; } // Throttled slow-client warning, shared by the queue-drop and heavy-shed // paths (at most one line per client per kDropWarnIntervalSeconds). - if (outcome.dropped > 0 || outcome.shed_heavy) { + 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 = false; { @@ -389,7 +394,7 @@ bool WebSocketMiddleware::send_binary( } } if (should_warn) { - if (outcome.shed_heavy) { + 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); @@ -397,7 +402,7 @@ bool WebSocketMiddleware::send_binary( } } - return outcome.accepted; + return outcome.result; } void WebSocketMiddleware::drop_pending(const std::string& client_identity) { 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 88523c9..b2a5594 100644 --- a/docs/API.md +++ b/docs/API.md @@ -551,9 +551,10 @@ 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-heavy-flag)), so a continuous stream of large frames cannot fill the backlog and evict small-topic frames. (One-shot -latched-replay frames are the exception: they are delivered reliably at normal -priority so a late subscriber always receives the retained sample, and may -therefore briefly occupy the backlog — but they are not a continuous stream.) +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: @@ -649,8 +650,9 @@ its own **heavy** frame with header flag bit 0 (`0x1`) set, while smaller messag stay aggregated in a single unflagged frame. This is purely a framing change — the payload format is identical, and a publish -cycle may now emit several binary frames (one light frame plus one per heavy -message) instead of one. Clients that do not care about the distinction can ignore +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. Clients that do not care about the distinction can ignore the flag and decode every frame the same way (each frame is self-describing via its `message_count`). Clients that do care may use bit 0 to, for example, surface heavy-topic drop indicators. The flag never affects decoding, so old clients remain diff --git a/fastdds/src/main.cpp b/fastdds/src/main.cpp index a9365ea..bbd6b20 100644 --- a/fastdds/src/main.cpp +++ b/fastdds/src/main.cpp @@ -107,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); @@ -118,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()), - static_cast(heavy_frame_threshold_bytes)); + 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 9ffbce9..2f5b71b 100644 --- a/ros2/src/main.cpp +++ b/ros2/src/main.cpp @@ -102,6 +102,14 @@ int main(int argc, char** argv) { 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); @@ -135,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()), - static_cast(heavy_frame_threshold_bytes)); + 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"); @@ -197,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 a17d5ab..f419961 100644 --- a/rti/src/main.cpp +++ b/rti/src/main.cpp @@ -112,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); @@ -123,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()), - static_cast(heavy_frame_threshold_bytes)); + 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 index ac8116d..2928938 100644 --- a/tests/unit/test_backpressure.cpp +++ b/tests/unit/test_backpressure.cpp @@ -83,7 +83,7 @@ class BackpressureTest : public ::testing::Test { TEST_F(BackpressureTest, SendsImmediatelyWhenSocketHasRoom) { auto out = run(FramePriority::kNormal, frame_of(100)); - EXPECT_TRUE(out.accepted); + EXPECT_EQ(out.result, SendResult::kDelivered); EXPECT_EQ(out.frames_flushed, 0u); ASSERT_EQ(sent_.size(), 1u); EXPECT_EQ(sent_[0].size(), 100u); @@ -97,6 +97,7 @@ TEST_F(BackpressureTest, FlushStopsWhenWatermarkReachedMidFlush) { } 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); @@ -105,8 +106,7 @@ TEST_F(BackpressureTest, FlushStopsWhenWatermarkReachedMidFlush) { TEST_F(BackpressureTest, HeavyFrameShedWhenCongested) { buffered_ = watermark_; // already congested auto out = run(FramePriority::kHeavy, frame_of(5000)); - EXPECT_TRUE(out.shed_heavy); - EXPECT_TRUE(out.accepted); + EXPECT_EQ(out.result, SendResult::kShed); EXPECT_TRUE(sent_.empty()); EXPECT_TRUE(backlog_.empty()) << "a heavy frame must be shed, never queued"; } @@ -114,8 +114,7 @@ TEST_F(BackpressureTest, HeavyFrameShedWhenCongested) { TEST_F(BackpressureTest, NormalFrameQueuedWhenCongested) { buffered_ = watermark_; // congested auto out = run(FramePriority::kNormal, frame_of(100)); - EXPECT_TRUE(out.accepted); - EXPECT_FALSE(out.shed_heavy); + EXPECT_EQ(out.result, SendResult::kQueued); EXPECT_TRUE(sent_.empty()); ASSERT_EQ(backlog_.size(), 1u); } @@ -126,8 +125,7 @@ TEST_F(BackpressureTest, SendFailureDuringFlushIsNotAccepted) { 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_FALSE(out.accepted); - EXPECT_TRUE(out.send_failed); + EXPECT_EQ(out.result, SendResult::kClientGone); } TEST_F(BackpressureTest, FlushDrainsFullyThenSendsCurrentWhenRoom) { @@ -137,7 +135,7 @@ TEST_F(BackpressureTest, FlushDrainsFullyThenSendsCurrentWhenRoom) { backlog_.push_back(frame_of(10)); } auto out = run(FramePriority::kNormal, frame_of(10)); - EXPECT_TRUE(out.accepted); + EXPECT_EQ(out.result, SendResult::kDelivered); EXPECT_EQ(out.frames_flushed, 3u); EXPECT_TRUE(backlog_.empty()); EXPECT_EQ(sent_.size(), 4u); // 3 flushed + current @@ -146,8 +144,7 @@ TEST_F(BackpressureTest, FlushDrainsFullyThenSendsCurrentWhenRoom) { 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_TRUE(out.accepted); - EXPECT_FALSE(out.shed_heavy); + EXPECT_EQ(out.result, SendResult::kDelivered); ASSERT_EQ(sent_.size(), 1u); EXPECT_EQ(sent_[0].size(), 5000u); } @@ -159,8 +156,7 @@ TEST_F(BackpressureTest, QueueClientGoneReturnsNotAccepted) { 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_FALSE(out.accepted); - EXPECT_TRUE(out.send_failed); + EXPECT_EQ(out.result, SendResult::kClientGone); } TEST_F(BackpressureTest, FlushBoundedByMaxFlush) { @@ -175,3 +171,24 @@ TEST_F(BackpressureTest, FlushBoundedByMaxFlush) { 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 c476b64..5f6f0ab 100644 --- a/tests/unit/test_bridge_server.cpp +++ b/tests/unit/test_bridge_server.cpp @@ -85,23 +85,32 @@ class MockMiddleware : public MiddlewareInterface { return true; } - bool send_binary( - const std::string& client_identity, const std::vector& data, - FramePriority priority = FramePriority::kNormal) override { - // Test seam: simulate a client vanishing right before a heavy frame is - // delivered (send returns false), so partial-send stats can be exercised. - if (fail_heavy_sends_ && data.size() >= 16) { + SendResult send_binary( + const std::string& client_identity, const std::vector& data, FramePriority priority) override { + bool is_heavy_frame = false; + if (data.size() >= 16) { uint32_t flags = 0; std::memcpy(&flags, data.data() + 12, sizeof(flags)); - if ((flags & kFrameFlagHeavy) != 0) { - return false; - } + is_heavy_frame = (flags & kFrameFlagHeavy) != 0; + } + // 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) 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) { + std::lock_guard lock(binary_mutex_); + binary_priorities_.push_back(priority); + return SendResult::kShed; } log_send(SendKind::kBinary, client_identity); std::lock_guard lock(binary_mutex_); binary_sends_.emplace_back(client_identity, data); binary_priorities_.push_back(priority); - return true; + return SendResult::kDelivered; } bool is_ready() const override { @@ -194,12 +203,20 @@ class MockMiddleware : public MiddlewareInterface { return binary_priorities_; } - /// When enabled, send_binary() returns false for frames marked kFrameFlagHeavy - /// (simulating the client disconnecting before the heavy frame is delivered). + /// 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). std::vector get_replies(const std::string& client_id) { std::lock_guard lock(reply_mutex_); @@ -253,6 +270,7 @@ class MockMiddleware : public MiddlewareInterface { 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_; @@ -504,7 +522,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 { @@ -1995,14 +2014,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()); } @@ -2361,7 +2380,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"}}); @@ -2384,7 +2403,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"}}); @@ -2412,7 +2431,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"}}); @@ -2599,7 +2618,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"}}); @@ -3152,8 +3171,8 @@ 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_, 19999, 10.0, 50.0, WhitelistFilter{}, - /*heavy_frame_threshold_bytes=*/64); + 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"}}); @@ -3207,8 +3226,8 @@ 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_, 19999, 10.0, 50.0, WhitelistFilter{}, - /*heavy_frame_threshold_bytes=*/0); + 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"}}); @@ -3239,8 +3258,8 @@ TEST_F(BridgeServerTest, PartialSendDoesNotOvercountForwardStats) { // 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_, 19999, 10.0, 50.0, WhitelistFilter{}, - /*heavy_frame_threshold_bytes=*/64); + 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"}}); @@ -3270,8 +3289,8 @@ 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_, 19999, 10.0, 50.0, WhitelistFilter{}, - /*heavy_frame_threshold_bytes=*/64); + 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"}}); @@ -3311,8 +3330,8 @@ 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_, 19999, 10.0, 50.0, WhitelistFilter{}, - /*heavy_frame_threshold_bytes=*/64); + 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"}}); @@ -3342,8 +3361,8 @@ TEST_F(BridgeServerTest, LargeLatchedSampleReplayedUnflagged) { // 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_, 19999, 10.0, 50.0, WhitelistFilter{}, - /*heavy_frame_threshold_bytes=*/64); + 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"}}); @@ -3379,8 +3398,8 @@ 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_, 19999, 10.0, 50.0, WhitelistFilter{}, - /*heavy_frame_threshold_bytes=*/64); + 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"}}); @@ -3408,3 +3427,157 @@ TEST_F(BridgeServerTest, HeavyFramesSentWithHeavyPriority) { EXPECT_EQ(priorities[i], heavy_frame ? FramePriority::kHeavy : FramePriority::kNormal); } } + +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(); + ASSERT_EQ(sends.size(), 2u) << "rate gate must admit exactly 2 of the 3 heavy messages"; + for (const auto& [cid, frame] : sends) { + EXPECT_NE(frame_flags(frame) & kFrameFlagHeavy, 0u) << "each admitted heavy message is its own frame"; + auto msgs = decode_frame(frame); + 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(); + ASSERT_EQ(sends.size(), 3u) << "1 light + 2 heavy"; + std::set heavy_topics; + std::set light_topics; + for (const auto& [cid, frame] : sends) { + auto msgs = decode_frame(frame); + if ((frame_flags(frame) & kFrameFlagHeavy) != 0) { + 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_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(); From 6ce5dd325f61b0d00b80cc51e89b4e807a47cb67 Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Wed, 8 Jul 2026 20:40:02 +0200 Subject: [PATCH 8/8] =?UTF-8?q?fix(protocol):=20don't=20wire-flag=20heavy?= =?UTF-8?q?=20frames=20=E2=80=94=20preserve=20plugin=20compatibility?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The size-class split stamped kFrameFlagHeavy (flags bit 0) on heavy frames, but the official PlotJuggler plugin's decoder (data_stream_pj_bridge/ pj_bridge_protocol.cpp) rejects ANY frame with flags != 0 ("must be zero in current protocol version") — so every existing plugin build would silently drop heavy frames, i.e. exactly the large topics (pointclouds) the split protects. Heaviness is not a wire property: the split into separate frames and the shed-before-transmit are driven entirely by the in-memory GroupFrame.is_heavy / FramePriority. So heavy frames now ship with flags == 0 (finalize() instead of finalize(kFrameFlagHeavy)), identical on the wire to any other frame — the plugin accepts them and the isolation/shedding benefit is unchanged. kFrameFlagHeavy and the finalize(flags) API are retained, reserved for a future capability-negotiated rollout. Verified the plugin decoder otherwise tolerates this PR: it ignores the server capability list (so size_class_frames is harmless) and decodes each binary message independently (so multiple frames per publish cycle are fine). protocol_version is unchanged. Tests reclassify light/heavy frames by send priority instead of the wire flag (and assert flags == 0 on every frame as a plugin-compatibility guard). Docs updated: flags field is reserved/always-0, size-class section documents the no-wire-flag decision. 268 tests pass. Co-Authored-By: Claude Opus 4.8 --- app/include/pj_bridge/protocol_constants.hpp | 10 ++-- app/src/bridge_server.cpp | 9 ++- docs/API.md | 29 ++++++---- tests/unit/test_bridge_server.cpp | 61 +++++++++++--------- 4 files changed, 65 insertions(+), 44 deletions(-) diff --git a/app/include/pj_bridge/protocol_constants.hpp b/app/include/pj_bridge/protocol_constants.hpp index 7c0191d..af57f3c 100644 --- a/app/include/pj_bridge/protocol_constants.hpp +++ b/app/include/pj_bridge/protocol_constants.hpp @@ -33,10 +33,12 @@ static constexpr uint32_t kBinaryFrameMagic = 0x42524A50; /// Size of the binary frame header in bytes static constexpr size_t kBinaryHeaderSize = 16; -/// Binary frame header flag bits (offset 12 of the 16-byte header). Bit 0 -/// marks a "heavy" frame carrying an isolated large/size-class message -/// (see docs/API.md); remaining bits are reserved = 0. Old clients ignore -/// this field, so the flag is purely additive. +/// 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 diff --git a/app/src/bridge_server.cpp b/app/src/bridge_server.cpp index bda2ff4..fe3551e 100644 --- a/app/src/bridge_server.cpp +++ b/app/src/bridge_server.cpp @@ -1090,11 +1090,16 @@ void BridgeServer::publish_aggregated_messages() { 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). + // 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(kFrameFlagHeavy); + heavy_frame.compressed_data = heavy_serializer.finalize(); heavy_frame.msg_count = 1; heavy_frame.client_ids = client_ids; heavy_frame.is_heavy = true; diff --git a/docs/API.md b/docs/API.md index b2a5594..60ddb9a 100644 --- a/docs/API.md +++ b/docs/API.md @@ -549,7 +549,7 @@ configurable: 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-heavy-flag)), so a continuous stream of +[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 @@ -622,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 | Bit 0 (`0x1`) = heavy frame (see [Size-class frames](#size-class-frames-heavy-flag)); other bits reserved = 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) @@ -639,21 +639,26 @@ For each message: The magic bytes allow clients to validate frame integrity before decompression. -### Size-class frames (heavy flag) +### 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 **heavy** frame with header flag bit 0 (`0x1`) set, while smaller messages -stay aggregated in a single unflagged frame. +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. Clients that do not care about the distinction can ignore -the flag and decode every frame the same way (each frame is self-describing via its -`message_count`). Clients that do care may use bit 0 to, for example, surface -heavy-topic drop indicators. The flag never affects decoding, so old clients remain -compatible and `protocol_version` is unchanged. +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/tests/unit/test_bridge_server.cpp b/tests/unit/test_bridge_server.cpp index 5f6f0ab..7d0b62e 100644 --- a/tests/unit/test_bridge_server.cpp +++ b/tests/unit/test_bridge_server.cpp @@ -87,25 +87,22 @@ class MockMiddleware : public MiddlewareInterface { SendResult send_binary( const std::string& client_identity, const std::vector& data, FramePriority priority) override { - bool is_heavy_frame = false; - if (data.size() >= 16) { - uint32_t flags = 0; - std::memcpy(&flags, data.data() + 12, sizeof(flags)); - is_heavy_frame = (flags & kFrameFlagHeavy) != 0; - } + // 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) so the bridge's "shed frames aren't counted as - // forwarded" rule can be verified without a real socket. + // (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) { - std::lock_guard lock(binary_mutex_); - binary_priorities_.push_back(priority); 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); @@ -3193,20 +3190,24 @@ TEST_F(BridgeServerTest, PublishSplitsHeavyTopicIntoOwnFrame) { 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 the header heavy flag, not by emission order. + // 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) { - if ((frame_flags(sends[i].second) & kFrameFlagHeavy) != 0) { + 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 frame (flags == 0) found"; - ASSERT_GE(heavy_idx, 0) << "no heavy frame (kFrameFlagHeavy) found"; + 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"); @@ -3310,13 +3311,14 @@ TEST_F(BridgeServerTest, HeavyFrameThresholdBoundaryIsInclusive) { 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 (const auto& [cid, frame] : sends) { - auto msgs = decode_frame(frame); + for (size_t i = 0; i < sends.size(); ++i) { + auto msgs = decode_frame(sends[i].second); ASSERT_EQ(msgs.size(), 1u); - if ((frame_flags(frame) & kFrameFlagHeavy) != 0) { + if (priorities[i] == FramePriority::kHeavy) { heavy_is_at = (msgs[0].topic == "/at"); } else { light_is_below = (msgs[0].topic == "/below"); @@ -3349,8 +3351,9 @@ TEST_F(BridgeServerTest, OnlyHeavyTopicEmitsNoLightFrame) { 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_NE(frame_flags(sends[0].second) & kFrameFlagHeavy, 0u); + EXPECT_EQ(priorities[0], FramePriority::kHeavy); auto msgs = decode_frame(sends[0].second); ASSERT_EQ(msgs.size(), 1u); EXPECT_EQ(msgs[0].topic, "/big"); @@ -3422,9 +3425,13 @@ TEST_F(BridgeServerTest, HeavyFramesSentWithHeavyPriority) { 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) { - const bool heavy_frame = (frame_flags(sends[i].second) & kFrameFlagHeavy) != 0; - EXPECT_EQ(priorities[i], heavy_frame ? FramePriority::kHeavy : FramePriority::kNormal); + 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"); } } @@ -3452,10 +3459,11 @@ TEST_F(BridgeServerTest, RateLimitedHeavyTopicRespectsRateGate) { 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 (const auto& [cid, frame] : sends) { - EXPECT_NE(frame_flags(frame) & kFrameFlagHeavy, 0u) << "each admitted heavy message is its own frame"; - auto msgs = decode_frame(frame); + 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"); } @@ -3530,12 +3538,13 @@ TEST_F(BridgeServerTest, MultipleHeavyTopicsEachGetOwnFrame) { 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 (const auto& [cid, frame] : sends) { - auto msgs = decode_frame(frame); - if ((frame_flags(frame) & kFrameFlagHeavy) != 0) { + 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 {