From 7c144cc096fb2ba3846a3c03117fa2d4ee5658a3 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Sat, 12 Sep 2026 14:10:15 +0000 Subject: [PATCH 1/8] test(sync): retain relay listeners through startup The SDK local relay probes and releases an available port before run binds it again. Another socket can claim that port in between, causing AddrInUse in parallel relay-connection tests. Passing port zero alone also leaves the SDK URL pointing to zero in the locked dependency version. Use a test-only listener bound to loopback port zero and retain it for the fixture lifetime. Upgrade HTTP connections and pass their streams to the SDK relay handler, preserving relay policies and subscription behavior. Own connection tasks in a JoinSet and abort them with the fixture. Move all ten affected test relay startups to this fixture; production is unchanged. Add coverage that starts 32 relays concurrently and connects to their unique actual bound addresses, using bounded handshake/close deadlines. Do not disable tests, serialize the suite, or retry occupied ports. Validation: all 65 relay-connection tests pass, including the previously failing tail-group restoration test and the concurrent-listener regression. Full Rust formatting and whitespace checks pass. Backport applies cleanly to 3.0.2. Full nixpkgs host validation remains pending. Assisted-by: Codex (GPT-6) --- src/sync/relay_connection.rs | 34 +++--- src/sync/relay_connection/test_relay.rs | 136 ++++++++++++++++++++++++ 2 files changed, 150 insertions(+), 20 deletions(-) create mode 100644 src/sync/relay_connection/test_relay.rs diff --git a/src/sync/relay_connection.rs b/src/sync/relay_connection.rs index 429fa94..75e175b 100644 --- a/src/sync/relay_connection.rs +++ b/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"); diff --git a/src/sync/relay_connection/test_relay.rs b/src/sync/relay_connection/test_relay.rs new file mode 100644 index 0000000..c7cde68 --- /dev/null +++ b/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| { + 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"); + } +} From 6d05e7eb0541d10018f4ecef593ff4d8c58af4a3 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Sat, 12 Sep 2026 14:48:16 +0000 Subject: [PATCH 2/8] test: remove scheduling margins from expiry assertions Short real-time TTLs and sleep-based ordering made cache and purgatory tests sensitive to host load. Age private cache timestamps explicitly and compare queue or persistence deadlines against timestamps bracketing the operation. Production clocks and expiry policy are unchanged. Retain tests for both cache tiers, remaining TTL, queue deadlines and persisted expiry values. Validation: 33 rejected-index and 132 purgatory tests passed in the offline repository development shell. Assisted-by: Codex (GPT-6) --- src/purgatory/persistence.rs | 55 ++++++++------------ src/purgatory/sync/queue.rs | 26 +++++----- src/sync/rejected_index.rs | 97 ++++++++++++++++++++++++------------ 3 files changed, 98 insertions(+), 80 deletions(-) diff --git a/src/purgatory/persistence.rs b/src/purgatory/persistence.rs index 7fca2cf..67cd129 100644 --- a/src/purgatory/persistence.rs +++ b/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)); } } diff --git a/src/purgatory/sync/queue.rs b/src/purgatory/sync/queue.rs index 3226f47..be56f15 100644 --- a/src/purgatory/sync/queue.rs +++ b/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)); } } diff --git a/src/sync/rejected_index.rs b/src/sync/rejected_index.rs index 718fca3..0808a4e 100644 --- a/src/sync/rejected_index.rs +++ b/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 From 1187cc062e4cc93e39cbcee9f9187641aebf5a1f Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Sat, 12 Sep 2026 14:48:16 +0000 Subject: [PATCH 3/8] test(git): synchronize streaming assertions with explicit gates Sleeping fake Git processes and assumed HTTP frame boundaries made streaming checks depend on scheduler timing. Hold fake Git behind an owned loopback gate until the test observes the expected prefix; collect bytes independently of frame splits and explicitly release subprocess progress. Keep terminal-flush, promotion and cleanup ordering assertions. Gate waits have bounded deadlines; no production streaming code changes are included. Validation: all three git_response_streaming tests passed. Assisted-by: Codex (GPT-6) --- tests/git_response_streaming.rs | 260 ++++++++++++++------------------ 1 file changed, 117 insertions(+), 143 deletions(-) diff --git a/tests/git_response_streaming.rs b/tests/git_response_streaming.rs index f669493..bd64639 100644 --- a/tests/git_response_streaming.rs +++ b/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(body: &mut B) -> Vec +where + B: Body + 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(body: &mut B, streamed: &mut Vec) +where + B: Body + 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 { 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"); From 75da9a8446d79a38b84ce1590a4ed6801ba08bf7 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Sat, 12 Sep 2026 14:48:16 +0000 Subject: [PATCH 4/8] test: preserve subprocess and recovery fixture listeners Releasing a reservation before spawn or recovery allowed unrelated tests to take the same port. Transfer the socket across exec through a private Unix test-only protocol, retaining a parent copy across Grasp restarts. Validate the inherited descriptor and keep it out of Git and SSH descendants. Offline identity fixtures retain an accept-and-close endpoint until recovery. Mock relay listeners and upgraded connections remain owned until shutdown. Use real HTTP or SDK readiness instead of grace sleeps; startup errors in the owned-listener path are not retried. Normal server startup and deployment configuration are unchanged. The capability probe permits older external harnesses to remain compatible. Validation: listener adoption unit tests passed; relay_identity passed with recovery coverage, MockRelay shutdown passed, and a real subprocess restart plus a cross-repository ngit harness smoke test passed. Full platform builds remain host validation. Assisted-by: Codex (GPT-6) --- src/main.rs | 16 +++ src/server.rs | 8 +- src/test_listener.rs | 93 +++++++++++++++++ tests/common/mock_relay.rs | 88 +++++++++++++--- tests/common/port.rs | 209 ++++++++++++++++++++++++------------- tests/common/relay.rs | 129 ++++++++--------------- tests/relay_identity.rs | 47 +++++++-- 7 files changed, 407 insertions(+), 183 deletions(-) create mode 100644 src/test_listener.rs diff --git a/src/main.rs b/src/main.rs index 33d7936..b1fcb2a 100644 --- a/src/main.rs +++ b/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 diff --git a/src/server.rs b/src/server.rs index d9ae464..3a6d85e 100644 --- a/src/server.rs +++ b/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 { + pub async fn start(config: Config) -> Result { // 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 { let local_addr = listener.local_addr()?; config.bind_address = local_addr.to_string(); if config.domain.is_empty() { diff --git a/src/test_listener.rs b/src/test_listener.rs new file mode 100644 index 0000000..744999e --- /dev/null +++ b/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> { + 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 { + 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()); + } +} diff --git a/tests/common/mock_relay.rs b/tests/common/mock_relay.rs index ebbed89..2af1225 100644 --- a/tests/common/mock_relay.rs +++ b/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) -> 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, custom_nip11: Option, + upgrades: Arc>>, ) -> Result>, 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") diff --git a/tests/common/port.rs b/tests/common/port.rs index 042f6ec..5de7c4f 100644 --- a/tests/common/port.rs +++ b/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 { 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, + task: Option>, +} + +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"); + } } diff --git a/tests/common/relay.rs b/tests/common/relay.rs index a097d36..cd3f909 100644 --- a/tests/common/relay.rs +++ b/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(); } diff --git a/tests/relay_identity.rs b/tests/relay_identity.rs index 1e920a2..b15db00 100644 --- a/tests/relay_identity.rs +++ b/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"); +} From cf2ec36a68896ff3f6ca0fe79230457e70b8ce31 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Sat, 12 Sep 2026 14:48:26 +0000 Subject: [PATCH 5/8] test(sync): wait for observable fixture readiness Fixed startup and promotion sleeps could expire before asynchronous work completed. Wait with bounded deadlines for event visibility and an actual connected gauge; connection attempts and health do not prove readiness. Preserve URL schemes when constructing metrics endpoints, retain ownership of unavailable endpoints, and remove stopped sources instead of replacing them with a live placeholder. Keep immediate assertions where completion already guarantees the result. Production sync policy is unchanged. Validation: shared sync helper tests passed; relay_identity and the fixture lifecycle target compiled and passed with the revised common helpers. Full sync and deletion integration execution remains host validation. Assisted-by: Codex (GPT-6) --- tests/common/nip09_helpers.rs | 103 ++++---------- tests/common/sync_helpers.rs | 250 +++++++++++++++++----------------- 2 files changed, 151 insertions(+), 202 deletions(-) diff --git a/tests/common/nip09_helpers.rs b/tests/common/nip09_helpers.rs index 36d7e4b..51c5672 100644 --- a/tests/common/nip09_helpers.rs +++ b/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) } diff --git a/tests/common/sync_helpers.rs b/tests/common/sync_helpers.rs index e7dd955..75e9ae9 100644 --- a/tests/common/sync_helpers.rs +++ b/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::() { - 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::() { - // 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 { - // 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, + source_relays: Vec>, syncing_relay: Option, - #[allow(dead_code)] - nowhere_url: Option, + unavailable_endpoint: Option, } 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"); From 3c77e76a1f0f2a4aa693375a60e411a9f5a30108 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Sat, 12 Sep 2026 14:48:26 +0000 Subject: [PATCH 6/8] test(git): own request tasks and avoid pipe backpressure deadlocks Detached HTTP tasks could outlive stopped Git fixtures. Sequential stdin and output handling could deadlock when a child filled stdout or stderr before consuming input. Own connection tasks, cancel Git children with their request, and drain both outputs concurrently with stdin. Keep fixture HTTP semantics and repository layout unchanged. A bound owned listener replaces the unnecessary startup grace loop. Explicit stop joins cancelled tasks while synchronous drop aborts the owner. Validation: all seven shared SmartGitServer tests passed, including live connection shutdown and one-MiB concurrent pipe traffic. Assisted-by: Codex (GPT-6) --- tests/common/git_server.rs | 214 +++++++++++++++++++++++-------------- 1 file changed, 131 insertions(+), 83 deletions(-) diff --git a/tests/common/git_server.rs b/tests/common/git_server.rs index edc986e..7ca5a4d 100644 --- a/tests/common/git_server.rs +++ b/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>, 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>, 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 { + 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> { + 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, + ) { + 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 From 59a37b660ab358c05dc7c1beaaa689c9bf17781f Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Sat, 12 Sep 2026 14:48:26 +0000 Subject: [PATCH 7/8] test: cancel proxy connections when fixtures stop Stopping accept loops left detached relay/proxy sessions alive. Track HTTP, WebSocket upgrade and forwarding tasks under their owning fixture, cancel them on shutdown, and drain cancellation before explicit stop returns. Preserve censoring, rate limits, authentication and simulated disconnect behavior. Add regressions that observe a live protocol exchange before asserting the connection closes on stop, all under bounded deadlines. Validation: fixture lifecycle checks passed for censoring, REQ/NEG limiting, flapping and setup-drop relays. Auth-gating and upload proxy shutdown regressions passed through relay_identity's common helper tests. Assisted-by: Codex (GPT-6) --- tests/common/auth_gating_relay.rs | 69 ++++++++++++++++------ tests/common/censoring_proxy.rs | 10 +++- tests/common/flapping_relay.rs | 10 +++- tests/common/neg_limiting_proxy.rs | 10 +++- tests/common/req_limiting_proxy.rs | 10 +++- tests/common/setup_drop_relay.rs | 10 +++- tests/common/upload_pack_counting_proxy.rs | 46 ++++++++++++++- tests/fixture_lifecycle.rs | 58 ++++++++++++++++++ 8 files changed, 198 insertions(+), 25 deletions(-) create mode 100644 tests/fixture_lifecycle.rs diff --git a/tests/common/auth_gating_relay.rs b/tests/common/auth_gating_relay.rs index 647ba23..93cd681 100644 --- a/tests/common/auth_gating_relay.rs +++ b/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>>, ) -> Result>, 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> = None; - let mut backend_task: Option> = 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>, >, 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"); + } +} diff --git a/tests/common/censoring_proxy.rs b/tests/common/censoring_proxy.rs index a986e79..179f2bf 100644 --- a/tests/common/censoring_proxy.rs +++ b/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(()); } diff --git a/tests/common/flapping_relay.rs b/tests/common/flapping_relay.rs index cd08627..897a5fb 100644 --- a/tests/common/flapping_relay.rs +++ b/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(()); } diff --git a/tests/common/neg_limiting_proxy.rs b/tests/common/neg_limiting_proxy.rs index aa4e8b6..f94a7a0 100644 --- a/tests/common/neg_limiting_proxy.rs +++ b/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(()); } diff --git a/tests/common/req_limiting_proxy.rs b/tests/common/req_limiting_proxy.rs index 22eaa35..1d40dd3 100644 --- a/tests/common/req_limiting_proxy.rs +++ b/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(()); } diff --git a/tests/common/setup_drop_relay.rs b/tests/common/setup_drop_relay.rs index ab27b5e..411b0ff 100644 --- a/tests/common/setup_drop_relay.rs +++ b/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(()); } diff --git a/tests/common/upload_pack_counting_proxy.rs b/tests/common/upload_pack_counting_proxy.rs index 8f38e30..8516c3a 100644 --- a/tests/common/upload_pack_counting_proxy.rs +++ b/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\ diff --git a/tests/fixture_lifecycle.rs b/tests/fixture_lifecycle.rs new file mode 100644 index 0000000..266ca1e --- /dev/null +++ b/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); +} From cac1256fb5777a19fb484a366065f4f905fa8029 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Sat, 12 Sep 2026 14:49:09 +0000 Subject: [PATCH 8/8] docs(test): record fixture ownership and timing contracts Document the private test listener protocol, bounded observable readiness, connection ownership and controlled timing assertions so new fixtures do not reintroduce the same races. Replace advice to rerun transient failures with instructions to preserve evidence and diagnose the underlying cause. The listener environment is test-only rather than deployment configuration; normal service configuration and the NixOS module are deliberately unchanged. Validation: checked guidance against the implementation and targeted test commands; formatting and whitespace checks pass. Assisted-by: Codex (GPT-6) --- AGENTS.md | 8 +++++-- docs/how-to/test-fixtures.md | 41 ++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) create mode 100644 docs/how-to/test-fixtures.md diff --git a/AGENTS.md b/AGENTS.md index 869c99a..4645e9c 100644 --- a/AGENTS.md +++ b/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 diff --git a/docs/how-to/test-fixtures.md b/docs/how-to/test-fixtures.md new file mode 100644 index 0000000..1f693d3 --- /dev/null +++ b/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.