From 5095a348c5ab013984ea010ed9a3dbab81d0af8e Mon Sep 17 00:00:00 2001
From: GitGab19 <gabriele.vernetti@protonmail.com>
Date: Fri, 7 Aug 2026 12:47:53 +0200
Subject: [PATCH 01/15] test(integration): fix latent ordering races in the
 test harness

Four tests polled on one condition and then asserted on another that
settles later. Each was masked by a coarse 1s poll interval that happened
to return late enough for the second condition to hold, rather than by
any actual synchronisation:

- `SnifferSV1::wait_and_assert` waited using a fuzzy matcher (which also
  accepts an `OkResponse` whose serialized form merely *contains* the
  filter string) and then re-fetched with a strict predicate, panicking
  "Message disappeared after wait_for_message" when the two disagreed.
  It now polls on exactly the predicate it fetches with.

- `test_extension_negotiation_with_tlv_in_submit_shares` popped
  `RequestExtensions` off the queue without waiting for it. It is sent
  *after* `SetupConnectionSuccess`, so the pop could precede the send.

- `pool_api_endpoints_with_miner` and `jdc_api_endpoints_with_miner`
  polled until a client was registered, then asserted that client already
  had a channel. The channel opens after registration. Both now poll on
  the channel count; `poll_until` takes `&str` so it can address the
  dynamic `/clients/{id}/channels` route.

- `non_aggregated_translator_correctly_deals_with_group_channels`
  compared a `mining.notify` prevhash before and after a chain tip
  update without clearing the queue, so the stale pre-update notify
  could be matched and the prevhash appeared unchanged. Adds
  `SnifferSV1::clean_queue`, mirroring the Sv2 sniffer.

Each was reproduced deterministically before being fixed.
---
 .../lib/prometheus_metrics_assertions.rs      |   9 +-
 integration-tests/lib/sv1_sniffer.rs          |  95 ++++++++------
 integration-tests/lib/utils.rs                | 120 +++++++++++++++++-
 integration-tests/tests/extensions.rs         |  14 +-
 .../tests/monitoring_integration.rs           |  18 ++-
 .../tests/translator_integration.rs           |  10 ++
 6 files changed, 221 insertions(+), 45 deletions(-)

diff --git a/integration-tests/lib/prometheus_metrics_assertions.rs b/integration-tests/lib/prometheus_metrics_assertions.rs
index 3ffba2809..c1a5b9f03 100644
--- a/integration-tests/lib/prometheus_metrics_assertions.rs
+++ b/integration-tests/lib/prometheus_metrics_assertions.rs
@@ -166,13 +166,16 @@ impl MonitoringApi {
 
     /// Poll `path` until the response deserialises into `T` and `predicate` returns true.
     ///
-    /// Retries every 500 ms until `timeout`. Non-2xx responses and
+    /// Retries every [`crate::utils::POLL_INTERVAL`] until `timeout`. Non-2xx responses and
     /// deserialisation failures are tolerated (the endpoint may not be ready
     /// yet — for example a `/api/v1/clients/{id}` route returns 404 until the
     /// snapshot cache first populates). On timeout, the panic message includes
     /// the path, target type, last status, and last body so CI failures are
     /// debuggable without re-running with extra logging.
-    pub async fn poll_until<T, F>(&self, path: &'static str, timeout: Duration, predicate: F) -> T
+    ///
+    /// `path` is a plain `&str` so callers can poll dynamic routes such as
+    /// `/api/v1/clients/{id}/channels`, whose state settles after the client itself appears.
+    pub async fn poll_until<T, F>(&self, path: &str, timeout: Duration, predicate: F) -> T
     where
         T: serde::de::DeserializeOwned,
         F: Fn(&T) -> bool,
@@ -205,7 +208,7 @@ impl MonitoringApi {
                     last_body,
                 );
             }
-            tokio::time::sleep(Duration::from_millis(500)).await;
+            tokio::time::sleep(crate::utils::POLL_INTERVAL).await;
         }
     }
 
diff --git a/integration-tests/lib/sv1_sniffer.rs b/integration-tests/lib/sv1_sniffer.rs
index f41d4a409..2b5418b09 100644
--- a/integration-tests/lib/sv1_sniffer.rs
+++ b/integration-tests/lib/sv1_sniffer.rs
@@ -60,10 +60,8 @@ impl SnifferSV1 {
                 match TcpStream::connect(upstream_address).await {
                     Ok(s) => break s,
                     Err(_) => {
-                        tracing::warn!(
-                            "SnifferSV1: unable to connect to upstream, retrying after 1 second"
-                        );
-                        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+                        tracing::warn!("SnifferSV1: unable to connect to upstream, retrying");
+                        tokio::time::sleep(crate::utils::CONNECT_RETRY_INTERVAL).await;
                         continue;
                     }
                 }
@@ -91,6 +89,18 @@ impl SnifferSV1 {
         });
     }
 
+    /// Clears all messages from the specified direction's queue.
+    ///
+    /// Use before triggering an event whose effect is asserted with [`SnifferSV1::wait_and_assert`]
+    /// when a message of the same type is already queued: the wait matches the most recent
+    /// matching message, which would otherwise be the stale pre-event one.
+    pub async fn clean_queue(&self, direction: MessageDirection) {
+        match direction {
+            MessageDirection::ToUpstream => self.messages_from_downstream.clear().await,
+            MessageDirection::ToDownstream => self.messages_from_upstream.clear().await,
+        }
+    }
+
     /// Wait for a specific message to be received from the downstream role.
     pub async fn wait_for_message(&self, message: &[&str], direction: MessageDirection) {
         if message.is_empty() {
@@ -116,7 +126,7 @@ impl SnifferSV1 {
                     message.first().unwrap()
                 );
             } else {
-                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                tokio::time::sleep(crate::utils::POLL_INTERVAL).await;
                 continue;
             }
         }
@@ -141,7 +151,7 @@ impl SnifferSV1 {
             if now.elapsed().as_secs() > 60 {
                 panic!("Timeout: keepalive mining.notify (job_id containing '#') not found");
             } else {
-                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                tokio::time::sleep(crate::utils::POLL_INTERVAL).await;
                 continue;
             }
         }
