TCPStreamPusher: hard backpressure cap so a wedged writer can't hang the run

The peer-liveness timeout only catches a *silent* writer. A misbehaving writer
that keeps sending BUSY heartbeats while never draining (e.g. a permanently
wedged filesystem) would otherwise block SendAll -- and, through it, the queued
SendImage path and the end-of-run frame_transformation_futures.get() -- forever.

Add a progress-based cap in SendAll: if no bytes leave the socket for
max_backpressure_timeout (default 60s, tunable via SetMaxBackpressureTimeout)
the connection is declared dead regardless of heartbeats. It is one global cap,
enforced everywhere SendAll runs, so it bounds both mid-run stalls and
finalization. Generous relative to the 15s liveness window, since a heartbeating
peer is given more grace than a silent one -- but finite.

Add TCPImageCommTest_WedgedWriter_DroppedByBackpressureCap: a writer that ACKs
START then stalls forever while heartbeating (cap 1.5s, liveness 5s) must have
its connection dropped, and neither the producers nor EndDataCollection may
hang. Verified to hang (timeout) with the cap disabled.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

image_pusher/TCPStreamPusher.cpp

@@ -197,16 +197,19 @@ bool TCPStreamPusher::IsConnectionAlive(const Connection& c) const {
 bool TCPStreamPusher::SendAll(Connection& c, const void* buf, size_t len) {
     const auto* p = static_cast<const uint8_t*>(buf);
     size_t sent = 0;
+    int64_t last_progress_ns = SteadyNowNs();
 
     while (sent < len) {
         const int local_fd = c.fd.load();
         if (local_fd < 0 || c.broken)
             return false;
 
+        const int64_t now_ns = SteadyNowNs();
+
         // Treat backpressure as fatal only if the peer shows NO sign of life. A busy
         // writer keeps refreshing last_peer_activity_ns via BUSY heartbeats / ACKs, so
-        // we wait through arbitrarily long stalls; only a silent peer trips the timeout.
-        const int64_t silent_ns = SteadyNowNs() - c.last_peer_activity_ns.load(std::memory_order_relaxed);
+        // we wait through long stalls; only a silent peer trips this timeout.
+        const int64_t silent_ns = now_ns - c.last_peer_activity_ns.load(std::memory_order_relaxed);
         if (silent_ns > std::chrono::duration_cast<std::chrono::nanoseconds>(peer_liveness_timeout).count()) {
             logger.Warning("No liveness from writer on socket " + std::to_string(c.socket_number) +
                            " for " + std::to_string(silent_ns / 1000000) + " ms while sending; marking broken");
@@ -215,6 +218,19 @@ bool TCPStreamPusher::SendAll(Connection& c, const void* buf, size_t len) {
             return false;
         }
 
+        // Hard backpressure cap: a writer may keep heartbeating yet never drain (e.g. a
+        // permanently wedged filesystem). If no bytes leave the socket for this long, give
+        // up regardless of heartbeats so the run -- and its finalization -- cannot hang.
+        if (now_ns - last_progress_ns >
+            std::chrono::duration_cast<std::chrono::nanoseconds>(max_backpressure_timeout).count()) {
+            logger.Warning("No send progress to writer on socket " + std::to_string(c.socket_number) +
+                           " for " + std::to_string((now_ns - last_progress_ns) / 1000000) +
+                           " ms despite heartbeats; marking broken");
+            c.broken = true;
+            CloseFd(c.fd);
+            return false;
+        }
+
         pollfd pfd{};
         pfd.fd = local_fd;
         pfd.events = POLLOUT;
@@ -247,6 +263,7 @@ bool TCPStreamPusher::SendAll(Connection& c, const void* buf, size_t len) {
         }
 
         sent += static_cast<size_t>(rc);
+        last_progress_ns = SteadyNowNs();
     }
 
     return true;

image_pusher/TCPStreamPusher.h

@@ -109,6 +109,12 @@ class TCPStreamPusher : public ImagePusher {
     // every ~250 ms via BUSY heartbeats (and via DATA ACKs), so genuine backpressure of any
     // duration is tolerated; only a truly silent (frozen/dead) peer trips this.
     std::chrono::milliseconds peer_liveness_timeout{15000};
+    // Hard upper bound on backpressure: if the socket accepts no bytes for this long the
+    // writer is wedged and is declared dead even if it keeps heartbeating, so a
+    // misbehaving writer cannot block the run (or its finalization) forever. Generous
+    // relative to peer_liveness_timeout, since a heartbeating peer is given more grace
+    // than a silent one — but still finite.
+    std::chrono::milliseconds max_backpressure_timeout{60000};
 
     int64_t images_per_file = 1;
     uint64_t run_number = 0;
@@ -160,6 +166,12 @@ public:
     /// before data collection starts.
     void SetPeerLivenessTimeout(std::chrono::milliseconds t) { peer_liveness_timeout = t; }
 
+    /// Hard upper bound on backpressure. Even while the peer keeps heartbeating, if no
+    /// bytes can be sent for this long the writer is declared dead so a wedged writer
+    /// cannot block the run or its finalization forever. Must be set before data
+    /// collection starts.
+    void SetMaxBackpressureTimeout(std::chrono::milliseconds t) { max_backpressure_timeout = t; }
+
     std::vector<std::string> GetAddress() const override { return {endpoint}; }
 
     /// Returns the number of currently connected writers (can be called at any time)

tests/TCPImagePusherTest.cpp

@@ -930,3 +930,62 @@ TEST_CASE("TCPImageCommTest_StalledWriter_SurvivesViaHeartbeat", "[TCP]") {
 
     image_buffer.Finalize(std::chrono::seconds(5));
 }
+
+TEST_CASE("TCPImageCommTest_WedgedWriter_DroppedByBackpressureCap", "[TCP]") {
+    // A writer that keeps heartbeating but never drains (e.g. a permanently wedged
+    // filesystem) must not block the run or its finalization forever. The hard
+    // backpressure cap tears the connection down even though BUSY keeps arriving, and
+    // well before the (longer) peer-liveness timeout that those heartbeats keep at bay.
+    constexpr int64_t N = 1000;
+    constexpr auto liveness = std::chrono::milliseconds(5000);       // kept fresh by heartbeats
+    constexpr auto max_backpressure = std::chrono::milliseconds(1500);
+
+    TCPStreamPusher pusher("tcp://127.0.0.1:*", 1, 16 * 1024);
+    pusher.SetPeerLivenessTimeout(liveness);
+    pusher.SetMaxBackpressureTimeout(max_backpressure);
+
+    StallableWriterDouble writer(pusher.GetAddress()[0], 16 * 1024);   // never released
+    REQUIRE(writer.Connected());
+
+    for (int attempt = 0; attempt < 200 && pusher.GetConnectedWriters() < 1; ++attempt)
+        std::this_thread::sleep_for(std::chrono::milliseconds(25));
+    REQUIRE(pusher.GetConnectedWriters() == 1);
+
+    ImageBuffer image_buffer(16 * 1024 * 1024);
+    image_buffer.StartMeasurement(static_cast<size_t>(4096));
+
+    StartMessage start{.images_per_file = 1000, .write_master_file = true};
+    pusher.StartDataCollection(start);   // writer ACKs START, then stalls forever
+
+    auto sender = std::async(std::launch::async, [&] {
+        for (int64_t i = 0; i < N; i++) {
+            ZeroCopyReturnValue *slot = nullptr;
+            while ((slot = image_buffer.GetImageSlot()) == nullptr)
+                std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            std::memset(slot->GetImage(), 0, 256);
+            slot->SetImageNumber(i);
+            slot->SetImageSize(256);
+            slot->ReadyToSend();
+            pusher.SendImage(*slot);
+        }
+    });
+
+    // The cap must fire and drop the connection despite continuous heartbeats.
+    bool dropped = false;
+    for (int attempt = 0; attempt < 400 && !dropped; ++attempt) {
+        if (pusher.GetConnectedWriters() == 0)
+            dropped = true;
+        else
+            std::this_thread::sleep_for(std::chrono::milliseconds(25));
+    }
+    CHECK(dropped);
+
+    // Neither the producers nor finalization may hang once the writer is wedged.
+    REQUIRE(sender.wait_for(std::chrono::seconds(10)) == std::future_status::ready);
+    sender.get();
+
+    EndMessage end{};
+    CHECK(pusher.EndDataCollection(end) == false);   // bounded, and reports failure
+
+    image_buffer.Finalize(std::chrono::seconds(5));
+}