Isolate heavy topics into size-class frames + shed under backpressure (Phase 1)

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,

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

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 |

app/include/pj_bridge/bridge_server.hpp

@@ -28,6 +28,7 @@
 
 #include "pj_bridge/message_buffer.hpp"
 #include "pj_bridge/middleware/middleware_interface.hpp"
+#include "pj_bridge/protocol_constants.hpp"
 #include "pj_bridge/session_manager.hpp"
 #include "pj_bridge/subscription_manager_interface.hpp"
 #include "pj_bridge/topic_source_interface.hpp"
@@ -47,6 +48,21 @@ namespace pj_bridge {
  * Thread-safe for concurrent client connections.
  * Event loop is driven externally (no internal timers).
  */
+
+/// Tunable configuration for BridgeServer (backend-agnostic). Bundled into one
+/// struct so entry points construct the server with a single named aggregate
+/// rather than a long positional argument list.
+struct BridgeServerConfig {
+  int port = 9090;                 ///< WebSocket port
+  double session_timeout = 10.0;   ///< client session timeout, seconds
+  double publish_rate = 50.0;      ///< message aggregation/publish rate, Hz
+  WhitelistFilter whitelist = {};  ///< topic whitelist (default: matches everything)
+  /// Per-message byte size at or above which a topic's message is isolated into
+  /// its own size-class ("heavy") frame instead of being aggregated with light
+  /// topics. 0 disables splitting (single aggregated frame). Default: 256 KiB.
+  size_t heavy_frame_threshold_bytes = kDefaultHeavyFrameThresholdBytes;
+};
+
 class BridgeServer {
  public:
   struct StatsSnapshot {
@@ -61,16 +77,12 @@ class BridgeServer {
    * @param topic_source Backend-specific topic discovery and schema provider
    * @param subscription_manager Backend-specific subscription manager
    * @param middleware Middleware interface for network communication
-   * @param port Server port (default: 9090)
-   * @param session_timeout Session timeout in seconds (default: 10.0)
-   * @param publish_rate Message aggregation publish rate in Hz (default: 50.0)
-   * @param whitelist Topic whitelist filter (default: matches everything)
+   * @param config Tunable server configuration (see BridgeServerConfig)
    */
   explicit BridgeServer(
       std::shared_ptr<TopicSourceInterface> topic_source,
       std::shared_ptr<SubscriptionManagerInterface> subscription_manager,
-      std::shared_ptr<MiddlewareInterface> middleware, int port = 9090, double session_timeout = 10.0,
-      double publish_rate = 50.0, WhitelistFilter whitelist = {});
+      std::shared_ptr<MiddlewareInterface> middleware, BridgeServerConfig config = {});
 
   /// Shuts down middleware before members are destroyed, preventing
   /// disconnect callbacks from firing into a partially destroyed object.
@@ -211,6 +223,9 @@ class BridgeServer {
   double session_timeout_;
   double publish_rate_;
   WhitelistFilter whitelist_;
+  // Per-message byte size at or above which a topic is isolated into its own
+  // size-class ("heavy") frame; 0 disables splitting. See publish_aggregated_messages().
+  size_t heavy_frame_threshold_bytes_;
 
   // State
   std::atomic<bool> initialized_;

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<uint8_t> finalize();
+  std::vector<uint8_t> finalize(uint32_t flags = 0);
 
   /**
    * @brief Compress data using ZSTD (compression level 1)

app/include/pj_bridge/middleware/backpressure.hpp

@@ -0,0 +1,105 @@
+/*
+ * Copyright (C) 2026 Davide Faconti
+ *
+ * This file is part of pj_bridge.
+ *
+ * pj_bridge is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * pj_bridge is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with pj_bridge. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include <cstddef>
+#include <optional>
+#include <vector>
+
+#include "pj_bridge/middleware/middleware_interface.hpp"
+
+namespace pj_bridge {
+
+/// Outcome of run_backpressure() for one outgoing frame. The disposition is a
+/// single discriminant (no contradictory flag combinations are representable);
+/// the counters are companion data.
+struct SendOutcome {
+  SendResult result = SendResult::kClientGone;  ///< how the current frame was handled
+  size_t frames_flushed = 0;                    ///< backlog frames flushed to the socket this call
+  size_t dropped = 0;                           ///< backlog frames evicted on overflow (only when result == kQueued)
+};
+
+/// Socket-agnostic backpressure policy shared by the send path. The caller
+/// injects the socket/queue primitives (as callables — templated to avoid
+/// std::function type-erasure on the hot path) so the policy is unit-testable
+/// without a real connection:
+///   - @p buffered_amount : `size_t()`                    — current socket buffer bytes
+///   - @p pop_pending     : `std::optional<vector<uint8_t>>()` — pop oldest queued frame (nullopt if empty)
+///   - @p send            : `bool(const vector<uint8_t>&)` — transmit a frame; false = client gone
+///   - @p queue_pending   : `std::optional<size_t>(const vector<uint8_t>&)` — enqueue (drop-oldest),
+///                          returns #dropped, or nullopt if the client is gone
+///
+/// Policy: first flush the backlog while the socket has room (re-checking the
+/// watermark each iteration so an already-congested socket is never fed
+/// further, and never flushing more than @p max_flush frames so a concurrent
+/// producer cannot make one call flush forever); then handle the current
+/// frame — send it if there is room (kDelivered), else drop a kHeavy frame
+/// before transmit (kShed) or enqueue a kNormal frame (kQueued). A vanished
+/// client surfaces as kClientGone.
+///
+/// @param max_flush upper bound on frames flushed this call — pass the backlog
+///        size observed at call start.
+template <class BufferedAmount, class PopPending, class Send, class QueuePending>
+SendOutcome run_backpressure(
+    FramePriority priority, const std::vector<uint8_t>& frame, size_t watermark, size_t max_flush,
+    const BufferedAmount& buffered_amount, const PopPending& pop_pending, const Send& send,
+    const QueuePending& queue_pending) {
+  SendOutcome out;
+
+  // Flush queued frames to the socket, re-checking the watermark each iteration
+  // so a socket that fills up mid-flush is never fed further (the flush-recheck
+  // fix: the old loop computed the flush count once and could dump a burst of
+  // stale frames onto an already-congested socket). Bounded by max_flush so a
+  // producer enqueueing concurrently cannot keep this single call flushing.
+  while (out.frames_flushed < max_flush && buffered_amount() < watermark) {
+    std::optional<std::vector<uint8_t>> queued = pop_pending();
+    if (!queued) {
+      break;
+    }
+    if (!send(*queued)) {
+      out.result = SendResult::kClientGone;
+      return out;
+    }
+    out.frames_flushed++;
+  }
+
+  // Handle the current frame against the live socket buffer.
+  if (buffered_amount() < watermark) {
+    out.result = send(frame) ? SendResult::kDelivered : SendResult::kClientGone;
+    return out;
+  }
+
+  // Socket congested: shed heavy frames before transmit; queue normal frames.
+  if (priority == FramePriority::kHeavy) {
+    out.result = SendResult::kShed;
+    return out;
+  }
+
+  std::optional<size_t> dropped = queue_pending(frame);
+  if (!dropped) {
+    out.result = SendResult::kClientGone;
+    return out;
+  }
+  out.dropped = *dropped;
+  out.result = SendResult::kQueued;
+  return out;
+}
+
+}  // namespace pj_bridge

app/include/pj_bridge/middleware/middleware_interface.hpp

@@ -28,6 +28,23 @@
 
 namespace pj_bridge {
 
+/// Delivery priority for a per-client binary frame. Under socket congestion a
+/// `kHeavy` (large/size-class) frame is dropped before transmit rather than
+/// queued, so one big frame cannot starve the small frames behind it; a
+/// `kNormal` frame instead falls back to the queue-with-drop-oldest backlog.
+/// (When the socket has room, both are sent immediately.) See docs/API.md.
+enum class FramePriority { kNormal, kHeavy };
+
+/// Outcome of send_binary(): how the transport handled the frame. Only
+/// `kDelivered` and `kQueued` are (or will be) put on the wire; `kShed` and
+/// `kClientGone` are never delivered, so callers must NOT count them as sent.
+enum class SendResult {
+  kDelivered,   ///< written to the socket now (or flushed from the backlog)
+  kQueued,      ///< enqueued for later delivery (kNormal frame, socket congested)
+  kShed,        ///< dropped before transmit (kHeavy frame, socket congested)
+  kClientGone,  ///< the client disconnected; nothing was sent
+};
+
 /// Abstract transport layer between BridgeServer and clients.
 ///
 /// Implementations handle connection management and bidirectional messaging.
@@ -65,8 +82,14 @@ class MiddlewareInterface {
   virtual bool publish_data(const std::vector<uint8_t>& 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<uint8_t>& data) = 0;
+  /// @param priority kHeavy frames are shed before transmit under congestion
+  ///        instead of queued (default kNormal preserves the legacy behavior).
+  /// @return how the frame was handled (see SendResult). Callers counting
+  ///         forwarded bytes/messages must treat only kDelivered/kQueued as
+  ///         forwarded — kShed and kClientGone never reach the client.
+  virtual SendResult send_binary(
+      const std::string& client_identity, const std::vector<uint8_t>& 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

app/include/pj_bridge/middleware/websocket_middleware.hpp

@@ -33,6 +33,7 @@
 #include <unordered_map>
 #include <vector>
 
+#include "pj_bridge/middleware/backpressure.hpp"
 #include "pj_bridge/middleware/bounded_frame_queue.hpp"
 #include "pj_bridge/middleware/middleware_interface.hpp"
 
@@ -48,7 +49,13 @@ struct TlsConfig {
 
 class WebSocketMiddleware : public MiddlewareInterface {
  public:
-  explicit WebSocketMiddleware(size_t client_backlog_size = 100, std::optional<TlsConfig> 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<TlsConfig> tls = std::nullopt,
+      size_t socket_buffer_watermark = kSocketBufferHighWatermark);
   ~WebSocketMiddleware() override;
 
   WebSocketMiddleware(const WebSocketMiddleware&) = delete;
@@ -61,7 +68,11 @@ class WebSocketMiddleware : public MiddlewareInterface {
   bool receive_request(std::vector<uint8_t>& data, std::string& client_identity) override;
   bool send_reply(const std::string& client_identity, const std::vector<uint8_t>& data) override;
   bool publish_data(const std::vector<uint8_t>& data) override;
-  bool send_binary(const std::string& client_identity, const std::vector<uint8_t>& data) override;
+  // NOTE: the FramePriority default lives only on the base MiddlewareInterface
+  // declaration (defaults are bound statically, so repeating it here could
+  // silently diverge).
+  SendResult send_binary(
+      const std::string& client_identity, const std::vector<uint8_t>& data, FramePriority priority) override;
   bool is_ready() const override;
   void set_on_connect(ConnectionCallback callback) override;
   void set_on_disconnect(ConnectionCallback callback) override;
@@ -71,6 +82,17 @@ class WebSocketMiddleware : public MiddlewareInterface {
   /// across all clients (currently connected and already disconnected).
   uint64_t dropped_frame_count() const;
 
+  /// Total number of kHeavy frames shed before transmit under congestion
+  /// (dropped instead of queued), summed over the middleware's lifetime.
+  uint64_t heavy_shed_count() const;
+
+  /// Lossy-send policy watermark adapted from foxglove_bridge (MIT License,
+  /// Copyright (c) Foxglove Technologies Inc): once a client's outgoing socket
+  /// buffer reaches this many bytes, further frames are queued (kNormal) or shed
+  /// (kHeavy) instead of blocking or disconnecting the client. Public so entry
+  /// points can sanity-check a configured heavy-frame threshold against it.
+  static constexpr size_t kSocketBufferHighWatermark = 1u << 20;  // 1 MiB
+
  private:
   struct IncomingRequest {
     std::string client_id;
@@ -98,7 +120,13 @@ class WebSocketMiddleware : public MiddlewareInterface {
   // total). Guarded by clients_mutex_.
   uint64_t dropped_from_disconnected_{0};
 
+  // Lifetime count of kHeavy frames shed before transmit under congestion
+  // (dropped rather than queued). Distinct from dropped_frame_count(), which
+  // counts backlog-overflow drops of kNormal frames. Guarded by clients_mutex_.
+  uint64_t heavy_shed_total_{0};
+
   size_t client_backlog_size_;
+  size_t socket_buffer_watermark_;
   std::optional<TlsConfig> tls_;
 
   ConnectionCallback on_connect_;
@@ -111,12 +139,6 @@ class WebSocketMiddleware : public MiddlewareInterface {
 
   static constexpr int kShutdownTimeoutSeconds = 3;
   static constexpr size_t kMaxIncomingQueueSize = 1024;
-
-  // Lossy-send policy adapted from foxglove_bridge (MIT License, Copyright
-  // (c) Foxglove Technologies Inc): once a client's outgoing socket buffer
-  // exceeds this watermark, new frames are queued (dropping the oldest on
-  // overflow) instead of blocking or disconnecting the client.
-  static constexpr size_t kSocketBufferHighWatermark = 1u << 20;  // 1 MiB
   static constexpr int kDropWarnIntervalSeconds = 30;
 };
 

app/include/pj_bridge/protocol_constants.hpp

@@ -33,6 +33,21 @@ static constexpr uint32_t kBinaryFrameMagic = 0x42524A50;
 /// Size of the binary frame header in bytes
 static constexpr size_t kBinaryHeaderSize = 16;
 
+/// Binary frame header flag bit (offset 12 of the 16-byte header) reserved for a
+/// future "heavy" (isolated large/size-class message) marker. NOT currently
+/// emitted: existing PlotJuggler plugins reject any frame with flags != 0, so
+/// heavy frames ship unflagged (flags == 0) and heaviness is conveyed
+/// server-side via FramePriority instead. Reserved here for a future
+/// capability-negotiated rollout (see docs/API.md).
+static constexpr uint32_t kFrameFlagHeavy = 0x1;
+
+/// Default per-message byte threshold at or above which a topic's message is
+/// isolated into its own "heavy" size-class frame instead of being aggregated
+/// with light topics (see docs/API.md). Chosen comfortably below the 1 MiB
+/// socket high-watermark and well above typical scalar/odom/tf frames. A
+/// threshold of 0 disables splitting (single aggregated frame, legacy behavior).
+static constexpr size_t kDefaultHeavyFrameThresholdBytes = 256 * 1024;  // 256 KiB
+
 /// Schema encoding identifier for ROS2 message definitions
 inline constexpr const char* kSchemaEncodingRos2Msg = "ros2msg";
 
@@ -49,6 +64,7 @@ inline constexpr const char* kServerCapabilities[] = {
     "latched_replay",        // retained samples replayed after subscribe/resume
     "topics_changed",        // pushed topic advertisement (subscribe_topic_updates)
     "per_topic_rate_limit",  // subscribe entries accept {name, max_rate_hz}
+    "size_class_frames",     // large topics isolated into own frames (header flag bit0 = heavy)
 };
 
 }  // namespace pj_bridge