diff --git a/apps/maple-agent/app/src/desktop/local_api_tests.rs b/apps/maple-agent/app/src/desktop/local_api_tests.rs index 1bc6a18a8..5546c0aba 100644 --- a/apps/maple-agent/app/src/desktop/local_api_tests.rs +++ b/apps/maple-agent/app/src/desktop/local_api_tests.rs @@ -168,6 +168,20 @@ impl FixtureHost { self.status_holds.lock().unwrap().push_back(hold); release } + + fn listener_task(&self) -> Option { + self.runtime + .as_ref() + .unwrap() + .block_on(self.manager.listener_task_for_test()) + } + + fn assert_api_stopped(&self) { + let status = self.manager.status(); + assert_eq!(status.state, LocalApiState::Stopped); + assert_eq!(status.active_requests, 0); + assert!(self.listener_task().is_none()); + } } impl Drop for FixtureHost { @@ -228,12 +242,12 @@ fn toggle(cx: &mut gpui::VisualTestContext) { cx.simulate_click(bounds.center(), gpui::Modifiers::default()); } -fn assert_socket(port: u16, open: bool) { +fn assert_socket_open(port: u16) { let result = std::net::TcpStream::connect_timeout( &std::net::SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port)), Duration::from_secs(1), ); - assert_eq!(result.is_ok(), open, "port {port}"); + assert!(result.is_ok(), "port {port}"); } #[gpui::test] @@ -242,16 +256,17 @@ fn selected_execution_computers_cannot_move_settings_listener_off_the_controller ) { let settings_lock = crate::settings::SETTINGS_IO_LOCK.lock(); cx.executor().allow_parking(); - // Reserve all three before releasing any, so a port cannot be reused by - // another fixture merely because the OS selected the same ephemeral port. - let reservations: Vec<_> = (0..3) - .map(|_| std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap()) - .collect(); - let ports: Vec<_> = reservations - .iter() - .map(|socket| socket.local_addr().unwrap().port()) - .collect(); - drop(reservations); + // Keep the never-started remote ports reserved throughout the test. Their + // manager ownership, rather than global port availability, proves no API + // started there. Only the controller needs its reservation released. + let reservations: [std::net::TcpListener; 3] = std::array::from_fn(|_| { + std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap() + }); + let ports = reservations + .each_ref() + .map(|socket| socket.local_addr().unwrap().port()); + let [local_reservation, _east_reservation, _west_reservation] = reservations; + drop(local_reservation); let local = FixtureHost::new("controller", ports[0]); let east = FixtureHost::new("east", ports[1]); let west = FixtureHost::new("west", ports[2]); @@ -276,9 +291,11 @@ fn selected_execution_computers_cannot_move_settings_listener_off_the_controller local.manager.status().state == LocalApiState::Running && local.count("set_local_api_config") == 1 }); - assert_socket(ports[0], true); - assert_socket(ports[1], false); - assert_socket(ports[2], false); + assert_socket_open(ports[0]); + let listener_task = local.listener_task().expect("controller's listening task"); + assert!(!listener_task.is_finished()); + east.assert_api_stopped(); + west.assert_api_stopped(); assert_eq!(east.api_calls(), 0); assert_eq!(west.api_calls(), 0); @@ -289,9 +306,12 @@ fn selected_execution_computers_cannot_move_settings_listener_off_the_controller && local.count("set_local_api_config") == 2 }); assert!(!local.manager.status().config.enabled); - assert_socket(ports[0], false); - assert_socket(ports[1], false); - assert_socket(ports[2], false); + // The original accept task owns its listener by value. Completion proves + // that socket was dropped even if another test has already reused its port. + assert!(listener_task.is_finished()); + local.assert_api_stopped(); + east.assert_api_stopped(); + west.assert_api_stopped(); assert!(local.count("get_local_api_status") >= 2); assert_eq!(east.api_calls(), 0); assert_eq!(west.api_calls(), 0); diff --git a/apps/maple-agent/app/src/host/local_api_tests.rs b/apps/maple-agent/app/src/host/local_api_tests.rs index 8c64eb993..61b91e54e 100644 --- a/apps/maple-agent/app/src/host/local_api_tests.rs +++ b/apps/maple-agent/app/src/host/local_api_tests.rs @@ -5,6 +5,13 @@ use crate::local_api::{LocalApiConfig, LocalApiState, LocalApiStatus}; use maple_agent::store::records::{CredentialTokens, SecretString}; use maple_agent::store::{DamageReason, RefusalReason, StorePaths}; use serde_json::json; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +struct ListenerProbe { + port: u16, + task: tokio::task::AbortHandle, + connection: tokio::net::TcpStream, +} #[test] fn unavailable_stores_keep_api_status_and_emergency_stop_but_reject_configuration() { @@ -115,7 +122,7 @@ impl HostFixture { } } - async fn start_listener(&self) -> u16 { + async fn start_listener(&self) -> ListenerProbe { let reservation = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap(); let port = reservation.local_addr().unwrap().port(); drop(reservation); @@ -149,7 +156,31 @@ impl HostFixture { .status(), reqwest::StatusCode::UNAUTHORIZED ); - port + let task = self + .backend + .local_api + .listener_task_for_test() + .await + .expect("the running API must own its accept task"); + assert!(!task.is_finished()); + let mut connection = tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port)) + .await + .unwrap(); + connection + .write_all(format!("GET /health HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\n\r\n").as_bytes()) + .await + .unwrap(); + let mut prefix = [0; 12]; + tokio::time::timeout(Duration::from_secs(3), connection.read_exact(&mut prefix)) + .await + .expect("the original listener must accept the probe connection") + .unwrap(); + assert_eq!(&prefix, b"HTTP/1.1 200"); + ListenerProbe { + port, + task, + connection, + } } async fn finish(self) { @@ -163,13 +194,26 @@ impl HostFixture { } } -async fn assert_closed(port: u16) { +async fn assert_closed(listener: &mut ListenerProbe) { + // The accept task owns its TcpListener by value. Its completion proves + // that exact socket was dropped, even if another test now owns the port. assert!( - tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port)) - .await - .is_err(), - "the listening socket must be gone when cleanup acknowledges success" + listener.task.is_finished(), + "the original accept task must finish before cleanup acknowledges success" ); + tokio::time::timeout(Duration::from_secs(3), async { + let mut data = [0; 1024]; + loop { + // Drain any health response bytes before observing EOF/reset on + // the connection accepted by the original server before Stop. + match listener.connection.read(&mut data).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } + }) + .await + .expect("cleanup must close the original accepted connection"); } #[tokio::test] @@ -178,7 +222,8 @@ async fn emergency_stop_over_ipc_closes_the_real_listener_and_preserves_requeste let backend = Arc::new(LocalBackend::isolated_for_test(root.path())); credentials(&backend); let fixture = HostFixture::new(root.path(), backend).await; - let port = fixture.start_listener().await; + let mut listener = fixture.start_listener().await; + let port = listener.port; let status: LocalApiStatus = fixture .client .call("get_local_api_status", ()) @@ -190,9 +235,10 @@ async fn emergency_stop_over_ipc_closes_the_real_listener_and_preserves_requeste let stopped: LocalApiStatus = fixture.client.call("stop_local_api", ()).await.unwrap(); assert_eq!(stopped.state, LocalApiState::Stopped); + assert_eq!(stopped.active_requests, 0); assert!(stopped.config.enabled); assert_eq!(stopped.config.port, port); - assert_closed(port).await; + assert_closed(&mut listener).await; // The UI switch has a different contract from emergency Stop: Off is // durable and must not require a network request just to stop spending. @@ -232,7 +278,6 @@ async fn saved_account_and_enabled_preference_do_not_start_before_authentication let account = store.account_id(USER).unwrap().unwrap(); let reservation = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap(); let port = reservation.local_addr().unwrap().port(); - drop(reservation); store .set_account_setting( account, @@ -251,7 +296,7 @@ async fn saved_account_and_enabled_preference_do_not_start_before_authentication .expect("resume must not try to validate a cached account implicitly") .unwrap(); assert_eq!(backend.local_api.status().state, LocalApiState::Stopped); - assert_closed(port).await; + assert!(backend.local_api.listener_task_for_test().await.is_none()); assert_eq!( store.account_setting(account, "local_api.config").unwrap(), Some(json!({"enabled":true,"port":port})), @@ -335,14 +380,14 @@ async fn logout_closes_the_listener_without_relying_on_the_account_event_pump() let backend = Arc::new(LocalBackend::isolated_for_test(root.path())); credentials(&backend); let fixture = HostFixture::new(root.path(), backend).await; - let port = fixture.start_listener().await; + let mut listener = fixture.start_listener().await; fixture .client .call::<()>("logout_and_clear", (USER,)) .await .unwrap(); assert!(fixture.backend.saved_user_id().is_none()); - assert_closed(port).await; + assert_closed(&mut listener).await; let stopped: LocalApiStatus = fixture .client .call("get_local_api_status", ()) @@ -358,7 +403,7 @@ async fn emergency_stop_during_confirmed_erasure_survives_a_precommit_cleanup_re let backend = Arc::new(LocalBackend::isolated_for_test(root.path())); credentials(&backend); let fixture = HostFixture::new(root.path(), backend).await; - let port = fixture.start_listener().await; + let mut listener = fixture.start_listener().await; let generation = fixture.store.store_id().unwrap(); let epoch = fixture.backend.local_api_control_epoch(); let spills = maple_agent::store::spill::SpillRoot::new(fixture.store.paths()).unwrap(); @@ -390,7 +435,7 @@ async fn emergency_stop_during_confirmed_erasure_survives_a_precommit_cleanup_re .await .expect("maintenance must suspend the listener before waiting for readers"); assert_eq!(fixture.backend.local_api_control_epoch(), epoch); - assert_closed(port).await; + assert_closed(&mut listener).await; let stopped: LocalApiStatus = tokio::time::timeout( Duration::from_secs(3), fixture.client.call("stop_local_api", ()), @@ -407,7 +452,7 @@ async fn emergency_stop_during_confirmed_erasure_survives_a_precommit_cleanup_re assert_eq!(fixture.backend.local_api_control_epoch(), epoch + 1); assert_eq!(fixture.store.store_id().unwrap(), generation); assert_eq!(fixture.store.state(), StoreState::Open); - assert_closed(port).await; + assert_closed(&mut listener).await; let stopped: LocalApiStatus = fixture .client .call("get_local_api_status", ()) @@ -425,7 +470,7 @@ async fn emergency_stop_during_confirmed_erasure_survives_a_precommit_cleanup_re fixture.backend.local_api.status().state, LocalApiState::Stopped ); - assert_closed(port).await; + assert_closed(&mut listener).await; std::fs::remove_file(spills.path()).unwrap(); fixture.finish().await; } @@ -435,7 +480,7 @@ async fn owner_shutdown_closes_the_listener_before_releasing_local_data() { let root = directory(); let backend = Arc::new(LocalBackend::isolated_for_test(root.path())); let fixture = HostFixture::new(root.path(), backend).await; - let port = fixture.start_listener().await; + let mut listener = fixture.start_listener().await; tokio::time::timeout( Duration::from_secs(3), fixture @@ -446,7 +491,7 @@ async fn owner_shutdown_closes_the_listener_before_releasing_local_data() { .expect("a local listener must not hold an account reader through shutdown") .unwrap(); assert_ne!(fixture.store.state(), StoreState::Open); - assert_closed(port).await; + assert_closed(&mut listener).await; fixture.finish().await; } diff --git a/apps/maple-agent/app/src/local_api.rs b/apps/maple-agent/app/src/local_api.rs index dea822711..39076e262 100644 --- a/apps/maple-agent/app/src/local_api.rs +++ b/apps/maple-agent/app/src/local_api.rs @@ -219,6 +219,16 @@ impl LocalApiManager { self.inner.state.lock().unwrap().generation } + #[cfg(test)] + pub async fn listener_task_for_test(&self) -> Option { + self.inner + .running + .lock() + .await + .as_ref() + .map(|running| running.task.abort_handle()) + } + pub fn stays_running(&self) -> bool { matches!( self.status().state, @@ -630,6 +640,15 @@ mod tests { format!("http://127.0.0.1:{}{path}", config.port) } + async fn listener_task(manager: &LocalApiManager) -> tokio::task::AbortHandle { + let task = manager + .listener_task_for_test() + .await + .expect("the running manager must own its accept task"); + assert!(!task.is_finished()); + task + } + async fn start( manager: &LocalApiManager, config: LocalApiConfig, @@ -703,6 +722,7 @@ mod tests { ) .await; assert_eq!(running.state, LocalApiState::Running); + let task = listener_task(&manager).await; assert!(manager.stays_running()); assert!(running.revision > status.revision); assert_eq!( @@ -719,9 +739,7 @@ mod tests { assert_eq!(stopped.owner, None); assert_eq!(stopped.state, LocalApiState::Stopped); assert!(!manager.stays_running()); - TcpListener::bind((Ipv4Addr::LOCALHOST, config.port)) - .await - .unwrap(); + assert!(task.is_finished()); } #[tokio::test] @@ -813,6 +831,7 @@ mod tests { .unwrap() .unwrap(); assert!(String::from_utf8_lossy(&response[..read]).contains("503")); + let task = listener_task(&manager).await; let stopped = tokio::time::timeout(Duration::from_secs(3), manager.stop()) .await .unwrap(); @@ -821,9 +840,7 @@ mod tests { while let Some(result) = callers.join_next().await { assert!(result.unwrap().is_err()); } - TcpListener::bind((Ipv4Addr::LOCALHOST, config.port)) - .await - .unwrap(); + assert!(task.is_finished()); } #[tokio::test] @@ -861,6 +878,7 @@ mod tests { .unwrap(); wait_for(&manager, |status| status.active_requests == 1).await; assert!(!dropped.load(Ordering::SeqCst)); + let task = listener_task(&manager).await; // Do not consume the stream. Connection cancellation must interrupt a // peer whose receive window fills, not wait for another body poll. let stopped = tokio::time::timeout(Duration::from_secs(3), manager.stop()) @@ -869,9 +887,7 @@ mod tests { assert_eq!(stopped.active_requests, 0); assert!(dropped.load(Ordering::SeqCst)); drop(response); - TcpListener::bind((Ipv4Addr::LOCALHOST, config.port)) - .await - .unwrap(); + assert!(task.is_finished()); } #[tokio::test] @@ -1034,9 +1050,7 @@ mod tests { .unwrap(); assert_eq!(delayed.state, LocalApiState::Stopped); assert_eq!(delayed.owner, None); - TcpListener::bind((Ipv4Addr::LOCALHOST, config.port)) - .await - .unwrap(); + assert!(manager.listener_task_for_test().await.is_none()); let disabled = LocalApiConfig { enabled: false, ..config @@ -1060,11 +1074,16 @@ mod tests { .await .unwrap(); stream.write_all(b"GET / HTTP/1.1\r\n").await.unwrap(); + let task = listener_task(&manager).await; drop(manager); + tokio::time::timeout(Duration::from_secs(3), async { + while !task.is_finished() { + tokio::task::yield_now().await; + } + }) + .await + .expect("dropping the manager must finish its owned accept task"); assert_closed(stream).await; - TcpListener::bind((Ipv4Addr::LOCALHOST, config.port)) - .await - .unwrap(); } #[tokio::test]