image_pusher: do not hold connections_mutex across sends and joins

The header has said since it was written that blocking queue operations
must never run under connections_mutex; three code paths did exactly that.
KeepaliveThread held it while sending a keepalive to every connection,
which blocks until the peer-liveness or backpressure timeout - so one
half-dead writer socket could stall SendImage and every /statistics poll
for up to a minute, from an idle-time heartbeat. AcceptorThread and
StartDataCollection held it across RemoveDeadConnections, which joins a
writer thread that may itself be inside such a send.

RemoveDeadConnections is split in two: DetachDeadConnections unlinks them
from the pool under the mutex, which is quick, and CloseDeadConnections
tears them down afterwards with the mutex released - safe because they are
no longer reachable by anyone else. The keepalive loop copies the pool out
and sends outside the lock, the pattern EndDataCollection already used.

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

image_pusher/TCPStreamPusher.cpp

@@ -554,11 +554,25 @@ void TCPStreamPusher::AcceptorThread() {
         if (new_fd < 0)
             continue;
 
-        std::lock_guard lg(connections_mutex);
+        std::vector<std::shared_ptr<Connection>> dead;
+        bool rejected = false;
+        size_t total = 0;
+        {
+            std::lock_guard lg(connections_mutex);
+            dead = DetachDeadConnections();
 
-        RemoveDeadConnections();
+            if (connections.size() >= max_connections) {
+                rejected = true;
+            } else {
+                SetupNewConnection(new_fd, next_socket_number++);
+                total = connections.size();
+            }
+        }
 
-        if (connections.size() >= max_connections) {
+        // Outside the lock: this joins writer threads that may be mid-send.
+        CloseDeadConnections(dead);
+
+        if (rejected) {
             logger.Warning("Max connections (" + std::to_string(max_connections) +
                           ") reached, rejecting new connection");
             shutdown(new_fd, SHUT_RDWR);
@@ -566,9 +580,8 @@ void TCPStreamPusher::AcceptorThread() {
             continue;
         }
 
-        SetupNewConnection(new_fd, next_socket_number++);
         logger.Info("Accepted writer connection (socket_number=" + std::to_string(next_socket_number - 1) +
-                   ", total=" + std::to_string(connections.size()) + ")");
+                   ", total=" + std::to_string(total) + ")");
     }
 }
 
@@ -605,29 +618,40 @@ void TCPStreamPusher::SetupNewConnection(int new_fd, uint32_t socket_number) {
     connections.emplace_back(std::move(c));
 }
 
-void TCPStreamPusher::RemoveDeadConnections() {
-    // Must be called with connections_mutex held.
-    // We move dead connections out, release the mutex implicitly (caller still holds it),
-    // then join their futures. Actually — we can join right here since PersistentAckThread
-    // doesn't take connections_mutex, so no deadlock.
+std::vector<std::shared_ptr<TCPStreamPusher::Connection>> TCPStreamPusher::DetachDeadConnections() {
+    // Must be called with connections_mutex held. Only unlinks them from the pool - which is quick -
+    // and hands them back for the caller to close once the mutex is released. Tearing them down here
+    // would join a writer thread that can be sitting in a send for up to max_backpressure_timeout,
+    // with connections_mutex held the whole time.
+    std::vector<std::shared_ptr<Connection>> dead;
     auto it = connections.begin();
     while (it != connections.end()) {
         auto c = *it;
         if (c->broken || !c->connected || !IsConnectionAlive(*c)) {
             c->connected = false;
             c->broken = true;
-            StopDataCollectionThreads(*c);
-            CloseFd(c->fd);
-
-            if (c->persistent_ack_future.valid())
-                c->persistent_ack_future.get();
-
-            logger.Info("Removed dead connection (socket_number=" + std::to_string(c->socket_number) + ")");
+            dead.push_back(c);
             it = connections.erase(it);
         } else {
             ++it;
         }
     }
+    return dead;
+}
+
+void TCPStreamPusher::CloseDeadConnections(const std::vector<std::shared_ptr<Connection>> &dead) {
+    // Blocking: joins the writer thread, which may be inside a send. Never call with
+    // connections_mutex held. Safe to run unlocked - these are already out of the pool, so nobody
+    // else can reach them.
+    for (const auto &c : dead) {
+        StopDataCollectionThreads(*c);
+        CloseFd(c->fd);
+
+        if (c->persistent_ack_future.valid())
+            c->persistent_ack_future.get();
+
+        logger.Info("Removed dead connection (socket_number=" + std::to_string(c->socket_number) + ")");
+    }
 }
 
 void TCPStreamPusher::KeepaliveThread() {
@@ -642,8 +666,16 @@ void TCPStreamPusher::KeepaliveThread() {
         if (data_collection_active)
             continue;
 
-        std::lock_guard lg(connections_mutex);
-        for (auto& cptr : connections) {
+        // Copy the pool out and send with the mutex released. SendFrame blocks until the peer
+        // liveness or backpressure timeout, and holding connections_mutex across that stalls
+        // SendImage and every /statistics poll behind an idle-time keepalive.
+        std::vector<std::shared_ptr<Connection>> local_connections;
+        {
+            std::lock_guard lg(connections_mutex);
+            local_connections = connections;
+        }
+
+        for (auto& cptr : local_connections) {
             auto& c = *cptr;
             if (c.broken || !c.connected)
                 continue;
@@ -657,7 +689,12 @@ void TCPStreamPusher::KeepaliveThread() {
             }
         }
 
-        RemoveDeadConnections();
+        std::vector<std::shared_ptr<Connection>> dead;
+        {
+            std::lock_guard lg(connections_mutex);
+            dead = DetachDeadConnections();
+        }
+        CloseDeadConnections(dead);
     }
 }
 
@@ -799,12 +836,26 @@ void TCPStreamPusher::StartDataCollection(StartMessage& message) {
     total_data_acked_bad.store(0, std::memory_order_relaxed);
     total_data_acked_total.store(0, std::memory_order_relaxed);
 
+    // Stopping a writer thread joins it, and closing a detached connection joins it too, so both
+    // happen with connections_mutex released; the mutex is only taken to read and write the pool.
+    std::vector<std::shared_ptr<Connection>> current;
+    {
+        std::lock_guard lg(connections_mutex);
+        current = connections;
+    }
+    for (auto& c : current)
+        StopDataCollectionThreads(*c);
+
+    std::vector<std::shared_ptr<Connection>> dead;
+    {
+        std::lock_guard lg(connections_mutex);
+        dead = DetachDeadConnections();
+    }
+    CloseDeadConnections(dead);
+
     std::vector<std::shared_ptr<Connection>> local_connections;
     {
         std::lock_guard lg(connections_mutex);
-        for (auto& c : connections)
-            StopDataCollectionThreads(*c);
-        RemoveDeadConnections();
         if (connections.empty())
             throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "No writers connected to " + endpoint);
 

image_pusher/TCPStreamPusher.h

@@ -145,7 +145,10 @@ class TCPStreamPusher : public ImagePusher {
     void KeepaliveThread();
 
     void SetupNewConnection(int new_fd, uint32_t socket_number);
-    void RemoveDeadConnections();
+    // Unlink dead connections from the pool (connections_mutex held) and close them (mutex released).
+    // Split because closing joins a writer thread that can be blocked in a send.
+    std::vector<std::shared_ptr<Connection>> DetachDeadConnections();
+    void CloseDeadConnections(const std::vector<std::shared_ptr<Connection>> &dead);
     void TearDownConnection(Connection& c);
 
     void StartDataCollectionThreads(Connection& c);