image_pusher: serialise tearing a connection down

Two futures per connection are joined from more than one path: the acceptor reaping a
dead connection, and the control plane starting or ending a run. Calling get() on one
future from two threads at once is undefined and invalidates it, and the guard against
it was a non-atomic test-then-set of an atomic flag - `if (!c.active) return; c.active
= false;` - so both callers could pass it.

Holding connections_mutex across the joins is what the previous commit removed on
purpose, and rightly: the joins block on a writer that may be inside a send. So the
lock is per connection and covers only the teardown. Nothing it joins takes it, so it
cannot deadlock.

The comment claiming a detached connection is unreachable by anyone else was wrong: a
control-plane call that copied the pool before the erase still holds a shared_ptr to it.
That is precisely how the two teardowns meet.

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

image_pusher/TCPStreamPusher.cpp

@@ -166,6 +166,13 @@ void TCPStreamPusher::TearDownConnection(Connection& c) {
     c.connected = false;
     c.broken = true;
     CloseFd(c.fd);
+    JoinPersistentAck(c);
+}
+
+// The persistent ack reader is joined both here and by the acceptor when it reaps a dead
+// connection, so the join takes the same per-connection lock the writer join does.
+void TCPStreamPusher::JoinPersistentAck(Connection& c) {
+    std::lock_guard teardown(c.teardown_mutex);
     if (c.persistent_ack_future.valid())
         c.persistent_ack_future.get();
 }
@@ -641,14 +648,14 @@ std::vector<std::shared_ptr<TCPStreamPusher::Connection>> TCPStreamPusher::Detac
 
 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.
+    // connections_mutex held. These are out of the pool, but a control-plane call that copied the
+    // pool before the erase still holds a shared_ptr to them, so the joins below take the
+    // connection's own teardown lock rather than assuming exclusive access.
     for (const auto &c : dead) {
         StopDataCollectionThreads(*c);
         CloseFd(c->fd);
 
-        if (c->persistent_ack_future.valid())
-            c->persistent_ack_future.get();
+        JoinPersistentAck(*c);
 
         logger.Info("Removed dead connection (socket_number=" + std::to_string(c->socket_number) + ")");
     }
@@ -732,6 +739,10 @@ void TCPStreamPusher::StartDataCollectionThreads(Connection& c) {
 }
 
 void TCPStreamPusher::StopDataCollectionThreads(Connection& c) {
+    // One thread at a time per connection: the acceptor can be reaping this connection while the
+    // control plane is starting or ending a run on it, and both end in writer_future.get().
+    std::lock_guard teardown(c.teardown_mutex);
+
     if (!c.active)
         return;
 

image_pusher/TCPStreamPusher.h

@@ -53,6 +53,12 @@ class TCPStreamPusher : public ImagePusher {
         // Persistent ack/keepalive reader (runs as long as the connection is alive)
         std::future<void> persistent_ack_future;
 
+        // Serialises tearing this connection down. Both futures below are joined from more than one
+        // path - the acceptor reaping a dead connection, and the control plane starting or ending a
+        // run - and calling get() on one future from two threads is undefined and invalidates it.
+        // Held only around the joins, never around a send, and no thread it joins takes it.
+        std::mutex teardown_mutex;
+
         std::mutex send_mutex;
         std::mutex ack_mutex;
         std::condition_variable ack_cv;
@@ -153,6 +159,7 @@ class TCPStreamPusher : public ImagePusher {
 
     void StartDataCollectionThreads(Connection& c);
     void StopDataCollectionThreads(Connection& c);
+    void JoinPersistentAck(Connection& c);
 
     bool WaitForAck(Connection& c, TCPFrameType ack_for, std::chrono::milliseconds timeout, std::string* error_text);
     bool WaitForEndAck(Connection& c, std::chrono::milliseconds liveness_timeout, std::string* error_text);