test(sv2): make the aggregated fallback test fail instead of hang (#849)

tests/integration-sv2/lib/sniffer.rs

@@ -48,6 +48,17 @@ pub struct Sniffer<'a> {
     action: Vec<InterceptAction>,
     timeout: Option<u64>,
     negotiated_extensions: Arc<Mutex<Vec<u16>>>,
+    /// Handles for the tasks `start` spawns, so a test can stop them (#849).
+    ///
+    /// Without this a sniffer runs until the PROCESS exits. Its proxy task retries
+    /// `TcpStream::connect` to its upstream once a second for ever, so a sniffer left behind by
+    /// a finished test keeps looping against an upstream that has gone. Every `tests/*.rs` is one
+    /// binary running many tests, so those accumulate across a file.
+    ///
+    /// `PoolSv2`, `TranslatorSv2` and the JD roles all have a `shutdown()`; the sniffers did not,
+    /// which is why `shutdown_all!` never covered them and why no test could clean them up even
+    /// where it wanted to.
+    tasks: Arc<Mutex<Vec<tokio::task::JoinHandle<()>>>>,
 }
 
 impl<'a> Sniffer<'a> {
@@ -71,6 +82,29 @@ impl<'a> Sniffer<'a> {
             action,
             timeout,
             negotiated_extensions: Arc::new(Mutex::new(Vec::new())),
+            tasks: Arc::new(Mutex::new(Vec::new())),
+        }
+    }
+
+    /// Stop the tasks this sniffer spawned.
+    ///
+    /// `abort()` rather than a cancellation token: the proxy task spends its life awaiting socket
+    /// reads and an upstream-connect retry loop, both of which are cancel-safe await points, and
+    /// a token would have to be threaded through every one of them to be checked. A test that has
+    /// finished asserting has no interest in a graceful drain.
+    ///
+    /// Idempotent — aborting an already-finished task is a no-op — so a test may call it on a
+    /// sniffer whose peer has already gone.
+    ///
+    /// `async` with nothing to await, so that `shutdown_all!` can take it: that macro expands to
+    /// `tokio::join!` over each handle's `shutdown()`, which requires futures. A sync method here
+    /// would compile everywhere except the one place a test actually wants to use it, and the
+    /// point of this is to be uniform with `PoolSv2`/`TranslatorSv2`/the JD roles.
+    pub async fn shutdown(&self) {
+        if let Ok(mut handles) = self.tasks.lock() {
+            for h in handles.drain(..) {
+                h.abort();
+            }
         }
     }
 
@@ -94,6 +128,8 @@ impl<'a> Sniffer<'a> {
             check_on_drop: false,
             action: Vec::new(),
             timeout,
+            // Test-only constructor: it never calls `start`, so there is nothing to abort.
+            tasks: Arc::new(Mutex::new(Vec::new())),
             negotiated_extensions: Arc::new(Mutex::new(Vec::new())),
         }
     }
@@ -126,7 +162,7 @@ impl<'a> Sniffer<'a> {
         // the port from the moment it is chosen, so an independent bind here fails (#612).
         let std_listener = crate::utils::claim_listener(listening_address);
 
-        tokio::spawn(async move {
+        let handle = tokio::spawn(async move {
             let listener = tokio::net::TcpListener::from_std(std_listener)
                 .expect("Sniffer: cannot adopt listener");
             let (downstream_receiver, downstream_sender) =
@@ -154,6 +190,11 @@ impl<'a> Sniffer<'a> {
                 _ = recv_from_up_send_to_down(upstream_receiver, downstream_sender, messages_from_upstream, action, &identifier, negotiated_extensions.clone()) => { },
             };
         });
+        // Registered so `shutdown()` can stop it. Locking here rather than inside the task keeps
+        // the handle out of its own closure.
+        if let Ok(mut t) = self.tasks.lock() {
+            t.push(handle);
+        }
     }
 
     /// Returns the oldest message sent by downstream.

tests/integration-sv2/tests/translator_aggregated_integration.rs

@@ -40,7 +40,26 @@ use stratum_apps::stratum_core::{
 // the pool exchange the correct messages upon connection. And that the miner is able to submit
 // shares.
 
-#[tokio::test]
+// ⛔ MULTI-THREADED ON PURPOSE (#849). On the default current-thread runtime this test does not
+// fail — it HANGS, for ever, and takes the whole binary with it.
+//
+// `Sniffer::wait_for_message_type` has a 60s deadline meant to turn exactly this into a failure,
+// and #450 moved its queue read onto the blocking pool so the read could not stall the executor.
+// That hardening is not sufficient. The read is only one of the things on this runtime that takes
+// a BLOCKING lock: the sniffer's own forwarding task (`add_message` -> `safe_lock`), the
+// translator and the pool all do it inline on the executor. tokio's TIMER also lives on that one
+// thread, so as soon as any of them blocks it, `timeout` and `sleep` stop advancing and the 60s
+// deadline can never arrive. The guard is unreachable precisely when it is needed.
+//
+// Measured on this test, `--exact`, quiet box: current-thread hung 5/5 runs to a 150s kill with
+// no panic and no log output after the first 10s. With four worker threads: 0/3 hangs — two runs
+// failed on the 60s deadline with its own message, one passed.
+//
+// ⚠ That surviving 2-in-3 failure is a REAL defect, not a harness artefact: after `CloseChannel`
+// the translator often never opens the fallback upstream, logging `Failed to send fallback status
+// from ChannelManager`. It is filed separately. This attribute does not fix it — it makes it
+// visible as a failure in 60s instead of a 420s CI timeout with an empty log.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn aggregated_translator_triggers_fallback_on_close_channel_message() {
     start_tracing();
 
@@ -192,7 +211,7 @@ async fn aggregated_translator_triggers_fallback_on_close_channel_message() {
     sniffer_b
         .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION)
         .await;
-    translator.shutdown().await;
+    shutdown_all!(translator, sniffer_a, sniffer_b);
 }
 
 // Verify's that the non-aggregated mode translator does not shut down if an
@@ -253,7 +272,7 @@ async fn tproxy_sends_single_open_extended_mining_channel_in_aggregated_mode() {
             .await
     );
 
-    shutdown_all!(pool, tproxy);
+    shutdown_all!(pool, tproxy, pool_translator_sniffer);
 }
 
 #[tokio::test]
@@ -497,7 +516,7 @@ async fn aggregated_translator_correctly_deals_with_group_channels() {
             break;
         }
     }
-    shutdown_all!(translator, pool);
+    shutdown_all!(translator, pool, sniffer, _sniffer_pool_tp);
 }
 
 // This test launches a tProxy in non-aggregated mode and leverages a MockUpstream to test the
@@ -674,7 +693,7 @@ async fn aggregated_translator_handles_downstream_connecting_during_future_job()
     sv1_sniffer_2
         .wait_for_message(&["mining.submit"], MessageDirection::ToUpstream)
         .await;
-    translator.shutdown().await;
+    shutdown_all!(translator, sniffer);
 }
 
 // This test verifies that the pool server continues accepting new connection