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. 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/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/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/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 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"); + } +} 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/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/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 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/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/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/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/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/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"); 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); +} 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"); 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"); +}