test(server): hold dead-endpoint ports so parallel tests cannot bind them

services/ravel-server/tests/alerting_e2e.rs

@@ -10,6 +10,9 @@
 
 #![allow(clippy::expect_used, clippy::unwrap_used)]
 
+#[path = "support/refused_addr.rs"]
+mod refused_addr;
+
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicI64, AtomicU16, Ordering};
@@ -1452,16 +1455,11 @@ async fn a_rule_removed_from_config_never_repeats() {
 
 // --- Partial federated coverage freezes a rule (ADR-0071 amendment) --------
 
-/// An address that was bound then immediately released: a connection to it is
-/// refused, so a remote pointed here is unavailable at evaluation time. Same
-/// device `federation_e2e.rs` uses for its dead-cluster cases.
-async fn dead_endpoint() -> String {
-    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
-        .await
-        .expect("bind dead");
-    let addr = listener.local_addr().expect("dead local addr");
-    drop(listener);
-    addr.to_string()
+/// An address whose connections are refused, so a remote pointed here is
+/// unavailable at evaluation time. Same device `federation_e2e.rs` uses for its
+/// dead-cluster cases.
+fn dead_endpoint() -> String {
+    refused_addr::refused_addr().to_string()
 }
 
 /// An evaluator whose PromQL engine federates to one `skip_unavailable = true`
@@ -1477,7 +1475,7 @@ async fn evaluator_with_dead_remote(
     clock: TestClock,
     rules: Vec<Rule>,
 ) -> AlertEvaluator {
-    let endpoint = dead_endpoint().await;
+    let endpoint = dead_endpoint();
     let cluster_config = RemoteClusterConfig {
         name: "west".to_string(),
         endpoint: endpoint.clone(),

services/ravel-server/tests/analytics_endpoint.rs

@@ -12,6 +12,9 @@
 #[path = "support/fragment_tls.rs"]
 mod fragment_tls;
 
+#[path = "support/refused_addr.rs"]
+mod refused_addr;
+
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
@@ -812,16 +815,11 @@ async fn analytics_submits_one_audit_event_per_executed_query() {
 const DEAD_CLUSTER: &str = "west";
 const GATE_OPERATOR_CRED: &str = "operator-west-credential";
 
-/// An address bound then immediately released: a connection to it is refused,
-/// so a remote pointed here is unavailable at query time (the same technique
-/// `federation_handler_warnings.rs` uses).
-async fn dead_endpoint() -> String {
-    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
-        .await
-        .expect("bind dead");
-    let addr = listener.local_addr().expect("dead local addr");
-    drop(listener);
-    addr.to_string()
+/// An address whose connections are refused, so a remote pointed here is
+/// unavailable at query time (the same technique `federation_handler_warnings.rs`
+/// uses).
+fn dead_endpoint() -> String {
+    refused_addr::refused_addr().to_string()
 }
 
 /// A `Federation` over one unavailable remote cluster that opted into
@@ -870,7 +868,7 @@ async fn federated_app(metric: &str, samples: &[(i64, f64)]) -> Router {
     .await;
     let catalog =
         Arc::new(Catalog::new(Arc::clone(&store), CatalogConfig::default()).expect("catalog"));
-    let federation = Arc::new(dead_federation(DEAD_CLUSTER, &dead_endpoint().await));
+    let federation = Arc::new(dead_federation(DEAD_CLUSTER, &dead_endpoint()));
     let engine =
         QueryEngine::new(catalog, store, EngineConfig::default()).with_federation(federation);
     router(AnalyticsState {

services/ravel-server/tests/federation_e2e.rs

@@ -40,6 +40,9 @@
 
 #![allow(clippy::expect_used, clippy::unwrap_used)]
 
+#[path = "support/refused_addr.rs"]
+mod refused_addr;
+
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
@@ -371,15 +374,10 @@ async fn spawn_remote(
     }
 }
 
-/// An address that was bound then immediately released: a connection to it is
-/// refused, so a remote pointed here is unavailable at query time.
-async fn dead_endpoint() -> String {
-    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
-        .await
-        .expect("bind dead");
-    let addr = listener.local_addr().expect("dead local addr");
-    drop(listener);
-    addr.to_string()
+/// An address whose connections are refused, so a remote pointed here is
+/// unavailable at query time.
+fn dead_endpoint() -> String {
+    refused_addr::refused_addr().to_string()
 }
 
 /// Build a coordinator-side [`RemoteCluster`] dialed by the production
@@ -500,7 +498,7 @@ async fn federated_series_discovery_returns_union_and_warns_on_skip() {
     let east = spawn_remote(east_store, OPERATOR_CRED, &acme).await;
 
     // A dead cluster "west", skip_unavailable=true.
-    let west_endpoint = dead_endpoint().await;
+    let west_endpoint = dead_endpoint();
 
     let clusters = vec![
         make_remote_cluster(
@@ -726,7 +724,7 @@ async fn federated_query_marks_skipped_cluster_in_warnings() {
     let east = spawn_remote(east_store, OPERATOR_CRED, &acme).await;
 
     // A dead cluster "west", also skip_unavailable=true.
-    let west_endpoint = dead_endpoint().await;
+    let west_endpoint = dead_endpoint();
 
     let clusters = vec![
         make_remote_cluster(

services/ravel-server/tests/federation_handler_warnings.rs

@@ -16,6 +16,9 @@
 
 #![allow(clippy::expect_used, clippy::unwrap_used)]
 
+#[path = "support/refused_addr.rs"]
+mod refused_addr;
+
 use std::collections::HashSet;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
@@ -124,15 +127,10 @@ async fn publish_series(store: &dyn ObjectStoreBackend, tenant: &TenantId, base_
         .expect("publish");
 }
 
-/// An address bound then immediately released: a connection to it is refused, so
-/// a remote pointed here is unavailable at query time.
-async fn dead_endpoint() -> String {
-    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
-        .await
-        .expect("bind dead");
-    let addr = listener.local_addr().expect("dead local addr");
-    drop(listener);
-    addr.to_string()
+/// An address whose connections are refused, so a remote pointed here is
+/// unavailable at query time.
+fn dead_endpoint() -> String {
+    refused_addr::refused_addr().to_string()
 }
 
 /// A `Federation` over one unavailable remote cluster with `skip_unavailable =
@@ -176,7 +174,7 @@ async fn promql_with_federation(store: Arc<dyn ObjectStoreBackend>, tenant: &Ten
     .expect("catalog");
     let mut tokens = std::collections::HashMap::new();
     tokens.insert(TOKEN.to_string(), tenant.clone());
-    let federation = Arc::new(dead_federation("west", &dead_endpoint().await));
+    let federation = Arc::new(dead_federation("west", &dead_endpoint()));
 
     let app_state = build_app_state(
         catalog,

services/ravel-server/tests/metrics_endpoint.rs

@@ -610,11 +610,7 @@ async fn metrics_render_get_unverified_and_control_plane_counters() {
     assert_eq!(served.load(Ordering::Relaxed), 3);
     assert_eq!(served_bytes.load(Ordering::Relaxed), 568);
 
-    let closed = tokio::net::TcpListener::bind("127.0.0.1:0")
-        .await
-        .expect("bind");
-    let closed_base = format!("http://{}", closed.local_addr().expect("addr"));
-    drop(closed);
+    let closed_base = format!("http://{}", refused_addr::refused_addr());
     s3_store_on(closed_base, &metrics)
         .bucket_protection_report(&BucketProtectionParams::default())
         .await;
@@ -1687,6 +1683,9 @@ async fn drain_overrun_and_residue_families_render() {
 #[path = "support/strict_flush.rs"]
 mod strict_flush;
 
+#[path = "support/refused_addr.rs"]
+mod refused_addr;
+
 fn spawn_strict_export(
     server: &strict_flush::PolicyServer,
 ) -> tokio::task::JoinHandle<reqwest::Response> {

services/ravel-server/tests/support/refused_addr.rs

@@ -0,0 +1,34 @@
+//! A loopback address whose connections are refused for the life of the test
+//! process. Included with `#[path]` by each test that needs a dead endpoint.
+//!
+//! Binding `127.0.0.1:0`, reading the port and dropping the listener frees the
+//! port for any parallel test to bind before the dial, which then gets a
+//! response instead of a refusal. Here the port stays bound by a socket that
+//! never listens: a connect to it is refused at once, and with neither
+//! `SO_REUSEADDR` nor `SO_REUSEPORT` set no other socket can bind it while the
+//! process lives.
+
+#![allow(dead_code, clippy::expect_used)]
+
+use std::net::SocketAddr;
+use std::sync::OnceLock;
+
+use tokio::net::TcpSocket;
+
+static REFUSED: OnceLock<(TcpSocket, SocketAddr)> = OnceLock::new();
+
+/// The process-wide refused address.
+pub fn refused_addr() -> SocketAddr {
+    REFUSED
+        .get_or_init(|| {
+            let socket = TcpSocket::new_v4().expect("refused socket");
+            socket.set_reuseaddr(false).expect("clear SO_REUSEADDR");
+            socket.set_reuseport(false).expect("clear SO_REUSEPORT");
+            socket
+                .bind(SocketAddr::from(([127, 0, 0, 1], 0)))
+                .expect("bind refused socket");
+            let addr = socket.local_addr().expect("refused local addr");
+            (socket, addr)
+        })
+        .1
+}