From 5a09b325fd92508e4f239d16d7909bd86ed83fb2 Mon Sep 17 00:00:00 2001 From: Defenwycke Date: Mon, 7 Sep 2026 02:15:47 +0100 Subject: [PATCH 1/2] test(sv2): give `Sniffer` a `shutdown()` (#849) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WIP — the method and task registry only; nothing calls it yet. A sniffer's proxy task runs until the PROCESS exits, and it retries `TcpStream::connect` to its upstream once a second for ever. Every `tests/*.rs` is one binary running many tests, so a sniffer left behind by a finished test keeps looping against an upstream that has gone, and they accumulate across a file. That is a candidate cause for #849 item 3 (`translator_aggregated_integration`: four tests pass with `--exact`, hang when run together). `PoolSv2`, `TranslatorSv2` and the JD roles all have a `shutdown()`; the sniffers did not, which is why `shutdown_all!` never covered them. ⚠ Not yet verified against the hang — the hypothesis is untested. Claude-Session: https://claude.ai/code/session_01XjzQeoCkuKx3vb4amkzAdT --- tests/integration-sv2/lib/sniffer.rs | 38 +++++++++++++++++++++++++++- 1 file changed, 37 insertions(+), 1 deletion(-) diff --git a/tests/integration-sv2/lib/sniffer.rs b/tests/integration-sv2/lib/sniffer.rs index 613329dfd..1fddecab2 100644 --- a/tests/integration-sv2/lib/sniffer.rs +++ b/tests/integration-sv2/lib/sniffer.rs @@ -48,6 +48,17 @@ pub struct Sniffer<'a> { action: Vec, timeout: Option, negotiated_extensions: Arc>>, + /// 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>>>, } impl<'a> Sniffer<'a> { @@ -71,6 +82,24 @@ 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. + pub fn shutdown(&self) { + if let Ok(mut handles) = self.tasks.lock() { + for h in handles.drain(..) { + h.abort(); + } } } @@ -94,6 +123,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 +157,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 +185,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. From 57edd01182a978f89315d8ec7e4dd2b4572e4b31 Mon Sep 17 00:00:00 2001 From: Defenwycke Date: Mon, 7 Sep 2026 03:24:04 +0100 Subject: [PATCH 2/2] test(sv2): make the aggregated fallback test fail instead of hang (#849) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `aggregated_translator_triggers_fallback_on_close_channel_message` does not fail when it goes wrong — it hangs, for ever, and takes the whole binary with it. Measured with `--exact` on an idle box: 5 runs out of 5 killed at a 150s cap, no panic, no log line from the test process after the first 10 seconds. This corrects the record in `ci.yml` and #849, which both say the four aggregated tests "pass INDIVIDUALLY" and hang only when run together. They do not. This one hangs alone, and no predecessor is involved: t1-then-t3 hung, t3-alone hung, and a passing run happens roughly 1 time in 3 either way. `Sniffer::wait_for_message_type` already carries 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 is necessary and not sufficient. `#[tokio::test]` is a current-thread runtime and the queue read is only one of the things on it that takes a blocking lock — the sniffer's own forwarding task (`add_message` -> `safe_lock`), the translator and the pool all take one inline. tokio's timer lives on that same single thread, so once any of them blocks it, `timeout` and `sleep` stop advancing and the deadline can never arrive. The guard is unreachable precisely when it is needed. Giving this test four worker threads makes the deadline reachable again: 0 hangs in 4 runs, two of which failed on the 60s deadline with its own message, two of which passed. ⚠ The surviving failure is REAL and is not fixed here: after `CloseChannel` the translator often never opens the fallback upstream, logging `Failed to send fallback status from ChannelManager`. It reproduces roughly 2 runs in 3 and was invisible for as long as the target hung rather than failed. Filed separately. This target still must not gate until that is fixed and proven green on CI. Also finishes the sniffer shutdown from the previous commit. `shutdown()` is now async so `shutdown_all!` can take it — that macro expands to `tokio::join!` over each handle's `shutdown()`, so a sync method could not be used by the one thing meant to call it — and every sniffer in this file is now shut down with its pool and translator rather than left running to the end of the process. --- tests/integration-sv2/lib/sniffer.rs | 7 ++++- .../translator_aggregated_integration.rs | 29 +++++++++++++++---- 2 files changed, 30 insertions(+), 6 deletions(-) diff --git a/tests/integration-sv2/lib/sniffer.rs b/tests/integration-sv2/lib/sniffer.rs index 1fddecab2..faf4d87d1 100644 --- a/tests/integration-sv2/lib/sniffer.rs +++ b/tests/integration-sv2/lib/sniffer.rs @@ -95,7 +95,12 @@ impl<'a> Sniffer<'a> { /// /// Idempotent — aborting an already-finished task is a no-op — so a test may call it on a /// sniffer whose peer has already gone. - pub fn shutdown(&self) { + /// + /// `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(); diff --git a/tests/integration-sv2/tests/translator_aggregated_integration.rs b/tests/integration-sv2/tests/translator_aggregated_integration.rs index 9961d21a6..e7373036d 100644 --- a/tests/integration-sv2/tests/translator_aggregated_integration.rs +++ b/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