Merge #9698b1f9: test: eliminate shared fixture races and timing-depend…

test: eliminate shared fixture races and timing-dependent assertions

nostr:nevent1qgsx2lyl2e4zvfadwcvkd9fkrcwczj7mf858hy85mwqclwgut8wpg2spz3mhxue69uhhyetvv9ujumn8d96zuer9wcq3yamnwvaz7tm8d96xummnw3ezucm0d5q3kamnwvaz7tmwva5hgtnyv9hxxmmwwashjer9wchxxmmdqqsfdx93lyfyyuvw79qg8qe4eec4gjxkqasu3glrrsljhdgle3hw5fqtx4mfz

PR-Author: DanConwayDev's Agent
nostr:npub1v47f74n2ycn66asev62nv8sas99akj0g0wg0fkup37u3ckwuzs4q7cwtp0

CoverNote:

Parallel tests could lose released relay ports, leave connections alive after fixture shutdown, block Git children on full pipes, or infer readiness from scheduler-dependent sleeps. This revision extends the existing relay-port PR into one consolidated reliability proposal for Grasp.

- Retain SDK relay listeners through startup and exercise 32 concurrent fixtures.
- Transfer reserved loopback sockets into Grasp subprocesses through a private Unix test protocol, retain the address across restart, and keep offline/recovering endpoints continuously owned.
- Own HTTP, WebSocket upgrade and backend-forwarding tasks across the mock relay, auth gate, Git servers and proxy fixtures. Explicit shutdown cancels and joins requests; Git subprocess I/O drains concurrently.
- Wait for event visibility and actual connected-state metrics; correct metrics URL construction and stopped-source bookkeeping.
- Replace expiry scheduling margins with explicit timestamps and streaming sleeps/frame assumptions with observable gates and byte-level assertions.

Tests remain enabled and parallel. Normal server binding and service configuration are unchanged; inherited listener support requires NGIT_TEST=1 and validates a listening loopback socket. The private capability probe lets the companion ngit harness retain compatibility with older releases.

Targeted Linux validation in the repository development shell: 65 relay-connection tests, 33 rejected-index tests, 132 purgatory tests, all 3 streaming tests, 7 shared Git-server tests, and relay-identity/recovery tests passed. Final fixture_lifecycle passed 57 tests; the moved subprocess-restart regression passed separately. Scoped Clippy for fixture_lifecycle and relay_identity passed with warnings denied. A smoke test queried this Grasp binary using the modified ngit harness. Formatting and whitespace checks pass.

These checks are scoped, not a claim of full workspace, nixpkgs sandbox or Darwin validation. Full host builds remain outstanding for the new changes. External Buzz service fixtures are outside this PR. The separate ngit reliability proposal addresses the equivalent harness and Blossom timing issues.

AI assistance: implemented and reviewed with Codex (GPT-6).

AGENTS.md

@@ -121,10 +121,14 @@ nix develop -c cargo test -p grasp-audit --lib specific_test_name -- --nocapture
 ### Troubleshooting
 
 **Buffer Size Errors:**
-If you see mpsc channel buffer size panics on first test run, this is usually transient. Simply run the tests again.
+Capture the failing test and panic before retrying. Diagnose the fixture or
+capacity assumption; a successful rerun alone does not establish correctness.
 
 **Port Conflicts:**
-Both `TestRelay` and `test-ngit-relay.sh` use random ports to avoid conflicts. If you see port errors, ensure no stale processes are running.
+`TestRelay` transfers an owned loopback listener into the subprocess and
+retains it across restarts. Do not release a reservation and rebind its port.
+See [test fixture guidance](docs/how-to/test-fixtures.md) for readiness,
+shutdown, and timing rules. Audit external server scripts separately.
 
 ## Code Patterns
 

docs/how-to/test-fixtures.md

@@ -0,0 +1,41 @@
+# Reliable test fixtures
+
+Run tests through the repository development shell. Keep parallel checks
+enabled: each fixture owns its socket, tasks, subprocesses and temporary data.
+
+## Socket ownership
+
+Allocate loopback port zero and transfer the bound listener into the server.
+Do not bind, inspect the port, drop the socket and bind that port again.
+Subprocess fixtures pass `NGIT_TEST_LISTENER_FD` with `NGIT_TEST=1` on Unix.
+The binary advertises this private protocol through
+`--internal-test-listener-support`; it is not deployment configuration and is
+not exposed through the NixOS module or example service environment.
+
+`TestRelay` retains a parent copy across same-address restarts. Offline relay
+scenarios use `UnavailableEndpoint`, which accepts and closes connections
+while retaining the socket, then transfers that socket to the recovered
+server. This prevents another parallel test from taking the offline address.
+
+## Observable readiness and shutdown
+
+Wait for HTTP readiness, connected-state metrics, or the expected event with
+a bounded deadline. A successful TCP connection, a connection attempt counter,
+and an arbitrary grace period do not prove that a server is ready.
+
+Connection tasks belong to their fixture's accept loop. Stopping the fixture
+cancels and joins them; dropping it cancels the owner. Git fixture subprocesses
+are cancelled with their request, and all three pipes are driven concurrently.
+
+## Time and streaming assertions
+
+Age private cache timestamps explicitly and bracket wall-clock timestamps
+before and after the operation. Streaming tests coordinate fake Git with an
+owned loopback gate, collect bytes independently of HTTP frame boundaries,
+and retain assertions about terminal flush and post-push promotion ordering.
+Fixed sleeps remain appropriate only when elapsed time is itself under test;
+polling an observable condition must always have a bounded deadline.
+
+Targeted regressions include `fixture_lifecycle`, `relay_identity`,
+`git_response_streaming`, and the shared Git-server tests in `sync`.
+Full package validation remains necessary after these scoped checks.

src/main.rs

@@ -15,6 +15,8 @@ use ngit_grasp::{
 };
 
 mod docs_export;
+#[cfg(unix)]
+mod test_listener;
 
 /// Top-level CLI dispatcher.
 ///
