TCP stream: tolerate writer backpressure via BUSY heartbeats

A slow filesystem could stall the writer's consume pipeline, propagating
TCP backpressure to the pusher. The pusher then treated that backpressure
as a dead peer and force-closed the connection mid-run; the writer
reconnected as a brand-new connection outside the active session, so the
rest of the run was silently dropped and the half-written HDF5 file was
later finalized with holes.

Replace the throughput/progress-based send timeout with a peer-liveness
timeout:

- Add TCPFrameType::BUSY (wire version 2 -> 3).
- TCPImagePuller runs a heartbeat thread that sends BUSY every 250ms on a
  thread independent of the (possibly stalled) write path, so liveness
  keeps flowing during deep stalls. A send_mutex serializes
  ACK/pong/heartbeat writes.
- TCPStreamPusher refreshes last_peer_activity_ns on every inbound frame
  and only declares a connection dead after peer_liveness_timeout (15s) of
  complete silence, tolerating arbitrarily long backpressure while still
  catching a genuinely dead peer (and immediate EPIPE/ECONNRESET).
- Re-key both backpressure waits (the SendAll data path and the post-END
  WaitForEndAck) onto this liveness signal instead of byte-progress /
  DATA-ACK-progress, so a slow final flush at END is tolerated too.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

image_puller/TCPImagePuller.cpp

@@ -54,6 +54,7 @@ TCPImagePuller::TCPImagePuller(const std::string &tcp_addr,
 
     receiver_thread = std::thread(&TCPImagePuller::ReceiverThread, this);
     cbor_thread = std::thread(&TCPImagePuller::CBORThread, this);
+    heartbeat_thread = std::thread(&TCPImagePuller::HeartbeatThread, this);
 
     if (!repub_address.empty()) {
         repub_socket = std::make_unique<ZMQSocket>(ZMQSocketType::Push);
@@ -91,6 +92,8 @@ bool TCPImagePuller::SendAll(const void *buf, size_t len) {
 }
 
 bool TCPImagePuller::SendAck(const PullerAckMessage &ack) {
+    std::lock_guard lg(send_mutex);
+
     TcpFrameHeader h{};
     h.type = static_cast<uint16_t>(TCPFrameType::ACK);
     h.run_number = ack.run_number;
@@ -319,7 +322,12 @@ void TCPImagePuller::ReceiverThread() {
                 TcpFrameHeader pong{};
                 pong.type = static_cast<uint16_t>(TCPFrameType::KEEPALIVE);
                 pong.payload_size = 0;
-                if (!SendAll(&pong, sizeof(pong))) {
+                bool pong_ok;
+                {
+                    std::lock_guard lg(send_mutex);
+                    pong_ok = SendAll(&pong, sizeof(pong));
+                }
+                if (!pong_ok) {
                     logger.Info("Keepalive pong send failed, reconnecting to " + addr);
                     CloseSocket();
                     std::this_thread::sleep_for(std::chrono::milliseconds(20));
@@ -365,6 +373,37 @@ void TCPImagePuller::ReceiverThread() {
     cbor_fifo.PutBlocking(ImagePullerOutput{});
 }
 
+void TCPImagePuller::HeartbeatThread() {
+    // While connected, periodically tell the pusher we are alive even if the
+    // consuming pipeline is stalled (e.g. blocked on a slow filesystem). This lets
+    // the pusher distinguish a busy-but-healthy writer from a dead one and keep
+    // waiting through arbitrarily long backpressure instead of dropping the run.
+    while (!disconnect) {
+        // Sleep in small slices so shutdown stays prompt.
+        for (int i = 0; i < 5 && !disconnect; i++)
+            std::this_thread::sleep_for(HeartbeatInterval / 5);
+        if (disconnect)
+            break;
+
+        {
+            std::unique_lock ul(fd_mutex);
+            if (fd < 0)
+                continue; // Not connected; ReceiverThread is (re)establishing the link.
+        }
+
+        TcpFrameHeader h{};
+        h.type = static_cast<uint16_t>(TCPFrameType::BUSY);
+        h.payload_size = 0;
+        h.ack_fifo_occupancy = cbor_fifo.GetCurrentUtilization();
+        h.ack_fifo_max_occupancy = cbor_fifo.Size();
+
+        // Best effort: a failure here just means the socket is gone, which
+        // ReceiverThread will detect and reconnect on its own.
+        std::lock_guard lg(send_mutex);
+        SendAll(&h, sizeof(h));
+    }
+}
+
 void TCPImagePuller::Disconnect() {
     if (disconnect.exchange(true))
         return;
@@ -377,4 +416,6 @@ void TCPImagePuller::Disconnect() {
         cbor_thread.join();
     if (repub_thread.joinable())
         repub_thread.join();
+    if (heartbeat_thread.joinable())
+        heartbeat_thread.join();
 }