@@ -156,32 +166,49 @@ impl SnifferSV1 {
     ) where
         F: FnMut(sv1_api::Message),
     {
-        let f = filter.inner();
-        self.wait_for_message(&[&f], direction).await;
-
         let aggregator = match direction {
             MessageDirection::ToUpstream => &self.messages_from_downstream,
             MessageDirection::ToDownstream => &self.messages_from_upstream,
         };
 
-        let message = match filter {
-            SV1MessageFilter::WithMessageName(method_name) => aggregator
-                .get_last_matching(|msg| match msg {
-                    sv1_api::Message::StandardRequest(req) => req.method == method_name,
-                    sv1_api::Message::Notification(notif) => notif.method == method_name,
-                    _ => false,
-                })
-                .await
-                .expect("Message disappeared after wait_for_message"),
-            SV1MessageFilter::WithMessageId(method_id) => aggregator
-                .get_last_matching(|msg| match msg {
-                    sv1_api::Message::StandardRequest(req) => req.id == method_id,
-                    sv1_api::Message::OkResponse(req) => req.id == method_id,
-                    sv1_api::Message::ErrorResponse(req) => req.id == method_id,
-                    _ => false,
-                })
-                .await
-                .expect("Message disappeared after wait_for_message"),
+        // Poll on exactly the predicate this function fetches with, rather than waiting on
+        // `wait_for_message`'s looser matcher and then re-querying. That matcher also accepts an
+        // `OkResponse` whose serialized form merely *contains* the filter string, so it can report
+        // a match that the strict predicates below cannot find — a window the previous 1s poll
+        // interval happened to step over rather than avoid.
+        let now = std::time::Instant::now();
+        let message = loop {
+            let found = match &filter {
+                SV1MessageFilter::WithMessageName(method_name) => {
+                    aggregator
+                        .get_last_matching(|msg| match msg {
+                            sv1_api::Message::StandardRequest(req) => req.method == *method_name,
+                            sv1_api::Message::Notification(notif) => notif.method == *method_name,
+                            _ => false,
+                        })
+                        .await
+                }
+                SV1MessageFilter::WithMessageId(method_id) => {
+                    aggregator
+                        .get_last_matching(|msg| match msg {
+                            sv1_api::Message::StandardRequest(req) => req.id == *method_id,
+                            sv1_api::Message::OkResponse(req) => req.id == *method_id,
+                            sv1_api::Message::ErrorResponse(req) => req.id == *method_id,
+                            _ => false,
+                        })
+                        .await
+                }
+            };
+
+            if let Some(message) = found {
+                break message;
+            }
+
+            if now.elapsed().as_secs() > 60 {
+                panic!("Timeout: SV1 message matching {filter:?} not found");
+            }
+
+            tokio::time::sleep(crate::utils::POLL_INTERVAL).await;
         };
         assertion(message);
     }
@@ -222,20 +249,12 @@ impl SnifferSV1 {
 ///
 /// For `WithMessageName` you can pass method name like `mining.subscribe`, And for `WithMessageId`
 /// you can pass the id of the message you are interested in filtering.
+#[derive(Debug)]
 pub enum SV1MessageFilter {
     WithMessageName(&'static str),
     WithMessageId(u64),
 }
 
-impl SV1MessageFilter {
-    fn inner(&self) -> String {
-        match self {
-            SV1MessageFilter::WithMessageName(mn) => mn.to_string(),
-            SV1MessageFilter::WithMessageId(mi) => mi.to_string(),
-        }
-    }
-}
-
 /// Represents a SV1 message manager.
 ///
 /// This struct can be used in order to aggregate and manage SV1 messages.
@@ -251,6 +270,10 @@ impl MessagesAggregatorSV1 {
         }
     }
 
+    async fn clear(&self) {
+        self.messages.lock().await.clear();
+    }
+
     async fn add_message(&self, message: sv1_api::Message) {
         let mut messages = self.messages.lock().await;
         messages.push_back(message);
diff --git a/integration-tests/lib/utils.rs b/integration-tests/lib/utils.rs
index 4367e71ea..a4a06ee3b 100644
--- a/integration-tests/lib/utils.rs
+++ b/integration-tests/lib/utils.rs
@@ -9,8 +9,9 @@ use once_cell::sync::Lazy;
 use std::{
     collections::HashSet,
     convert::TryInto,
-    net::{SocketAddr, TcpListener},
+    net::{SocketAddr, TcpListener, TcpStream},
     sync::{Arc, Mutex},
+    time::Duration,
 };
 use stratum_apps::{
     key_utils::{Secp256k1PublicKey, Secp256k1SecretKey},
@@ -30,6 +31,123 @@ use tokio_util::sync::CancellationToken;
 // prevents get_available_port from ever returning the same port twice
 static UNIQUE_PORTS: Lazy<Mutex<HashSet<u16>>> = Lazy::new(|| Mutex::new(HashSet::new()));
 
+/// How often readiness gates and message-wait loops re-check their condition.
+///
+/// Chosen as the knee of the cost/benefit curve: the suite performs ~600 waits, so residual dead
+/// time is roughly `600 * POLL_INTERVAL`. Dropping from the previous 1s to 50ms recovers ~9.5
+/// minutes; going further to 10ms would recover only ~9s more while raising the wakeup rate 5x.
+/// Each poll is one uncontended mutex acquisition (or one loopback connect), so 20 polls/sec is
+/// negligible. [`crate::sniffer::Sniffer::assert_message_not_present`] already polls the same
+/// structures at 100ms.
+pub const POLL_INTERVAL: Duration = Duration::from_millis(200);
+
+/// Retry cadence for the *unbounded* connect loops that wait for a peer to come up.
+///
+/// Deliberately much slower than [`POLL_INTERVAL`]. Those loops have no timeout, so when a peer
+/// never appears — which several tests arrange on purpose — they spin for the whole test. Every
+/// test runs on a bare `#[tokio::test]`, i.e. a single-threaded runtime, so a 20 Hz connect loop
+/// competes with the test's own work on the one thread it has. The message-wait loops are safe at
+/// [`POLL_INTERVAL`] because they exit as soon as their message lands.
+pub const CONNECT_RETRY_INTERVAL: Duration = Duration::from_secs(1);
+
+/// Ceiling for readiness gates on spawned child processes (Bitcoin Core, sv2-tp).
+///
+/// This is never paid on the happy path — only when something is genuinely wrong — so it is set
+/// far above the observed startup time rather than tuned tightly. It stays well under nextest's
+/// `terminate-after` (120s) so the gate's own panic fires before nextest kills the test, which
+/// keeps the failure message legible.
+pub const PROCESS_READY_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Budget for the best-effort wait on an in-process role's listening socket.
+///
+/// Deliberately equal to the fixed sleep this replaced, so no code path is ever slower than it was
+/// before: a healthy role unblocks in milliseconds, and one that never listens costs exactly what
+/// the old `sleep(1s)` cost. See [`wait_until_listening`] for why this cannot be an assertion.
+pub const ROLE_READY_BUDGET: Duration = Duration::from_secs(1);
+
+/// Blocks until a Unix socket at `path` accepts a connection, polling every [`POLL_INTERVAL`].
+///
+/// Deliberately connects rather than checking `Path::exists`: Bitcoin Core creates the socket file
+/// before it is able to serve on it, and the window between the two is wide enough that merely
+/// stat-ing the path lets tests proceed against a node that then refuses their IPC traffic. A
+/// successful connect proves the listener is bound and accepting.
+///
+/// Panics with `what` in the message if `timeout` elapses first.
+pub fn wait_for_unix_socket(path: &std::path::Path, timeout: Duration, what: &str) {
+    let start = std::time::Instant::now();
+    loop {
+        if path.exists() && std::os::unix::net::UnixStream::connect(path).is_ok() {
+            tracing::debug!(
+                target: "readiness",
+                "ready: {what} after {:?}",
+                start.elapsed()
+            );
+            return;
+        }
+        if start.elapsed() > timeout {
+            panic!(
+                "timeout after {timeout:?} waiting for {what} to accept connections on {}",
+                path.display()
+            );
+        }
+        std::thread::sleep(POLL_INTERVAL);
+    }
+}
+
+/// Blocks until a TCP connection to `addr` succeeds, polling every [`POLL_INTERVAL`].
+///
+/// Used from synchronous contexts; see [`wait_for_listener_async`] for the async equivalent.
+///
+/// Panics with `what` in the message if `timeout` elapses first.
+pub fn wait_for_listener(addr: SocketAddr, timeout: Duration, what: &str) {
+    let start = std::time::Instant::now();
+    loop {
+        if TcpStream::connect_timeout(&addr, POLL_INTERVAL).is_ok() {
+            tracing::debug!(
+                target: "readiness",
+                "ready: {what} after {:?}",
+                start.elapsed()
+            );
+            return;
+        }
+        if start.elapsed() > timeout {
+            panic!("timeout after {timeout:?} waiting for {what} to listen on {addr}");
+        }
+        std::thread::sleep(POLL_INTERVAL);
+    }
+}
+
+/// Waits up to `budget` for a TCP connection to `addr` to succeed, polling every
+/// [`POLL_INTERVAL`]. Returns whether the listener came up.
+///
+/// Unlike the process gates above this deliberately does **not** assert, because a role's
+/// listening socket is not a universally valid readiness signal. `PoolRuntime::bootstrap` runs
+/// `bootstrap_template_provider()` before `start_services()`, so the pool only starts listening
+/// once its template-distribution handshake completes — and tests that intercept that handshake
+/// prevent it from ever listening, by design. Giving up quietly keeps those tests behaving as they
+/// did while still letting healthy roles unblock in milliseconds.
+pub async fn wait_until_listening(addr: SocketAddr, budget: Duration, what: &str) -> bool {
+    let start = std::time::Instant::now();
+    loop {
+        if tokio::net::TcpStream::connect(addr).await.is_ok() {
+            tracing::debug!(
+                target: "readiness",
+                "ready: {what} after {:?}",
+                start.elapsed()
+            );
+            return true;
+        }
+        if start.elapsed() > budget {
+            tracing::debug!(
+                target: "readiness",
+                "{what} not listening on {addr} within {budget:?}, continuing anyway"
+            );
+            return false;
+        }
+        tokio::time::sleep(POLL_INTERVAL).await;
+    }
+}
+
 pub fn get_available_address() -> SocketAddr {
     let port = get_available_port();
     SocketAddr::from(([127, 0, 0, 1], port))
diff --git a/integration-tests/tests/extensions.rs b/integration-tests/tests/extensions.rs
index c59000c1e..08559e557 100644
--- a/integration-tests/tests/extensions.rs
+++ b/integration-tests/tests/extensions.rs
@@ -14,7 +14,10 @@ use integration_tests_sv2::{interceptor::MessageDirection, template_provider::Di
 use stratum_apps::stratum_core::{
     binary_sv2::Seq064KOwned,
     common_messages_sv2::*,
-    extensions_sv2::{EXTENSION_TYPE_WORKER_HASHRATE_TRACKING, TLV_FIELD_TYPE_USER_IDENTITY},
+    extensions_sv2::{
+        EXTENSION_TYPE_WORKER_HASHRATE_TRACKING, TLV_FIELD_TYPE_USER_IDENTITY,
+        extensions_negotiation::MESSAGE_TYPE_REQUEST_EXTENSIONS,
+    },
     mining_sv2::*,
 };
 use tracing::info;
@@ -76,6 +79,15 @@ async fn test_extension_negotiation_with_tlv_in_submit_shares() {
         )
         .await;
 
+    // RequestExtensions is sent *after* SetupConnectionSuccess, so wait for it rather than
+    // assuming it has already been queued by the time the success message lands.
+    pool_translator_sniffer
+        .wait_for_message_type(
+            MessageDirection::ToUpstream,
+            MESSAGE_TYPE_REQUEST_EXTENSIONS,
+        )
+        .await;
+
     // Verify RequestExtensions includes extension 0x0002
     let request_extensions_msg = match pool_translator_sniffer.next_message_from_downstream() {
         Some((
diff --git a/integration-tests/tests/monitoring_integration.rs b/integration-tests/tests/monitoring_integration.rs
index 48feb97b1..49194eddc 100644
--- a/integration-tests/tests/monitoring_integration.rs
+++ b/integration-tests/tests/monitoring_integration.rs
@@ -449,9 +449,14 @@ async fn pool_api_endpoints_with_miner() {
     let client: Sv2ClientResponse = pool_mon.fetch_typed(&routes::client_by_id(client_id)).await;
     assert_eq!(client.client_id, client_id);
 
-    // /api/v1/clients/{id}/channels — at least one channel.
+    // /api/v1/clients/{id}/channels — at least one channel. The channel opens after the client
+    // itself is registered, so poll for it rather than reading once.
     let channels: Sv2ClientChannelsResponse = pool_mon
-        .fetch_typed(&routes::client_channels(client_id))
+        .poll_until(
+            &routes::client_channels(client_id),
+            METRIC_POLL_TIMEOUT,
+            |r: &Sv2ClientChannelsResponse| r.total_standard + r.total_extended >= 1,
+        )
         .await;
     assert_eq!(channels.client_id, client_id);
     assert!(
@@ -759,9 +764,14 @@ async fn jdc_api_endpoints_with_miner() {
     let client: Sv2ClientResponse = jdc_mon.fetch_typed(&routes::client_by_id(client_id)).await;
     assert_eq!(client.client_id, client_id);
 
-    // /api/v1/clients/{id}/channels — tProxy opens an extended channel via JDC.
+    // /api/v1/clients/{id}/channels — tProxy opens an extended channel via JDC. That channel
+    // opens after the client is registered, so poll for it rather than reading once.
     let channels: Sv2ClientChannelsResponse = jdc_mon
-        .fetch_typed(&routes::client_channels(client_id))
+        .poll_until(
+            &routes::client_channels(client_id),
+            METRIC_POLL_TIMEOUT,
+            |r: &Sv2ClientChannelsResponse| r.total_extended >= 1,
+        )
         .await;
     assert_eq!(channels.client_id, client_id);
     assert!(
diff --git a/integration-tests/tests/translator_integration.rs b/integration-tests/tests/translator_integration.rs
index 84989eebb..b24edc949 100644
--- a/integration-tests/tests/translator_integration.rs
+++ b/integration-tests/tests/translator_integration.rs
@@ -1218,6 +1218,16 @@ async fn non_aggregated_translator_correctly_deals_with_group_channels() {
             .expect("Failed to capture prevhash before chain tip update")
     };
 
+    // Drop the mining.notify messages captured above. The assertion below matches the most recent
+    // mining.notify, and the SV1 translation of the new prevhash reaches each miner asynchronously
+    // after the SV2 SetNewPrevHash — so without clearing, the stale pre-update notify is what gets
+    // matched and the prevhash appears unchanged.
+    for sv1_sniffer in sv1_sniffers.iter() {
+        sv1_sniffer
+            .clean_queue(MessageDirection::ToDownstream)
+            .await;
+    }
+
     // now let's force a chain tip update, so we trigger a NewExtendedMiningJob + SetNewPrevHash
     // message pair
     tp.generate_blocks(1);

From a3a036a1ea1dd17a3fd95e1b9395481fd819ed30 Mon Sep 17 00:00:00 2001
From: GitGab19 <gabriele.vernetti@protonmail.com>
Date: Fri, 7 Aug 2026 12:48:18 +0200
Subject: [PATCH 02/15] test(integration): poll on a 200ms interval instead of
 1s
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The suite spent most of its wall clock asleep. Bucketing every timestamp
gap across a full run showed ~613s of dead time in loops that checked a
condition, missed, and then slept a full second — so a message arriving
in 30ms still cost ~1000ms.

Introduces two intervals in `utils` rather than one:

- `POLL_INTERVAL` (200ms) for message-wait loops, which exit as soon as
  their message lands.
- `CONNECT_RETRY_INTERVAL` (1s) for the unbounded connect-retry loops,
  which have no timeout and spin for as long as a peer is absent —
  something several tests arrange deliberately. Every test runs on a bare
  `#[tokio::test]`, i.e. a single-threaded runtime, so those loops must
  not busy-poll against the test's own work.

200ms is not a tuning artefact. Measured across full runs: 1s leaves
~613s of dead time, 200ms leaves ~87s, and 50ms leaves ~61s. The residual
at 50ms is real message latency rather than poll delay, so 200ms already
captures ~96% of everything recoverable and anything in the 150-250ms
range lands in the same place.

50ms was also actively harmful: it starved the single-threaded runtime
and produced a 120s hang (nextest's slow-timeout terminate-after) in 5 of
6 runs. At 200ms: 0 hangs in 7 runs.
---
 integration-tests/lib/mining_device/mod.rs |  4 ++--
 integration-tests/lib/mock_roles.rs        |  6 ++----
 integration-tests/lib/mod.rs               |  7 +++----
 integration-tests/lib/sniffer.rs           | 17 ++++++++---------
 4 files changed, 15 insertions(+), 19 deletions(-)

diff --git a/integration-tests/lib/mining_device/mod.rs b/integration-tests/lib/mining_device/mod.rs
index cdc45be4d..c17a48a51 100644
--- a/integration-tests/lib/mining_device/mod.rs
+++ b/integration-tests/lib/mining_device/mod.rs
@@ -120,10 +120,10 @@ pub async fn connect(
                 Ok(socket) => break socket,
                 Err(e) => {
                     error!(
-                        "Failed to connect to Upstream role at {}, retrying in 5s: {}",
+                        "Failed to connect to Upstream role at {}, retrying: {}",
                         address, e
                     );
-                    tokio::time::sleep(Duration::from_secs(5)).await;
+                    tokio::time::sleep(crate::utils::CONNECT_RETRY_INTERVAL).await;
                 }
             },
             Err(_) => {
diff --git a/integration-tests/lib/mock_roles.rs b/integration-tests/lib/mock_roles.rs
index e9a9ebdf9..b97a46e05 100644
--- a/integration-tests/lib/mock_roles.rs
+++ b/integration-tests/lib/mock_roles.rs
@@ -67,10 +67,8 @@ impl MockDownstream {
             match TcpStream::connect(upstream_address).await {
                 Ok(stream) => break stream,
                 Err(_) => {
-                    tracing::warn!(
-                        "MockDownstream: unable to connect to upstream, retrying after 1 second"
-                    );
-                    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+                    tracing::warn!("MockDownstream: unable to connect to upstream, retrying");
+                    tokio::time::sleep(crate::utils::CONNECT_RETRY_INTERVAL).await;
                 }
             }
         })
diff --git a/integration-tests/lib/mod.rs b/integration-tests/lib/mod.rs
index fd840b5cd..5a9d7094a 100644
--- a/integration-tests/lib/mod.rs
+++ b/integration-tests/lib/mod.rs
@@ -9,7 +9,6 @@ use pool_sv2::PoolSv2;
 use std::{
     convert::TryFrom,
     net::{Ipv4Addr, SocketAddr},
-    time::Duration,
 };
 use stratum_apps::{
     bitcoin_core_sv2::runtime_api::BitcoinCoreVersion,
@@ -20,7 +19,7 @@ use stratum_apps::{
 use tracing::Level;
 use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt};
 use translator_sv2::TranslatorSv2;
-use utils::get_available_address;
+use utils::{ROLE_READY_BUDGET, get_available_address};
 
 pub mod interceptor;
 pub mod message_aggregator;
@@ -172,7 +171,7 @@ pub async fn start_pool(
     tokio::spawn(async move {
         _ = pool_clone.start().await;
     });
-    tokio::time::sleep(Duration::from_secs(1)).await;
+    tokio::time::sleep(ROLE_READY_BUDGET).await;
     (pool, listening_address, monitoring_address)
 }
 
@@ -360,7 +359,7 @@ pub async fn start_pool_with_jds(
     tokio::spawn(async move {
         _ = pool_clone.start().await;
     });
-    tokio::time::sleep(Duration::from_secs(1)).await;
+    tokio::time::sleep(ROLE_READY_BUDGET).await;
     (pool, pool_address, jds_address, monitoring_address)
 }
 
diff --git a/integration-tests/lib/sniffer.rs b/integration-tests/lib/sniffer.rs
index 36984277b..470971908 100644
--- a/integration-tests/lib/sniffer.rs
+++ b/integration-tests/lib/sniffer.rs
@@ -3,8 +3,8 @@ use crate::{
     message_aggregator::MessagesAggregator,
     types::MsgType,
     utils::{
-        create_downstream, create_upstream, recv_from_down_send_to_up, recv_from_up_send_to_down,
-        wait_for_client,
+        CONNECT_RETRY_INTERVAL, POLL_INTERVAL, create_downstream, create_upstream,
+        recv_from_down_send_to_up, recv_from_up_send_to_down, wait_for_client,
     },
 };
 use std::{
@@ -96,11 +96,12 @@ impl<'a> Sniffer<'a> {
                     Ok(stream) => break stream,
                     Err(_) => {
                         tracing::warn!(
-                            "Sniffer {}: unable to connect to upstream {}, retrying after 1 second",
+                            "Sniffer {}: unable to connect to upstream {}, retrying in {:?}",
                             identifier,
-                            upstream_address
+                            upstream_address,
+                            CONNECT_RETRY_INTERVAL
                         );
-                        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+                        tokio::time::sleep(CONNECT_RETRY_INTERVAL).await;
                     }
                 }
             })
@@ -193,8 +194,7 @@ impl<'a> Sniffer<'a> {
                 );
             }
 
-            // sleep to reduce async lock contention
-            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+            tokio::time::sleep(POLL_INTERVAL).await;
         }
     }
 
@@ -257,8 +257,7 @@ impl<'a> Sniffer<'a> {
                 );
             }
 
-            // sleep to reduce async lock contention
-            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+            tokio::time::sleep(POLL_INTERVAL).await;
         }
     }
 

From 27f1bf9c792189527eb8f08057e498447fd471b5 Mon Sep 17 00:00:00 2001
From: GitGab19 <gabriele.vernetti@protonmail.com>
Date: Fri, 7 Aug 2026 12:48:42 +0200
Subject: [PATCH 03/15] test(integration): gate on readiness instead of fixed
 sleeps, drop datadirs
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`TemplateProvider::start` slept a flat 2s after spawning Bitcoin Core and
a further 3s after spawning sv2-tp — 5s per template provider, ~79 times
per run, whether or not either process was ready.

A fixed sleep couples two things that should be independent: how long you
wait when the machine is slow (safety) and how long you wait when it is
fast (speed). Polling separates them, so the ceilings here are *more*
generous than the sleeps they replace (30s) while the common path returns
as soon as the process is actually serving. Measured: Bitcoin Core IPC
socket ready in ~110us, sv2-tp in ~151ms.

Both gates prove serviceability rather than existence. The IPC gate
connects to `node.sock` rather than checking that the path exists,
because datadirs are keyed by port and a stale socket file from an
earlier test can satisfy a `Path::exists` check while refusing traffic —
that failure mode caused `tdp_io_integration_v30x` to run for 108s
against a node that never answered, and `jdp_io_integration_v30x` to time
out entirely.

Note that a connect probe is only safe against a listener that ascribes
no meaning to a bare connection. The pool is not such a listener — it
accepts every connection as a protocol session — so `start_pool` keeps a
plain sleep. Probing it created phantom downstreams that failed setup and
hung `jds_isolates_state_for_colliding_request_ids_across_downstreams` in
4 of 10 runs.

Also removes each node's datadir on drop. Nothing cleaned them up, so a
full run left ~1.5GB of `.bitcoin-{port}` directories behind and
successive runs accumulated until the filesystem filled. Retention is
still available via SV2_KEEP_TEST_DATADIR for post-mortem inspection of
debug.log and chainstate, since that is plausibly why a persistent
`staticdir` was chosen in the first place.
---
 integration-tests/lib/template_provider.rs | 67 +++++++++++++++++++---
 1 file changed, 59 insertions(+), 8 deletions(-)

diff --git a/integration-tests/lib/template_provider.rs b/integration-tests/lib/template_provider.rs
index fdabc1369..139db73b5 100644
--- a/integration-tests/lib/template_provider.rs
+++ b/integration-tests/lib/template_provider.rs
@@ -2,6 +2,7 @@ use corepc_node::{Conf, ConnectParams, Node, types::GetBlockchainInfo};
 use std::{
     env,
     fs::create_dir_all,
+    net::SocketAddr,
     path::PathBuf,
     process::{Child, Command, Stdio},
 };
@@ -11,7 +12,9 @@ use stratum_apps::{
 };
 use tracing::warn;
 
-use crate::utils::{fs_utils, http, tarball};
+use crate::utils::{
+    PROCESS_READY_TIMEOUT, fs_utils, http, tarball, wait_for_listener, wait_for_unix_socket,
+};
 
 const VERSION_SV2_TP: &str = "1.1.0";
 const BITCOIN_CORE_V30X: &str = "30.2";
@@ -252,17 +255,25 @@ impl BitcoinCore {
             }
         };
 
-        // Wait for Bitcoin Core to fully start and create IPC socket
-        std::thread::sleep(std::time::Duration::from_secs(2));
-
         let is_signet = conf.network == "signet";
         let data_dir = conf.staticdir.clone().expect("staticdir should be set");
 
-        BitcoinCore {
+        let core = BitcoinCore {
             bitcoind,
             data_dir,
             is_signet,
-        }
+        };
+
+        // Wait for Bitcoin Core to create the IPC socket that sv2-tp connects to. Polling the
+        // socket path is the direct readiness signal, so this returns as soon as Core is actually
+        // up instead of always paying a fixed sleep.
+        wait_for_unix_socket(
+            &core.ipc_socket_path(),
+            PROCESS_READY_TIMEOUT,
+            "Bitcoin Core IPC socket",
+        );
+
+        core
     }
 
     /// Mine `n` blocks.
@@ -348,6 +359,41 @@ impl BitcoinCore {
     }
 }
 
+/// Set to keep each node's datadir after the test finishes, for post-mortem inspection of
+/// `debug.log`, chainstate and wallet.
+///
+/// Retained datadirs are not cleaned up by anything else, and a full suite run creates one per
+/// node — roughly 1.5 GB — so this is opt-in rather than the default. Left on, successive runs
+/// accumulate until the filesystem fills, which on a tmpfs-backed `/tmp` means RAM and on a CI
+/// runner means a disk with only a few GB spare.
+const KEEP_DATADIR_ENV: &str = "SV2_KEEP_TEST_DATADIR";
+
+impl Drop for BitcoinCore {
+    fn drop(&mut self) {
+        if env::var_os(KEEP_DATADIR_ENV).is_some() {
+            warn!(
+                "{KEEP_DATADIR_ENV} set, keeping test datadir {}",
+                self.data_dir.display()
+            );
+            return;
+        }
+
+        // The node must be stopped before its files are removed. `Drop` for this struct runs
+        // *before* its fields are dropped, so `bitcoind` is still live here; `Node::drop` would
+        // otherwise try a graceful RPC shutdown against a datadir that no longer exists.
+        let _ = self.bitcoind.stop();
+
+        match std::fs::remove_dir_all(&self.data_dir) {
+            Ok(()) => {}
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+            Err(e) => warn!(
+                "failed to remove test datadir {}: {e}",
+                self.data_dir.display()
+            ),
+        }
+    }
+}
+
 /// Represents a template provider using Bitcoin Core with IPC and standalone sv2-tp.
 ///
 /// This implementation launches two separate processes:
@@ -427,8 +473,13 @@ impl TemplateProvider {
             .spawn()
             .expect("Failed to start sv2-tp process");
 
-        // Wait for sv2-tp to start and connect to Bitcoin Core
-        std::thread::sleep(std::time::Duration::from_secs(3));
+        // Wait for sv2-tp to start and accept connections on its Sv2 port. A successful connect is
+        // the direct readiness signal, so this returns as soon as sv2-tp is actually serving.
+        wait_for_listener(
+            SocketAddr::from(([127, 0, 0, 1], port)),
+            PROCESS_READY_TIMEOUT,
+            "sv2-tp",
+        );
 
         TemplateProvider {
             bitcoin_core,

From 92f0e9574376dae3c69ec850422e2cc2c0b1f449 Mon Sep 17 00:00:00 2001
From: GitGab19 <gabriele.vernetti@protonmail.com>
Date: Fri, 7 Aug 2026 13:38:10 +0200
Subject: [PATCH 04/15] test(integration): stop probing Bitcoin Core's IPC
 socket with a connect
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The IPC readiness gate connected to node.sock and dropped the connection
immediately. Core's libmultiprocess layer sets TCP_NODELAY on every
accepted connection, so that setsockopt ran against a socket which was
already gone. Linux tolerates it; macOS returns EINVAL, which surfaced as

  mp/proxy.cpp:45: error: Uncaught exception in daemonized task.;
  exception = kj/async-io-unix.c++:1365: failed:
  setsocketopt(IPPROTO_TCP, TCP_NODELAY): Invalid argument

and took down Core's IPC listener. sv2-tp could then never connect, so it
never bound its Sv2 port, and the gate after it failed with

  timeout after 30s waiting for sv2-tp to listen on 127.0.0.1:49215

This is the rule already documented for start_pool in efc4dca — a connect
probe is only safe against a listener that ascribes no meaning to a bare
connection — applied to a capnp RPC endpoint, where it plainly does not
hold. It went unnoticed because every run validating this branch was on
Linux.

The gate now waits for the socket to appear rather than connecting to it.
Stat-ing was previously rejected because a stale node.sock from an earlier
test on the same port could satisfy it, so this also removes any leftover
datadir for the port before starting the node, which closes that window
directly instead of by probing around it.
---
 integration-tests/lib/template_provider.rs | 30 +++++++++++++++----
 integration-tests/lib/utils.rs             | 35 ++++++++++++----------
 2 files changed, 44 insertions(+), 21 deletions(-)

diff --git a/integration-tests/lib/template_provider.rs b/integration-tests/lib/template_provider.rs
index 139db73b5..f199fc894 100644
--- a/integration-tests/lib/template_provider.rs
+++ b/integration-tests/lib/template_provider.rs
@@ -13,7 +13,7 @@ use stratum_apps::{
 use tracing::warn;
 
 use crate::utils::{
-    PROCESS_READY_TIMEOUT, fs_utils, http, tarball, wait_for_listener, wait_for_unix_socket,
+    PROCESS_READY_TIMEOUT, fs_utils, http, tarball, wait_for_listener, wait_for_path,
 };
 
 const VERSION_SV2_TP: &str = "1.1.0";
@@ -119,6 +119,16 @@ impl BitcoinCore {
         let staticdir = format!(".bitcoin-{port}");
         conf.staticdir = Some(data_dir.join(staticdir.clone()));
 
+        // Remove any datadir left from an earlier test that drew the same port. Ports come from
+        // the OS, so reuse is rare but possible, and `BitcoinCore`'s cleanup does not run if a
+        // previous run was killed. Without this a stale `node.sock` can satisfy the readiness
+        // check below while belonging to a node that is long gone.
+        if let Err(e) = std::fs::remove_dir_all(conf.staticdir.as_ref().unwrap())
+            && e.kind() != std::io::ErrorKind::NotFound
+        {
+            warn!("failed to clear stale datadir for port {port}: {e}");
+        }
+
         let max_tip_age_arg = format!("-maxtipage={SIGNET_FIXTURE_MAX_TIP_AGE_SECS}");
         match difficulty_level {
             DifficultyLevel::Low => {
@@ -264,10 +274,20 @@ impl BitcoinCore {
             is_signet,
         };
 
-        // Wait for Bitcoin Core to create the IPC socket that sv2-tp connects to. Polling the
-        // socket path is the direct readiness signal, so this returns as soon as Core is actually
-        // up instead of always paying a fixed sleep.
-        wait_for_unix_socket(
+        // Wait for Bitcoin Core to create the IPC socket that sv2-tp connects to.
+        //
+        // This waits for the socket to *appear* and deliberately does not connect to it. Core's
+        // libmultiprocess layer sets TCP_NODELAY on every accepted connection; probing with a
+        // connect that is immediately dropped makes that setsockopt run against a socket which is
+        // already gone, which returns EINVAL on macOS and kills Core's IPC listener with an
+        // "Uncaught exception in daemonized task". sv2-tp can then never connect. A capnp RPC
+        // endpoint ascribes meaning to a bare connection, so it must not be used as a liveness
+        // probe.
+        //
+        // Waiting on the path alone is sound here because any datadir left over from an earlier
+        // test using this port is removed above, so the socket that appears can only be the one
+        // this node just created.
+        wait_for_path(
             &core.ipc_socket_path(),
             PROCESS_READY_TIMEOUT,
             "Bitcoin Core IPC socket",
diff --git a/integration-tests/lib/utils.rs b/integration-tests/lib/utils.rs
index a4a06ee3b..13182a60c 100644
--- a/integration-tests/lib/utils.rs
+++ b/integration-tests/lib/utils.rs
@@ -65,33 +65,36 @@ pub const PROCESS_READY_TIMEOUT: Duration = Duration::from_secs(30);
 /// the old `sleep(1s)` cost. See [`wait_until_listening`] for why this cannot be an assertion.
 pub const ROLE_READY_BUDGET: Duration = Duration::from_secs(1);
 
-/// Blocks until a Unix socket at `path` accepts a connection, polling every [`POLL_INTERVAL`].
+/// Blocks until `path` exists, polling every [`POLL_INTERVAL`].
 ///
-/// Deliberately connects rather than checking `Path::exists`: Bitcoin Core creates the socket file
-/// before it is able to serve on it, and the window between the two is wide enough that merely
-/// stat-ing the path lets tests proceed against a node that then refuses their IPC traffic. A
-/// successful connect proves the listener is bound and accepting.
+/// Deliberately stats the path rather than connecting to it. This gates on Bitcoin Core's IPC
+/// socket, and a capnp RPC endpoint ascribes meaning to a bare connection: Core's libmultiprocess
+/// layer sets TCP_NODELAY on each accepted connection, so a probe that connects and immediately
+/// drops makes that setsockopt run against a socket that is already gone. On macOS that returns
+/// EINVAL and takes down Core's IPC listener with an "Uncaught exception in daemonized task",
+/// after which sv2-tp can never connect. Linux tolerates the same call, so this only reproduces
+/// on macOS.
+///
+/// Stat-ing is sound provided callers remove any datadir left over from an earlier test on the
+/// same port, so the socket that appears can only belong to the node just started.
 ///
 /// Panics with `what` in the message if `timeout` elapses first.
-pub fn wait_for_unix_socket(path: &std::path::Path, timeout: Duration, what: &str) {
+pub fn wait_for_path(path: &std::path::Path, timeout: Duration, what: &str) {
     let start = std::time::Instant::now();
-    loop {
-        if path.exists() && std::os::unix::net::UnixStream::connect(path).is_ok() {
-            tracing::debug!(
-                target: "readiness",
-                "ready: {what} after {:?}",
-                start.elapsed()
-            );
-            return;
-        }
+    while !path.exists() {
         if start.elapsed() > timeout {
             panic!(
-                "timeout after {timeout:?} waiting for {what} to accept connections on {}",
+                "timeout after {timeout:?} waiting for {what} (path {} never appeared)",
                 path.display()
             );
         }
         std::thread::sleep(POLL_INTERVAL);
     }
+    tracing::debug!(
+        target: "readiness",
+        "ready: {what} after {:?}",
+        start.elapsed()
+    );
 }
 
 /// Blocks until a TCP connection to `addr` succeeds, polling every [`POLL_INTERVAL`].

From d7a7ca27a7a5106ea317cfb9a45170dbcbea0cb2 Mon Sep 17 00:00:00 2001
From: plebhash <plebhash@proton.me>
Date: Sat, 8 Aug 2026 11:24:59 -0300
Subject: [PATCH 05/15] test(integration): make port allocation race-free
 across test processes

Replace the per-process-only UNIQUE_PORTS in-process dedup set with
flock-based per-port lockfiles held for the process lifetime.  Two
concurrent nextest test processes probing bind(0) to find free ports
can now never pick the same port because the non-blocking exclusive
flock is forced to fail on the loser.

Bitcoin Core datadirs are keyed by port (.bitcoin-{port}), so a
port collision meant two nodes sharing one datadir, not just a bind
error.  The lockfiles live under $TMPDIR/sv2-it-ports/ and are
released by the kernel on exit (including kill -9), so stale files
are harmless and never need cleanup.
---
 integration-tests/Cargo.toml   |  1 +
 integration-tests/lib/utils.rs | 70 ++++++++++++++++++++++++++--------
 2 files changed, 56 insertions(+), 15 deletions(-)

diff --git a/integration-tests/Cargo.toml b/integration-tests/Cargo.toml
index 301f5115d..789b35293 100644
--- a/integration-tests/Cargo.toml
+++ b/integration-tests/Cargo.toml
@@ -31,6 +31,7 @@ hex = "0.4.3"
 clap = { version = "^4.5.4", features = ["derive"] }
 serde_json = "1"
 serde = { version = "1", features = ["derive"] }
+libc = "0.2"
 
 # Direct dependencies kept only for the embedded `mining_device` module.
 # Remove this block when removing:
diff --git a/integration-tests/lib/utils.rs b/integration-tests/lib/utils.rs
index 13182a60c..67b87ed01 100644
--- a/integration-tests/lib/utils.rs
+++ b/integration-tests/lib/utils.rs
@@ -7,9 +7,10 @@ use crate::{
 use async_channel::{Receiver, Sender};
 use once_cell::sync::Lazy;
 use std::{
-    collections::HashSet,
     convert::TryInto,
+    fs, io,
     net::{SocketAddr, TcpListener, TcpStream},
+    os::fd::AsRawFd,
     sync::{Arc, Mutex},
     time::Duration,
 };
@@ -28,8 +29,36 @@ use stratum_apps::{
 };
 use tokio_util::sync::CancellationToken;
 
-// prevents get_available_port from ever returning the same port twice
-static UNIQUE_PORTS: Lazy<Mutex<HashSet<u16>>> = Lazy::new(|| Mutex::new(HashSet::new()));
+/// Advisory per-port lockfiles held for the process lifetime so no two concurrent
+/// test processes can claim the same port. The kernel releases flock on exit (even
+/// after `kill -9`), so stale lockfiles are harmless.
+static HELD_LOCKS: Lazy<Mutex<Vec<std::fs::File>>> = Lazy::new(|| Mutex::new(Vec::new()));
+
+fn lockfile_for(port: u16) -> std::path::PathBuf {
+    std::env::temp_dir()
+        .join("sv2-it-ports")
+        .join(format!("{port}.lock"))
+}
+
+fn try_lock_port(port: u16) -> io::Result<std::fs::File> {
+    let path = lockfile_for(port);
+    if let Some(parent) = path.parent() {
+        fs::create_dir_all(parent)?;
+    }
+    let file = std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(&path)?;
+    // Non-blocking exclusive lock: EAGAIN → another process holds this port.
+    // SAFETY: fd is valid, flock is async-signal-safe on both Linux and macOS.
+    let fd = file.as_raw_fd();
+    if unsafe { libc::flock(fd, libc::LOCK_EX | libc::LOCK_NB) } != 0 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(file)
+}
 
 /// How often readiness gates and message-wait loops re-check their condition.
 ///
@@ -151,23 +180,34 @@ pub async fn wait_until_listening(addr: SocketAddr, budget: Duration, what: &str
     }
 }
 
+/// Allocate a loopback address whose port is exclusively reserved across
+/// concurrent test processes.
+///
+/// Probes with `bind(0)` then holds a per-port `flock` lockfile in
+/// `$TMPDIR/sv2-it-ports/` for the process lifetime — no TOCTOU window
+/// between probe and bind.  Lockfiles are released by the kernel on exit.
 pub fn get_available_address() -> SocketAddr {
-    let port = get_available_port();
-    SocketAddr::from(([127, 0, 0, 1], port))
+    SocketAddr::from(([127, 0, 0, 1], get_available_port()))
 }
 
 fn get_available_port() -> u16 {
-    let mut unique_ports = UNIQUE_PORTS.lock().unwrap();
-
     loop {
-        let port = TcpListener::bind("127.0.0.1:0")
-            .unwrap()
-            .local_addr()
-            .unwrap()
-            .port();
-        if !unique_ports.contains(&port) {
-            unique_ports.insert(port);
-            return port;
+        let probe = TcpListener::bind("127.0.0.1:0")
+            .expect("bind(0) for port probe");
+        let port = probe.local_addr().expect("probe local_addr").port();
+        match try_lock_port(port) {
+            Ok(lock_handle) => {
+                HELD_LOCKS.lock().expect("ports lock").push(lock_handle);
+                drop(probe);
+                return port;
+            }
+            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
+                drop(probe);
+            }
+            Err(e) => {
+                drop(probe);
+                panic!("failed to lock port {port}: {e}");
+            }
         }
     }
 }

From ad5a3ebd1800501988f1152ffc0ad766ea7b9ccb Mon Sep 17 00:00:00 2001
From: plebhash <plebhash@proton.me>
Date: Sat, 8 Aug 2026 11:29:35 -0300
Subject: [PATCH 06/15] test(integration): serialize template-provider artifact
 downloads across processes

Concurrent nextest processes on a cold CI runner would all see the
bitcoin-core / sv2-tp / high_diff_chain directories as missing and
simultaneously download+unpack into the same shared tree.

Guard each artifact with a blocking exclusive flock so only the first
process actually downloads; the rest wait on the lock, then skip via
an inner exists() re-check.  Also give tarball::unpack a pid-unique
temp filename so two processes can never share the same staging file
even if the lock is bypassed.
---
 integration-tests/lib/template_provider.rs | 94 ++++++++++++++++++----
 integration-tests/lib/utils.rs             |  5 +-
 2 files changed, 80 insertions(+), 19 deletions(-)

diff --git a/integration-tests/lib/template_provider.rs b/integration-tests/lib/template_provider.rs
index f199fc894..ec35e6427 100644
--- a/integration-tests/lib/template_provider.rs
+++ b/integration-tests/lib/template_provider.rs
@@ -15,6 +15,40 @@ use tracing::warn;
 use crate::utils::{
     PROCESS_READY_TIMEOUT, fs_utils, http, tarball, wait_for_listener, wait_for_path,
 };
+use std::os::fd::AsRawFd;
+
+/// Acquire a blocking exclusive flock on `lock_path`, run `f`, then release.
+///
+/// Used to serialize download+unpack of shared artifacts (bitcoin-core,
+/// sv2-tp, high_diff_chain) across concurrently executing nextest processes.
+/// The lock is held only for the window where the artifact is missing; every
+/// process re-checks `exists()` inside the guard so only the first one
+/// actually downloads.
+fn with_exclusive_lock(lock_path: &std::path::Path, f: impl FnOnce()) {
+    if let Some(parent) = lock_path.parent() {
+        std::fs::create_dir_all(parent).ok();
+    }
+    let file = std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(lock_path)
+        .expect("open lockfile for artifact download");
+    let fd = file.as_raw_fd();
+    loop {
+        // SAFETY: fd is valid, flock is async-signal-safe on Linux and macOS.
+        if unsafe { libc::flock(fd, libc::LOCK_EX) } == 0 {
+            break;
+        }
+        let e = std::io::Error::last_os_error();
+        if e.kind() != std::io::ErrorKind::Interrupted {
+            panic!("flock on artifact lockfile: {e}");
+        }
+    }
+    f();
+    // lock released when file is dropped
+}
 
 const VERSION_SV2_TP: &str = "1.1.0";
 const BITCOIN_CORE_V30X: &str = "30.2";
@@ -163,22 +197,30 @@ impl BitcoinCore {
                 let signet_datadir = data_dir.join(staticdir.clone()).join("signet");
                 create_dir_all(signet_datadir.clone()).expect("Failed to create signet directory");
 
-                // Download and cache high difficulty chain if not exists
+                // Download and cache high difficulty chain if not exists.
+                // Guarded with a flock so concurrent test processes don't race
+                // on the unpack.
                 if !high_diff_chain_dir.exists() {
-                    let local_tarball =
-                        current_dir.join("resources").join("high_diff_chain.tar.gz");
-                    let tarball_bytes = if local_tarball.exists() {
-                        warn!("Using local high_diff_chain.tar.gz");
-                        tarball::read_from_file(local_tarball.to_str().unwrap())
-                    } else {
-                        warn!("Downloading high_diff_chain for the testing session...");
-                        //this is pinning to the commit right before the current one, where I added
-                        //the tar.gz file with the chain data.
-                        let url = "https://raw.githubusercontent.com/stratum-mining/sv2-apps/eb41b790626fb51ce55e74be8fa0b4f07d4029bf/integration-tests/resources/high_diff_chain.tar.gz";
-                        http::make_get_request(url, 5)
-                    };
-
-                    tarball::unpack(&tarball_bytes, &bin_dir);
+                    with_exclusive_lock(
+                        &bin_dir.join(".locks").join("high_diff_chain.lock"),
+                        || {
+                            if !high_diff_chain_dir.exists() {
+                                let local_tarball = current_dir
+                                    .join("resources")
+                                    .join("high_diff_chain.tar.gz");
+                                let tarball_bytes = if local_tarball.exists() {
+                                    warn!("Using local high_diff_chain.tar.gz");
+                                    tarball::read_from_file(local_tarball.to_str().unwrap())
+                                } else {
+                                    warn!("Downloading high_diff_chain for the testing session...");
+                                    let url = "https://raw.githubusercontent.com/stratum-mining/sv2-apps/eb41b790626fb51ce55e74be8fa0b4f07d4029bf/integration-tests/resources/high_diff_chain.tar.gz";
+                                    http::make_get_request(url, 5)
+                                };
+
+                                tarball::unpack(&tarball_bytes, &bin_dir);
+                            }
+                        },
+                    );
                 }
 
                 // Copy high difficulty signet data into signet datadir
@@ -197,6 +239,12 @@ impl BitcoinCore {
         let bitcoin_cli_bin = bitcoin_home.join("bin").join("bitcoin-cli");
 
         if !bitcoin_node_bin.exists() {
+            with_exclusive_lock(
+                &bin_dir
+                    .join(".locks")
+                    .join(format!("bitcoin-{bitcoin_core_version}.lock")),
+                || {
+                    if !bitcoin_node_bin.exists() {
             let tarball_bytes = match env::var("BITCOIN_CORE_TARBALL_FILE") {
                 Ok(path) => tarball::read_from_file(&path),
                 Err(_) => {
@@ -238,7 +286,10 @@ impl BitcoinCore {
                         .expect("Failed to sign Bitcoin Core binary");
                 }
             }
-        }
+                    } // inner re-check: bin still missing?
+                }, // closure
+            ); // with_exclusive_lock
+        } // outer exists() guard
 
         // Add IPC and basic args
         conf.args.extend(vec![
@@ -442,6 +493,12 @@ impl TemplateProvider {
         let sv2_tp_bin = sv2_tp_home.join("bin").join("sv2-tp");
 
         if !sv2_tp_bin.exists() {
+            with_exclusive_lock(
+                &bin_dir
+                    .join(".locks")
+                    .join(format!("sv2-tp-{VERSION_SV2_TP}.lock")),
+                || {
+                    if !sv2_tp_bin.exists() {
             let tarball_bytes = match env::var("SV2TP_TARBALL_FILE") {
                 Ok(path) => tarball::read_from_file(&path),
                 Err(_) => {
@@ -470,7 +527,10 @@ impl TemplateProvider {
                     .output()
                     .expect("Failed to sign sv2-tp binary");
             }
-        }
+                    } // inner re-check: bin still missing?
+                }, // closure
+            ); // with_exclusive_lock
+        } // outer exists() guard
 
         // Launch sv2-tp process
         let datadir = bitcoin_core.data_dir();
diff --git a/integration-tests/lib/utils.rs b/integration-tests/lib/utils.rs
index 67b87ed01..3b46ded30 100644
--- a/integration-tests/lib/utils.rs
+++ b/integration-tests/lib/utils.rs
@@ -599,8 +599,9 @@ pub mod tarball {
     pub fn unpack(tarball_bytes: &[u8], destination: &Path) {
         use std::{io::Write as IoWrite, process::Command};
 
-        // Write tarball bytes to a temp file
-        let temp_tarball = destination.join("temp.tar.gz");
+        // Use pid-unique temp name so concurrent test processes can't share it.
+        let pid = std::process::id();
+        let temp_tarball = destination.join(format!("temp-{pid}.tar.gz"));
         let mut temp_file = File::create(&temp_tarball).unwrap();
         temp_file.write_all(tarball_bytes).unwrap();
         drop(temp_file);

From 72693ee77ea9c72a914c695f2b5e664e3af036d2 Mon Sep 17 00:00:00 2001
From: plebhash <plebhash@proton.me>
Date: Sat, 8 Aug 2026 11:29:57 -0300
Subject: [PATCH 07/15] test(integration): pin harness mining devices to one
 worker thread

The mining_device auto mode spawns (logical_cpus - 1) hashing threads per
test process.  With test-threads > 1 this oversubscribes the CI runner and
makes timing-sensitive share-rate assertions flaky (SHARES_PER_MINUTE).

Call set_cores(1) in the test-harness helper start_mining_device_sv2 so
every test that mines through the harness gets a single hashing thread.
The standalone binary and benches (not part of ITFCI) keep auto mode.
---
 integration-tests/lib/mod.rs | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/integration-tests/lib/mod.rs b/integration-tests/lib/mod.rs
index 5a9d7094a..a0b89e4ed 100644
--- a/integration-tests/lib/mod.rs
+++ b/integration-tests/lib/mod.rs
@@ -555,6 +555,9 @@ pub fn start_mining_device_sv2(
     nominal_hashrate_multiplier: Option<f32>,
     single_submit: bool,
 ) {
+    // ponytail: pin to 1 thread so concurrent test processes don't
+    // oversubscribe the CI runner's CPUs and flake share-rate assertions.
+    crate::mining_device::set_cores(1);
     tokio::spawn(async move {
         crate::mining_device::connect(
             upstream.to_string(),

From 532b720db88a8286a16c3bf1691af3e7b154a18a Mon Sep 17 00:00:00 2001
From: plebhash <plebhash@proton.me>
Date: Sat, 8 Aug 2026 11:30:19 -0300
Subject: [PATCH 08/15] test(integration): run nextest at num-cpus instead of
 serial

Replace test-threads = 1 with test-threads = "num-cpus" so the
Integration Tests CI drains the job queue faster on free-tier runners
(4 vCPU on ubuntu-latest, 3 on macos-latest per GitHub docs).

Comment documents the NEXTEST_TEST_THREADS=1 override for local
sequential debugging.
---
 integration-tests/.config/nextest.toml | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/integration-tests/.config/nextest.toml b/integration-tests/.config/nextest.toml
index 5f5d26a09..1b31f5f1a 100644
--- a/integration-tests/.config/nextest.toml
+++ b/integration-tests/.config/nextest.toml
@@ -5,8 +5,14 @@
 # and that's a reliable indication that we shouldn't merge this PR
 retries = { backoff = "fixed", count = 3, delay = "2s" }
 
-# only run one test at a time, which allows a human-friendly experience for inspecting logs
-test-threads = 1
+# Number of test processes to run concurrently.
+#
+# "num-cpus" matches the runner (4 vCPU on ubuntu-latest, 3 vCPU on
+# macos-latest — both free-tier public repos per GitHub docs).
+#
+# For sequential human-readable logs (one test at a time), override locally:
+#     NEXTEST_TEST_THREADS=1 cargo nextest run ...
+test-threads = "num-cpus"
 
 # label as slow if a test runs for more than 60s
 # kill it after 120s

From 5760cd78489f16b65768a9dc89ec24c74aa7192b Mon Sep 17 00:00:00 2001
From: plebhash <plebhash@proton.me>
Date: Sat, 8 Aug 2026 11:30:19 -0300
Subject: [PATCH 09/15] ci(integration): drop --nocapture so concurrent output
 stays readable

With test-threads > 1, live stdout from N interleaving tests is
unreadable.  nextest captures per-test output and prints it only for
failing/flaky tests at the end.  The RUST_LOG=debug logs remain
captured; re-run with --no-capture is documented in the step comment
for manual triage.
---
 .github/workflows/integration-tests.yaml | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/integration-tests.yaml b/.github/workflows/integration-tests.yaml
index 68bb35be9..c43de27ea 100644
--- a/.github/workflows/integration-tests.yaml
+++ b/.github/workflows/integration-tests.yaml
@@ -48,6 +48,9 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: cargo binstall --no-confirm --disable-telemetry --disable-strategies quick-install --locked cargo-nextest@0.9.100
 
+# Failure triage: re-run a single test with captured logs shown live:
+      #   RUST_LOG=debug cargo nextest run --manifest-path=integration-tests/Cargo.toml \
+      #     -E 'test(<name>)' --no-capture
       - name: Integration Tests
         run: |
-         RUST_BACKTRACE=1 RUST_LOG=debug cargo nextest run --manifest-path=integration-tests/Cargo.toml --nocapture
+          RUST_BACKTRACE=1 RUST_LOG=debug cargo nextest run --manifest-path=integration-tests/Cargo.toml

From 40db1adf568f58de17ab629d2d3bb66887428c37 Mon Sep 17 00:00:00 2001
From: plebhash <plebhash@proton.me>
Date: Sat, 8 Aug 2026 11:31:06 -0300
Subject: [PATCH 10/15] style(integration): cargo fmt + Cargo.lock update for
 libc dep

---
 integration-tests/Cargo.lock               |   1 +
 integration-tests/lib/template_provider.rs | 138 +++++++++++----------
 integration-tests/lib/utils.rs             |   3 +-
 3 files changed, 73 insertions(+), 69 deletions(-)

diff --git a/integration-tests/Cargo.lock b/integration-tests/Cargo.lock
index 71923cf17..63b6566d9 100644
--- a/integration-tests/Cargo.lock
+++ b/integration-tests/Cargo.lock
@@ -2678,6 +2678,7 @@ dependencies = [
  "half",
  "hex",
  "jd_client_sv2",
+ "libc",
  "minreq",
  "num-format",
  "num_cpus",
diff --git a/integration-tests/lib/template_provider.rs b/integration-tests/lib/template_provider.rs
index ec35e6427..6fcf96a32 100644
--- a/integration-tests/lib/template_provider.rs
+++ b/integration-tests/lib/template_provider.rs
@@ -205,9 +205,8 @@ impl BitcoinCore {
                         &bin_dir.join(".locks").join("high_diff_chain.lock"),
                         || {
                             if !high_diff_chain_dir.exists() {
-                                let local_tarball = current_dir
-                                    .join("resources")
-                                    .join("high_diff_chain.tar.gz");
+                                let local_tarball =
+                                    current_dir.join("resources").join("high_diff_chain.tar.gz");
                                 let tarball_bytes = if local_tarball.exists() {
                                     warn!("Using local high_diff_chain.tar.gz");
                                     tarball::read_from_file(local_tarball.to_str().unwrap())
@@ -245,47 +244,47 @@ impl BitcoinCore {
                     .join(format!("bitcoin-{bitcoin_core_version}.lock")),
                 || {
                     if !bitcoin_node_bin.exists() {
-            let tarball_bytes = match env::var("BITCOIN_CORE_TARBALL_FILE") {
-                Ok(path) => tarball::read_from_file(&path),
-                Err(_) => {
-                    warn!(
-                        "Downloading Bitcoin Core {} for the testing session. This could take a while...",
-                        bitcoin_core_version
-                    );
-                    let download_endpoint = env::var("BITCOIN_CORE_DOWNLOAD_ENDPOINT")
+                        let tarball_bytes = match env::var("BITCOIN_CORE_TARBALL_FILE") {
+                            Ok(path) => tarball::read_from_file(&path),
+                            Err(_) => {
+                                warn!(
+                                    "Downloading Bitcoin Core {} for the testing session. This could take a while...",
+                                    bitcoin_core_version
+                                );
+                                let download_endpoint = env::var("BITCOIN_CORE_DOWNLOAD_ENDPOINT")
                         .unwrap_or_else(|_| {
                             format!(
                                 "https://bitcoincore.org/bin/bitcoin-core-{bitcoin_core_version}"
                             )
                         });
-                    let url = format!("{download_endpoint}/{bitcoin_filename}");
-                    http::make_get_request(&url, 5)
-                }
-            };
-
-            if let Some(parent) = bitcoin_home.parent() {
-                create_dir_all(parent).unwrap();
-            }
-
-            tarball::unpack(&tarball_bytes, &bin_dir);
-
-            assert!(
-                bitcoin_node_bin.exists(),
-                "Bitcoin Core node binary not found after unpack in {}",
-                bitcoin_home.display()
-            );
-
-            // Sign the binaries on macOS
-            if os == "macos" {
-                for bin in &[&bitcoin_node_bin, &bitcoin_cli_bin] {
-                    std::process::Command::new("codesign")
-                        .arg("--sign")
-                        .arg("-")
-                        .arg(bin)
-                        .output()
-                        .expect("Failed to sign Bitcoin Core binary");
-                }
-            }
+                                let url = format!("{download_endpoint}/{bitcoin_filename}");
+                                http::make_get_request(&url, 5)
+                            }
+                        };
+
+                        if let Some(parent) = bitcoin_home.parent() {
+                            create_dir_all(parent).unwrap();
+                        }
+
+                        tarball::unpack(&tarball_bytes, &bin_dir);
+
+                        assert!(
+                            bitcoin_node_bin.exists(),
+                            "Bitcoin Core node binary not found after unpack in {}",
+                            bitcoin_home.display()
+                        );
+
+                        // Sign the binaries on macOS
+                        if os == "macos" {
+                            for bin in &[&bitcoin_node_bin, &bitcoin_cli_bin] {
+                                std::process::Command::new("codesign")
+                                    .arg("--sign")
+                                    .arg("-")
+                                    .arg(bin)
+                                    .output()
+                                    .expect("Failed to sign Bitcoin Core binary");
+                            }
+                        }
                     } // inner re-check: bin still missing?
                 }, // closure
             ); // with_exclusive_lock
@@ -499,34 +498,39 @@ impl TemplateProvider {
                     .join(format!("sv2-tp-{VERSION_SV2_TP}.lock")),
                 || {
                     if !sv2_tp_bin.exists() {
-            let tarball_bytes = match env::var("SV2TP_TARBALL_FILE") {
-                Ok(path) => tarball::read_from_file(&path),
-                Err(_) => {
-                    warn!("Downloading sv2-tp for the testing session. This could take a while...");
-                    let download_endpoint =
-                        env::var("SV2TP_DOWNLOAD_ENDPOINT").unwrap_or_else(|_| {
-                            "https://github.com/stratum-mining/sv2-tp/releases/download".to_owned()
-                        });
-                    let url = format!("{download_endpoint}/v{VERSION_SV2_TP}/{sv2_tp_filename}");
-                    http::make_get_request(&url, 5)
-                }
-            };
-
-            if let Some(parent) = sv2_tp_home.parent() {
-                create_dir_all(parent).unwrap();
-            }
-
-            tarball::unpack(&tarball_bytes, &bin_dir);
-
-            // Sign the binary on macOS
-            if os == "macos" {
-                std::process::Command::new("codesign")
-                    .arg("--sign")
-                    .arg("-")
-                    .arg(&sv2_tp_bin)
-                    .output()
-                    .expect("Failed to sign sv2-tp binary");
-            }
+                        let tarball_bytes = match env::var("SV2TP_TARBALL_FILE") {
+                            Ok(path) => tarball::read_from_file(&path),
+                            Err(_) => {
+                                warn!(
+                                    "Downloading sv2-tp for the testing session. This could take a while..."
+                                );
+                                let download_endpoint = env::var("SV2TP_DOWNLOAD_ENDPOINT")
+                                    .unwrap_or_else(|_| {
+                                        "https://github.com/stratum-mining/sv2-tp/releases/download"
+                                            .to_owned()
+                                    });
+                                let url = format!(
+                                    "{download_endpoint}/v{VERSION_SV2_TP}/{sv2_tp_filename}"
+                                );
+                                http::make_get_request(&url, 5)
+                            }
+                        };
+
+                        if let Some(parent) = sv2_tp_home.parent() {
+                            create_dir_all(parent).unwrap();
+                        }
+
+                        tarball::unpack(&tarball_bytes, &bin_dir);
+
+                        // Sign the binary on macOS
+                        if os == "macos" {
+                            std::process::Command::new("codesign")
+                                .arg("--sign")
+                                .arg("-")
+                                .arg(&sv2_tp_bin)
+                                .output()
+                                .expect("Failed to sign sv2-tp binary");
+                        }
                     } // inner re-check: bin still missing?
                 }, // closure
             ); // with_exclusive_lock
diff --git a/integration-tests/lib/utils.rs b/integration-tests/lib/utils.rs
index 3b46ded30..0a98077e4 100644
--- a/integration-tests/lib/utils.rs
+++ b/integration-tests/lib/utils.rs
@@ -192,8 +192,7 @@ pub fn get_available_address() -> SocketAddr {
 
 fn get_available_port() -> u16 {
     loop {
-        let probe = TcpListener::bind("127.0.0.1:0")
-            .expect("bind(0) for port probe");
+        let probe = TcpListener::bind("127.0.0.1:0").expect("bind(0) for port probe");
         let port = probe.local_addr().expect("probe local_addr").port();
         match try_lock_port(port) {
             Ok(lock_handle) => {

From 77c7884def97f6b019142160782c7a7dfc46b9fb Mon Sep 17 00:00:00 2001
From: GitGab19 <gabriele.vernetti@protonmail.com>
Date: Mon, 10 Aug 2026 12:42:24 +0200
Subject: [PATCH 11/15] test(integration): harden concurrent test execution

---
 .github/workflows/integration-tests.yaml    |   2 +-
 integration-tests/lib/mod.rs                |   4 +-
 integration-tests/lib/sv1_minerd/error.rs   |   9 +
 integration-tests/lib/sv1_minerd/process.rs | 173 +++++++++++++-------
 integration-tests/lib/template_provider.rs  |  35 +---
 integration-tests/lib/utils.rs              |  42 ++++-
 6 files changed, 164 insertions(+), 101 deletions(-)

diff --git a/.github/workflows/integration-tests.yaml b/.github/workflows/integration-tests.yaml
index c43de27ea..6c78e44cf 100644
--- a/.github/workflows/integration-tests.yaml
+++ b/.github/workflows/integration-tests.yaml
@@ -48,7 +48,7 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: cargo binstall --no-confirm --disable-telemetry --disable-strategies quick-install --locked cargo-nextest@0.9.100
 
-# Failure triage: re-run a single test with captured logs shown live:
+      # Failure triage: re-run a single test with captured logs shown live:
       #   RUST_LOG=debug cargo nextest run --manifest-path=integration-tests/Cargo.toml \
       #     -E 'test(<name>)' --no-capture
       - name: Integration Tests
diff --git a/integration-tests/lib/mod.rs b/integration-tests/lib/mod.rs
index a0b89e4ed..de2df13ef 100644
--- a/integration-tests/lib/mod.rs
+++ b/integration-tests/lib/mod.rs
@@ -555,8 +555,8 @@ pub fn start_mining_device_sv2(
     nominal_hashrate_multiplier: Option<f32>,
     single_submit: bool,
 ) {
-    // ponytail: pin to 1 thread so concurrent test processes don't
-    // oversubscribe the CI runner's CPUs and flake share-rate assertions.
+    // Pin to one thread so concurrent test processes do not oversubscribe the CI runner's CPUs
+    // and make share-rate assertions flaky.
     crate::mining_device::set_cores(1);
     tokio::spawn(async move {
         crate::mining_device::connect(
diff --git a/integration-tests/lib/sv1_minerd/error.rs b/integration-tests/lib/sv1_minerd/error.rs
index 6ad643984..213c644dd 100644
--- a/integration-tests/lib/sv1_minerd/error.rs
+++ b/integration-tests/lib/sv1_minerd/error.rs
@@ -19,6 +19,8 @@ pub enum MinerdError {
     InvalidConfiguration(String),
     /// Failed to parse hashrate from minerd benchmark output
     HashrateParseError,
+    /// Timed out before minerd produced a benchmark hashrate
+    HashrateMeasurementTimeout(std::time::Duration),
     /// Mutex was poisoned
     MutexPoisoned,
     /// OS or Architecture not supported
@@ -38,6 +40,12 @@ impl fmt::Display for MinerdError {
             MinerdError::HashrateParseError => {
                 write!(f, "Failed to parse hashrate from minerd benchmark output")
             }
+            MinerdError::HashrateMeasurementTimeout(timeout) => {
+                write!(
+                    f,
+                    "Timed out after {timeout:?} waiting for minerd benchmark output"
+                )
+            }
             MinerdError::MutexPoisoned => write!(f, "Mutex was poisoned"),
             MinerdError::OsArchNotSupported(msg) => {
                 write!(f, "OS or architecture not supported: {msg}")
@@ -57,6 +65,7 @@ impl std::error::Error for MinerdError {
             | MinerdError::ProcessNotRunning
             | MinerdError::InvalidConfiguration(_)
             | MinerdError::HashrateParseError
+            | MinerdError::HashrateMeasurementTimeout(_)
             | MinerdError::MutexPoisoned => None,
             MinerdError::OsArchNotSupported(_) => None,
         }
diff --git a/integration-tests/lib/sv1_minerd/process.rs b/integration-tests/lib/sv1_minerd/process.rs
index 9a70c3807..42857c406 100644
--- a/integration-tests/lib/sv1_minerd/process.rs
+++ b/integration-tests/lib/sv1_minerd/process.rs
@@ -3,6 +3,7 @@ use std::{
     net::SocketAddr,
     path::PathBuf,
     sync::{Arc, Mutex},
+    time::{Duration, Instant},
 };
 use tokio::{
     io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
@@ -12,11 +13,48 @@ use tokio::{
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info};
 
-use crate::utils::{http, tarball};
+use crate::utils::{http, tarball, with_exclusive_lock};
 
 use super::error::MinerdError;
 
 const VERSION_MINERD: &str = "2.5.1";
+const HASHRATE_MEASUREMENT_TIMEOUT: Duration = Duration::from_secs(15);
+const PROCESS_EXIT_TIMEOUT: Duration = Duration::from_secs(2);
+const PROCESS_EXIT_POLL_INTERVAL: Duration = Duration::from_millis(10);
+
+fn kill_and_reap_process(child: &mut TokioChild, what: &str) -> bool {
+    match child.try_wait() {
+        Ok(Some(_)) => return true,
+        Ok(None) => {}
+        Err(e) => {
+            error!("Failed to inspect {what} process before shutdown: {e}");
+            return false;
+        }
+    }
+
+    if let Err(e) = child.start_kill() {
+        error!("Failed to kill {what} process: {e}");
+        return false;
+    }
+
+    let deadline = Instant::now() + PROCESS_EXIT_TIMEOUT;
+    loop {
+        match child.try_wait() {
+            Ok(Some(_)) => return true,
+            Ok(None) if Instant::now() < deadline => {
+                std::thread::sleep(PROCESS_EXIT_POLL_INTERVAL);
+            }
+            Ok(None) => {
+                error!("Timed out waiting for {what} process to exit");
+                return false;
+            }
+            Err(e) => {
+                error!("Failed to reap {what} process: {e}");
+                return false;
+            }
+        }
+    }
+}
 
 fn get_minerd_filename(os: &str, arch: &str) -> Result<String, MinerdError> {
     match (os, arch) {
@@ -70,25 +108,33 @@ impl MinerdProcess {
         let arch = std::env::consts::ARCH;
         let download_filename = get_minerd_filename(os, arch)?;
 
-        if !minerd_dir.exists() {
-            fs::create_dir_all(minerd_dir.clone()).expect("failed to create minerd directory");
-            let download_endpoint = format!(
-                "https://github.com/stratum-mining/cpuminer/releases/download/v{VERSION_MINERD}/"
-            );
-            let url = format!("{download_endpoint}{download_filename}");
-            let tarball_bytes = http::make_get_request(&url, 5);
-            tarball::unpack(&tarball_bytes, &minerd_dir);
-        }
-
         let minerd_binary = minerd_dir.join("minerd");
-
-        if os == "macos" {
-            std::process::Command::new("codesign")
-                .arg("--sign")
-                .arg("-")
-                .arg(&minerd_binary)
-                .output()
-                .expect("failed to sign minerd binary");
+        if !minerd_binary.exists() {
+            with_exclusive_lock(
+                &minerd_dir
+                    .join(".locks")
+                    .join(format!("minerd-{VERSION_MINERD}.lock")),
+                || {
+                    if !minerd_binary.exists() {
+                        fs::create_dir_all(&minerd_dir).expect("failed to create minerd directory");
+                        let download_endpoint = format!(
+                            "https://github.com/stratum-mining/cpuminer/releases/download/v{VERSION_MINERD}/"
+                        );
+                        let url = format!("{download_endpoint}{download_filename}");
+                        let tarball_bytes = http::make_get_request(&url, 5);
+                        tarball::unpack(&tarball_bytes, &minerd_dir);
+
+                        if os == "macos" {
+                            std::process::Command::new("codesign")
+                                .arg("--sign")
+                                .arg("-")
+                                .arg(&minerd_binary)
+                                .output()
+                                .expect("failed to sign minerd binary");
+                        }
+                    }
+                },
+            );
         }
 
         // Bind to local address for the proxy
@@ -412,53 +458,57 @@ impl MinerdProcess {
             MinerdError::ProcessSpawn(std::io::Error::other("Failed to get stderr"))
         })?;
 
-        // Give minerd some time to run and produce hashrate output
-        tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
-
-        // Kill the benchmark process
-        if let Err(e) = child.kill().await {
-            error!("Failed to kill benchmark process: {}", e);
-        }
-
-        // Read and parse the output from stderr
         let mut reader = BufReader::new(stderr);
         let mut line = String::new();
-        let mut hashrate_hashes_per_sec = None;
-
-        // Read output lines to find hashrate information
         let mut all_output = Vec::new();
-        while let Ok(bytes_read) = reader.read_line(&mut line).await {
-            if bytes_read == 0 {
-                break;
-            }
+        let measurement = tokio::time::timeout(HASHRATE_MEASUREMENT_TIMEOUT, async {
+            loop {
+                line.clear();
+                let bytes_read = reader.read_line(&mut line).await?;
+                if bytes_read == 0 {
+                    return Err(MinerdError::HashrateParseError);
+                }
 
-            let line_trimmed = line.trim();
-            all_output.push(line_trimmed.to_string());
-            debug!("Benchmark output: {}", line_trimmed);
-
-            // Parse hashrate from lines like:
-            // "[2025-08-29 20:10:39] thread 0: 2097152 hashes, 1441 khash/s"
-            // "[2025-08-29 20:10:39] Total: 1441 khash/s"
-            if let Some(hashrate_khash) = parse_hashrate_from_benchmark_line(line_trimmed) {
-                info!("Detected benchmark hashrate: {} khash/s", hashrate_khash);
-                // Convert khash/s to hashes/s (multiply by 1000)
-                hashrate_hashes_per_sec = Some(hashrate_khash * 1000.0);
-                // We can break after finding the first hashrate measurement
-                break;
+                let line_trimmed = line.trim();
+                all_output.push(line_trimmed.to_string());
+                debug!("Benchmark output: {line_trimmed}");
+
+                // minerd writes both per-thread and total measurements to stderr. Either is a
+                // valid readiness signal; waiting for one avoids a fixed timing assumption under
+                // concurrent CPU load.
+                if let Some(hashrate_khash) = parse_hashrate_from_benchmark_line(line_trimmed) {
+                    info!("Detected benchmark hashrate: {hashrate_khash} khash/s");
+                    return Ok(hashrate_khash * 1000.0);
+                }
             }
+        })
+        .await;
 
-            line.clear();
+        match child.try_wait() {
+            Ok(Some(_)) => {}
+            Ok(None) => {
+                if let Err(e) = child.kill().await {
+                    error!("Failed to kill and reap benchmark process: {e}");
+                }
+            }
+            Err(e) => error!("Failed to inspect benchmark process during cleanup: {e}"),
         }
 
-        // If we couldn't parse hashrate, log all output for debugging
-        if hashrate_hashes_per_sec.is_none() {
+        let result = match measurement {
+            Ok(result) => result,
+            Err(_) => Err(MinerdError::HashrateMeasurementTimeout(
+                HASHRATE_MEASUREMENT_TIMEOUT,
+            )),
+        };
+
+        if result.is_err() {
             error!("Failed to parse hashrate from minerd benchmark output. Full output:");
             for (i, line) in all_output.iter().enumerate() {
                 error!("  Line {}: {}", i + 1, line);
             }
         }
 
-        hashrate_hashes_per_sec.ok_or(MinerdError::HashrateParseError)
+        result
     }
 }
 
@@ -467,18 +517,17 @@ impl Drop for MinerdProcess {
         // Trigger cancellation to signal all tasks to stop
         self.cancellation_token.cancel();
 
-        match self.process.lock() {
-            Ok(mut process_guard) => {
-                if let Some(mut process) = process_guard.take() {
-                    if let Err(e) = process.start_kill() {
-                        error!("Error killing minerd process on drop: {}", e);
-                    } else {
-                        info!("minerd process killed on drop");
-                    }
-                }
-            }
+        let process = match self.process.lock() {
+            Ok(mut process_guard) => process_guard.take(),
             Err(_) => {
                 error!("Mutex poisoned in Drop implementation, cannot kill process cleanly");
+                None
+            }
+        };
+
+        if let Some(mut process) = process {
+            if kill_and_reap_process(&mut process, "minerd") {
+                info!("minerd process killed and reaped on drop");
             }
         }
     }
diff --git a/integration-tests/lib/template_provider.rs b/integration-tests/lib/template_provider.rs
index 6fcf96a32..8799b52e0 100644
--- a/integration-tests/lib/template_provider.rs
+++ b/integration-tests/lib/template_provider.rs
@@ -14,41 +14,8 @@ use tracing::warn;
 
 use crate::utils::{
     PROCESS_READY_TIMEOUT, fs_utils, http, tarball, wait_for_listener, wait_for_path,
+    with_exclusive_lock,
 };
-use std::os::fd::AsRawFd;
-
-/// Acquire a blocking exclusive flock on `lock_path`, run `f`, then release.
-///
-/// Used to serialize download+unpack of shared artifacts (bitcoin-core,
-/// sv2-tp, high_diff_chain) across concurrently executing nextest processes.
-/// The lock is held only for the window where the artifact is missing; every
-/// process re-checks `exists()` inside the guard so only the first one
-/// actually downloads.
-fn with_exclusive_lock(lock_path: &std::path::Path, f: impl FnOnce()) {
-    if let Some(parent) = lock_path.parent() {
-        std::fs::create_dir_all(parent).ok();
-    }
-    let file = std::fs::OpenOptions::new()
-        .read(true)
-        .write(true)
-        .create(true)
-        .truncate(false)
-        .open(lock_path)
-        .expect("open lockfile for artifact download");
-    let fd = file.as_raw_fd();
-    loop {
-        // SAFETY: fd is valid, flock is async-signal-safe on Linux and macOS.
-        if unsafe { libc::flock(fd, libc::LOCK_EX) } == 0 {
-            break;
-        }
-        let e = std::io::Error::last_os_error();
-        if e.kind() != std::io::ErrorKind::Interrupted {
-            panic!("flock on artifact lockfile: {e}");
-        }
-    }
-    f();
-    // lock released when file is dropped
-}
 
 const VERSION_SV2_TP: &str = "1.1.0";
 const BITCOIN_CORE_V30X: &str = "30.2";
diff --git a/integration-tests/lib/utils.rs b/integration-tests/lib/utils.rs
index 0a98077e4..194bc529e 100644
--- a/integration-tests/lib/utils.rs
+++ b/integration-tests/lib/utils.rs
@@ -11,6 +11,7 @@ use std::{
     fs, io,
     net::{SocketAddr, TcpListener, TcpStream},
     os::fd::AsRawFd,
+    path::Path,
     sync::{Arc, Mutex},
     time::Duration,
 };
@@ -60,6 +61,42 @@ fn try_lock_port(port: u16) -> io::Result<std::fs::File> {
     Ok(file)
 }
 
+/// Acquires a blocking exclusive file lock, runs `f`, and releases the lock.
+///
+/// This serializes initialization of artifacts shared by concurrent nextest processes. Callers
+/// must check whether their artifact exists again inside `f`: another process may have created it
+/// while this process was waiting for the lock.
+pub fn with_exclusive_lock(lock_path: &Path, f: impl FnOnce()) {
+    if let Some(parent) = lock_path.parent() {
+        fs::create_dir_all(parent).unwrap_or_else(|e| {
+            panic!(
+                "failed to create lockfile directory {}: {e}",
+                parent.display()
+            )
+        });
+    }
+    let file = std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(lock_path)
+        .unwrap_or_else(|e| panic!("failed to open lockfile {}: {e}", lock_path.display()));
+    let fd = file.as_raw_fd();
+    loop {
+        // SAFETY: `fd` remains valid for the lifetime of `file`; `flock` is available on every
+        // platform supported by the integration-test harness (Linux and macOS).
+        if unsafe { libc::flock(fd, libc::LOCK_EX) } == 0 {
+            break;
+        }
+        let e = io::Error::last_os_error();
+        if e.kind() != io::ErrorKind::Interrupted {
+            panic!("failed to lock {}: {e}", lock_path.display());
+        }
+    }
+    f();
+}
+
 /// How often readiness gates and message-wait loops re-check their condition.
 ///
 /// Chosen as the knee of the cost/benefit curve: the suite performs ~600 waits, so residual dead
@@ -184,8 +221,9 @@ pub async fn wait_until_listening(addr: SocketAddr, budget: Duration, what: &str
 /// concurrent test processes.
 ///
 /// Probes with `bind(0)` then holds a per-port `flock` lockfile in
-/// `$TMPDIR/sv2-it-ports/` for the process lifetime — no TOCTOU window
-/// between probe and bind.  Lockfiles are released by the kernel on exit.
+/// `$TMPDIR/sv2-it-ports/` for the process lifetime. Every integration-test process follows this
+/// protocol, so another test cannot claim the port after the probe is dropped and before its role
+/// binds. Lockfiles are released by the kernel on exit.
 pub fn get_available_address() -> SocketAddr {
     SocketAddr::from(([127, 0, 0, 1], get_available_port()))
 }

From 2705cba9bfbadec08f620f011b8768da5bd4ee43 Mon Sep 17 00:00:00 2001
From: GitGab19 <gabriele.vernetti@protonmail.com>
Date: Mon, 10 Aug 2026 13:08:17 +0200
Subject: [PATCH 12/15] test(translator): tolerate vardiff updates before
 shares

---
 integration-tests/tests/translator_integration.rs | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/integration-tests/tests/translator_integration.rs b/integration-tests/tests/translator_integration.rs
index b24edc949..0cfec6340 100644
--- a/integration-tests/tests/translator_integration.rs
+++ b/integration-tests/tests/translator_integration.rs
@@ -1132,6 +1132,10 @@ async fn non_aggregated_translator_correctly_deals_with_group_channels() {
                 _,
                 AnyMessageOwned::Mining(parsers_sv2::MiningOwned::SubmitSharesExtended(msg)),
             )) => msg,
+            // Vardiff may enqueue an UpdateChannel before the share on loaded CI runners.
+            Some((_, AnyMessageOwned::Mining(parsers_sv2::MiningOwned::UpdateChannel(_)))) => {
+                continue;
+            }
             msg => panic!("Expected SubmitSharesExtended message, found: {:?}", msg),
         };
 
@@ -1182,6 +1186,10 @@ async fn non_aggregated_translator_correctly_deals_with_group_channels() {
                 _,
                 AnyMessageOwned::Mining(parsers_sv2::MiningOwned::SubmitSharesExtended(msg)),
             )) => msg,
+            // Vardiff may enqueue an UpdateChannel before the share on loaded CI runners.
+            Some((_, AnyMessageOwned::Mining(parsers_sv2::MiningOwned::UpdateChannel(_)))) => {
+                continue;
+            }
             msg => panic!("Expected SubmitSharesExtended message, found: {:?}", msg),
         };
 
@@ -1303,6 +1311,10 @@ async fn non_aggregated_translator_correctly_deals_with_group_channels() {
                 _,
                 AnyMessageOwned::Mining(parsers_sv2::MiningOwned::SubmitSharesExtended(msg)),
             )) => msg,
+            // Vardiff may enqueue an UpdateChannel before the share on loaded CI runners.
+            Some((_, AnyMessageOwned::Mining(parsers_sv2::MiningOwned::UpdateChannel(_)))) => {
+                continue;
+            }
             msg => panic!("Expected SubmitSharesExtended message, found: {:?}", msg),
         };
 

From f2788e92d4a960daaad0117d960734288e5d7bfd Mon Sep 17 00:00:00 2001
From: GitGab19 <gabriele.vernetti@protonmail.com>
Date: Mon, 10 Aug 2026 16:03:05 +0200
Subject: [PATCH 13/15] ci: cache Cargo build artifacts in compile-heavy jobs

Use a common target directory for workspace commands and configure
rust-cache from the integration-tests workspace.

Replace ineffective caches and fix the MSRV cache ordering while
keeping the existing checks unchanged.
---
 .github/workflows/ci.yaml                |  8 ++++++++
 .github/workflows/integration-tests.yaml |  9 +++++++++
 .github/workflows/lockfiles.yaml         |  8 ++++++++
 .github/workflows/msrv.yaml              |  6 +++++-
 .github/workflows/semver-check.yaml      | 20 ++++++--------------
 .github/workflows/tests.yaml             | 21 ++++++---------------
 6 files changed, 42 insertions(+), 30 deletions(-)

diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index c26e4119b..1537336c6 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -15,6 +15,8 @@ jobs:
   clippy:
     name: Clippy Check
     runs-on: ubuntu-latest
+    env:
+      CARGO_TARGET_DIR: ${{ github.workspace }}/target
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
@@ -27,6 +29,12 @@ jobs:
         with:
           components: clippy
 
+      - name: Cache Rust build
+        uses: Swatinem/rust-cache@v2
+        with:
+          cache-bin: "false"
+          workspaces: integration-tests -> ../target
+
       - name: Clippy check
         run: |
           cargo clippy --manifest-path=stratum-apps/Cargo.toml --all-features -- -D warnings
diff --git a/.github/workflows/integration-tests.yaml b/.github/workflows/integration-tests.yaml
index 6c78e44cf..ad716801c 100644
--- a/.github/workflows/integration-tests.yaml
+++ b/.github/workflows/integration-tests.yaml
@@ -8,6 +8,9 @@ on:
 
 name: Integration Tests
 
+env:
+  CARGO_TARGET_DIR: ${{ github.workspace }}/target
+
 jobs:
   ci:
     runs-on: ${{ matrix.os }}
@@ -27,6 +30,12 @@ jobs:
       - name: Install Rust
         uses: dtolnay/rust-toolchain@1.88
 
+      - name: Cache Rust build
+        uses: Swatinem/rust-cache@v2
+        with:
+          cache-bin: "false"
+          workspaces: integration-tests -> ../target
+
       - name: Install capnp dependencies (Ubuntu)
         if: matrix.os == 'ubuntu-latest'
         run: sudo apt-get install -y capnproto libcapnp-dev
diff --git a/.github/workflows/lockfiles.yaml b/.github/workflows/lockfiles.yaml
index 632947ff6..2f1b1a1e1 100644
--- a/.github/workflows/lockfiles.yaml
+++ b/.github/workflows/lockfiles.yaml
@@ -11,6 +11,8 @@ on:
 jobs:
   build:
     runs-on: ubuntu-latest
+    env:
+      CARGO_TARGET_DIR: ${{ github.workspace }}/target
 
     steps:
       - uses: actions/checkout@v4
@@ -21,6 +23,12 @@ jobs:
       - name: Install Rust
         uses: dtolnay/rust-toolchain@stable
 
+      - name: Cache Rust build
+        uses: Swatinem/rust-cache@v2
+        with:
+          cache-bin: "false"
+          workspaces: integration-tests -> ../target
+
       - name: Build with locked dependencies
         run: |
           cargo build --manifest-path=stratum-apps/Cargo.toml --all-features --locked
diff --git a/.github/workflows/msrv.yaml b/.github/workflows/msrv.yaml
index 6a3d53ed2..c89728c7a 100644
--- a/.github/workflows/msrv.yaml
+++ b/.github/workflows/msrv.yaml
@@ -13,12 +13,16 @@ jobs:
     runs-on: ubuntu-latest
     env:
       CARGO_INCREMENTAL: 0
+      CARGO_TARGET_DIR: ${{ github.workspace }}/target
       RUSTFLAGS: -C debuginfo=0
 
     steps:
       - uses: actions/checkout@v4
-      - uses: Swatinem/rust-cache@v1.2.0
       - uses: dtolnay/rust-toolchain@1.88
+      - uses: Swatinem/rust-cache@v2
+        with:
+          cache-bin: "false"
+          workspaces: integration-tests -> ../target
 
       - name: Install capnp dependencies
         run: sudo apt-get install -y capnproto libcapnp-dev
diff --git a/.github/workflows/semver-check.yaml b/.github/workflows/semver-check.yaml
index 43d273712..050b76c14 100644
--- a/.github/workflows/semver-check.yaml
+++ b/.github/workflows/semver-check.yaml
@@ -11,6 +11,8 @@ on:
 jobs:
   semver-check:
     runs-on: ubuntu-latest
+    env:
+      CARGO_TARGET_DIR: ${{ github.workspace }}/target
 
     steps:
       - name: Checkout repository
@@ -22,21 +24,11 @@ jobs:
       - name: Install capnp dependencies
         run: sudo apt-get install -y capnproto libcapnp-dev
 
-      - name: Cache Cargo registry
-        uses: actions/cache@v4
+      - name: Cache Rust build
+        uses: Swatinem/rust-cache@v2
         with:
-          path: ~/.cargo/registry
-          key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }}
-          restore-keys: |
-            ${{ runner.os }}-cargo-registry-
-
-      - name: Cache Cargo index
-        uses: actions/cache@v4
-        with:
-          path: ~/.cargo/git
-          key: ${{ runner.os }}-cargo-index-${{ hashFiles('**/Cargo.lock') }}
-          restore-keys: |
-            ${{ runner.os }}-cargo-index-
+          cache-bin: "false"
+          workspaces: integration-tests -> ../target
 
       - name: Install dependencies
         run: sudo apt-get update && sudo apt-get install -y cmake
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index 448db8e66..0bffcd54a 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -8,6 +8,7 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
+  CARGO_TARGET_DIR: ${{ github.workspace }}/target
 
 jobs:
   test:
@@ -25,23 +26,13 @@ jobs:
       - name: Install capnp dependencies (macOS)
         if: matrix.os == 'macos-latest'
         run: brew install capnp
-      - name: Cache cargo registry
-        uses: actions/cache@v4
-        with:
-          path: ~/.cargo/registry
-          key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }}
-      - name: Cache cargo index
-        uses: actions/cache@v4
-        with:
-          path: ~/.cargo/git
-          key: ${{ runner.os }}-cargo-index-${{ hashFiles('**/Cargo.lock') }}
-      - name: Cache cargo build
-        uses: actions/cache@v4
-        with:
-          path: target
-          key: ${{ runner.os }}-cargo-build-target-${{ hashFiles('**/Cargo.lock') }}
       - name: Install Rust
         uses: dtolnay/rust-toolchain@1.88
+      - name: Cache Rust build
+        uses: Swatinem/rust-cache@v2
+        with:
+          cache-bin: "false"
+          workspaces: integration-tests -> ../target
       - name: Test stratum-apps workspace
         run: cargo test --manifest-path=stratum-apps/Cargo.toml --all-features
       - name: Test pool

From 1111a69f06edf56d06707b8b6fc7d828280616b1 Mon Sep 17 00:00:00 2001
From: GitGab19 <gabriele.vernetti@protonmail.com>
Date: Mon, 10 Aug 2026 17:19:10 +0200
Subject: [PATCH 14/15] test(integration): harden asynchronous message
 assertions

---
 .../tests/monitoring_integration.rs           |  2 ++
 .../tests/translator_integration.rs           | 29 +++++++++++++------
 2 files changed, 22 insertions(+), 9 deletions(-)

diff --git a/integration-tests/tests/monitoring_integration.rs b/integration-tests/tests/monitoring_integration.rs
index 49194eddc..dba883a38 100644
--- a/integration-tests/tests/monitoring_integration.rs
+++ b/integration-tests/tests/monitoring_integration.rs
@@ -565,6 +565,7 @@ async fn tproxy_api_endpoints_with_miner() {
     let global: GlobalInfo = tproxy_mon
         .poll_until(routes::GLOBAL, METRIC_POLL_TIMEOUT, |r: &GlobalInfo| {
             r.sv1_clients.as_ref().is_some_and(|c| c.total_clients >= 1)
+                && r.server.as_ref().is_some_and(|s| s.extended_channels >= 1)
         })
         .await;
     let server_summary = global
@@ -703,6 +704,7 @@ async fn jdc_api_endpoints_with_miner() {
     let global: GlobalInfo = jdc_mon
         .poll_until(routes::GLOBAL, METRIC_POLL_TIMEOUT, |r: &GlobalInfo| {
             r.sv2_clients.as_ref().is_some_and(|c| c.total_clients >= 1)
+                && r.server.as_ref().is_some_and(|s| s.extended_channels >= 1)
         })
         .await;
     let server_summary = global
diff --git a/integration-tests/tests/translator_integration.rs b/integration-tests/tests/translator_integration.rs
index 0cfec6340..d804045e2 100644
--- a/integration-tests/tests/translator_integration.rs
+++ b/integration-tests/tests/translator_integration.rs
@@ -1257,15 +1257,26 @@ async fn non_aggregated_translator_correctly_deals_with_group_channels() {
         EXPECTED_GROUP_CHANNEL_ID
     );
 
-    sniffer
-        .wait_for_message_type(
-            MessageDirection::ToDownstream,
-            MESSAGE_TYPE_MINING_SET_NEW_PREV_HASH,
-        )
-        .await;
-    let set_new_prev_hash = match sniffer.next_message_from_upstream() {
-        Some((_, AnyMessageOwned::Mining(parsers_sv2::MiningOwned::SetNewPrevHash(msg)))) => msg,
-        msg => panic!("Expected SetNewPrevHash message, found: {:?}", msg),
+    let set_new_prev_hash = loop {
+        sniffer
+            .wait_for_message_type(
+                MessageDirection::ToDownstream,
+                MESSAGE_TYPE_MINING_SET_NEW_PREV_HASH,
+            )
+            .await;
+        match sniffer.next_message_from_upstream() {
+            Some((_, AnyMessageOwned::Mining(parsers_sv2::MiningOwned::SetNewPrevHash(msg)))) => {
+                break msg;
+            }
+            // A chain-tip update may enqueue multiple group-channel jobs before SetNewPrevHash.
+            Some((
+                _,
+                AnyMessageOwned::Mining(parsers_sv2::MiningOwned::NewExtendedMiningJob(msg)),
+            )) => {
+                assert_eq!(msg.channel_id, EXPECTED_GROUP_CHANNEL_ID);
+            }
+            msg => panic!("Expected SetNewPrevHash message, found: {:?}", msg),
+        }
     };
     assert_eq!(set_new_prev_hash.channel_id, EXPECTED_GROUP_CHANNEL_ID);
 

From 30b7dfbd2a84f6387d0de4c326de766e9a3d40bd Mon Sep 17 00:00:00 2001
From: GitGab19 <gabriele.vernetti@protonmail.com>
Date: Tue, 11 Aug 2026 10:55:45 +0200
Subject: [PATCH 15/15] test(integration): close remaining parallel execution
 races

Publish downloaded artifacts atomically after extraction and signing, retain
the minerd proxy listener until startup, and tolerate vardiff SetTarget
messages at the affected assertion sites.

Also fix Cargo cache invalidation across workspaces, remove the unused
readiness probe, and correct stale diagnostics and documentation.
---
 .github/workflows/ci.yaml                     |   2 +
 .github/workflows/integration-tests.yaml      |   2 +
 .github/workflows/lockfiles.yaml              |   2 +
 .github/workflows/msrv.yaml                   |   2 +
 .github/workflows/semver-check.yaml           |   2 +
 .github/workflows/tests.yaml                  |   2 +
 integration-tests/lib/mod.rs                  |   6 +-
 integration-tests/lib/sv1_minerd/process.rs   |  68 +++++--
 integration-tests/lib/template_provider.rs    | 137 +++++++++----
 integration-tests/lib/utils.rs                | 187 +++++++++++++-----
 integration-tests/tests/extensions.rs         |  10 +-
 .../tests/translator_integration.rs           |  56 ++++--
 12 files changed, 340 insertions(+), 136 deletions(-)

diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 1537336c6..878c42869 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -33,6 +33,8 @@ jobs:
         uses: Swatinem/rust-cache@v2
         with:
           cache-bin: "false"
+          # Hash every Cargo workspace without registering the shared target more than once.
+          key: cargo-inputs-${{ hashFiles('**/Cargo.toml', '**/Cargo.lock') }}
           workspaces: integration-tests -> ../target
 
       - name: Clippy check
diff --git a/.github/workflows/integration-tests.yaml b/.github/workflows/integration-tests.yaml
index ad716801c..398b0f4a9 100644
--- a/.github/workflows/integration-tests.yaml
+++ b/.github/workflows/integration-tests.yaml
@@ -34,6 +34,8 @@ jobs:
         uses: Swatinem/rust-cache@v2
         with:
           cache-bin: "false"
+          # Hash every Cargo workspace without registering the shared target more than once.
+          key: cargo-inputs-${{ hashFiles('**/Cargo.toml', '**/Cargo.lock') }}
           workspaces: integration-tests -> ../target
 
       - name: Install capnp dependencies (Ubuntu)
diff --git a/.github/workflows/lockfiles.yaml b/.github/workflows/lockfiles.yaml
index 2f1b1a1e1..1799e6807 100644
--- a/.github/workflows/lockfiles.yaml
+++ b/.github/workflows/lockfiles.yaml
@@ -27,6 +27,8 @@ jobs:
         uses: Swatinem/rust-cache@v2
         with:
           cache-bin: "false"
+          # Hash every Cargo workspace without registering the shared target more than once.
+          key: cargo-inputs-${{ hashFiles('**/Cargo.toml', '**/Cargo.lock') }}
           workspaces: integration-tests -> ../target
 
       - name: Build with locked dependencies
diff --git a/.github/workflows/msrv.yaml b/.github/workflows/msrv.yaml
index c89728c7a..b727b8169 100644
--- a/.github/workflows/msrv.yaml
+++ b/.github/workflows/msrv.yaml
@@ -22,6 +22,8 @@ jobs:
       - uses: Swatinem/rust-cache@v2
         with:
           cache-bin: "false"
+          # Hash every Cargo workspace without registering the shared target more than once.
+          key: cargo-inputs-${{ hashFiles('**/Cargo.toml', '**/Cargo.lock') }}
           workspaces: integration-tests -> ../target
 
       - name: Install capnp dependencies
diff --git a/.github/workflows/semver-check.yaml b/.github/workflows/semver-check.yaml
index 050b76c14..6681b2d2b 100644
--- a/.github/workflows/semver-check.yaml
+++ b/.github/workflows/semver-check.yaml
@@ -28,6 +28,8 @@ jobs:
         uses: Swatinem/rust-cache@v2
         with:
           cache-bin: "false"
+          # Hash every Cargo workspace without registering the shared target more than once.
+          key: cargo-inputs-${{ hashFiles('**/Cargo.toml', '**/Cargo.lock') }}
           workspaces: integration-tests -> ../target
 
       - name: Install dependencies
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index 0bffcd54a..61aa9da3e 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -32,6 +32,8 @@ jobs:
         uses: Swatinem/rust-cache@v2
         with:
           cache-bin: "false"
+          # Hash every Cargo workspace without registering the shared target more than once.
+          key: cargo-inputs-${{ hashFiles('**/Cargo.toml', '**/Cargo.lock') }}
           workspaces: integration-tests -> ../target
       - name: Test stratum-apps workspace
         run: cargo test --manifest-path=stratum-apps/Cargo.toml --all-features
diff --git a/integration-tests/lib/mod.rs b/integration-tests/lib/mod.rs
index de2df13ef..cc3d2cec7 100644
--- a/integration-tests/lib/mod.rs
+++ b/integration-tests/lib/mod.rs
@@ -19,7 +19,7 @@ use stratum_apps::{
 use tracing::Level;
 use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt};
 use translator_sv2::TranslatorSv2;
-use utils::{ROLE_READY_BUDGET, get_available_address};
+use utils::{ROLE_STARTUP_DELAY, get_available_address};
 
 pub mod interceptor;
 pub mod message_aggregator;
@@ -171,7 +171,7 @@ pub async fn start_pool(
     tokio::spawn(async move {
         _ = pool_clone.start().await;
     });
-    tokio::time::sleep(ROLE_READY_BUDGET).await;
+    tokio::time::sleep(ROLE_STARTUP_DELAY).await;
     (pool, listening_address, monitoring_address)
 }
 
@@ -359,7 +359,7 @@ pub async fn start_pool_with_jds(
     tokio::spawn(async move {
         _ = pool_clone.start().await;
     });
-    tokio::time::sleep(ROLE_READY_BUDGET).await;
+    tokio::time::sleep(ROLE_STARTUP_DELAY).await;
     (pool, pool_address, jds_address, monitoring_address)
 }
 
diff --git a/integration-tests/lib/sv1_minerd/process.rs b/integration-tests/lib/sv1_minerd/process.rs
index 42857c406..13486d80b 100644
--- a/integration-tests/lib/sv1_minerd/process.rs
+++ b/integration-tests/lib/sv1_minerd/process.rs
@@ -1,7 +1,6 @@
 use std::{
-    fs,
     net::SocketAddr,
-    path::PathBuf,
+    path::{Path, PathBuf},
     sync::{Arc, Mutex},
     time::{Duration, Instant},
 };
@@ -87,6 +86,8 @@ pub struct MinerdProcess {
     process: Arc<Mutex<Option<TokioChild>>>,
     /// Address where the wrapper listens for minerd connections
     local_address: SocketAddr,
+    /// Listener retained until the proxy starts so the port is continuously reserved
+    proxy_listener: Option<TcpListener>,
     /// Address of the upstream mining server
     upstream_address: SocketAddr,
     /// Whether to kill the process after the first mining.submit
@@ -116,22 +117,41 @@ impl MinerdProcess {
                     .join(format!("minerd-{VERSION_MINERD}.lock")),
                 || {
                     if !minerd_binary.exists() {
-                        fs::create_dir_all(&minerd_dir).expect("failed to create minerd directory");
                         let download_endpoint = format!(
                             "https://github.com/stratum-mining/cpuminer/releases/download/v{VERSION_MINERD}/"
                         );
                         let url = format!("{download_endpoint}{download_filename}");
                         let tarball_bytes = http::make_get_request(&url, 5);
-                        tarball::unpack(&tarball_bytes, &minerd_dir);
-
-                        if os == "macos" {
-                            std::process::Command::new("codesign")
-                                .arg("--sign")
-                                .arg("-")
-                                .arg(&minerd_binary)
-                                .output()
-                                .expect("failed to sign minerd binary");
-                        }
+                        tarball::unpack_path_atomically(
+                            &tarball_bytes,
+                            &minerd_dir,
+                            Path::new("minerd"),
+                            |staged_binary| {
+                                if os == "macos" {
+                                    let signature_is_valid = std::process::Command::new("codesign")
+                                        .arg("--verify")
+                                        .arg(staged_binary)
+                                        .output()
+                                        .expect("failed to verify minerd binary signature")
+                                        .status
+                                        .success();
+                                    if !signature_is_valid {
+                                        let output = std::process::Command::new("codesign")
+                                            .arg("--force")
+                                            .arg("--sign")
+                                            .arg("-")
+                                            .arg(staged_binary)
+                                            .output()
+                                            .expect("failed to sign minerd binary");
+                                        assert!(
+                                            output.status.success(),
+                                            "failed to sign minerd binary: {}",
+                                            String::from_utf8_lossy(&output.stderr)
+                                        );
+                                    }
+                                }
+                            },
+                        );
                     }
                 },
             );
@@ -148,6 +168,7 @@ impl MinerdProcess {
             minerd_binary,
             process: Arc::new(Mutex::new(None)),
             local_address,
+            proxy_listener: Some(listener),
             upstream_address,
             single_submit,
             cancellation_token: CancellationToken::new(),
@@ -218,9 +239,12 @@ impl MinerdProcess {
 
     /// Starts the TCP proxy to intercept communications between minerd and the upstream server
     pub async fn start_tcp_proxy(&mut self) -> Result<(), MinerdError> {
-        let listener = TcpListener::bind(self.local_address)
-            .await
-            .map_err(MinerdError::ProxySetup)?;
+        let listener = self.proxy_listener.take().ok_or_else(|| {
+            MinerdError::ProxySetup(std::io::Error::new(
+                std::io::ErrorKind::AlreadyExists,
+                "minerd TCP proxy has already started",
+            ))
+        })?;
         let upstream_address = self.upstream_address;
         let single_submit = self.single_submit;
         let process = Arc::clone(&self.process);
@@ -601,6 +625,18 @@ mod tests {
         assert!(hashrate > 0.0);
     }
 
+    #[tokio::test]
+    async fn proxy_port_remains_reserved_until_proxy_starts() {
+        let minerd_process = MinerdProcess::new(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)), false)
+            .await
+            .unwrap();
+        let local_address = minerd_process.local_address();
+
+        assert!(TcpListener::bind(local_address).await.is_err());
+        drop(minerd_process);
+        assert!(TcpListener::bind(local_address).await.is_ok());
+    }
+
     #[test]
     fn test_parse_hashrate_from_benchmark_line() {
         // Test the parsing logic with known good inputs
diff --git a/integration-tests/lib/template_provider.rs b/integration-tests/lib/template_provider.rs
index 8799b52e0..9b50cf8f6 100644
--- a/integration-tests/lib/template_provider.rs
+++ b/integration-tests/lib/template_provider.rs
@@ -3,7 +3,7 @@ use std::{
     env,
     fs::create_dir_all,
     net::SocketAddr,
-    path::PathBuf,
+    path::{Path, PathBuf},
     process::{Child, Command, Stdio},
 };
 use stratum_apps::{
@@ -183,7 +183,12 @@ impl BitcoinCore {
                                     http::make_get_request(url, 5)
                                 };
 
-                                tarball::unpack(&tarball_bytes, &bin_dir);
+                                tarball::unpack_path_atomically(
+                                    &tarball_bytes,
+                                    &bin_dir,
+                                    Path::new("high_diff_chain"),
+                                    |_| {},
+                                );
                             }
                         },
                     );
@@ -200,9 +205,9 @@ impl BitcoinCore {
         let os = env::consts::OS;
         let arch = env::consts::ARCH;
         let bitcoin_filename = get_bitcoin_core_filename(os, arch, bitcoin_core_version);
-        let bitcoin_home = bin_dir.join(format!("bitcoin-{bitcoin_core_version}"));
+        let bitcoin_dirname = format!("bitcoin-{bitcoin_core_version}");
+        let bitcoin_home = bin_dir.join(&bitcoin_dirname);
         let bitcoin_node_bin = bitcoin_home.join("libexec").join("bitcoin-node");
-        let bitcoin_cli_bin = bitcoin_home.join("bin").join("bitcoin-cli");
 
         if !bitcoin_node_bin.exists() {
             with_exclusive_lock(
@@ -229,29 +234,54 @@ impl BitcoinCore {
                             }
                         };
 
-                        if let Some(parent) = bitcoin_home.parent() {
-                            create_dir_all(parent).unwrap();
-                        }
-
-                        tarball::unpack(&tarball_bytes, &bin_dir);
+                        tarball::unpack_path_atomically(
+                            &tarball_bytes,
+                            &bin_dir,
+                            Path::new(&bitcoin_dirname),
+                            |staged_home| {
+                                let staged_node_bin =
+                                    staged_home.join("libexec").join("bitcoin-node");
+                                let staged_cli_bin = staged_home.join("bin").join("bitcoin-cli");
+                                assert!(
+                                    staged_node_bin.exists(),
+                                    "Bitcoin Core node binary not found after unpack in {}",
+                                    staged_home.display()
+                                );
 
-                        assert!(
-                            bitcoin_node_bin.exists(),
-                            "Bitcoin Core node binary not found after unpack in {}",
-                            bitcoin_home.display()
+                                // Preserve valid upstream signatures; otherwise sign before
+                                // publication so readers only see ready binaries.
+                                if os == "macos" {
+                                    for bin in [&staged_node_bin, &staged_cli_bin] {
+                                        let signature_is_valid = std::process::Command::new(
+                                            "codesign",
+                                        )
+                                        .arg("--verify")
+                                        .arg(bin)
+                                        .output()
+                                        .expect("Failed to verify Bitcoin Core binary signature")
+                                        .status
+                                        .success();
+                                        if signature_is_valid {
+                                            continue;
+                                        }
+
+                                        let output = std::process::Command::new("codesign")
+                                            .arg("--force")
+                                            .arg("--sign")
+                                            .arg("-")
+                                            .arg(bin)
+                                            .output()
+                                            .expect("Failed to sign Bitcoin Core binary");
+                                        assert!(
+                                            output.status.success(),
+                                            "Failed to sign Bitcoin Core binary {}: {}",
+                                            bin.display(),
+                                            String::from_utf8_lossy(&output.stderr)
+                                        );
+                                    }
+                                }
+                            },
                         );
-
-                        // Sign the binaries on macOS
-                        if os == "macos" {
-                            for bin in &[&bitcoin_node_bin, &bitcoin_cli_bin] {
-                                std::process::Command::new("codesign")
-                                    .arg("--sign")
-                                    .arg("-")
-                                    .arg(bin)
-                                    .output()
-                                    .expect("Failed to sign Bitcoin Core binary");
-                            }
-                        }
                     } // inner re-check: bin still missing?
                 }, // closure
             ); // with_exclusive_lock
@@ -455,7 +485,8 @@ impl TemplateProvider {
         let os = env::consts::OS;
         let arch = env::consts::ARCH;
         let sv2_tp_filename = get_sv2_tp_filename(os, arch);
-        let sv2_tp_home = bin_dir.join(format!("sv2-tp-{VERSION_SV2_TP}"));
+        let sv2_tp_dirname = format!("sv2-tp-{VERSION_SV2_TP}");
+        let sv2_tp_home = bin_dir.join(&sv2_tp_dirname);
         let sv2_tp_bin = sv2_tp_home.join("bin").join("sv2-tp");
 
         if !sv2_tp_bin.exists() {
@@ -483,21 +514,45 @@ impl TemplateProvider {
                             }
                         };
 
-                        if let Some(parent) = sv2_tp_home.parent() {
-                            create_dir_all(parent).unwrap();
-                        }
-
-                        tarball::unpack(&tarball_bytes, &bin_dir);
-
-                        // Sign the binary on macOS
-                        if os == "macos" {
-                            std::process::Command::new("codesign")
-                                .arg("--sign")
-                                .arg("-")
-                                .arg(&sv2_tp_bin)
-                                .output()
-                                .expect("Failed to sign sv2-tp binary");
-                        }
+                        tarball::unpack_path_atomically(
+                            &tarball_bytes,
+                            &bin_dir,
+                            Path::new(&sv2_tp_dirname),
+                            |staged_home| {
+                                let staged_binary = staged_home.join("bin").join("sv2-tp");
+                                assert!(
+                                    staged_binary.exists(),
+                                    "sv2-tp binary not found after unpack in {}",
+                                    staged_home.display()
+                                );
+
+                                // Preserve a valid upstream signature; otherwise sign before
+                                // publication so readers only see a ready binary.
+                                if os == "macos" {
+                                    let signature_is_valid = std::process::Command::new("codesign")
+                                        .arg("--verify")
+                                        .arg(&staged_binary)
+                                        .output()
+                                        .expect("Failed to verify sv2-tp binary signature")
+                                        .status
+                                        .success();
+                                    if !signature_is_valid {
+                                        let output = std::process::Command::new("codesign")
+                                            .arg("--force")
+                                            .arg("--sign")
+                                            .arg("-")
+                                            .arg(&staged_binary)
+                                            .output()
+                                            .expect("Failed to sign sv2-tp binary");
+                                        assert!(
+                                            output.status.success(),
+                                            "Failed to sign sv2-tp binary: {}",
+                                            String::from_utf8_lossy(&output.stderr)
+                                        );
+                                    }
+                                }
+                            },
+                        );
                     } // inner re-check: bin still missing?
                 }, // closure
             ); // with_exclusive_lock
diff --git a/integration-tests/lib/utils.rs b/integration-tests/lib/utils.rs
index 194bc529e..860a8c16c 100644
--- a/integration-tests/lib/utils.rs
+++ b/integration-tests/lib/utils.rs
@@ -99,21 +99,18 @@ pub fn with_exclusive_lock(lock_path: &Path, f: impl FnOnce()) {
 
 /// How often readiness gates and message-wait loops re-check their condition.
 ///
-/// Chosen as the knee of the cost/benefit curve: the suite performs ~600 waits, so residual dead
-/// time is roughly `600 * POLL_INTERVAL`. Dropping from the previous 1s to 50ms recovers ~9.5
-/// minutes; going further to 10ms would recover only ~9s more while raising the wakeup rate 5x.
-/// Each poll is one uncontended mutex acquisition (or one loopback connect), so 20 polls/sec is
-/// negligible. [`crate::sniffer::Sniffer::assert_message_not_present`] already polls the same
-/// structures at 100ms.
+/// At 200ms each waiter checks five times per second. This materially reduces residual latency
+/// across the suite's hundreds of waits compared with the previous one-second cadence, without
+/// making every single-threaded test runtime wake twenty or more times per second.
 pub const POLL_INTERVAL: Duration = Duration::from_millis(200);
 
 /// Retry cadence for the *unbounded* connect loops that wait for a peer to come up.
 ///
 /// Deliberately much slower than [`POLL_INTERVAL`]. Those loops have no timeout, so when a peer
 /// never appears — which several tests arrange on purpose — they spin for the whole test. Every
-/// test runs on a bare `#[tokio::test]`, i.e. a single-threaded runtime, so a 20 Hz connect loop
-/// competes with the test's own work on the one thread it has. The message-wait loops are safe at
-/// [`POLL_INTERVAL`] because they exit as soon as their message lands.
+/// test runs on a bare `#[tokio::test]`, i.e. a single-threaded runtime, so even a 5 Hz connect
+/// loop competes with the test's own work on the one thread it has. The message-wait loops are safe
+/// at [`POLL_INTERVAL`] because they exit as soon as their message lands.
 pub const CONNECT_RETRY_INTERVAL: Duration = Duration::from_secs(1);
 
 /// Ceiling for readiness gates on spawned child processes (Bitcoin Core, sv2-tp).
@@ -124,12 +121,13 @@ pub const CONNECT_RETRY_INTERVAL: Duration = Duration::from_secs(1);
 /// keeps the failure message legible.
 pub const PROCESS_READY_TIMEOUT: Duration = Duration::from_secs(30);
 
-/// Budget for the best-effort wait on an in-process role's listening socket.
+/// Fixed startup grace period for in-process pool roles.
 ///
-/// Deliberately equal to the fixed sleep this replaced, so no code path is ever slower than it was
-/// before: a healthy role unblocks in milliseconds, and one that never listens costs exactly what
-/// the old `sleep(1s)` cost. See [`wait_until_listening`] for why this cannot be an assertion.
-pub const ROLE_READY_BUDGET: Duration = Duration::from_secs(1);
+/// A TCP readiness probe is unsafe here because the pool treats every accepted connection as a
+/// protocol session. Probing created phantom downstreams and hung tests that intentionally
+/// intercept setup, so startup retains the original one-second delay until the role exposes an
+/// internal readiness signal.
+pub const ROLE_STARTUP_DELAY: Duration = Duration::from_secs(1);
 
 /// Blocks until `path` exists, polling every [`POLL_INTERVAL`].
 ///
@@ -165,7 +163,7 @@ pub fn wait_for_path(path: &std::path::Path, timeout: Duration, what: &str) {
 
 /// Blocks until a TCP connection to `addr` succeeds, polling every [`POLL_INTERVAL`].
 ///
-/// Used from synchronous contexts; see [`wait_for_listener_async`] for the async equivalent.
+/// Used from synchronous process-startup contexts.
 ///
 /// Panics with `what` in the message if `timeout` elapses first.
 pub fn wait_for_listener(addr: SocketAddr, timeout: Duration, what: &str) {
@@ -186,37 +184,6 @@ pub fn wait_for_listener(addr: SocketAddr, timeout: Duration, what: &str) {
     }
 }
 
-/// Waits up to `budget` for a TCP connection to `addr` to succeed, polling every
-/// [`POLL_INTERVAL`]. Returns whether the listener came up.
-///
-/// Unlike the process gates above this deliberately does **not** assert, because a role's
-/// listening socket is not a universally valid readiness signal. `PoolRuntime::bootstrap` runs
-/// `bootstrap_template_provider()` before `start_services()`, so the pool only starts listening
-/// once its template-distribution handshake completes — and tests that intercept that handshake
-/// prevent it from ever listening, by design. Giving up quietly keeps those tests behaving as they
-/// did while still letting healthy roles unblock in milliseconds.
-pub async fn wait_until_listening(addr: SocketAddr, budget: Duration, what: &str) -> bool {
-    let start = std::time::Instant::now();
-    loop {
-        if tokio::net::TcpStream::connect(addr).await.is_ok() {
-            tracing::debug!(
-                target: "readiness",
-                "ready: {what} after {:?}",
-                start.elapsed()
-            );
-            return true;
-        }
-        if start.elapsed() > budget {
-            tracing::debug!(
-                target: "readiness",
-                "{what} not listening on {addr} within {budget:?}, continuing anyway"
-            );
-            return false;
-        }
-        tokio::time::sleep(POLL_INTERVAL).await;
-    }
-}
-
 /// Allocate a loopback address whose port is exclusively reserved across
 /// concurrent test processes.
 ///
@@ -618,11 +585,22 @@ pub mod http {
 
 pub mod tarball {
     use std::{
-        fs::File,
+        fs::{self, File},
         io::{BufReader, Read},
-        path::Path,
+        path::{Path, PathBuf},
+        sync::atomic::{AtomicU64, Ordering},
     };
 
+    static NEXT_STAGING_ID: AtomicU64 = AtomicU64::new(0);
+
+    struct StagingDirectory(PathBuf);
+
+    impl Drop for StagingDirectory {
+        fn drop(&mut self) {
+            fs::remove_dir_all(&self.0).ok();
+        }
+    }
+
     pub fn read_from_file(path: &str) -> Vec<u8> {
         let file = File::open(path).unwrap_or_else(|_| {
             panic!("Cannot find {path:?} specified with env var BITCOIND_TARBALL_FILE")
@@ -661,6 +639,119 @@ pub mod tarball {
         // Clean up temp tarball
         std::fs::remove_file(&temp_tarball).ok();
     }
+
+    /// Extracts `relative_path` into a private staging directory, prepares it, then atomically
+    /// publishes it beneath `destination`.
+    ///
+    /// The staging directory is created on the same filesystem as the final path, so a successful
+    /// rename makes the complete file or directory visible in one operation. Callers must serialize
+    /// publication and check that the final path is still absent while holding that lock.
+    pub fn unpack_path_atomically(
+        tarball_bytes: &[u8],
+        destination: &Path,
+        relative_path: &Path,
+        prepare: impl FnOnce(&Path),
+    ) {
+        assert!(
+            !relative_path.is_absolute(),
+            "published tarball path must be relative"
+        );
+        fs::create_dir_all(destination).unwrap_or_else(|e| {
+            panic!(
+                "failed to create artifact directory {}: {e}",
+                destination.display()
+            )
+        });
+
+        let artifact_name = relative_path
+            .file_name()
+            .expect("published tarball path must have a file name")
+            .to_string_lossy();
+        let staging = loop {
+            let id = NEXT_STAGING_ID.fetch_add(1, Ordering::Relaxed);
+            let path = destination.join(format!(
+                ".{artifact_name}.staging-{}-{id}",
+                std::process::id()
+            ));
+            match fs::create_dir(&path) {
+                Ok(()) => break StagingDirectory(path),
+                Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
+                Err(e) => panic!("failed to create staging directory {}: {e}", path.display()),
+            }
+        };
+
+        unpack(tarball_bytes, &staging.0);
+        let staged_path = staging.0.join(relative_path);
+        assert!(
+            staged_path.exists(),
+            "artifact {} not found after unpack in {}",
+            relative_path.display(),
+            staging.0.display()
+        );
+        prepare(&staged_path);
+
+        let final_path = destination.join(relative_path);
+        if let Some(parent) = final_path.parent() {
+            fs::create_dir_all(parent).unwrap_or_else(|e| {
+                panic!(
+                    "failed to create artifact parent directory {}: {e}",
+                    parent.display()
+                )
+            });
+        }
+        fs::rename(&staged_path, &final_path).unwrap_or_else(|e| {
+            panic!(
+                "failed to publish artifact {} to {}: {e}",
+                staged_path.display(),
+                final_path.display()
+            )
+        });
+    }
+
+    #[cfg(test)]
+    mod tests {
+        use super::*;
+        use std::process::Command;
+
+        #[test]
+        fn atomic_unpack_publishes_only_after_preparation() {
+            let id = NEXT_STAGING_ID.fetch_add(1, Ordering::Relaxed);
+            let root = StagingDirectory(std::env::temp_dir().join(format!(
+                "sv2-atomic-unpack-test-{}-{id}",
+                std::process::id()
+            )));
+            let source = root.0.join("source");
+            let artifact = source.join("artifact");
+            fs::create_dir_all(&artifact).unwrap();
+            fs::write(artifact.join("payload"), b"complete").unwrap();
+
+            let archive = root.0.join("artifact.tar.gz");
+            let status = Command::new("tar")
+                .arg("-czf")
+                .arg(&archive)
+                .arg("-C")
+                .arg(&source)
+                .arg("artifact")
+                .status()
+                .expect("failed to create test tarball");
+            assert!(status.success(), "failed to create test tarball");
+
+            let destination = root.0.join("destination");
+            let final_path = destination.join("artifact");
+            unpack_path_atomically(
+                &fs::read(&archive).unwrap(),
+                &destination,
+                Path::new("artifact"),
+                |staged_path| {
+                    assert!(!final_path.exists());
+                    fs::write(staged_path.join("prepared"), b"yes").unwrap();
+                },
+            );
+
+            assert_eq!(fs::read(final_path.join("payload")).unwrap(), b"complete");
+            assert_eq!(fs::read(final_path.join("prepared")).unwrap(), b"yes");
+        }
+    }
 }
 
 pub mod fs_utils {
diff --git a/integration-tests/tests/extensions.rs b/integration-tests/tests/extensions.rs
index 08559e557..d4941807a 100644
--- a/integration-tests/tests/extensions.rs
+++ b/integration-tests/tests/extensions.rs
@@ -96,10 +96,7 @@ async fn test_extension_negotiation_with_tlv_in_submit_shares() {
                 ExtensionsNegotiationOwned::RequestExtensions(msg),
             )),
         )) => msg,
-        _ => panic!(
-            "received unexpected message: {:?}",
-            pool_translator_sniffer.next_message_from_downstream()
-        ),
+        msg => panic!("received unexpected message: {msg:?}"),
     };
     assert_eq!(
         request_extensions_msg.requested_extensions,
@@ -137,10 +134,7 @@ async fn test_extension_negotiation_with_tlv_in_submit_shares() {
             let user_identity = msg.user_identity.as_utf8_or_hex();
             assert_eq!(user_identity, "user_identity.miner1".to_string());
         }
-        _ => panic!(
-            "received unexpected message: {:?}",
-            pool_translator_sniffer.next_message_from_downstream()
-        ),
+        msg => panic!("received unexpected message: {msg:?}"),
     }
 
     pool_translator_sniffer
diff --git a/integration-tests/tests/translator_integration.rs b/integration-tests/tests/translator_integration.rs
index d804045e2..af585037a 100644
--- a/integration-tests/tests/translator_integration.rs
+++ b/integration-tests/tests/translator_integration.rs
@@ -1155,17 +1155,24 @@ async fn non_aggregated_translator_correctly_deals_with_group_channels() {
     // that's actually directed to the group channel ID, and not each individual channel
     tp.create_mempool_transaction().unwrap();
 
-    sniffer
-        .wait_for_message_type(
-            MessageDirection::ToDownstream,
-            MESSAGE_TYPE_NEW_EXTENDED_MINING_JOB,
-        )
-        .await;
-    let new_extended_mining_job = match sniffer.next_message_from_upstream() {
-        Some((_, AnyMessageOwned::Mining(parsers_sv2::MiningOwned::NewExtendedMiningJob(msg)))) => {
-            msg
+    let new_extended_mining_job = loop {
+        sniffer
+            .wait_for_message_type(
+                MessageDirection::ToDownstream,
+                MESSAGE_TYPE_NEW_EXTENDED_MINING_JOB,
+            )
+            .await;
+        match sniffer.next_message_from_upstream() {
+            Some((
+                _,
+                AnyMessageOwned::Mining(parsers_sv2::MiningOwned::NewExtendedMiningJob(msg)),
+            )) => {
+                break msg;
+            }
+            // Every vardiff UpdateChannel is answered with SetTarget on this queue.
+            Some((_, AnyMessageOwned::Mining(parsers_sv2::MiningOwned::SetTarget(_)))) => continue,
+            msg => panic!("Expected NewExtendedMiningJob message, found: {:?}", msg),
         }
-        msg => panic!("Expected NewExtendedMiningJob message, found: {:?}", msg),
     };
     assert_eq!(
         new_extended_mining_job.channel_id,
@@ -1240,17 +1247,24 @@ async fn non_aggregated_translator_correctly_deals_with_group_channels() {
     // message pair
     tp.generate_blocks(1);
 
-    sniffer
-        .wait_for_message_type(
-            MessageDirection::ToDownstream,
-            MESSAGE_TYPE_NEW_EXTENDED_MINING_JOB,
-        )
-        .await;
-    let new_extended_mining_job = match sniffer.next_message_from_upstream() {
-        Some((_, AnyMessageOwned::Mining(parsers_sv2::MiningOwned::NewExtendedMiningJob(msg)))) => {
-            msg
+    let new_extended_mining_job = loop {
+        sniffer
+            .wait_for_message_type(
+                MessageDirection::ToDownstream,
+                MESSAGE_TYPE_NEW_EXTENDED_MINING_JOB,
+            )
+            .await;
+        match sniffer.next_message_from_upstream() {
+            Some((
+                _,
+                AnyMessageOwned::Mining(parsers_sv2::MiningOwned::NewExtendedMiningJob(msg)),
+            )) => {
+                break msg;
+            }
+            // Every vardiff UpdateChannel is answered with SetTarget on this queue.
+            Some((_, AnyMessageOwned::Mining(parsers_sv2::MiningOwned::SetTarget(_)))) => continue,
+            msg => panic!("Expected NewExtendedMiningJob message, found: {:?}", msg),
         }
-        msg => panic!("Expected NewExtendedMiningJob message, found: {:?}", msg),
     };
     assert_eq!(
         new_extended_mining_job.channel_id,
@@ -1275,6 +1289,8 @@ async fn non_aggregated_translator_correctly_deals_with_group_channels() {
             )) => {
                 assert_eq!(msg.channel_id, EXPECTED_GROUP_CHANNEL_ID);
             }
+            // Every vardiff UpdateChannel is answered with SetTarget on this queue.
+            Some((_, AnyMessageOwned::Mining(parsers_sv2::MiningOwned::SetTarget(_)))) => continue,
             msg => panic!("Expected SetNewPrevHash message, found: {:?}", msg),
         }
     };