@@ -45,6 +47,14 @@ enum Cli {
 
 #[tokio::main]
 async fn main() -> Result<()> {
+    // Harnesses probe this before handing a reserved Unix listener to the child.
+    if std::env::args_os().nth(1).as_deref() == Some(OsStr::new("--internal-test-listener-support"))
+    {
+        #[cfg(unix)]
+        println!("ngit-test-listener-v1");
+        return Ok(());
+    }
+
     // Documentation builds need only the static command model. Dispatch before
     // dotenv, clap parsing, secret discovery, filesystem access, or relay
     // startup so exporting metadata is safe in an isolated build environment.
@@ -123,6 +133,12 @@ async fn run_relay(config: Config) -> Result<()> {
         "Starting ngit-grasp"
     );
 
+    #[cfg(unix)]
+    let server = match test_listener::from_env()? {
+        Some(listener) => RelayServer::start_with_listener(config, listener).await?,
+        None => RelayServer::start(config).await?,
+    };
+    #[cfg(not(unix))]
     let server = RelayServer::start(config).await?;
 
     server.run_until(shutdown_signal()).await

src/purgatory/persistence.rs

@@ -106,7 +106,6 @@ pub fn offset_to_instant(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::thread;
     use std::time::Duration;
 
     #[test]
@@ -117,17 +116,14 @@ mod tests {
 
         let offset = instant_to_offset(future, now_system, now_instant);
 
-        // Should be approximately 60 seconds (within tolerance)
-        assert!(offset.as_secs() >= 59 && offset.as_secs() <= 61);
+        assert_eq!(offset, Duration::from_secs(60));
     }
 
     #[test]
     fn test_instant_to_offset_past() {
         let now_system = SystemTime::now();
-        let past_instant = Instant::now();
-        // Simulate some time passing
-        thread::sleep(Duration::from_millis(10));
         let now_instant = Instant::now();
+        let past_instant = now_instant - Duration::from_secs(10);
 
         let offset = instant_to_offset(past_instant, now_system, now_instant);
 
@@ -137,22 +133,9 @@ mod tests {
 
     #[test]
     fn test_offset_to_instant_with_time_remaining() {
-        let saved_at = SystemTime::now();
+        let saved_at = SystemTime::now() - Duration::from_secs(10);
         let offset = Duration::from_secs(60);
-
-        // Simulate a very short downtime (< 10ms)
-        thread::sleep(Duration::from_millis(5));
-
-        let now_instant = Instant::now();
-        let restored = offset_to_instant(offset, saved_at, now_instant);
-
-        // Should be approximately 60 seconds in the future
-        let remaining = restored.duration_since(now_instant);
-        assert!(
-            remaining.as_secs() >= 59 && remaining.as_secs() <= 61,
-            "Expected ~60s, got {}s",
-            remaining.as_secs()
-        );
+        assert_restored_within_call_bounds(offset, saved_at, Instant::now());
     }
 
     #[test]
@@ -178,21 +161,23 @@ mod tests {
         // Convert to offset
         let offset = instant_to_offset(future, now_system, now_instant);
 
-        // Immediately convert back (minimal downtime)
-        let restored = offset_to_instant(offset, now_system, now_instant);
+        assert_restored_within_call_bounds(offset, now_system, now_instant);
+    }
 
-        // Should be very close to the original future instant
-        let diff = if restored > future {
-            restored.duration_since(future)
-        } else {
-            future.duration_since(restored)
+    // The conversion reads SystemTime internally. Bound that read with actual
+    // observations instead of assuming a maximum scheduling delay.
+    fn assert_restored_within_call_bounds(
+        offset: Duration,
+        saved_at: SystemTime,
+        reference: Instant,
+    ) {
+        let before = SystemTime::now();
+        let restored = offset_to_instant(offset, saved_at, reference);
+        let after = SystemTime::now();
+        let remaining_at = |now: SystemTime| {
+            offset.saturating_sub(now.duration_since(saved_at).unwrap_or(Duration::ZERO))
         };
-
-        // Allow for small timing differences (< 100ms)
-        assert!(
-            diff < Duration::from_millis(100),
-            "Round trip should preserve instant within 100ms, got {}ms",
-            diff.as_millis()
-        );
+        assert!(restored >= reference + remaining_at(after));
+        assert!(restored <= reference + remaining_at(before));
     }
 }

src/purgatory/sync/queue.rs

@@ -135,17 +135,17 @@ mod tests {
         entry.attempt_count = 5;
         entry.next_attempt = Instant::now() + Duration::from_secs(120);
 
-        // New event arrives with shorter delay
+        // Capture the call bounds instead of assuming the scheduler runs promptly.
+        let original_next = entry.next_attempt;
+        let before = Instant::now();
         entry.on_new_event(Duration::from_secs(10));
+        let after = Instant::now();
 
         // Attempt count should be reset
         assert_eq!(entry.attempt_count, 0);
 
-        // next_attempt should be updated to the sooner time
-        // (within a small tolerance for test timing)
-        let expected = Instant::now() + Duration::from_secs(10);
-        assert!(entry.next_attempt <= expected + Duration::from_millis(100));
-        assert!(entry.next_attempt >= expected - Duration::from_millis(100));
+        assert!(entry.next_attempt >= original_next.min(before + Duration::from_secs(10)));
+        assert!(entry.next_attempt <= original_next.min(after + Duration::from_secs(10)));
     }
 
     #[test]
@@ -160,16 +160,14 @@ mod tests {
         assert_eq!(entry.attempt_count, 0);
 
         // But next_attempt should not be pushed back
-        assert!(entry.next_attempt <= original_next + Duration::from_millis(100));
+        assert_eq!(entry.next_attempt, original_next);
     }
 
     #[test]
     fn is_ready_checks_both_conditions() {
         let mut entry = SyncQueueEntry::new(Duration::from_secs(0));
 
-        // Should be ready initially (no delay, not in progress)
-        // Note: there might be a tiny delay, so we wait a moment
-        std::thread::sleep(Duration::from_millis(10));
+        // A zero delay is already ready; monotonic time cannot move backward.
         assert!(entry.is_ready());
 
         // Mark as in progress - should not be ready
@@ -185,12 +183,13 @@ mod tests {
     #[test]
     fn on_sync_complete_increments_and_schedules() {
         let mut entry = SyncQueueEntry::new(Duration::from_secs(0));
-        std::thread::sleep(Duration::from_millis(10)); // Ensure next_attempt has passed
 
         entry.in_progress = true;
         entry.attempt_count = 0;
 
+        let before = Instant::now();
         entry.on_sync_complete();
+        let after = Instant::now();
 
         // Should no longer be in progress
         assert!(!entry.in_progress);
@@ -199,8 +198,7 @@ mod tests {
         assert_eq!(entry.attempt_count, 1);
 
         // Next attempt should be scheduled with backoff (20s for attempt 1)
-        let expected = Instant::now() + Duration::from_secs(20);
-        assert!(entry.next_attempt >= expected - Duration::from_millis(100));
-        assert!(entry.next_attempt <= expected + Duration::from_millis(100));
+        assert!(entry.next_attempt >= before + Duration::from_secs(20));
+        assert!(entry.next_attempt <= after + Duration::from_secs(20));
     }
 }

src/server.rs

@@ -82,7 +82,7 @@ impl RelayServer {
     /// Expects `config.relay_owner_nsec` to be set (see [`Config::load`])
     /// and does **not** install a tracing subscriber — that is the
     /// caller's concern.
-    pub async fn start(mut config: Config) -> Result<Self> {
+    pub async fn start(config: Config) -> Result<Self> {
         // Bind first so kernel-assigned ports are resolved before any
         // component captures the domain / bind address.
         let requested: SocketAddr = config
@@ -92,6 +92,12 @@ impl RelayServer {
         let listener = TcpListener::bind(&requested)
             .await
             .with_context(|| format!("failed to bind {}", requested))?;
+        Self::start_with_listener(config, listener).await
+    }
+
+    /// Start from an already bound listener without releasing its address.
+    /// Used by subprocess test fixtures to preserve their port reservation.
+    pub async fn start_with_listener(mut config: Config, listener: TcpListener) -> Result<Self> {
         let local_addr = listener.local_addr()?;
         config.bind_address = local_addr.to_string();
         if config.domain.is_empty() {

src/sync/rejected_index.rs

@@ -1507,6 +1507,13 @@ mod tests {
         keys.sign_event(unsigned).unwrap()
     }
 
+    fn expire_hot_cache(cache: &HotCache) {
+        let expired_at = Instant::now() - cache.expiry_duration;
+        for entry in cache.entries.write().unwrap().values_mut() {
+            entry.cached_at = expired_at;
+        }
+    }
+
     fn simulate_checkpoint_downtime(path: &Path, downtime: Duration) {
         let json = std::fs::read_to_string(path).expect("read rejected-event checkpoint");
         let mut state: RejectedCacheState =
@@ -1598,7 +1605,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_hot_cache_expires_after_duration() {
-        let cache = HotCache::new(Duration::from_millis(50));
+        let cache = HotCache::new(Duration::from_secs(120));
         let event = create_test_event().await;
 
         cache.add(
@@ -1611,8 +1618,8 @@ mod tests {
 
         assert!(cache.contains(&event.id));
 
-        // Wait for expiry
-        std::thread::sleep(Duration::from_millis(60));
+        // Move the cache timestamp to its expiry boundary without waiting.
+        expire_hot_cache(&cache);
 
         let expired = cache.cleanup_expired();
         assert_eq!(expired, 1);
@@ -1698,14 +1705,21 @@ mod tests {
 
     #[tokio::test]
     async fn test_unrecoverable_ids_expire_with_cold_bound() {
-        let index = RejectedEventsIndex::new(Duration::from_millis(10), Duration::from_millis(50));
+        let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
         let event = create_test_event().await;
 
         index.add_unrecoverable(event.id, 30617);
         assert!(index.contains(&event.id));
 
-        // Passage of time is the behaviour under test (bounded expiry)
-        std::thread::sleep(Duration::from_millis(60));
+        // Set the rejection timestamp at the cold expiry boundary.
+        index
+            .unrecoverable
+            .entries
+            .write()
+            .unwrap()
+            .get_mut(&event.id)
+            .unwrap()
+            .rejected_at = Instant::now() - index.unrecoverable.expiry_duration;
 
         assert!(!index.contains(&event.id));
         assert_eq!(index.cleanup_expired_unrecoverable(), 1);
@@ -1793,8 +1807,8 @@ mod tests {
     #[tokio::test]
     async fn test_cleanup_expired_both_tiers() {
         let index = RejectedEventsIndex::new(
-            Duration::from_millis(50),  // Hot cache expires quickly
-            Duration::from_millis(100), // Cold index expires slower
+            Duration::from_secs(120),    // Hot cache expires quickly
+            Duration::from_secs(604800), // Cold index expires slower
         );
         let event = create_test_event().await;
 
@@ -1805,15 +1819,22 @@ mod tests {
             RejectionReason::DoesNotListService,
         );
 
-        // Wait for hot cache to expire
-        std::thread::sleep(Duration::from_millis(60));
+        // Expire only the hot tier, independently of scheduler timing.
+        expire_hot_cache(&index.hot_cache);
 
         let (hot_expired, cold_expired) = index.cleanup_expired_for_type("announcement");
         assert_eq!(hot_expired, 1);
         assert_eq!(cold_expired, 0); // Not expired yet
 
-        // Wait for cold index to expire
-        std::thread::sleep(Duration::from_millis(50));
+        // Now expire the remaining cold entry explicitly.
+        index
+            .cold_index
+            .entries
+            .write()
+            .unwrap()
+            .get_mut(&event.id)
+            .unwrap()
+            .rejected_at = Instant::now() - index.cold_index.expiry_duration;
 
         let (hot_expired, cold_expired) = index.cleanup_expired_for_type("announcement");
         assert_eq!(hot_expired, 0); // Already cleaned up
@@ -1822,8 +1843,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_hot_cache_miss_after_expiry() {
-        let index =
-            RejectedEventsIndex::new(Duration::from_millis(50), Duration::from_secs(604800));
+        let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
         let event = create_test_event().await;
         let pubkey = event.pubkey;
         let identifier = "test-repo".to_string();
@@ -1835,8 +1855,8 @@ mod tests {
             RejectionReason::MaintainerNotYetValid,
         );
 
-        // Wait for hot cache to expire
-        std::thread::sleep(Duration::from_millis(60));
+        // Expire only the hot tier, independently of scheduler timing.
+        expire_hot_cache(&index.hot_cache);
 
         let (removed, hot_events) =
             index.invalidate_and_get(&pubkey, &identifier, Some(EventType::Announcement));
@@ -1847,8 +1867,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_expired_dependency_candidate_keeps_id_for_targeted_refetch() {
-        let index =
-            RejectedEventsIndex::new(Duration::from_millis(50), Duration::from_secs(604800));
+        let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
         let keys = Keys::generate();
         let dependency_event = keys
             .sign_event(
@@ -1877,7 +1896,7 @@ mod tests {
             RejectionReason::Other,
         );
 
-        std::thread::sleep(Duration::from_millis(60));
+        expire_hot_cache(&index.hot_cache);
 
         let (event_ids, hot_events) =
             index.dependency_candidates(&pubkey, &identifier, Some(EventType::State));
@@ -2110,7 +2129,7 @@ mod tests {
         let state_path = temp_dir.path().join("rejected_cache.json");
 
         let index = RejectedEventsIndex::new(
-            Duration::from_millis(50),   // Hot cache expires quickly
+            Duration::from_secs(120),    // Hot cache expires quickly
             Duration::from_secs(604800), // Cold index lasts long
         );
         let event = create_test_event().await;
@@ -2123,8 +2142,8 @@ mod tests {
             RejectionReason::MaintainerNotYetValid,
         );
 
-        // Wait for hot cache to expire
-        std::thread::sleep(Duration::from_millis(60));
+        // Expire only the hot tier, independently of scheduler timing.
+        expire_hot_cache(&index.hot_cache);
         index.cleanup_expired_for_type("announcement");
 
         assert_eq!(index.hot_cache_len(), 0);
@@ -2135,7 +2154,7 @@ mod tests {
 
         // Restore into new index
         let index2 =
-            RejectedEventsIndex::new(Duration::from_millis(50), Duration::from_secs(604800));
+            RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
         index2.restore_from_disk(&state_path).unwrap();
 
         // Verify only cold index restored (hot cache was empty)
@@ -2514,8 +2533,8 @@ mod tests {
         let temp_dir = tempfile::tempdir().unwrap();
         let state_path = temp_dir.path().join("rejected_cache.json");
 
-        // Create index with 2 second hot cache expiry
-        let index = RejectedEventsIndex::new(Duration::from_secs(2), Duration::from_secs(604800));
+        // Use the normal TTL and simulate the elapsed part explicitly.
+        let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
         let event = create_test_event().await;
 
         index.add_announcement(
@@ -2525,24 +2544,40 @@ mod tests {
             RejectionReason::DoesNotListService,
         );
 
-        // Wait 200ms (small fraction of TTL)
-        std::thread::sleep(Duration::from_millis(200));
+        // Save an entry that has already consumed part of its TTL.
+        index
+            .hot_cache
+            .entries
+            .write()
+            .unwrap()
+            .get_mut(&event.id)
+            .unwrap()
+            .cached_at = Instant::now() - Duration::from_secs(30);
 
         // Save to disk
         index.save_to_disk(&state_path).unwrap();
 
         // Immediately restore (minimal downtime)
-        let index2 = RejectedEventsIndex::new(Duration::from_secs(2), Duration::from_secs(604800));
+        let index2 =
+            RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
         index2.restore_from_disk(&state_path).unwrap();
 
-        // Event should still be retrievable (has ~1.8s remaining)
+        // Restoring must retain the elapsed age, not reset the TTL.
+        assert!(
+            index2.hot_cache.entries.read().unwrap()[&event.id]
+                .cached_at
+                .elapsed()
+                >= Duration::from_secs(30)
+        );
+
+        // The remaining TTL still permits retrieval.
         let events = index2
             .hot_cache
             .get_maintainer_events(&event.pubkey, "test-repo", None);
         assert_eq!(events.len(), 1);
 
-        // Wait 2 seconds (total 2.2s > 2s expiry)
-        std::thread::sleep(Duration::from_secs(2));
+        // Advance the restored entry to expiry without a wall-clock wait.
+        expire_hot_cache(&index2.hot_cache);
 
         // Now it should be expired
         let events = index2

src/sync/relay_connection.rs

@@ -2833,8 +2833,12 @@ impl RelayConnection {
     }
 }
 
+#[cfg(test)]
+mod test_relay;
+
 #[cfg(test)]
 mod tests {
+    use super::test_relay::TestRelay;
     use super::*;
 
     #[test]
@@ -3111,10 +3115,8 @@ mod tests {
 
     #[tokio::test]
     async fn fetch_events_targets_the_connections_exact_relay() {
-        let configured = LocalRelayBuilder::default().build();
-        configured.run().await.expect("start configured relay");
-        let other = LocalRelayBuilder::default().build();
-        other.run().await.expect("start other relay");
+        let configured = TestRelay::start(LocalRelayBuilder::default()).await;
+        let other = TestRelay::start(LocalRelayBuilder::default()).await;
 
         let expected = EventBuilder::new(Kind::TextNote, "only on the other relay")
             .finalize(&Keys::generate())
@@ -3167,8 +3169,7 @@ mod tests {
         const AUTHORS: usize = 3;
         const EVENTS_PER_AUTHOR: usize = 400;
 
-        let relay = LocalRelayBuilder::default().build();
-        relay.run().await.expect("start burst relay");
+        let relay = TestRelay::start(LocalRelayBuilder::default()).await;
         let mut authors = Vec::with_capacity(AUTHORS);
         for author_index in 0..AUTHORS {
             let keys = Keys::generate();
@@ -3229,8 +3230,7 @@ mod tests {
 
     #[tokio::test]
     async fn immediate_empty_eose_cannot_arrive_before_permit_registration() {
-        let relay = LocalRelayBuilder::default().build();
-        relay.run().await.expect("start empty relay");
+        let relay = TestRelay::start(LocalRelayBuilder::default()).await;
         let connection = RelayConnection::new(
             relay.url().await.to_string(),
             Some(Keys::generate()),
@@ -3276,8 +3276,7 @@ mod tests {
 
     #[tokio::test]
     async fn peer_closed_subscription_is_removed_from_sdk_registry() {
-        let relay = LocalRelayBuilder::default().build();
-        relay.run().await.expect("start registry relay");
+        let relay = TestRelay::start(LocalRelayBuilder::default()).await;
         let connection = RelayConnection::new(
             relay.url().await.to_string(),
             Some(Keys::generate()),
@@ -3311,8 +3310,7 @@ mod tests {
 
     #[tokio::test]
     async fn query_rate_closed_paces_later_wire_requests() {
-        let relay = LocalRelayBuilder::default().queries_per_minute(1).build();
-        relay.run().await.expect("start query-limited relay");
+        let relay = TestRelay::start(LocalRelayBuilder::default().queries_per_minute(1)).await;
         let connection = RelayConnection::new(
             relay.url().await.to_string(),
             Some(Keys::generate()),
@@ -3816,8 +3814,7 @@ mod tests {
 
     #[tokio::test]
     async fn minimum_churn_extension_keeps_full_and_auxiliary_subscriptions_open() {
-        let relay = LocalRelayBuilder::default().build();
-        relay.run().await.expect("start local relay");
+        let relay = TestRelay::start(LocalRelayBuilder::default()).await;
         let connection = RelayConnection::new(
             relay.url().await.to_string(),
             Some(Keys::generate()),
@@ -3879,8 +3876,7 @@ mod tests {
 
     #[tokio::test]
     async fn failed_minimum_churn_extension_restores_only_the_retired_tail() {
-        let relay = LocalRelayBuilder::default().build();
-        relay.run().await.expect("start local relay");
+        let relay = TestRelay::start(LocalRelayBuilder::default()).await;
         let connection = RelayConnection::new(
             relay.url().await.to_string(),
             Some(Keys::generate()),
@@ -3971,8 +3967,7 @@ mod tests {
 
     #[tokio::test]
     async fn partial_close_failure_restores_the_already_closed_tail_group() {
-        let relay = LocalRelayBuilder::default().build();
-        relay.run().await.expect("start local relay");
+        let relay = TestRelay::start(LocalRelayBuilder::default()).await;
         let connection = RelayConnection::new(
             relay.url().await.to_string(),
             Some(Keys::generate()),
@@ -4414,8 +4409,7 @@ mod tests {
 
     #[tokio::test]
     async fn immediate_transient_capacity_rejects_a_saturated_session() {
-        let relay = LocalRelayBuilder::default().build();
-        relay.run().await.expect("start local relay");
+        let relay = TestRelay::start(LocalRelayBuilder::default()).await;
         let relay_url = relay.url().await.to_string();
         let connection = permissive_connection(&relay_url, Keys::generate());
         connection.connect(3).await.expect("connect local relay");

src/sync/relay_connection/test_relay.rs

@@ -0,0 +1,136 @@
+//! Local relay fixture that owns its listener from allocation through shutdown.
+
+use std::convert::Infallible;
+use std::ops::Deref;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use http_body_util::Full;
+use hyper::body::Bytes;
+use hyper::service::service_fn;
+use hyper::{Request, Response};
+use hyper_util::rt::TokioIo;
+use nostr_sdk::prelude::{LocalRelay, LocalRelayBuilder};
+use tokio::net::TcpListener;
+use tokio::task::{JoinHandle, JoinSet};
+use tokio_tungstenite::tungstenite::handshake::derive_accept_key;
+
+pub(super) struct TestRelay {
+    relay: LocalRelay,
+    url: String,
+    task: JoinHandle<()>,
+}
+
+impl TestRelay {
+    pub(super) async fn start(builder: LocalRelayBuilder) -> Self {
+        // The SDK's run() probes and releases a port before binding it again.
+        // Keep this listener bound and give upgraded streams to the SDK instead.
+        let listener = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind test relay");
+        let url = format!("ws://{}", listener.local_addr().expect("relay address"));
+        let relay = builder.build();
+        let server = relay.clone();
+        let task = tokio::spawn(async move {
+            let mut connections = JoinSet::new();
+            loop {
+                tokio::select! {
+                    accepted = listener.accept() => {
+                        let (stream, addr) = accepted.expect("accept test relay connection");
+                        let relay = server.clone();
+                        connections.spawn(async move {
+                            let upgrade = Arc::new(Mutex::new(None));
+                            let requested_upgrade = upgrade.clone();
+                            let service = service_fn(move |req: Request<hyper::body::Incoming>| {
+                                let key = req.headers().get("sec-websocket-key").cloned();
+                                let response = if let Some(key) = key {
+                                    let accept = derive_accept_key(key.as_bytes());
+                                    *requested_upgrade.lock().expect("upgrade lock") =
+                                        Some(hyper::upgrade::on(req));
+                                    Response::builder()
+                                        .status(101)
+                                        .header("connection", "upgrade")
+                                        .header("upgrade", "websocket")
+                                        .header("sec-websocket-accept", accept)
+                                } else {
+                                    Response::builder().status(400)
+                                };
+                                async move {
+                                    Ok::<_, Infallible>(response.body(Full::new(Bytes::new())).unwrap())
+                                }
+                            });
+                            let http = hyper::server::conn::http1::Builder::new()
+                                .serve_connection(TokioIo::new(stream), service)
+                                .with_upgrades();
+                            if !matches!(tokio::time::timeout(Duration::from_secs(5), http).await, Ok(Ok(()))) {
+                                return;
+                            }
+                            let upgrade = upgrade.lock().expect("upgrade lock").take();
+                            if let Some(upgrade) = upgrade {
+                                if let Ok(Ok(stream)) = tokio::time::timeout(Duration::from_secs(5), upgrade).await {
+                                    let _ = relay.take_connection(TokioIo::new(stream), addr).await;
+                                }
+                            }
+                        });
+                    }
+                    result = connections.join_next(), if !connections.is_empty() => {
+                        result.expect("connection task").expect("connection task panicked");
+                    }
+                }
+            }
+        });
+        Self { relay, url, task }
+    }
+
+    pub(super) async fn url(&self) -> String {
+        self.url.clone()
+    }
+
+    pub(super) fn shutdown(&self) {
+        self.task.abort();
+        self.relay.shutdown();
+    }
+}
+
+impl Deref for TestRelay {
+    type Target = LocalRelay;
+
+    fn deref(&self) -> &Self::Target {
+        &self.relay
+    }
+}
+
+impl Drop for TestRelay {
+    fn drop(&mut self) {
+        // Aborting the listener also drops its JoinSet and aborts connections.
+        self.shutdown();
+    }
+}
+
+#[tokio::test]
+async fn concurrent_relays_keep_distinct_reachable_listeners() {
+    let mut starts = JoinSet::new();
+    for _ in 0..32 {
+        starts.spawn(TestRelay::start(LocalRelayBuilder::default()));
+    }
+    let mut relays = Vec::new();
+    let mut urls = std::collections::HashSet::new();
+    while let Some(result) = starts.join_next().await {
+        let relay = result.expect("start task");
+        assert!(urls.insert(relay.url().await.to_owned()));
+        relays.push(relay);
+    }
+    for relay in &relays {
+        let (mut socket, _) = tokio::time::timeout(
+            Duration::from_secs(5),
+            tokio_tungstenite::connect_async(relay.url().await),
+        )
+        .await
+        .expect("handshake deadline")
+        .expect("connect test relay");
+        tokio::time::timeout(Duration::from_secs(5), socket.close(None))
+            .await
+            .expect("close deadline")
+            .expect("close test connection");
+    }
+}

src/test_listener.rs

@@ -0,0 +1,93 @@
+//! Private listener handoff protocol for subprocess test fixtures.
+
+use std::os::fd::{FromRawFd, RawFd};
+
+use anyhow::{bail, Context, Result};
+use tokio::net::TcpListener;
+
+pub(crate) fn from_env() -> Result<Option<TcpListener>> {
+    let Some(fd) = std::env::var_os("NGIT_TEST_LISTENER_FD") else {
+        return Ok(None);
+    };
+    if std::env::var("NGIT_TEST").as_deref() != Ok("1") {
+        bail!("NGIT_TEST_LISTENER_FD is available only with NGIT_TEST=1");
+    }
+    let fd: RawFd = fd
+        .to_str()
+        .context("listener descriptor is not UTF-8")?
+        .parse()
+        .context("listener descriptor is not an integer")?;
+    listener_from_fd(fd).map(Some)
+}
+
+fn listener_from_fd(fd: RawFd) -> Result<TcpListener> {
+    if fd < 3 {
+        bail!("test listener must not use a standard stream descriptor");
+    }
+    // dup validates the descriptor and gives this function its own ownership.
+    // The inherited descriptor remains open until the test subprocess exits.
+    let owned = unsafe { libc::dup(fd) };
+    if owned < 0 {
+        return Err(std::io::Error::last_os_error()).context("duplicate test listener");
+    }
+    // SAFETY: dup returned a fresh owned descriptor, consumed exactly once.
+    let listener = unsafe { std::net::TcpListener::from_raw_fd(owned) };
+    // Do not leak either copy into Git/SSH children spawned by the relay.
+    for descriptor in [fd, owned] {
+        if unsafe { libc::fcntl(descriptor, libc::F_SETFD, libc::FD_CLOEXEC) } < 0 {
+            return Err(std::io::Error::last_os_error()).context("protect test listener from exec");
+        }
+    }
+    let mut accepting: libc::c_int = 0;
+    let mut length = std::mem::size_of_val(&accepting) as libc::socklen_t;
+    // SAFETY: both pointers reference live, correctly sized writable values.
+    let result = unsafe {
+        libc::getsockopt(
+            owned,
+            libc::SOL_SOCKET,
+            libc::SO_ACCEPTCONN,
+            (&mut accepting as *mut libc::c_int).cast(),
+            &mut length,
+        )
+    };
+    if result != 0 {
+        return Err(std::io::Error::last_os_error()).context("inspect test listener");
+    }
+    if accepting == 0 || !listener.local_addr()?.ip().is_loopback() {
+        bail!("test listener must be a listening loopback TCP socket");
+    }
+    listener.set_nonblocking(true)?;
+    TcpListener::from_std(listener).context("adopt test listener")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::os::fd::AsRawFd;
+
+    #[tokio::test]
+    async fn adoption_preserves_the_reserved_address() {
+        let reserved = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let address = reserved.local_addr().unwrap();
+        let listener = listener_from_fd(reserved.as_raw_fd()).unwrap();
+        drop(reserved);
+        assert_eq!(listener.local_addr().unwrap(), address);
+        assert_eq!(
+            std::net::TcpListener::bind(address).unwrap_err().kind(),
+            std::io::ErrorKind::AddrInUse
+        );
+        let client = tokio::net::TcpStream::connect(address).await.unwrap();
+        let (_, peer) = tokio::time::timeout(std::time::Duration::from_secs(5), listener.accept())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(peer, client.local_addr().unwrap());
+    }
+
+    #[tokio::test]
+    async fn adoption_rejects_non_listener_descriptors() {
+        let file = std::fs::File::open("/dev/null").unwrap();
+        assert!(listener_from_fd(file.as_raw_fd()).is_err());
+        assert!(listener_from_fd(-1).is_err());
+    }
+}

tests/common/auth_gating_relay.rs

@@ -35,6 +35,7 @@ use hyper_util::rt::TokioIo;
 use nostr_sdk::prelude::{Event, Keys, Kind, PublicKey};
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
+use tokio::task::JoinSet;
 use tokio_tungstenite::tungstenite::protocol::Role;
 use tokio_tungstenite::tungstenite::Message;
 use tokio_tungstenite::WebSocketStream;
@@ -82,6 +83,8 @@ impl AuthGatingRelay {
         let backend_url = backend_url.to_string();
         let accept_state = state.clone();
         let handle = tokio::spawn(async move {
+            let mut connections = JoinSet::new();
+            let upgrades = Arc::new(Mutex::new(JoinSet::new()));
             loop {
                 tokio::select! {
                     accepted = listener.accept() => {
@@ -89,11 +92,13 @@ impl AuthGatingRelay {
                         let backend_url = backend_url.clone();
                         let state = accept_state.clone();
                         let io = TokioIo::new(stream);
-                        tokio::spawn(async move {
+                        let upgrades = upgrades.clone();
+                        connections.spawn(async move {
                             let service = service_fn(move |req| {
+                                let upgrades = upgrades.clone();
                                 let backend_url = backend_url.clone();
                                 let state = state.clone();
-                                async move { handle_request(req, backend_url, mode, state).await }
+                                async move { handle_request(req, backend_url, mode, state, upgrades).await }
                             });
                             // Errors are expected when clients disconnect.
                             let _ = http1::Builder::new()
@@ -103,8 +108,12 @@ impl AuthGatingRelay {
                         });
                     }
                     _ = &mut shutdown_rx => break,
+                    _ = connections.join_next(), if !connections.is_empty() => {}
                 }
             }
+            connections.shutdown().await;
+            let mut upgrades = std::mem::take(&mut *upgrades.lock().unwrap());
+            upgrades.shutdown().await;
         });
 
         Self {
@@ -140,7 +149,10 @@ impl AuthGatingRelay {
             let _ = tx.send(());
         }
         if let Some(handle) = self.handle.take() {
-            let _ = handle.await;
+            tokio::time::timeout(std::time::Duration::from_secs(5), handle)
+                .await
+                .expect("auth gating relay tasks should stop")
+                .expect("auth gating relay server task should not panic");
         }
     }
 }
@@ -158,6 +170,7 @@ async fn handle_request(
     backend_url: String,
     mode: GateMode,
     state: GateState,
+    upgrades: Arc<Mutex<JoinSet<()>>>,
 ) -> Result<Response<Full<Bytes>>, hyper::Error> {
     let is_websocket = req
         .headers()
@@ -173,7 +186,9 @@ async fn handle_request(
             .map(str::to_string)
         {
             let accept_key = derive_accept_key(key.as_bytes());
-            tokio::spawn(async move {
+            let mut upgrades = upgrades.lock().unwrap();
+            while upgrades.try_join_next().is_some() {}
+            upgrades.spawn(async move {
                 match hyper::upgrade::on(req).await {
                     Ok(upgraded) => {
                         let ws = WebSocketStream::from_raw_socket(
@@ -251,7 +266,7 @@ async fn run_session(
 
     let mut authenticated = false;
     let mut backend: Option<SplitSink<_, Message>> = None;
-    let mut backend_task: Option<tokio::task::JoinHandle<()>> = None;
+    let mut backend_tasks = JoinSet::new();
 
     while let Some(message) = client_rx.next().await {
         let message = message.map_err(|e| format!("client read: {e}"))?;
@@ -273,7 +288,7 @@ async fn run_session(
                                 .map_err(|e| format!("backend connect failed: {e}"))?;
                         let (backend_tx, backend_rx) = backend_ws.split();
                         backend = Some(backend_tx);
-                        backend_task = Some(spawn_backend_forwarder(backend_rx, client_tx.clone()));
+                        backend_tasks.spawn(forward_backend(backend_rx, client_tx.clone()));
                     }
                     continue;
                 }
@@ -342,9 +357,7 @@ async fn run_session(
         }
     }
 
-    if let Some(task) = backend_task {
-        task.abort();
-    }
+    backend_tasks.shutdown().await;
     Ok(())
 }
 
@@ -395,20 +408,18 @@ async fn handle_auth(
     .await
 }
 
-fn spawn_backend_forwarder(
+async fn forward_backend(
     mut backend_rx: SplitStream<
         WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>,
     >,
     client_tx: SharedClientSink,
-) -> tokio::task::JoinHandle<()> {
-    tokio::spawn(async move {
-        while let Some(message) = backend_rx.next().await {
-            let Ok(message) = message else { break };
-            if client_tx.lock().await.send(message).await.is_err() {
-                break;
-            }
+) {
+    while let Some(message) = backend_rx.next().await {
+        let Ok(message) = message else { break };
+        if client_tx.lock().await.send(message).await.is_err() {
+            break;
         }
-    })
+    }
 }
 
 /// Derive the Sec-WebSocket-Accept key from the request key.
@@ -427,3 +438,25 @@ fn derive_accept_key(request_key: &[u8]) -> String {
         hash.as_byte_array(),
     )
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn stop_closes_active_websocket_sessions() {
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            let gate = AuthGatingRelay::start("ws://127.0.0.1:1", GateMode::Restricted).await;
+            let (mut client, _) = tokio_tungstenite::connect_async(gate.url()).await.unwrap();
+            let challenge = client.next().await.unwrap().unwrap();
+            assert!(challenge.to_text().unwrap().contains("AUTH"));
+            gate.stop().await;
+            assert!(matches!(
+                client.next().await,
+                None | Some(Err(_)) | Some(Ok(Message::Close(_)))
+            ));
+        })
+        .await
+        .expect("gate stop must close active sessions");
+    }
+}

tests/common/censoring_proxy.rs

@@ -61,6 +61,7 @@ impl CensoringProxy {
         let accept_dropped = dropped.clone();
 
         let handle = tokio::spawn(async move {
+            let mut connections = tokio::task::JoinSet::new();
             loop {
                 tokio::select! {
                     accepted = listener.accept() => {
@@ -68,7 +69,7 @@ impl CensoringProxy {
                         let backend_url = backend_url.clone();
                         let withheld = accept_withheld.clone();
                         let dropped = accept_dropped.clone();
-                        tokio::spawn(async move {
+                        connections.spawn(async move {
                             if let Err(error) =
                                 proxy_connection(stream, &backend_url, withheld, dropped).await
                             {
@@ -77,9 +78,13 @@ impl CensoringProxy {
                             }
                         });
                     }
+                    result = connections.join_next(), if !connections.is_empty() => {
+                        result.expect("connection task").expect("fixture connection panicked");
+                    }
                     _ = &mut shutdown_rx => break,
                 }
             }
+            connections.shutdown().await;
         });
 
         Self {
@@ -131,6 +136,9 @@ impl CensoringProxy {
 
 impl Drop for CensoringProxy {
     fn drop(&mut self) {
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+        }
         if let Some(tx) = self.shutdown_tx.take() {
             let _ = tx.send(());
         }

tests/common/flapping_relay.rs

@@ -31,13 +31,14 @@ impl FlappingRelay {
         let notify = connection_observed.clone();
 
         let handle = tokio::spawn(async move {
+            let mut connections = tokio::task::JoinSet::new();
             loop {
                 tokio::select! {
                     accepted = listener.accept() => {
                         let Ok((stream, _)) = accepted else { break };
                         let observed = observed.clone();
                         let notify = notify.clone();
-                        tokio::spawn(async move {
+                        connections.spawn(async move {
                             let Ok(mut websocket) = tokio_tungstenite::accept_async(stream).await
                             else {
                                 // NIP-11 HTTP probes are expected to fail this
@@ -61,9 +62,13 @@ impl FlappingRelay {
                             }
                         });
                     }
+                    result = connections.join_next(), if !connections.is_empty() => {
+                        result.expect("connection task").expect("fixture connection panicked");
+                    }
                     _ = &mut shutdown_rx => break,
                 }
             }
+            connections.shutdown().await;
         });
 
         Self {
@@ -112,6 +117,9 @@ impl FlappingRelay {
 
 impl Drop for FlappingRelay {
     fn drop(&mut self) {
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+        }
         if let Some(tx) = self.shutdown_tx.take() {
             let _ = tx.send(());
         }

tests/common/git_server.rs

@@ -150,6 +150,7 @@ impl SimpleGitServer {
         let repo_path = Arc::new(bare_repo_path);
 
         let handle = tokio::spawn(async move {
+            let mut connections = tokio::task::JoinSet::new();
             println!("[SmartGitServer] Server loop started on port {}", port);
             eprintln!("[SmartGitServer] Server loop started on port {}", port);
             loop {
@@ -161,7 +162,7 @@ impl SimpleGitServer {
                                 let repo_path = Arc::clone(&repo_path);
                                 let io = TokioIo::new(stream);
 
-                                tokio::spawn(async move {
+                                connections.spawn(async move {
                                     let service = service_fn(move |req| {
                                         let repo_path = Arc::clone(&repo_path);
                                         async move { handle_request(req, &repo_path).await }
@@ -183,18 +184,20 @@ impl SimpleGitServer {
                             }
                         }
                     }
+                    _ = connections.join_next(), if !connections.is_empty() => {},
                     _ = &mut shutdown_rx => {
                         // Shutdown signal received
                         break;
                     }
                 }
             }
+            connections.shutdown().await;
         });
 
         let url = format!("http://127.0.0.1:{}", port);
 
         // 7. Wait for server to be ready
-        wait_for_server_ready(port).await;
+        // The reserved listener is already bound; the runtime drives accepts.
 
         Self {
             shutdown_tx: Some(shutdown_tx),
@@ -236,8 +239,10 @@ impl Drop for SimpleGitServer {
         if let Some(tx) = self.shutdown_tx.take() {
             let _ = tx.send(());
         }
-        // Note: We can't await the handle in drop, but the temp_dir cleanup
-        // will happen automatically when _temp_dir is dropped
+        // Dropping the owner cancels every connection through its JoinSet.
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+        }
     }
 }
 
@@ -304,31 +309,6 @@ fn guess_content_type(path: &Path) -> &'static str {
     }
 }
 
-/// Wait for the server to be ready to accept connections.
-async fn wait_for_server_ready(port: u16) {
-    let max_attempts = 50; // 5 seconds total
-    let delay = std::time::Duration::from_millis(100);
-
-    for attempt in 0..max_attempts {
-        match tokio::net::TcpStream::connect(format!("127.0.0.1:{}", port)).await {
-            Ok(_) => {
-                // Connection successful, server is ready
-                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
-                return;
-            }
-            Err(_) => {
-                if attempt == max_attempts - 1 {
-                    panic!(
-                        "SimpleGitServer failed to start after {} attempts",
-                        max_attempts
-                    );
-                }
-                tokio::time::sleep(delay).await;
-            }
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -597,6 +577,7 @@ impl SmartGitServer {
         let repo_path = Arc::new(bare_repo_path);
 
         let handle = tokio::spawn(async move {
+            let mut connections = tokio::task::JoinSet::new();
             loop {
                 tokio::select! {
                     accept_result = listener.accept() => {
@@ -605,7 +586,7 @@ impl SmartGitServer {
                                 let repo_path = Arc::clone(&repo_path);
                                 let io = TokioIo::new(stream);
 
-                                tokio::spawn(async move {
+                                connections.spawn(async move {
                                     let service = service_fn(move |req| {
                                         let repo_path = Arc::clone(&repo_path);
                                         async move { handle_smart_request(req, &repo_path).await }
@@ -627,18 +608,20 @@ impl SmartGitServer {
                             }
                         }
                     }
+                    _ = connections.join_next(), if !connections.is_empty() => {},
                     _ = &mut shutdown_rx => {
                         // Shutdown signal received
                         break;
                     }
                 }
             }
+            connections.shutdown().await;
         });
 
         let url = format!("http://127.0.0.1:{}", port);
 
         // 6. Wait for server to be ready
-        wait_for_server_ready(port).await;
+        // The reserved listener is already bound; the runtime drives accepts.
 
         Self {
             shutdown_tx: Some(shutdown_tx),
@@ -680,8 +663,10 @@ impl Drop for SmartGitServer {
         if let Some(tx) = self.shutdown_tx.take() {
             let _ = tx.send(());
         }
-        // Note: We can't await the handle in drop, but the temp_dir cleanup
-        // will happen automatically when _temp_dir is dropped
+        // Dropping the owner cancels every connection through its JoinSet.
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+        }
     }
 }
 
@@ -755,11 +740,11 @@ async fn handle_info_refs_upload_pack(
     git_protocol_version: Option<&str>,
 ) -> Result<Response<Full<Bytes>>, hyper::Error> {
     use std::process::Stdio;
-    use tokio::io::AsyncReadExt;
     use tokio::process::Command as TokioCommand;
 
     // Spawn git upload-pack --advertise-refs
     let mut cmd = TokioCommand::from(grasp_audit::git_command());
+    cmd.kill_on_drop(true);
     cmd.arg("-c")
         .arg("uploadpack.allowReachableSHA1InWant=true")
         .arg("-c")
@@ -780,7 +765,7 @@ async fn handle_info_refs_upload_pack(
         .stdout(Stdio::piped())
         .stderr(Stdio::piped());
 
-    let mut child = match cmd.spawn() {
+    let child = match cmd.spawn() {
         Ok(child) => child,
         Err(e) => {
             eprintln!("Failed to spawn git upload-pack: {}", e);
@@ -791,20 +776,15 @@ async fn handle_info_refs_upload_pack(
         }
     };
 
-    // Read stdout
-    let mut output = Vec::new();
-    if let Some(mut stdout) = child.stdout.take() {
-        if let Err(e) = stdout.read_to_end(&mut output).await {
-            eprintln!("Failed to read git output: {}", e);
-        }
-    }
-
-    // Wait for process
-    let status = child.wait().await;
-    if let Ok(s) = &status {
-        if !s.success() {
-            eprintln!("git upload-pack --advertise-refs failed");
-        }
+    let output = match child.wait_with_output().await {
+        Ok(output) => output,
+        Err(error) => return Ok(git_io_failure(error)),
+    };
+    if !output.status.success() {
+        eprintln!(
+            "git upload-pack --advertise-refs failed: {}",
+            String::from_utf8_lossy(&output.stderr)
+        );
     }
 
     // Build response with pkt-line header
@@ -821,7 +801,7 @@ async fn handle_info_refs_upload_pack(
     response_body.extend_from_slice(b"0000");
 
     // Then the git output
-    response_body.extend_from_slice(&output);
+    response_body.extend_from_slice(&output.stdout);
 
     Ok(Response::builder()
         .status(StatusCode::OK)
@@ -844,7 +824,6 @@ async fn handle_upload_pack(
 ) -> Result<Response<Full<Bytes>>, hyper::Error> {
     use http_body_util::BodyExt;
     use std::process::Stdio;
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
     use tokio::process::Command as TokioCommand;
 
     // Read request body
@@ -852,6 +831,7 @@ async fn handle_upload_pack(
 
     // Spawn git upload-pack
     let mut cmd = TokioCommand::from(grasp_audit::git_command());
+    cmd.kill_on_drop(true);
     cmd.arg("-c")
         .arg("uploadpack.allowReachableSHA1InWant=true")
         .arg("-c")
@@ -871,7 +851,7 @@ async fn handle_upload_pack(
         .stdout(Stdio::piped())
         .stderr(Stdio::piped());
 
-    let mut child = match cmd.spawn() {
+    let child = match cmd.spawn() {
         Ok(child) => child,
         Err(e) => {
             eprintln!("Failed to spawn git upload-pack: {}", e);
@@ -882,51 +862,119 @@ async fn handle_upload_pack(
         }
     };
 
-    // Write request body to stdin
-    if let Some(mut stdin) = child.stdin.take() {
-        if let Err(e) = stdin.write_all(&body_bytes).await {
-            eprintln!("Failed to write to git stdin: {}", e);
-        }
-        // Close stdin to signal end of input
-        drop(stdin);
-    }
-
-    // Read stdout
-    let mut output = Vec::new();
-    if let Some(mut stdout) = child.stdout.take() {
-        if let Err(e) = stdout.read_to_end(&mut output).await {
-            eprintln!("Failed to read git output: {}", e);
-        }
-    }
-
-    // Read stderr for debugging
-    let mut stderr_output = Vec::new();
-    if let Some(mut stderr) = child.stderr.take() {
-        let _ = stderr.read_to_end(&mut stderr_output).await;
-    }
-
-    // Wait for process
-    let status = child.wait().await;
-    if let Ok(s) = &status {
-        if !s.success() {
-            let stderr_str = String::from_utf8_lossy(&stderr_output);
-            eprintln!("git upload-pack failed: {}", stderr_str);
-        }
+    let output = match collect_git_output(child, &body_bytes).await {
+        Ok(output) => output,
+        Err(error) => return Ok(git_io_failure(error)),
+    };
+    if !output.status.success() {
+        eprintln!(
+            "git upload-pack failed: {}",
+            String::from_utf8_lossy(&output.stderr)
+        );
     }
 
     Ok(Response::builder()
         .status(StatusCode::OK)
         .header("Content-Type", "application/x-git-upload-pack-result")
         .header("Cache-Control", "no-cache")
-        .body(Full::new(Bytes::from(output)))
+        .body(Full::new(Bytes::from(output.stdout)))
         .unwrap())
 }
 
+/// Write the request while draining both output pipes to prevent backpressure
+/// deadlocks. Cancelling the future kills the command's child process.
+async fn collect_git_output(
+    mut child: tokio::process::Child,
+    input: &[u8],
+) -> std::io::Result<std::process::Output> {
+    use tokio::io::AsyncWriteExt;
+    let stdin = child.stdin.take();
+    let write = async {
+        if let Some(mut stdin) = stdin {
+            stdin.write_all(input).await?;
+        }
+        Ok::<_, std::io::Error>(())
+    };
+    let (_, output) = tokio::try_join!(write, child.wait_with_output())?;
+    Ok(output)
+}
+
+fn git_io_failure(error: std::io::Error) -> Response<Full<Bytes>> {
+    eprintln!("Failed to collect git output: {error}");
+    Response::builder()
+        .status(StatusCode::INTERNAL_SERVER_ERROR)
+        .body(Full::new(Bytes::from("Failed to collect git output")))
+        .unwrap()
+}
+
 #[cfg(test)]
 mod smart_git_server_tests {
     use super::*;
     use crate::common::purgatory_helpers::{create_test_repo_with_commit, CommitVariant};
 
+    async fn assert_stop_closes_connection(
+        url: String,
+        stop: impl std::future::Future<Output = ()>,
+    ) {
+        use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            let address = url.strip_prefix("http://").unwrap();
+            let mut stream = tokio::net::TcpStream::connect(address).await.unwrap();
+            stream
+                .write_all(b"GET /missing HTTP/1.1\r\nHost: localhost\r\n\r\n")
+                .await
+                .unwrap();
+            let mut reader = BufReader::new(stream);
+            let mut status = String::new();
+            reader.read_line(&mut status).await.unwrap();
+            assert!(status.starts_with("HTTP/1.1 404"));
+            stop.await;
+            let mut remaining = Vec::new();
+            reader.read_to_end(&mut remaining).await.unwrap();
+        })
+        .await
+        .expect("stopped server must close keep-alive connections");
+    }
+
+    #[tokio::test]
+    async fn both_git_servers_close_owned_connections_on_stop() {
+        let directory = tempfile::tempdir().unwrap();
+        create_test_repo_with_commit(directory.path(), CommitVariant::StateTest).unwrap();
+        let simple = SimpleGitServer::start(directory.path()).await;
+        assert_stop_closes_connection(simple.url().to_string(), simple.stop()).await;
+        let smart = SmartGitServer::start(directory.path()).await;
+        assert_stop_closes_connection(smart.url().to_string(), smart.stop()).await;
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn git_io_drains_output_while_writing_input() {
+        use std::process::Stdio;
+        let directory = tempfile::tempdir().unwrap();
+        let payload = vec![b'x'; 1024 * 1024];
+        let path = directory.path().join("payload");
+        std::fs::write(&path, &payload).unwrap();
+        let child = tokio::process::Command::new("sh")
+            .args(["-c", "cat \"$1\"; cat \"$1\" >&2; cat", "test-io"])
+            .arg(&path)
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .kill_on_drop(true)
+            .spawn()
+            .unwrap();
+        let output = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            collect_git_output(child, &payload),
+        )
+        .await
+        .expect("all three pipes must make progress")
+        .unwrap();
+        assert!(output.status.success());
+        assert_eq!(output.stdout, payload.repeat(2));
+        assert_eq!(output.stderr, payload);
+    }
+
     #[tokio::test]
     async fn test_smart_git_server_starts_and_stops() {
         // Create a test repo

tests/common/mock_relay.rs

@@ -34,7 +34,7 @@
 //! - Does NOT perform any GRASP validation (no purgatory, no git data checks)
 
 use std::net::SocketAddr;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
 use http_body_util::Full;
 use hyper::body::Bytes;
@@ -46,6 +46,7 @@ use hyper_util::rt::TokioIo;
 use nostr_sdk::prelude::*;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
+use tokio::task::JoinSet;
 
 /// Mock Nostr relay that accepts all events without validation.
 ///
@@ -195,6 +196,25 @@ impl MockRelay {
         .await
     }
 
+    /// Recover on an owned listener, with initial events visible before accepts.
+    pub async fn start_on_listener(listener: std::net::TcpListener, events: Vec<Event>) -> Self {
+        let port = listener.local_addr().expect("mock listener address").port();
+        listener
+            .set_nonblocking(true)
+            .expect("nonblocking mock listener");
+        let listener = TcpListener::from_std(listener).expect("register mock listener");
+        Self::start_with_listener(
+            listener,
+            port,
+            RateLimit::default(),
+            None,
+            None,
+            events,
+            None,
+        )
+        .await
+    }
+
     /// Internal method to start the relay with an existing listener.
     #[allow(clippy::too_many_arguments)]
     async fn start_with_listener(
@@ -232,6 +252,8 @@ impl MockRelay {
         let server_relay = relay.clone();
 
         let handle = tokio::spawn(async move {
+            let mut connections = JoinSet::new();
+            let upgrades = Arc::new(Mutex::new(JoinSet::new()));
             loop {
                 tokio::select! {
                     accept_result = listener.accept() => {
@@ -242,8 +264,10 @@ impl MockRelay {
                                 let custom_nip11 = custom_nip11.clone();
                                 let io = TokioIo::new(stream);
 
-                                tokio::spawn(async move {
+                                let upgrades = upgrades.clone();
+                                connections.spawn(async move {
                                     let service = service_fn(move |req| {
+                                        let upgrades = upgrades.clone();
                                         let relay = relay.clone();
                                         let custom_nip11 = custom_nip11.clone();
                                         async move {
@@ -253,6 +277,7 @@ impl MockRelay {
                                                 remote_addr,
                                                 pagination,
                                                 custom_nip11,
+                                                upgrades,
                                             )
                                             .await
                                         }
@@ -275,12 +300,15 @@ impl MockRelay {
                             }
                         }
                     }
-                    _ = &mut shutdown_rx => {
-                        // Shutdown signal received
-                        break;
-                    }
+                    _ = &mut shutdown_rx => break,
+                    _ = connections.join_next(), if !connections.is_empty() => {}
                 }
             }
+            // Stop HTTP services before draining upgrades so no service can
+            // create another WebSocket task after the snapshot is taken.
+            connections.shutdown().await;
+            let mut upgrades = std::mem::take(&mut *upgrades.lock().unwrap());
+            upgrades.shutdown().await;
         });
 
         let url = format!("ws://127.0.0.1:{}", port);
@@ -321,13 +349,17 @@ impl MockRelay {
 
         // Wait for server task to complete
         if let Some(handle) = self.handle.take() {
-            let _ = handle.await;
+            tokio::time::timeout(std::time::Duration::from_secs(5), handle)
+                .await
+                .expect("mock relay tasks should stop")
+                .expect("mock relay server task should not panic");
         }
     }
 }
 
 impl Drop for MockRelay {
     fn drop(&mut self) {
+        self.relay.shutdown();
         // Send shutdown signal if not already sent
         if let Some(tx) = self.shutdown_tx.take() {
             let _ = tx.send(());
@@ -342,6 +374,7 @@ async fn handle_request(
     addr: SocketAddr,
     pagination: Option<PaginationConfig>,
     custom_nip11: Option<serde_json::Value>,
+    upgrades: Arc<Mutex<JoinSet<()>>>,
 ) -> Result<Response<Full<Bytes>>, hyper::Error> {
     // Check for WebSocket upgrade request
     let is_websocket = req
@@ -361,8 +394,10 @@ async fn handle_request(
         if let Some(key) = key {
             let accept_key = derive_accept_key(key.as_bytes());
 
-            // Spawn task to handle the upgraded connection
-            tokio::spawn(async move {
+            // Retain upgrade tasks until the fixture shuts down.
+            let mut upgrades = upgrades.lock().unwrap();
+            while upgrades.try_join_next().is_some() {}
+            upgrades.spawn(async move {
                 match hyper::upgrade::on(req).await {
                     Ok(upgraded) => {
                         if let Err(e) = relay.take_connection(TokioIo::new(upgraded), addr).await {
@@ -471,6 +506,33 @@ mod tests {
     use nostr_sdk::prelude::*;
     use std::time::Duration;
 
+    #[tokio::test]
+    async fn stop_closes_active_websocket_sessions() {
+        use futures_util::{SinkExt, StreamExt};
+        use tokio_tungstenite::tungstenite::Message;
+        tokio::time::timeout(Duration::from_secs(5), async {
+            let mock = MockRelay::start().await;
+            let (mut client, _) = tokio_tungstenite::connect_async(mock.url()).await.unwrap();
+            client
+                .send(Message::Text(r#"["REQ","stop-probe",{"limit":1}]"#.into()))
+                .await
+                .unwrap();
+            loop {
+                let message = client.next().await.unwrap().unwrap();
+                if message.to_text().is_ok_and(|text| text.contains("EOSE")) {
+                    break;
+                }
+            }
+            mock.stop().await;
+            assert!(matches!(
+                client.next().await,
+                None | Some(Err(_)) | Some(Ok(Message::Close(_)))
+            ));
+        })
+        .await
+        .expect("mock stop must close active sessions");
+    }
+
     #[tokio::test]
     async fn test_mock_relay_starts_and_stops() {
         let mock = MockRelay::start().await;
@@ -494,10 +556,10 @@ mod tests {
             .add_relay(mock.url())
             .await
             .expect("Failed to add relay");
-        client.connect().await;
-
-        // Wait for connection
-        tokio::time::sleep(Duration::from_millis(500)).await;
+        client
+            .try_connect_relay(mock.url(), Duration::from_secs(5))
+            .await
+            .expect("connect to mock relay");
 
         // Create and send a simple event
         let event = EventBuilder::new(Kind::TextNote, "Test note from MockRelay test")

tests/common/neg_limiting_proxy.rs

@@ -74,6 +74,7 @@ impl NegLimitingProxy {
         let accept_opened = opened.clone();
 
         let handle = tokio::spawn(async move {
+            let mut connections = tokio::task::JoinSet::new();
             loop {
                 tokio::select! {
                     accepted = listener.accept() => {
@@ -82,7 +83,7 @@ impl NegLimitingProxy {
                         let peak = accept_peak.clone();
                         let rejected = accept_rejected.clone();
                         let opened = accept_opened.clone();
-                        tokio::spawn(async move {
+                        connections.spawn(async move {
                             if let Err(error) = proxy_connection(
                                 stream,
                                 &backend_url,
@@ -98,9 +99,13 @@ impl NegLimitingProxy {
                             }
                         });
                     }
+                    result = connections.join_next(), if !connections.is_empty() => {
+                        result.expect("connection task").expect("fixture connection panicked");
+                    }
                     _ = &mut shutdown_rx => break,
                 }
             }
+            connections.shutdown().await;
         });
 
         Self {
@@ -146,6 +151,9 @@ impl NegLimitingProxy {
 
 impl Drop for NegLimitingProxy {
     fn drop(&mut self) {
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+        }
         if let Some(tx) = self.shutdown_tx.take() {
             let _ = tx.send(());
         }

tests/common/nip09_helpers.rs

@@ -19,6 +19,24 @@ use super::purgatory_helpers::{
 };
 use super::sync_helpers::create_repo_announcement;
 
+/// Wait for promotion rather than assuming the worker runs within a fixed delay.
+async fn wait_until_served(client: &AuditClient, event_id: EventId) {
+    tokio::time::timeout(Duration::from_secs(10), async {
+        loop {
+            if client
+                .is_event_on_relay(event_id)
+                .await
+                .expect("query promoted event")
+            {
+                return;
+            }
+            tokio::time::sleep(Duration::from_millis(50)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("event {event_id} was not served after git data arrived"));
+}
+
 /// Publish a repo announcement, submit a matching state event, and push the
 /// deterministic git data so the announcement is promoted out of purgatory and
 /// becomes queryable.
@@ -130,16 +148,7 @@ pub async fn publish_served_repo(client: &AuditClient, test_name: &str) -> (Even
         Err(e) => panic!("git push error while promoting repo: {}", e),
     }
 
-    // Give the relay a moment to promote the events out of purgatory.
-    tokio::time::sleep(Duration::from_millis(300)).await;
-
-    assert!(
-        client
-            .is_event_on_relay(announcement.id)
-            .await
-            .expect("query announcement"),
-        "announcement should be served after git data arrives"
-    );
+    wait_until_served(client, announcement.id).await;
 
     (announcement, repo_id)
 }
@@ -253,15 +262,7 @@ pub async fn publish_served_repo_with_maintainers(
         Err(e) => panic!("git push error while promoting repo: {}", e),
     }
 
-    tokio::time::sleep(Duration::from_millis(300)).await;
-
-    assert!(
-        client
-            .is_event_on_relay(announcement.id)
-            .await
-            .expect("query announcement"),
-        "announcement should be served after git data arrives"
-    );
+    wait_until_served(client, announcement.id).await;
 
     (announcement, repo_id)
 }
@@ -320,22 +321,8 @@ pub async fn publish_served_repo_with_state_event(
     push_to_relay(temp_dir.path(), &relay_domain, &npub, &repo_id)
         .expect("git push should promote announcement + state event out of purgatory");
 
-    tokio::time::sleep(Duration::from_millis(300)).await;
-
-    assert!(
-        client
-            .is_event_on_relay(announcement.id)
-            .await
-            .expect("query announcement"),
-        "announcement should be served after git data arrives"
-    );
-    assert!(
-        client
-            .is_event_on_relay(state_event.id)
-            .await
-            .expect("query state event"),
-        "state event should be served after git data arrives"
-    );
+    wait_until_served(client, announcement.id).await;
+    wait_until_served(client, state_event.id).await;
 
     (announcement, repo_id, state_event)
 }
@@ -406,22 +393,8 @@ pub async fn publish_served_repo_with_state_event_and_maintainers(
     push_to_relay(temp_dir.path(), &relay_domain, &npub, &repo_id)
         .expect("git push should promote announcement + state event out of purgatory");
 
-    tokio::time::sleep(Duration::from_millis(300)).await;
-
-    assert!(
-        client
-            .is_event_on_relay(announcement.id)
-            .await
-            .expect("query announcement"),
-        "announcement should be served after git data arrives"
-    );
-    assert!(
-        client
-            .is_event_on_relay(state_event.id)
-            .await
-            .expect("query state event"),
-        "state event should be served after git data arrives"
-    );
+    wait_until_served(client, announcement.id).await;
+    wait_until_served(client, state_event.id).await;
 
     (announcement, repo_id, state_event)
 }
@@ -483,8 +456,6 @@ pub async fn publish_served_announcement_for_identifier(
         .await
         .expect("relay should accept state event");
 
-    tokio::time::sleep(Duration::from_millis(300)).await;
-
     if client
         .is_event_on_relay(announcement.id)
         .await
@@ -536,13 +507,7 @@ pub async fn publish_served_announcement_for_identifier(
         Err(e) => panic!("git push error while promoting repo: {}", e),
     }
 
-    assert!(
-        client
-            .is_event_on_relay(announcement.id)
-            .await
-            .expect("query announcement"),
-        "announcement should be served after git data arrives"
-    );
+    wait_until_served(client, announcement.id).await;
 
     announcement
 }
@@ -599,8 +564,6 @@ pub async fn publish_served_announcement_with_state_for_identifier(
         .await
         .expect("relay should accept state event");
 
-    tokio::time::sleep(Duration::from_millis(300)).await;
-
     let announcement_served = client
         .is_event_on_relay(announcement.id)
         .await
@@ -656,20 +619,8 @@ pub async fn publish_served_announcement_with_state_for_identifier(
         Err(e) => panic!("git push error while promoting repo: {}", e),
     }
 
-    assert!(
-        client
-            .is_event_on_relay(announcement.id)
-            .await
-            .expect("query announcement"),
-        "announcement should be served after git data arrives"
-    );
-    assert!(
-        client
-            .is_event_on_relay(state_event.id)
-            .await
-            .expect("query state event"),
-        "state event should be served after git data arrives"
-    );
+    wait_until_served(client, announcement.id).await;
+    wait_until_served(client, state_event.id).await;
 
     (announcement, state_event)
 }

tests/common/port.rs

@@ -1,48 +1,13 @@
-//! Race-free port reservation for test fixtures.
+//! Bound listener ownership for test fixtures.
 //!
-//! ## The race
+//! Reserving a port and dropping its listener before the server binds leaves
+//! a race with every other concurrent test. Instead, retain the listener and
+//! transfer it directly to an in-process fixture or through the subprocess
+//! listener handoff protocol. Retain a clone across server restarts.
 //!
-//! The naive pattern — bind `127.0.0.1:0`, read the kernel-assigned port,
-//! drop the listener, hand the bare `u16` to whoever wants it — has a
-//! TOCTOU window between drop and the consumer's actual `bind`. During
-//! that window, anything else in the process (or, more rarely, another
-//! process) can be handed the same port by the kernel.
-//!
-//! The race is rare on lightly-loaded hardware but has been observed in
-//! CI and during development, and the failure mode
-//! (`Address already in use (os error 98)`) is a hard test fail with no
-//! useful information for the next debugger. The hazard is sharply worse
-//! in patterns that reserve a port well in advance of binding (e.g.
-//! pre-allocating a port to embed in an announcement event before
-//! starting the relay that will host it).
-//!
-//! Our in-process fixtures ([`MockRelay`], [`SmartGitServer`]) avoid the
-//! race entirely by **keeping the listener bound** and handing it
-//! straight to their tokio accept loop. That trick doesn't work for the
-//! [`TestRelay`] subprocess — `ngit-grasp` binds itself from
-//! `NGIT_BIND_ADDRESS`, and inheriting the pre-bound fd would require
-//! Unix-specific `pre_exec` plumbing we'd rather not own in the test
-//! harness.
-//!
-//! ## The reservation pattern
-//!
-//! Instead, [`reserve_port`] returns a [`PortReservation`] that **holds the
-//! bound `TcpListener`** until the caller is about to start the real
-//! service. While any reservation is live, no other call to
-//! `reserve_port` in this process can be handed the same port — the
-//! kernel won't reissue a port that is currently bound.
-//!
-//! The caller drops the reservation immediately before the real bind,
-//! shrinking the TOCTOU window from "however long the fixture takes to
-//! spawn" (or, worse, "however long the test takes to build the
-//! announcement event") to "a few microseconds inside the start
-//! function". The retry loop in [`crate::common::relay::TestRelay`]
-//! covers that residual window — defense-in-depth that has never been
-//! observed to fire in local stress runs.
-//!
-//! [`MockRelay`]: crate::common::mock_relay::MockRelay
-//! [`SmartGitServer`]: crate::common::git_server::SmartGitServer
-//! [`TestRelay`]: crate::common::relay::TestRelay
+//! [`UnavailableEndpoint`] accepts and closes connections while retaining the
+//! address. Recovery stops that accept loop before handing the same listener
+//! to the recovered service, so another test can never claim the port.
 
 use std::net::TcpListener;
 
@@ -50,17 +15,9 @@ use std::net::TcpListener;
 /// a live `TcpListener` so that no other [`reserve_port`] call in this
 /// process can be handed the same number.
 ///
-/// The reservation is released by:
-///
-/// - calling [`PortReservation::release`] to consume the reservation and
-///   return the port number (preferred — makes the release explicit at the
-///   call site), or
-/// - simply dropping the value (also fine, but the release point is then
-///   tied to lexical scope).
-///
-/// The caller should release **immediately** before the consuming service
-/// performs its own `bind` so that the TOCTOU window between
-/// reservation-release and service-bind is as small as possible.
+/// Transfer it with [`PortReservation::into_std_listener`] to start a service
+/// without releasing its address. Dropping it or calling `release` relinquishes
+/// the address and must not be used before starting a replacement service.
 #[derive(Debug)]
 pub struct PortReservation {
     port: u16,
@@ -71,10 +28,8 @@ pub struct PortReservation {
 
 /// Reserve a specific loopback port.
 ///
-/// Used by restart flows that must reuse an address already embedded in
-/// published events (e.g. a repository announcement naming the relay's
-/// domain). Fails while the port is still bound — callers should wait
-/// for the previous holder to exit and retry within a bounded deadline.
+/// Fails while another listener owns the address. Restart flows should retain
+/// and transfer their existing listener rather than binding again.
 pub fn reserve_specific(port: u16) -> std::io::Result<PortReservation> {
     let listener = TcpListener::bind(("127.0.0.1", port))?;
     Ok(PortReservation {
@@ -89,10 +44,23 @@ impl PortReservation {
         self.port
     }
 
-    /// Consume the reservation, dropping the underlying listener and
-    /// returning the port number. The port is now free for the caller's
-    /// real service to bind. Prefer this over relying on lexical drop —
-    /// it makes the release point explicit at the call site.
+    /// Transfer the bound listener without making its port available again.
+    pub fn into_std_listener(self) -> TcpListener {
+        self._listener
+    }
+
+    /// Wrap a listener retained across a subprocess restart.
+    pub fn from_listener(listener: TcpListener) -> Self {
+        Self {
+            port: listener
+                .local_addr()
+                .expect("reserved listener address")
+                .port(),
+            _listener: listener,
+        }
+    }
+
+    /// Relinquish the address. A later bind to this port is inherently racy.
     pub fn release(self) -> u16 {
         let port = self.port;
         // `self` is consumed; the listener inside is dropped here.
@@ -103,7 +71,7 @@ impl PortReservation {
 
 /// Bind `127.0.0.1:0`, capture the assigned port, and **keep the listener
 /// bound** inside the returned [`PortReservation`] until the caller
-/// releases it.
+/// transfers or drops it.
 ///
 /// While the reservation is live, no other `reserve_port` call in this
 /// process will be handed the same port. See module docs for why this
@@ -121,6 +89,79 @@ pub fn reserve_port() -> PortReservation {
     }
 }
 
+/// A continuously reserved endpoint that rejects protocol connections until
+/// its listener is transferred to a recovered service.
+pub struct UnavailableEndpoint {
+    listener: Option<TcpListener>,
+    task: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl Default for UnavailableEndpoint {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl UnavailableEndpoint {
+    pub fn new() -> Self {
+        Self::from_listener(reserve_port().into_std_listener())
+    }
+
+    pub fn from_listener(listener: TcpListener) -> Self {
+        listener
+            .set_nonblocking(true)
+            .expect("nonblocking unavailable endpoint");
+        let accepting = tokio::net::TcpListener::from_std(
+            listener.try_clone().expect("retain unavailable listener"),
+        )
+        .expect("register unavailable listener");
+        let task = tokio::spawn(async move {
+            loop {
+                let (stream, _) = accepting
+                    .accept()
+                    .await
+                    .expect("accept unavailable connection");
+                drop(stream);
+            }
+        });
+        Self {
+            listener: Some(listener),
+            task: Some(task),
+        }
+    }
+
+    pub fn port(&self) -> u16 {
+        self.listener
+            .as_ref()
+            .expect("owned unavailable listener")
+            .local_addr()
+            .expect("unavailable address")
+            .port()
+    }
+
+    pub async fn into_listener(mut self) -> TcpListener {
+        if let Some(task) = self.task.take() {
+            task.abort();
+            match tokio::time::timeout(std::time::Duration::from_secs(5), task)
+                .await
+                .expect("unavailable accept loop should stop before recovery")
+            {
+                Err(error) if error.is_cancelled() => {}
+                result => panic!("unavailable accept loop ended unexpectedly: {result:?}"),
+            }
+        }
+        self.listener.take().expect("transfer unavailable listener")
+    }
+}
+
+impl Drop for UnavailableEndpoint {
+    fn drop(&mut self) {
+        if let Some(task) = self.task.take() {
+            task.abort();
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -139,14 +180,36 @@ mod tests {
         assert_ne!(a.port(), c.port());
     }
 
-    // Note: there is intentionally no "released port is immediately
-    // bindable" unit test. Once released, the port re-enters the
-    // kernel's free pool, and under heavy parallel test load (where
-    // dozens of `reserve_port` / `TcpListener::bind("127.0.0.1:0")`
-    // calls are racing each other) another test can be handed that
-    // port number before this one rebinds. That race is exactly what
-    // `reserve_port` exists to suppress for the held-reservation
-    // window; once released, it is by design out of scope. The
-    // "bindable after release" property is implicitly exercised by
-    // every passing `TestRelay::start*` integration test.
+    #[tokio::test]
+    async fn unavailable_endpoint_preserves_address_through_recovery() {
+        use tokio::io::AsyncReadExt;
+        let endpoint = UnavailableEndpoint::new();
+        let address = endpoint.listener.as_ref().unwrap().local_addr().unwrap();
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            let mut client = tokio::net::TcpStream::connect(address).await.unwrap();
+            let mut byte = [0];
+            match client.read(&mut byte).await {
+                Ok(0) => {}
+                Err(error) if error.kind() == std::io::ErrorKind::ConnectionReset => {}
+                result => panic!("unavailable endpoint should close clients: {result:?}"),
+            }
+        })
+        .await
+        .expect("unavailable endpoint should reject connections");
+
+        let listener = endpoint.into_listener().await;
+        assert_eq!(listener.local_addr().unwrap(), address);
+        assert_eq!(
+            TcpListener::bind(address).unwrap_err().kind(),
+            std::io::ErrorKind::AddrInUse
+        );
+        let listener = tokio::net::TcpListener::from_std(listener).unwrap();
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            let client = tokio::net::TcpStream::connect(address).await.unwrap();
+            let (_, peer) = listener.accept().await.unwrap();
+            assert_eq!(peer, client.local_addr().unwrap());
+        })
+        .await
+        .expect("recovered listener should own every new connection");
+    }
 }

tests/common/relay.rs

@@ -16,7 +16,7 @@ use nostr_sdk::prelude::{Keys, ToBech32};
 use std::path::PathBuf;
 use std::process::{Child, Command, Stdio};
 use std::time::{Duration, Instant};
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::io::AsyncWriteExt;
 use tokio::time::sleep;
 
 use crate::common::port::{self, PortReservation};
@@ -26,31 +26,18 @@ use crate::common::port::{self, PortReservation};
 const READY_TIMEOUT: Duration = Duration::from_secs(5);
 /// How often to retry the HTTP probe while waiting for readiness.
 const READY_POLL: Duration = Duration::from_millis(100);
-/// Extra grace after the HTTP service responds before declaring the relay
-/// ready.
-const READY_GRACE: Duration = Duration::from_millis(100);
 /// Per-attempt timeout for the HTTP readiness probe once TCP connects.
 const READY_PROBE_TIMEOUT: Duration = Duration::from_secs(1);
-/// How many fresh port reservations to attempt before giving up. The
-/// subprocess binds itself from `NGIT_BIND_ADDRESS`, so there is a
-/// microsecond-scale TOCTOU window between [`PortReservation::release`]
-/// and the subprocess's own `bind`. If that window loses the race the
-/// subprocess exits before its TCP listener accepts; the readiness
-/// check picks that up via `try_wait` so we can retry on a fresh port
-/// instead of hanging the full readiness timeout.
-///
-/// In practice this loop has never been observed to fire in local
-/// stress testing — kept as defense-in-depth for CI / loaded hardware.
-const MAX_BIND_ATTEMPTS: usize = 5;
 
 /// Test relay fixture that manages relay lifecycle
 ///
 /// Automatically starts and stops the ngit-grasp relay for testing.
-/// Uses a kernel-assigned port held open by a [`PortReservation`] until
-/// just before subprocess spawn, eliminating the same-process port race
-/// that plagued the older "bind, drop, return port" pattern.
+/// Transfers a kernel-assigned listener to the child while retaining a
+/// parent copy, so startup and same-address restart never release the port.
 pub struct TestRelay {
     process: Child,
+    /// Parent copy retains the address across a same-endpoint restart.
+    listener: std::net::TcpListener,
     url: String,
     port: u16,
     /// Relay-owner identity configured in the subprocess.
@@ -725,49 +712,21 @@ impl TestRelay {
 
     /// Single entry point that drives the spawn+readiness loop.
     ///
-    /// Retries up to [`MAX_BIND_ATTEMPTS`] times if the subprocess exits
-    /// early — that's the signature of having lost the bind race in the
-    /// microseconds between [`PortReservation::release`] and the
-    /// subprocess's own `bind`. Each retry draws a brand-new
-    /// kernel-assigned port; two consecutive `AddrInUse` failures would
-    /// therefore require two independent races back to back.
-    async fn start_internal(initial_reservation: PortReservation, options: RelayOptions) -> Self {
-        let mut reservation = Some(initial_reservation);
-        for attempt in 1..=MAX_BIND_ATTEMPTS {
-            // Each attempt consumes the current reservation. On retry we
-            // re-acquire from the kernel — guaranteed to give us a port
-            // number different from any reservation currently held
-            // elsewhere in this process.
-            let r = reservation
-                .take()
-                .expect("reservation always present on attempt entry");
-            match Self::try_start_once(r, &options).await {
-                StartOutcome::Ready(relay) => return relay,
-                StartOutcome::EarlyExit { status } if attempt < MAX_BIND_ATTEMPTS => {
-                    eprintln!(
-                        "[TestRelay] ngit-grasp exited early on attempt \
-                         {attempt}/{MAX_BIND_ATTEMPTS} (status: {status:?}); \
-                         likely a port-bind race — retrying with a fresh port",
-                    );
-                    reservation = Some(port::reserve_port());
-                    continue;
-                }
-                StartOutcome::EarlyExit { status } => {
-                    panic!(
-                        "ngit-grasp subprocess exited early after {MAX_BIND_ATTEMPTS} attempts \
-                         (last exit status: {status:?}). If this is not a port-bind race, \
-                         check /tmp/relay-*.log for the relay's stdout."
-                    );
-                }
-            }
+    /// Transfer the reserved listener to the child. Startup errors are real
+    /// failures; no port is released and no address retry is needed.
+    async fn start_internal(reservation: PortReservation, options: RelayOptions) -> Self {
+        match Self::try_start_once(reservation, &options).await {
+            StartOutcome::Ready(relay) => relay,
+            StartOutcome::EarlyExit { status } => panic!(
+                "ngit-grasp exited before readiness (status: {status:?}); see /tmp/relay-*.log"
+            ),
         }
-        unreachable!("MAX_BIND_ATTEMPTS loop terminated without returning")
     }
 
     /// One attempt at spawning ngit-grasp on the given reservation and
     /// waiting for it to be ready. Returns [`StartOutcome::EarlyExit`]
     /// specifically when the subprocess died before the readiness probe
-    /// succeeded — the caller may retry in that case.
+    /// succeeded.
     async fn try_start_once(reservation: PortReservation, options: &RelayOptions) -> StartOutcome {
         let port = reservation.port();
         let bind_address = format!("127.0.0.1:{}", port);
@@ -960,16 +919,31 @@ impl TestRelay {
             );
         }
 
-        // Release the port reservation immediately before spawning the
-        // subprocess that will bind it. Holding the reservation through
-        // env-var setup above is what keeps any concurrent
-        // `reserve_port()` calls from picking this same number.
-        let _ = reservation.release();
+        // Transfer the reservation through exec and retain a parent copy
+        // for same-address restarts. The port is never released during startup.
+        let listener = reservation.into_std_listener();
+        #[cfg(unix)]
+        {
+            use std::os::{fd::AsRawFd, unix::process::CommandExt};
+            let fd = listener.as_raw_fd();
+            cmd.env("NGIT_TEST_LISTENER_FD", fd.to_string());
+            // SAFETY: fcntl is async-signal-safe; the parent owns the descriptor
+            // through spawn. The child adopts it before starting the server.
+            unsafe {
+                cmd.pre_exec(move || {
+                    if libc::fcntl(fd, libc::F_SETFD, 0) < 0 {
+                        return Err(std::io::Error::last_os_error());
+                    }
+                    Ok(())
+                });
+            }
+        }
 
         let process = cmd.spawn().expect("Failed to start relay process");
 
         let mut relay = Self {
             process,
+            listener,
             url,
             port,
             owner_keys: test_keys.clone(),
@@ -1035,27 +1009,13 @@ impl TestRelay {
     pub async fn restart(mut self) -> Self {
         let port = self.port;
         let options = self.options.clone();
-
+        let reservation = port::PortReservation::from_listener(
+            self.listener.try_clone().expect("retain restart listener"),
+        );
         let _ = self.process.kill();
         let _ = self.process.wait();
         drop(self);
 
-        // The kernel frees the port once the child is fully gone; retry
-        // the specific-port reservation within a bounded deadline.
-        let deadline = Instant::now() + Duration::from_secs(10);
-        let reservation = loop {
-            match port::reserve_specific(port) {
-                Ok(reservation) => break reservation,
-                Err(error) => {
-                    assert!(
-                        Instant::now() < deadline,
-                        "port {port} not released by stopped relay within deadline: {error}"
-                    );
-                    sleep(Duration::from_millis(50)).await;
-                }
-            }
-        };
-
         match Self::try_start_once(reservation, &options).await {
             StartOutcome::Ready(relay) => relay,
             StartOutcome::EarlyExit { status } => panic!(
@@ -1102,7 +1062,6 @@ impl TestRelay {
                 Ok(()) => {
                     // HTTP service handled a request successfully, so the
                     // accept loop, Hyper service, and relay wiring are ready.
-                    sleep(READY_GRACE).await;
                     return ReadyOutcome::Ready;
                 }
                 Err(_) if Instant::now() < deadline => {
@@ -1131,8 +1090,10 @@ impl TestRelay {
             );
             stream.write_all(request.as_bytes()).await?;
 
-            let mut response = [0_u8; 64];
-            let read = stream.read(&mut response).await?;
+            let mut response = Vec::new();
+            let mut reader = tokio::io::BufReader::new(stream);
+            let read =
+                tokio::io::AsyncBufReadExt::read_until(&mut reader, b'\n', &mut response).await?;
             if read == 0 {
                 return Err(std::io::Error::new(
                     std::io::ErrorKind::UnexpectedEof,
@@ -1163,13 +1124,7 @@ impl TestRelay {
 
     /// Stop the relay
     pub async fn stop(mut self) {
-        // Kill the process (gracefully if possible)
-        let _ = self.process.kill();
-
-        // Wait a bit for graceful shutdown
-        sleep(Duration::from_millis(100)).await;
-
-        // Force kill if still running
+        // kill() sends SIGKILL; reap directly instead of guessing a grace period.
         let _ = self.process.kill();
         let _ = self.process.wait();
     }

tests/common/req_limiting_proxy.rs

@@ -83,6 +83,7 @@ impl ReqLimitingProxy {
         let accept_opened = opened.clone();
 
         let handle = tokio::spawn(async move {
+            let mut connections = tokio::task::JoinSet::new();
             loop {
                 tokio::select! {
                     accepted = listener.accept() => {
@@ -91,7 +92,7 @@ impl ReqLimitingProxy {
                         let peak = accept_peak.clone();
                         let rejected = accept_rejected.clone();
                         let opened = accept_opened.clone();
-                        tokio::spawn(async move {
+                        connections.spawn(async move {
                             if let Err(error) = proxy_connection(
                                 stream,
                                 &backend_url,
@@ -107,9 +108,13 @@ impl ReqLimitingProxy {
                             }
                         });
                     }
+                    result = connections.join_next(), if !connections.is_empty() => {
+                        result.expect("connection task").expect("fixture connection panicked");
+                    }
                     _ = &mut shutdown_rx => break,
                 }
             }
+            connections.shutdown().await;
         });
 
         Self {
@@ -156,6 +161,9 @@ impl ReqLimitingProxy {
 
 impl Drop for ReqLimitingProxy {
     fn drop(&mut self) {
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+        }
         if let Some(tx) = self.shutdown_tx.take() {
             let _ = tx.send(());
         }

tests/common/setup_drop_relay.rs

@@ -34,6 +34,7 @@ impl SetupDropRelay {
         let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
 
         let handle = tokio::spawn(async move {
+            let mut connections = tokio::task::JoinSet::new();
             loop {
                 tokio::select! {
                     accepted = listener.accept() => {
@@ -43,7 +44,7 @@ impl SetupDropRelay {
                         let dropped_tx = dropped_tx.clone();
                         let dropped_rx = dropped_rx.clone();
                         let active_websockets = active_websockets.clone();
-                        tokio::spawn(async move {
+                        connections.spawn(async move {
                             handle_connection(
                                 stream,
                                 nip11_tx,
@@ -55,9 +56,13 @@ impl SetupDropRelay {
                             .await;
                         });
                     }
+                    result = connections.join_next(), if !connections.is_empty() => {
+                        result.expect("connection task").expect("fixture connection panicked");
+                    }
                     _ = &mut shutdown_rx => break,
                 }
             }
+            connections.shutdown().await;
         });
 
         Self {
@@ -142,6 +147,9 @@ async fn handle_connection(
 
 impl Drop for SetupDropRelay {
     fn drop(&mut self) {
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+        }
         if let Some(tx) = self.shutdown_tx.take() {
             let _ = tx.send(());
         }

tests/common/sync_helpers.rs

@@ -15,7 +15,7 @@ use std::time::Duration;
 
 use nostr_sdk::prelude::*;
 
-use super::port::{self, PortReservation};
+use super::port::{self, PortReservation, UnavailableEndpoint};
 use super::relay::TestRelay;
 
 const DESCENDANT_LIVE_LOG: &str = "Installed priority-bounded auxiliary live coverage";
@@ -476,80 +476,26 @@ pub async fn wait_for_sync_connection(
     expected_connections: usize,
     timeout: Duration,
 ) -> Result<(), String> {
-    // Convert ws:// URL to http:// for metrics endpoint
-    let http_url = syncing_relay_url
-        .replace("ws://", "http://")
-        .replace("/", "")
-        + "/metrics";
-
-    let start = std::time::Instant::now();
-    let poll_interval = Duration::from_millis(100);
-
-    while start.elapsed() < timeout {
-        // Fetch metrics
-        if let Ok(response) = reqwest::get(&http_url).await {
-            if let Ok(metrics) = response.text().await {
-                // Look for sync connection metrics
-                // The metric name pattern: ngit_sync_connections or similar
-                // We check for any indication of established connections
+    let wait = async {
+        loop {
+            if let Ok(metrics) = fetch_metrics(syncing_relay_url).await {
                 if check_sync_connections_in_metrics(&metrics, expected_connections) {
-                    return Ok(());
+                    return;
                 }
             }
+            tokio::time::sleep(Duration::from_millis(100)).await;
         }
-
-        tokio::time::sleep(poll_interval).await;
-    }
-
-    Err(format!(
-        "Timeout waiting for {} sync connection(s) on {} after {:?}",
-        expected_connections, syncing_relay_url, timeout
+    };
+    tokio::time::timeout(timeout, wait).await.map_err(|_| format!(
+        "Timeout waiting for {expected_connections} sync connection(s) on {syncing_relay_url} after {timeout:?}"
     ))
 }
 
-/// Check metrics string for expected number of sync connections.
-///
-/// Looks for various metric patterns that indicate sync connections:
-/// - ngit_sync_connections (gauge)
-/// - ngit_sync_relay_connections (gauge)
-/// - Any metric containing "sync" and "connection" with count > 0
+/// Connection attempts and health states do not establish a live connection.
 fn check_sync_connections_in_metrics(metrics: &str, expected: usize) -> bool {
-    // Parse metrics line by line looking for connection counts
-    for line in metrics.lines() {
-        // Skip comments and empty lines
-        if line.starts_with('#') || line.is_empty() {
-            continue;
-        }
-
-        // Look for sync connection metrics
-        // Format: metric_name{labels} value
-        // or: metric_name value
-        if line.contains("sync") && line.contains("connect") {
-            // Extract the value (last space-separated token)
-            if let Some(value_str) = line.split_whitespace().last() {
-                if let Ok(value) = value_str.parse::<f64>() {
-                    if value as usize >= expected {
-                        return true;
-                    }
-                }
-            }
-        }
-
-        // Also check for specific metric names that might indicate connections
-        // ngit_sync_health_state with value 1 or 2 (connecting/healthy)
-        if line.contains("ngit_sync_health") {
-            if let Some(value_str) = line.split_whitespace().last() {
-                if let Ok(value) = value_str.parse::<f64>() {
-                    // Health state > 0 typically means connection attempt or established
-                    if value > 0.0 && expected > 0 {
-                        return true;
-                    }
-                }
-            }
-        }
-    }
-
-    false
+    ParsedMetrics::parse(metrics)
+        .relays_connected_total()
+        .is_some_and(|connected| connected >= expected as i64)
 }
 
 // ============================================================================
@@ -677,10 +623,14 @@ pub fn repo_coord(keys: &Keys, identifier: &str) -> String {
 /// assert!(metrics.contains("ngit_sync_"));
 /// ```
 pub async fn fetch_metrics(relay_url: &str) -> Result<String, reqwest::Error> {
-    // Convert ws:// URL to http:// for metrics endpoint
-    let http_url = relay_url.replace("ws://", "http://").replace("/", "") + "/metrics";
+    reqwest::get(metrics_url(relay_url)).await?.text().await
+}
 
-    reqwest::get(&http_url).await?.text().await
+fn metrics_url(relay_url: &str) -> String {
+    let http_url = relay_url
+        .replacen("wss://", "https://", 1)
+        .replacen("ws://", "http://", 1);
+    format!("{}/metrics", http_url.trim_end_matches('/'))
 }
 
 // ============================================================================
@@ -837,10 +787,9 @@ impl ParsedMetrics {
 /// harness.stop_all().await;
 /// ```
 pub struct MetricsTestHarness {
-    source_relays: Vec<TestRelay>,
+    source_relays: Vec<Option<TestRelay>>,
     syncing_relay: Option<TestRelay>,
-    #[allow(dead_code)]
-    nowhere_url: Option<String>,
+    unavailable_endpoint: Option<UnavailableEndpoint>,
 }
 
 impl MetricsTestHarness {
@@ -848,34 +797,36 @@ impl MetricsTestHarness {
     pub async fn with_sources(count: usize) -> Self {
         let mut source_relays = Vec::new();
         for _ in 0..count {
-            source_relays.push(TestRelay::start().await);
+            source_relays.push(Some(TestRelay::start().await));
         }
 
         Self {
             source_relays,
             syncing_relay: None,
-            nowhere_url: None,
+            unavailable_endpoint: None,
         }
     }
 
     /// Get source relay URL
     pub fn source_url(&self, idx: usize) -> &str {
-        self.source_relays[idx].url()
+        self.source_relay(idx).url()
     }
 
     /// Get source relay domain (for announcement tags)
     pub fn source_domain(&self, idx: usize) -> String {
-        self.source_relays[idx].domain()
+        self.source_relay(idx).domain()
     }
 
     /// Get a reference to a source relay (for advanced test operations)
     pub fn source_relay(&self, idx: usize) -> &TestRelay {
-        &self.source_relays[idx]
+        self.source_relays[idx]
+            .as_ref()
+            .expect("source relay has been stopped")
     }
 
     /// Submit events to a specific source relay
     pub async fn submit_events(&self, source_idx: usize, events: &[Event]) -> Result<(), String> {
-        let relay = &self.source_relays[source_idx];
+        let relay = self.source_relay(source_idx);
         let keys = Keys::generate();
         let client = TestClient::new(relay.url(), keys).await?;
 
@@ -889,7 +840,7 @@ impl MetricsTestHarness {
 
     /// Start syncing relay pointing to source[idx]
     pub async fn start_syncing_relay(&mut self, source_idx: usize) {
-        let source_url = self.source_relays[source_idx].url().to_string();
+        let source_url = self.source_relay(source_idx).url().to_string();
         self.syncing_relay = Some(TestRelay::start_with_sync(Some(source_url)).await);
     }
 
@@ -904,37 +855,26 @@ impl MetricsTestHarness {
         source_idx: usize,
         reservation: PortReservation,
     ) {
-        let source_url = self.source_relays[source_idx].url().to_string();
+        let source_url = self.source_relay(source_idx).url().to_string();
         self.syncing_relay = Some(
             TestRelay::start_on_reservation_with_options(reservation, Some(source_url), false)
                 .await,
         );
     }
 
-    /// Start syncing relay pointing to random unused port (for failure tests)
+    /// Start syncing against an owned endpoint that closes every connection.
     pub async fn start_syncing_relay_to_nowhere(&mut self) {
-        let port = random_unused_port();
-        let nowhere_url = format!("ws://127.0.0.1:{}", port);
-        self.nowhere_url = Some(nowhere_url.clone());
+        let endpoint = UnavailableEndpoint::new();
+        let nowhere_url = format!("ws://127.0.0.1:{}", endpoint.port());
+        self.unavailable_endpoint = Some(endpoint);
         self.syncing_relay = Some(TestRelay::start_with_sync(Some(nowhere_url)).await);
     }
 
     /// Stop a source relay
     pub async fn stop_source(&mut self, source_idx: usize) {
-        // We need to take ownership to stop, so we swap with a new relay
-        // that we immediately stop. This is a workaround since TestRelay::stop
-        // takes self by value.
-        let relay = std::mem::replace(
-            &mut self.source_relays[source_idx],
-            TestRelay::start().await,
-        );
-        relay.stop().await;
-        // Stop the placeholder too
-        let placeholder = std::mem::replace(
-            &mut self.source_relays[source_idx],
-            TestRelay::start().await,
-        );
-        placeholder.stop().await;
+        if let Some(relay) = self.source_relays[source_idx].take() {
+            relay.stop().await;
+        }
     }
 
     /// Fetch and parse metrics from syncing relay
@@ -961,29 +901,41 @@ impl MetricsTestHarness {
         if let Some(relay) = self.syncing_relay.take() {
             relay.stop().await;
         }
-        for relay in self.source_relays.drain(..) {
+        for relay in self.source_relays.drain(..).flatten() {
             relay.stop().await;
         }
     }
 }
 
-// ============================================================================
-// Port Helpers
-// ============================================================================
-
-/// Get a random unused port by binding to port 0 and letting the OS assign one
-pub fn random_unused_port() -> u16 {
-    std::net::TcpListener::bind("127.0.0.1:0")
-        .expect("Failed to bind to random port")
-        .local_addr()
-        .expect("Failed to get local addr")
-        .port()
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[test]
+    fn metrics_url_preserves_scheme_and_base_path() {
+        assert_eq!(
+            metrics_url("ws://127.0.0.1:1234/"),
+            "http://127.0.0.1:1234/metrics"
+        );
+        assert_eq!(
+            metrics_url("wss://example.test/relay/"),
+            "https://example.test/relay/metrics"
+        );
+    }
+
+    #[test]
+    fn connection_readiness_ignores_attempts_and_health() {
+        assert!(!check_sync_connections_in_metrics("ngit_sync_connection_attempts_total 5\nngit_sync_health 1\nngit_sync_relays_connected_total 0", 1));
+        assert!(check_sync_connections_in_metrics(
+            "ngit_sync_relays_connected_total 2",
+            2
+        ));
+        assert!(!check_sync_connections_in_metrics(
+            "ngit_sync_relays_connected_total 1",
+            2
+        ));
+    }
+
     #[test]
     fn test_repo_coord_format() {
         let keys = Keys::generate();
@@ -1268,8 +1220,15 @@ pub async fn push_git_data_to_relay(
     push_to_relay(git_temp_dir.path(), &relay.domain(), &npub, identifier)
         .expect("Failed to push git data to relay");
 
-    // Brief wait for push processing
-    tokio::time::sleep(Duration::from_millis(500)).await;
+    assert!(
+        wait_for_event_on_relay(
+            relay.url(),
+            Filter::new().id(state_event.id),
+            Duration::from_secs(10)
+        )
+        .await,
+        "pushed state event must leave purgatory"
+    );
 
     git_temp_dir
 }
@@ -1379,7 +1338,15 @@ pub async fn push_unique_git_data_to_relay(
     push_to_relay(path, &relay.domain(), &npub, identifier)
         .expect("Failed to push git data to relay");
 
-    tokio::time::sleep(Duration::from_millis(500)).await;
+    assert!(
+        wait_for_event_on_relay(
+            relay.url(),
+            Filter::new().id(state_event.id),
+            Duration::from_secs(10)
+        )
+        .await,
+        "pushed state event must leave purgatory"
+    );
 
     git_temp_dir
 }
@@ -1464,8 +1431,25 @@ pub async fn setup_announcement_on_relay(
     push_to_relay(git_temp_dir.path(), &relay.domain(), &npub, identifier)
         .expect("Failed to push git data to relay");
 
-    // Brief wait for push processing
-    tokio::time::sleep(Duration::from_millis(500)).await;
+    assert!(
+        wait_for_event_on_relay(
+            relay.url(),
+            Filter::new().id(state_event.id),
+            Duration::from_secs(10)
+        )
+        .await,
+        "pushed state event must leave purgatory"
+    );
+
+    assert!(
+        wait_for_event_on_relay(
+            relay.url(),
+            Filter::new().id(announcement.id),
+            Duration::from_secs(10)
+        )
+        .await,
+        "pushed announcement must leave purgatory"
+    );
 
     (announcement, git_temp_dir)
 }
@@ -1587,7 +1571,15 @@ pub async fn run_sync_test(historic_events: &[Event], live_events: &[Event]) ->
         .expect("Failed to push git data to source relay");
 
     // 8. Wait for source relay to process the push and release events from purgatory
-    tokio::time::sleep(Duration::from_secs(2)).await;
+    assert!(
+        wait_for_event_on_relay(
+            source.url(),
+            Filter::new().id(announcement.id),
+            Duration::from_secs(10)
+        )
+        .await,
+        "source announcement must leave purgatory"
+    );
 
     // 9. Send historic events to source BEFORE syncing relay starts
     for event in historic_events {
@@ -1605,7 +1597,9 @@ pub async fn run_sync_test(historic_events: &[Event], live_events: &[Event]) ->
     .await;
 
     // 11. Wait for sync connection to establish
-    let _ = wait_for_sync_connection(syncing.url(), 1, Duration::from_secs(5)).await;
+    wait_for_sync_connection(syncing.url(), 1, Duration::from_secs(10))
+        .await
+        .expect("sync connection must establish before publishing live events");
 
     // 12. Send live events AFTER connection established
     for event in live_events {
@@ -1614,12 +1608,16 @@ pub async fn run_sync_test(historic_events: &[Event], live_events: &[Event]) ->
             .expect("Failed to send live event");
     }
 
-    // 13. Allow sync + purgatory promotion to complete on the syncing relay.
-    // The syncing relay receives the announcement (goes to purgatory) and state event.
-    // The purgatory sync loop (1s interval) fetches git data from source's clone URL
-    // (http://source-domain/npub/test-repo.git) and releases the announcement.
-    // We wait up to 8s to allow time for this.
-    tokio::time::sleep(Duration::from_secs(8)).await;
+    // 13. Observe announcement promotion rather than guessing a sync duration.
+    assert!(
+        wait_for_event_on_relay(
+            syncing.url(),
+            Filter::new().id(announcement.id),
+            Duration::from_secs(15)
+        )
+        .await,
+        "synced announcement must leave purgatory"
+    );
 
     // 14. Compute repo coordinate before moving keys
     let coordinate = repo_coord(&keys, "test-repo");

tests/common/upload_pack_counting_proxy.rs

@@ -31,6 +31,7 @@ use hyper::{Request, Response, StatusCode};
 use hyper_util::rt::TokioIo;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
+use tokio::task::JoinSet;
 
 /// One observed `POST /git-upload-pack` exchange.
 #[derive(Debug, Clone)]
@@ -77,6 +78,7 @@ impl UploadPackCountingProxy {
         let accept_info_refs = info_refs.clone();
 
         let handle = tokio::spawn(async move {
+            let mut connections = JoinSet::new();
             // One shared upstream client; keep it plain so request and
             // response bodies pass through byte-for-byte.
             let client = reqwest::Client::builder()
@@ -92,7 +94,7 @@ impl UploadPackCountingProxy {
                         let client = client.clone();
                         let exchanges = accept_exchanges.clone();
                         let info_refs = accept_info_refs.clone();
-                        tokio::spawn(async move {
+                        connections.spawn(async move {
                             let io = TokioIo::new(stream);
                             let service = service_fn(move |req| {
                                 let backend_url = backend_url.clone();
@@ -118,8 +120,10 @@ impl UploadPackCountingProxy {
                         });
                     }
                     _ = &mut shutdown_rx => break,
+                    _ = connections.join_next(), if !connections.is_empty() => {}
                 }
             }
+            connections.shutdown().await;
         });
 
         Self {
@@ -155,7 +159,10 @@ impl UploadPackCountingProxy {
             let _ = tx.send(());
         }
         if let Some(handle) = self.handle.take() {
-            let _ = handle.await;
+            tokio::time::timeout(std::time::Duration::from_secs(5), handle)
+                .await
+                .expect("upload-pack proxy tasks should stop")
+                .expect("upload-pack proxy server task should not panic");
         }
     }
 }
@@ -291,6 +298,41 @@ fn contains(haystack: &[u8], needle: &[u8]) -> bool {
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn stop_closes_active_forwarded_requests() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            let backend = TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let proxy = UploadPackCountingProxy::start(&format!(
+                "http://{}",
+                backend.local_addr().unwrap()
+            ))
+            .await;
+            let address = proxy.url().strip_prefix("http://").unwrap();
+            let mut client = tokio::net::TcpStream::connect(address).await.unwrap();
+            client
+                .write_all(
+                    b"GET /info/refs?service=git-upload-pack HTTP/1.1\r\nHost: localhost\r\n\r\n",
+                )
+                .await
+                .unwrap();
+            let (mut upstream, _) = backend.accept().await.unwrap();
+            let mut request = [0; 256];
+            assert!(upstream.read(&mut request).await.unwrap() > 0);
+            // The request has reached the backend, which deliberately retains
+            // the connection without replying while the proxy is stopped.
+            proxy.stop().await;
+            let mut byte = [0];
+            match client.read(&mut byte).await {
+                Ok(0) => {}
+                Err(error) if error.kind() == std::io::ErrorKind::ConnectionReset => {}
+                result => panic!("proxy retained an active request after stop: {result:?}"),
+            }
+        })
+        .await
+        .expect("proxy stop must close active requests");
+    }
+
     #[test]
     fn parse_wants_extracts_oids_from_pkt_lines() {
         let body = b"0032want aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n\

tests/fixture_lifecycle.rs

@@ -0,0 +1,58 @@
+//! Accepted connections must not outlive the fixture that owns them.
+mod common;
+
+use std::time::Duration;
+
+use common::{
+    censoring_proxy::CensoringProxy, flapping_relay::FlappingRelay,
+    neg_limiting_proxy::NegLimitingProxy, req_limiting_proxy::ReqLimitingProxy,
+    setup_drop_relay::SetupDropRelay, MockRelay,
+};
+use futures_util::StreamExt;
+
+macro_rules! check_shutdown {
+    ($fixture:expr) => {{
+        tokio::time::timeout(Duration::from_secs(10), async {
+            let fixture = $fixture;
+            let (mut connection, _) = tokio_tungstenite::connect_async(fixture.url())
+                .await
+                .expect("fixture must finish the WebSocket handshake");
+            fixture.stop().await;
+            while let Some(message) = connection.next().await {
+                if message.is_err() || message.unwrap().is_close() {
+                    break;
+                }
+            }
+        })
+        .await
+        .expect("fixture stop must close accepted connections");
+    }};
+}
+
+#[tokio::test]
+async fn censoring_proxy_closes_connections_on_stop() {
+    let backend = MockRelay::start().await;
+    check_shutdown!(CensoringProxy::start(backend.url()).await);
+}
+
+#[tokio::test]
+async fn req_limiting_proxy_closes_connections_on_stop() {
+    let backend = MockRelay::start().await;
+    check_shutdown!(ReqLimitingProxy::start(backend.url(), 2).await);
+}
+
+#[tokio::test]
+async fn neg_limiting_proxy_closes_connections_on_stop() {
+    let backend = MockRelay::start().await;
+    check_shutdown!(NegLimitingProxy::start(backend.url(), 2).await);
+}
+
+#[tokio::test]
+async fn flapping_relay_closes_connections_on_stop() {
+    check_shutdown!(FlappingRelay::start().await);
+}
+
+#[tokio::test]
+async fn setup_drop_relay_closes_connections_on_stop() {
+    check_shutdown!(SetupDropRelay::start().await);
+}

tests/git_response_streaming.rs

@@ -8,7 +8,7 @@ use std::time::Duration;
 use async_trait::async_trait;
 use clap::Parser;
 use http_body_util::BodyExt;
-use hyper::body::{Bytes, Frame};
+use hyper::body::{Body, Bytes, Frame};
 use ngit_grasp::config::Config;
 use ngit_grasp::git::handlers::handle_receive_pack;
 use ngit_grasp::git::sync::PurgatoryPromotionHooks;
@@ -24,6 +24,8 @@ use ngit_grasp::purgatory::Purgatory;
 use ngit_grasp::sync::rejected_index::RejectedEventsIndex;
 use nostr_sdk::prelude::LocalRelayBuilder;
 use nostr_sdk::prelude::*;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpListener;
 use tokio::sync::Semaphore;
 use tokio::time::timeout;
 
@@ -61,7 +63,8 @@ async fn receive_pack_response_streams_stdout_before_subprocess_exit() {
     let _env_lock = PATH_ENV_LOCK.lock().await;
 
     let fake_bin = tempfile::tempdir().expect("fake git bin tempdir");
-    write_fake_git(fake_bin.path());
+    let git_gate = GitGate::new().await;
+    write_fake_git_script(fake_bin.path(), false, git_gate.port());
     let _path = PathOverride::prepend(fake_bin.path());
 
     let repo = tempfile::tempdir().expect("repo tempdir");
@@ -91,39 +94,10 @@ async fn receive_pack_response_streams_stdout_before_subprocess_exit() {
 
     let mut body = response.into_body();
 
-    let first = timeout(Duration::from_secs(1), body.frame())
-        .await
-        .expect("first stdout chunk should arrive before fake git exits")
-        .expect("body should still be open")
-        .expect("first frame should not be an HTTP body error");
-    let first = frame_data(first);
-    assert!(
-        !first.is_empty(),
-        "first progress frame should contain data"
-    );
-
-    let no_second_yet = timeout(Duration::from_millis(250), body.frame()).await;
-    assert!(
-        no_second_yet.is_err(),
-        "body produced another frame while fake git was still sleeping; \
-         this test needs the first frame to be observed before subprocess EOF"
-    );
-
-    let mut streamed = first.to_vec();
-    loop {
-        let frame = timeout(Duration::from_secs(3), body.frame())
-            .await
-            .expect("remaining stdout should arrive after fake git wakes");
-        let Some(frame) = frame else {
-            break;
-        };
-        streamed.extend_from_slice(
-            &frame
-                .expect("remaining frame should not be an HTTP body error")
-                .into_data()
-                .expect("remaining frame should contain data"),
-        );
-    }
+    let mut streamed = read_first_progress(&mut body).await;
+    // Git cannot exit until this test releases it, regardless of scheduling.
+    git_gate.release().await;
+    finish_body(&mut body, &mut streamed).await;
     assert_eq!(streamed, b"first-progress\nsecond-progress\n");
 }
 
@@ -150,7 +124,8 @@ async fn receive_pack_terminal_flush_waits_for_purgatory_promotion() {
     let _env_lock = PATH_ENV_LOCK.lock().await;
 
     let fake_bin = tempfile::tempdir().expect("fake git bin tempdir");
-    write_fake_git_with_terminal_flush(fake_bin.path());
+    let git_gate = GitGate::new().await;
+    write_fake_git_script(fake_bin.path(), true, git_gate.port());
     let _path = PathOverride::prepend(fake_bin.path());
 
     let keys = Keys::generate();
@@ -203,12 +178,8 @@ async fn receive_pack_terminal_flush_waits_for_purgatory_promotion() {
     .expect("receive-pack handler should start fake subprocess");
 
     let mut body = response.into_body();
-    let first = timeout(Duration::from_secs(1), body.frame())
-        .await
-        .expect("receive-pack progress should stream before promotion")
-        .expect("body should contain progress")
-        .expect("progress frame should not be an HTTP body error");
-    let mut streamed = frame_data(first).to_vec();
+    let mut streamed = read_first_progress(&mut body).await;
+    git_gate.release().await;
 
     timeout(Duration::from_secs(3), entered.acquire())
         .await
@@ -225,56 +196,31 @@ async fn receive_pack_terminal_flush_waits_for_purgatory_promotion() {
         "announcement must not be queryable while promotion is blocked"
     );
 
-    while let Ok(Some(frame)) = timeout(Duration::from_millis(25), body.frame()).await {
-        streamed.extend_from_slice(
-            &frame
-                .expect("progress frame should not be an HTTP body error")
-                .into_data()
-                .expect("progress frame should contain data"),
-        );
-    }
-    assert!(
-        !streamed.ends_with(b"0000"),
-        "receive-pack terminal flush must remain hidden while promotion is blocked"
-    );
-
-    let keepalive = timeout(Duration::from_secs(6), body.frame())
-        .await
-        .expect("sideband keepalive should arrive during blocked promotion")
-        .expect("body should remain open while promotion is blocked")
-        .expect("keepalive should not be an HTTP body error");
-    streamed.extend_from_slice(
-        &keepalive
-            .into_data()
-            .expect("keepalive frame should contain data"),
-    );
-    assert!(
-        streamed
+    // Accumulate across arbitrary frame boundaries until the finalization
+    // keepalive is visible. Promotion remains blocked throughout this read.
+    timeout(Duration::from_secs(6), async {
+        while !streamed
             .windows(b"GRASP is finalizing the push\n".len())
-            .any(|window| window == b"GRASP is finalizing the push\n"),
-        "blocked post-push processing should emit sideband progress"
-    );
-    assert!(
-        !streamed.ends_with(b"0000"),
-        "keepalive must not expose the terminal flush"
-    );
+            .any(|window| window == b"GRASP is finalizing the push\n")
+        {
+            let frame = body
+                .frame()
+                .await
+                .expect("body should remain open while promotion is blocked")
+                .expect("progress frame should not be an HTTP body error");
+            streamed.extend_from_slice(&frame_data(frame));
+            assert!(
+                !streamed.ends_with(b"0000"),
+                "terminal flush must remain hidden while promotion is blocked"
+            );
+        }
+    })
+    .await
+    .expect("sideband keepalive should arrive during blocked promotion");
 
     release.add_permits(1);
 
-    loop {
-        let frame = timeout(Duration::from_secs(1), body.frame())
-            .await
-            .expect("response should finish after promotion is released");
-        let Some(frame) = frame else {
-            break;
-        };
-        streamed.extend_from_slice(
-            &frame
-                .expect("terminal frame should not be an HTTP body error")
-                .into_data()
-                .expect("terminal frame should contain data"),
-        );
-    }
+    finish_body(&mut body, &mut streamed).await;
 
     assert!(
         streamed.starts_with(b"first-progress\nsecond-progress\n"),
@@ -300,7 +246,8 @@ async fn prs_receive_pack_streams_stdout_before_cleanup_removes_empty_repo() {
     let _env_lock = PATH_ENV_LOCK.lock().await;
 
     let fake_bin = tempfile::tempdir().expect("fake git bin tempdir");
-    write_fake_git_with_terminal_flush(fake_bin.path());
+    let git_gate = GitGate::new().await;
+    write_fake_git_script(fake_bin.path(), true, git_gate.port());
     let _path = PathOverride::prepend(fake_bin.path());
 
     let keys = Keys::generate();
@@ -350,61 +297,92 @@ async fn prs_receive_pack_streams_stdout_before_cleanup_removes_empty_repo() {
 
     let mut body = response.into_body();
 
-    let first = timeout(Duration::from_secs(1), body.frame())
-        .await
-        .expect("first /prs/ stdout chunk should arrive before fake git exits")
-        .expect("/prs/ body should still be open")
-        .expect("first /prs/ frame should not be an HTTP body error");
-    let mut streamed = frame_data(first).to_vec();
-    assert!(
-        b"first-progress\n".starts_with(&streamed),
-        "the four-byte terminal look-behind may split the first progress chunk"
-    );
-
+    let mut streamed = read_first_progress(&mut body).await;
     assert!(
         repo_path.exists(),
         "/prs/ cleanup must not remove the repo before receive-pack exits"
     );
-
-    let no_second_yet = timeout(Duration::from_millis(250), body.frame()).await;
-    assert!(
-        no_second_yet.is_err(),
-        "/prs/ body produced another frame while fake git was still sleeping; \
-         this test needs the first frame to be observed before subprocess EOF"
-    );
-
-    let second = timeout(Duration::from_secs(3), body.frame())
-        .await
-        .expect("second /prs/ stdout chunk should arrive after fake git wakes")
-        .expect("/prs/ body should still be open for second chunk")
-        .expect("second /prs/ frame should not be an HTTP body error");
-    streamed.extend_from_slice(&frame_data(second));
-
-    // Family ref retention happens behind this boundary. Keep the deadline
-    // bounded but allow finalization the same scheduling headroom as the
-    // subprocess wake above.
-    let terminal = timeout(Duration::from_secs(3), body.frame())
-        .await
-        .expect("/prs/ terminal flush should arrive after cleanup")
-        .expect("/prs/ body should contain the terminal flush")
-        .expect("/prs/ terminal frame should not be an HTTP body error");
-    streamed.extend_from_slice(&frame_data(terminal));
+    git_gate.release().await;
+    finish_body(&mut body, &mut streamed).await;
     assert_eq!(streamed, b"first-progress\nsecond-progress\n0000");
 
-    let eof = timeout(Duration::from_secs(1), body.frame())
-        .await
-        .expect("/prs/ body should close after cleanup");
-    assert!(
-        eof.is_none(),
-        "/prs/ body should be closed after fake git exits"
-    );
-
     assert!(
         !repo_path.exists(),
         "/prs/ cleanup should remove zero-ref repo after receive-pack exits"
     );
 }
 
+// Keep the listener bound until Git has observed the parent's release.
+struct GitGate(TcpListener);
+
+impl GitGate {
+    async fn new() -> Self {
+        Self(
+            TcpListener::bind("127.0.0.1:0")
+                .await
+                .expect("bind fake git gate"),
+        )
+    }
+
+    fn port(&self) -> u16 {
+        self.0.local_addr().expect("fake git gate address").port()
+    }
+
+    async fn release(self) {
+        timeout(Duration::from_secs(3), async {
+            let (mut stream, _) = self.0.accept().await.expect("accept fake git gate");
+            stream
+                .write_all(b"release\n")
+                .await
+                .expect("release fake git");
+        })
+        .await
+        .expect("fake git should connect to its gate");
+    }
+}
+
+async fn read_first_progress<B>(body: &mut B) -> Vec<u8>
+where
+    B: Body<Data = Bytes> + Unpin,
+    B::Error: std::fmt::Debug,
+{
+    const FIRST: &[u8] = b"first-progress\n";
+    timeout(Duration::from_secs(3), async {
+        let mut streamed = Vec::new();
+        // Receive-pack can retain the last four bytes until further stdout
+        // arrives, and pipe reads need not match writes or HTTP frames.
+        while streamed.len() < FIRST.len() - 4 {
+            let frame = body
+                .frame()
+                .await
+                .expect("progress before fake git exit")
+                .expect("progress body frame");
+            streamed.extend_from_slice(&frame_data(frame));
+            assert!(
+                FIRST.starts_with(&streamed),
+                "unexpected initial git progress"
+            );
+        }
+        streamed
+    })
+    .await
+    .expect("first progress should stream before git is released")
+}
+
+async fn finish_body<B>(body: &mut B, streamed: &mut Vec<u8>)
+where
+    B: Body<Data = Bytes> + Unpin,
+    B::Error: std::fmt::Debug,
+{
+    timeout(Duration::from_secs(3), async {
+        while let Some(frame) = body.frame().await {
+            streamed.extend_from_slice(&frame_data(frame.expect("remaining body frame")));
+        }
+    })
+    .await
+    .expect("response should finish after fixture release");
+}
+
 fn frame_data(frame: Frame<Bytes>) -> Bytes {
     frame.into_data().expect("frame should contain data")
 }
@@ -452,15 +430,7 @@ fn test_write_policy(
     )
 }
 
-fn write_fake_git(bin_dir: &Path) {
-    write_fake_git_script(bin_dir, false);
-}
-
-fn write_fake_git_with_terminal_flush(bin_dir: &Path) {
-    write_fake_git_script(bin_dir, true);
-}
-
-fn write_fake_git_script(bin_dir: &Path, terminal_flush: bool) {
+fn write_fake_git_script(bin_dir: &Path, terminal_flush: bool, gate_port: u16) {
     let git_path = bin_dir.join("git");
     let terminal_flush = if terminal_flush {
         "printf '0000'\n"
@@ -481,8 +451,11 @@ done
 
 if [ "$is_receive_pack" = "1" ]; then
 cat >/dev/null
+exec 3<>/dev/tcp/127.0.0.1/__GATE_PORT__
 printf 'first-progress\n'
-sleep 2
+IFS= read -r -t 10 release <&3
+[ "$release" = release ]
+exec 3<&-
 printf 'second-progress\n'
 __TERMINAL_FLUSH__exit 0
 fi
@@ -513,7 +486,8 @@ fi
 echo "unsupported fake git invocation: $*" >&2
 exit 1
 "#
-        .replace("__TERMINAL_FLUSH__", terminal_flush),
+        .replace("__TERMINAL_FLUSH__", terminal_flush)
+        .replace("__GATE_PORT__", &gate_port.to_string()),
     )
     .expect("write fake git executable");
 

tests/relay_identity.rs

@@ -5,6 +5,7 @@ mod common;
 use std::collections::BTreeSet;
 use std::time::Duration;
 
+use common::port::UnavailableEndpoint;
 use common::{reserve_port, wait_for_event_on_relay, MockRelay, TestClient, TestRelay};
 use nostr::nips::nip65;
 use nostr_sdk::prelude::*;
@@ -244,9 +245,9 @@ async fn wiped_relay_adopts_identity_from_user_index_instead_of_publishing() {
 
 #[tokio::test]
 async fn identity_publication_defers_until_a_user_index_relay_is_reachable() {
-    // Release the reservation so connection attempts fail fast with refused;
-    // the MockRelay rebinds the same port later in the test.
-    let port = reserve_port().release();
+    // Reject protocol connections while retaining the recovery address.
+    let unavailable = UnavailableEndpoint::new();
+    let port = unavailable.port();
     let index_url = format!("ws://127.0.0.1:{port}");
 
     let relay = TestRelay::start_with_sync(Some(index_url)).await;
@@ -272,7 +273,7 @@ async fn identity_publication_defers_until_a_user_index_relay_is_reachable() {
 
     // Once an empty index becomes reachable, the generated identity is
     // released: seeded locally and published to the index.
-    let index = MockRelay::start_on_port(port).await;
+    let index = MockRelay::start_on_listener(unavailable.into_listener().await, Vec::new()).await;
     for url in [relay.url(), index.url()] {
         for kind in [Kind::Metadata, Kind::RelayList] {
             assert!(
@@ -297,9 +298,15 @@ async fn stored_identity_is_not_pushed_to_recovering_index_holding_an_identity()
     // Index A stays unreachable until it "recovers" already holding the
     // operator's customized profile; index B is reachable so the phase-1
     // index check can succeed without A.
-    let port_a = reserve_port().release();
-    let port_b = reserve_port().release();
-    let index_b = MockRelay::start_on_port(port_b).await;
+    let unavailable_a = UnavailableEndpoint::new();
+    let port_a = unavailable_a.port();
+    let listener_b = reserve_port().into_std_listener();
+    let port_b = listener_b.local_addr().expect("index B address").port();
+    let index_b = MockRelay::start_on_listener(
+        listener_b.try_clone().expect("retain index B listener"),
+        Vec::new(),
+    )
+    .await;
     let git_data = tempfile::tempdir().expect("git data dir");
     let relay_data = tempfile::tempdir().expect("relay data dir");
     let relay = TestRelay::start_on_reservation_persistent_user_index_relays(
@@ -338,8 +345,10 @@ async fn stored_identity_is_not_pushed_to_recovering_index_holding_an_identity()
     // identity events must wait for a successful index check before any
     // publication, then reach only relays confirmed to hold nothing.
     index_b.stop().await;
+    let unavailable_b = UnavailableEndpoint::from_listener(listener_b);
     let relay = relay.restart().await;
-    let index_b = MockRelay::start_on_port(port_b).await;
+    let index_b =
+        MockRelay::start_on_listener(unavailable_b.into_listener().await, Vec::new()).await;
     for kind in [Kind::Metadata, Kind::RelayList] {
         assert!(
             wait_for_event_on_relay(
@@ -356,7 +365,11 @@ async fn stored_identity_is_not_pushed_to_recovering_index_holding_an_identity()
     // A recovers already holding the customized profile. The per-relay
     // re-check before every send must adopt it locally instead of pushing
     // the stored (formerly generated) profile over it.
-    let index_a = MockRelay::start_on_port_with_events(port_a, vec![customized.clone()]).await;
+    let index_a = MockRelay::start_on_listener(
+        unavailable_a.into_listener().await,
+        vec![customized.clone()],
+    )
+    .await;
     assert!(
         wait_for_event_on_relay(
             relay.url(),
@@ -549,3 +562,19 @@ fn assert_identity_shape(relay: &TestRelay, events: &[Event]) {
     assert_eq!(advertised[0].0.to_string(), relay.url());
     assert_eq!(advertised[0].1, None, "unmarked means read and write");
 }
+
+#[tokio::test]
+async fn subprocess_restart_retains_the_reserved_endpoint() {
+    tokio::time::timeout(Duration::from_secs(20), async {
+        let relay = TestRelay::start().await;
+        let url = relay.url().to_string();
+        let relay = relay.restart().await;
+        assert_eq!(relay.url(), url);
+        let (_connection, _) = tokio_tungstenite::connect_async(relay.url())
+            .await
+            .expect("restarted subprocess must serve the inherited listener");
+        relay.stop().await;
+    })
+    .await
+    .expect("subprocess startup and restart must complete");
+}