From 6667985883cbfe7b9d07deac6e29eb860484a9d6 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 16 Oct 2026 15:04:42 +0900 Subject: [PATCH] fix: preserve test completion and fixture invariants under concurrency --- crates/worker-runtime/src/worker_source.rs | 10 +- .../src/working_directory/cleanup.rs | 11 ++- .../worker-runtime/src/workspace_request.rs | 93 ++++++++++++++++--- crates/worker/tests/controller_test.rs | 35 ++++++- .../tests/protocol_backpressure_tests.rs | 28 +++++- .../src/server_assignment_attention_tests.rs | 19 +++- crates/yoi/src/ticket_cli.rs | 2 + 7 files changed, 168 insertions(+), 30 deletions(-) diff --git a/crates/worker-runtime/src/worker_source.rs b/crates/worker-runtime/src/worker_source.rs index 18ec00cc7..a50fafc98 100644 --- a/crates/worker-runtime/src/worker_source.rs +++ b/crates/worker-runtime/src/worker_source.rs @@ -1816,10 +1816,12 @@ mod tests { } #[test] - fn runtime_owned_workspace_client_has_no_fixed_request_timeout() { - // This must exceed reqwest::blocking::Client's default 30-second timeout. - // A short delay only checks our Option field, not the effective HTTP deadline. - let (base_url, server) = delayed_workspace_response(Duration::from_secs(31)); + fn runtime_owned_workspace_client_returns_successful_response_without_request_timeout() { + // Effective deadline clearing is exercised with a short inherited deadline in + // workspace_request::tests::blocking_workspace_client_clears_initial_request_timeout. + let (base_url, server) = binary_loopback( + b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 11\r\nConnection: close\r\n\r\n{\"ok\":true}".to_vec(), + ); let client = RuntimeOwnedWorkspaceClient::new("workspace-a", base_url, "runtime-a", "worker-a"); assert_eq!(client.request_timeout, None); diff --git a/crates/worker-runtime/src/working_directory/cleanup.rs b/crates/worker-runtime/src/working_directory/cleanup.rs index 40b6957ee..a791cca85 100644 --- a/crates/worker-runtime/src/working_directory/cleanup.rs +++ b/crates/worker-runtime/src/working_directory/cleanup.rs @@ -1223,6 +1223,7 @@ pub(super) mod tests { } fn sized_fixture( + repo: &Path, size: u64, ) -> ( tempfile::TempDir, @@ -1230,7 +1231,9 @@ pub(super) mod tests { WorkingDirectoryBinding, BodyWatch, ) { - let (runtime, materializer, binding) = fixture(); + let runtime = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitMaterializer::new(runtime.path()); + let binding = materializer.create(&request(repo)).unwrap(); fs::write(binding.root.join(".git/info/exclude"), "target/\n").unwrap(); fs::create_dir(binding.root.join("target")).unwrap(); // Entry count remains constant, including .git and many ignored files. @@ -1256,10 +1259,14 @@ pub(super) mod tests { #[test] fn actual_cleanup_and_restart_retry_never_open_bodies_when_only_file_sizes_grow() { use io_observation::observe; + // Vary only file sizes, not Git object names/layout. Separate commits + // can have different object-prefix directory counts as timestamps change. + // Each case still owns an independent materialized checkout and Runtime. + let repo = create_clean_repo(); for failure in [None, Some(0), Some(10)] { let mut small_counts = None; for size in [256, 128 * 1024 * 1024] { - let (runtime, materializer, binding, mut watch) = sized_fixture(size); + let (runtime, materializer, binding, mut watch) = sized_fixture(repo.path(), size); let id = &binding.working_directory.id; let (first, first_io) = observe(failure, || materializer.cleanup_working_directory(id)); diff --git a/crates/worker-runtime/src/workspace_request.rs b/crates/worker-runtime/src/workspace_request.rs index 13beb9814..f008640cd 100644 --- a/crates/worker-runtime/src/workspace_request.rs +++ b/crates/worker-runtime/src/workspace_request.rs @@ -262,12 +262,14 @@ impl RuntimeWorkspaceRequestClient { request: RuntimeWorkspaceRequest, ) -> Result { let client = self.clone(); - std::thread::spawn(move || client.execute_blocking_inner(request, true)) - .join() - .map_err(|_| RuntimeWorkspaceRequestError::Transport { - message: "Workspace request thread panicked".to_string(), - timeout: false, - })? + std::thread::spawn(move || { + client.execute_blocking_inner(request, true, reqwest::blocking::Client::builder()) + }) + .join() + .map_err(|_| RuntimeWorkspaceRequestError::Transport { + message: "Workspace request thread panicked".to_string(), + timeout: false, + })? } /// Binary requests must retain their exact target and never follow redirects. @@ -276,18 +278,21 @@ impl RuntimeWorkspaceRequestClient { request: RuntimeWorkspaceRequest, ) -> Result { let client = self.clone(); - std::thread::spawn(move || client.execute_blocking_inner(request, false)) - .join() - .map_err(|_| RuntimeWorkspaceRequestError::Transport { - message: "Workspace binary request thread panicked".to_string(), - timeout: false, - })? + std::thread::spawn(move || { + client.execute_blocking_inner(request, false, reqwest::blocking::Client::builder()) + }) + .join() + .map_err(|_| RuntimeWorkspaceRequestError::Transport { + message: "Workspace binary request thread panicked".to_string(), + timeout: false, + })? } fn execute_blocking_inner( &self, request: RuntimeWorkspaceRequest, follow_redirects: bool, + client_builder: reqwest::blocking::ClientBuilder, ) -> Result { let prepared = self.prepare(&request)?; if !follow_redirects { @@ -304,7 +309,7 @@ impl RuntimeWorkspaceRequestClient { } // The blocking client defaults to 30 seconds. Apply None explicitly so // provider-owned command waits are not cut short by an HTTP deadline. - let mut client_builder = reqwest::blocking::Client::builder().timeout(request.timeout); + let mut client_builder = client_builder.timeout(request.timeout); if !follow_redirects { client_builder = client_builder.redirect(reqwest::redirect::Policy::none()); } @@ -425,6 +430,68 @@ mod tests { use super::*; use crate::auth::decode_runtime_request_source_claims; + // Component boundary: production request preparation, timeout policy and HTTP execution; + // only the initial client deadline and remote Workspace are supplied by the fixture. + #[tokio::test] + async fn blocking_workspace_client_clears_initial_request_timeout() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + tokio::time::timeout(Duration::from_secs(5), async { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .unwrap(); + let address = listener.local_addr().unwrap(); + let initial_timeout = Duration::from_millis(20); + let mut request = tokio::task::spawn_blocking(move || { + let builder = reqwest::blocking::Client::builder() + .no_proxy() + .timeout(initial_timeout); + RuntimeWorkspaceRequestClient::new("workspace-a", format!("http://{address}"), "runtime-a") + .execute_blocking_inner( + RuntimeWorkspaceRequest { + method: reqwest::Method::GET, + path_and_query: "/api/test".into(), + body: Vec::new(), + headers: HeaderMap::new(), + permission: "test".into(), + worker_id: None, + timeout: None, + max_response_bytes: 1024, + }, + true, + builder, + ) + }); + + let receive_and_cross_deadline = async { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut headers = Vec::new(); + while !headers.ends_with(b"\r\n\r\n") { + headers.push(stream.read_u8().await.unwrap()); + assert!(headers.len() <= 8192, "unbounded request headers"); + } + // Deliberately cross the inherited deadline after request arrival. + // This timer supplies the fault stimulus, not a completion signal. + tokio::time::sleep(initial_timeout * 2).await; + stream + }; + let mut stream = tokio::select! { + result = &mut request => panic!("request ended before response: {result:?}"), + stream = receive_and_cross_deadline => stream, + }; + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 11\r\nConnection: close\r\n\r\n{\"ok\":true}") + .await + .unwrap(); + drop(stream); + let response = request.await.unwrap().unwrap(); + assert_eq!(response.status, reqwest::StatusCode::OK); + assert_eq!(response.body, br#"{"ok":true}"#); + }) + .await + .expect("loopback request did not complete"); + } + #[test] fn route_issues_workspace_scoped_request_proof() { let identity = RuntimeIdentityMaterial::generate("runtime-a").unwrap(); diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index fa9e918c5..1938f0b18 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -1017,14 +1017,45 @@ async fn journal_backend_grant_installs_bound_tools_and_seals_saved_finish_befor }) .unwrap(); let handle = spawn_controller(worker).await; + let mut events = handle.subscribe(); + let submission_request_id = protocol::new_submission_request_id(); handle .send(Method::submit_text( - protocol::new_submission_request_id(), + submission_request_id.clone(), "Record only admitted public input.", )) .await .unwrap(); - wait_for_status(&handle, WorkerStatus::Idle).await; + // Initial Idle (or the first captured model request) does not prove the + // submitted tool loop completed. The exact Submit teardown fence follows + // persist_turn, so coverage and the sealed result retry are committed. + // This is a deadlock watchdog, not a turn-latency performance assertion. + tokio::time::timeout(std::time::Duration::from_secs(2), async { + let mut saw_run_end = false; + loop { + match events + .recv() + .await + .expect("Journal controller event stream closed") + { + Event::RunEnd { result } => { + assert_eq!(result, protocol::RunResult::Finished); + saw_run_end = true; + } + Event::WorkerState { snapshot } + if saw_run_end + && snapshot.catalog_status() == WorkerStatus::Idle + && snapshot.last_finished_submission_request_id.as_deref() + == Some(submission_request_id.as_str()) => + { + break; + } + _ => {} + } + } + }) + .await + .expect("Journal Submit did not finish and commit its history"); let request = wait_for_captured_request(&captured).await; let tools = request_tool_names(&request); assert_eq!( diff --git a/crates/workspace-server/src/server/tests/protocol_backpressure_tests.rs b/crates/workspace-server/src/server/tests/protocol_backpressure_tests.rs index 06759a67e..e6e3ea742 100644 --- a/crates/workspace-server/src/server/tests/protocol_backpressure_tests.rs +++ b/crates/workspace-server/src/server/tests/protocol_backpressure_tests.rs @@ -1,6 +1,15 @@ use super::*; +// Keep the short transport/session-lock liveness bounds separate from full +// lifecycle cleanup. The 10s output/Close deadlines below are asserted exactly +// with paused time; these wall-clock limits are deadlock watchdogs, not the +// normal-suite performance budget. const PROTOCOL_TEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2); +// Stop also runs Host admission, Runtime persistence and linked-Workdir storage +// publication. The parallel normal run exceeded 2s there after transport had +// already terminated. Check lock release separately at 2s, but allow this real +// lifecycle path scheduling/I/O headroom without replacing it with a stub. +const LIFECYCLE_CLEANUP_WATCHDOG: std::time::Duration = std::time::Duration::from_secs(15); // Completion replies are produced synchronously by the actual attached execution. // Unlike flooding the fixture's 32-entry broadcast bus, this reaches the Server's @@ -289,8 +298,13 @@ async fn browser_method_queue_overflow_fails_fast_and_does_not_pin_session_clean ); assert_eq!(receiver.len(), WORKER_PROTOCOL_METHOD_CAPACITY - 1); + let session_lock = current_worker_session_lock(&api, &identity); + let guard = tokio::time::timeout(PROTOCOL_TEST_TIMEOUT, session_lock.lock()) + .await + .expect("full method queue retained the session guard"); + drop(guard); let stopped = tokio::time::timeout( - PROTOCOL_TEST_TIMEOUT, + LIFECYCLE_CLEANUP_WATCHDOG, worker.stop( &WorkerOperationContext::Backend, WorkerLifecycleRequest { @@ -636,6 +650,7 @@ fn assert_bounded_unknown_delivery_close(messages: &[WsMessage]) { } async fn assert_same_worker_reconnect_and_stop( + api: &WorkspaceApi, worker: &WorkspaceWorker, context: &WorkerOperationContext, ) { @@ -649,8 +664,13 @@ async fn assert_same_worker_reconnect_and_stop( .expect("same Worker reconnect did not deliver a fresh snapshot"), Some(protocol::Event::Snapshot { .. }) )); + let session_lock = current_worker_session_lock(api, worker.identity()); + let guard = tokio::time::timeout(PROTOCOL_TEST_TIMEOUT, session_lock.lock()) + .await + .expect("terminated transport retained same Worker's session guard"); + drop(guard); let stopped = tokio::time::timeout( - PROTOCOL_TEST_TIMEOUT, + LIFECYCLE_CLEANUP_WATCHDOG, worker.stop( &WorkerOperationContext::Backend, WorkerLifecycleRequest { @@ -780,7 +800,7 @@ async fn assert_stalled_browser_output_reclaimed(explicit_termination: bool) { // SQLite jobs run on external threads; paused Tokio time cannot observe their // progress and may auto-advance a reconnect timeout before SQL completes. tokio::time::resume(); - assert_same_worker_reconnect_and_stop(&worker, &context).await; + assert_same_worker_reconnect_and_stop(&api, &worker, &context).await; } #[tokio::test(start_paused = true)] @@ -930,5 +950,5 @@ async fn stalled_remote_bridge_browser_cancel_overflow_reclaims_wire_and_allows_ )); } assert!(execution.protocol_methods().is_empty()); - assert_same_worker_reconnect_and_stop(&worker, &context).await; + assert_same_worker_reconnect_and_stop(&api, &worker, &context).await; } diff --git a/crates/workspace-server/src/server_assignment_attention_tests.rs b/crates/workspace-server/src/server_assignment_attention_tests.rs index ee20767bc..2f5c36ca8 100644 --- a/crates/workspace-server/src/server_assignment_attention_tests.rs +++ b/crates/workspace-server/src/server_assignment_attention_tests.rs @@ -155,8 +155,17 @@ async fn orchestrator_attention_barrier_callers_share_unsent_reservation() { #[tokio::test] async fn orchestrator_attention_input_rechecks_reserved_noop_before_and_after_comparison() { + // This handshake holds a reservation across the real guarded Restore path + // (including durable intent/config/Runtime writes). Its release deadline is + // a deadlock watchdog, not a 5s Restore performance contract: the parallel + // normal run exceeded that old bound before it could release the caller. + let restore_watchdog = std::time::Duration::from_secs(15); for pause_before_comparison in [true, false] { - let (_dir, api, execution) = attention_test_api().await; + // This case only needs assigned Ticket metadata, not a checkout. Keep + // the real API/storage and configured target, but skip Git init/commit. + let _dir = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(_dir.path()).await; + attention_assign_ticket(&api, "Attention backlog"); let started = start_attention_test_orchestrator(&api).await; let worker = RuntimeWorkerRef::new( EMBEDDED_WORKER_RUNTIME_ID, @@ -192,8 +201,8 @@ async fn orchestrator_attention_input_rechecks_reserved_noop_before_and_after_co if checks == 0 && pause_before_comparison { entered_tx.send(()).unwrap(); release_rx - .recv_timeout(std::time::Duration::from_secs(5)) - .unwrap(); + .recv_timeout(restore_watchdog) + .expect("guarded Restore did not release the reserved no-op"); } check_orchestrator_assignment_attention( &api_for_caller, @@ -210,8 +219,8 @@ async fn orchestrator_attention_input_rechecks_reserved_noop_before_and_after_co ); entered_tx.send(()).unwrap(); release_rx - .recv_timeout(std::time::Duration::from_secs(5)) - .unwrap(); + .recv_timeout(restore_watchdog) + .expect("guarded Restore did not release the reserved no-op"); } checks += 1; assert!(attention_pending(&api_for_caller).is_empty()); diff --git a/crates/yoi/src/ticket_cli.rs b/crates/yoi/src/ticket_cli.rs index a8dd9d9d1..92e9c8c7c 100644 --- a/crates/yoi/src/ticket_cli.rs +++ b/crates/yoi/src/ticket_cli.rs @@ -1313,6 +1313,7 @@ mod tests { priority: "normal".into(), labels: vec![], created_at: None, + created_by: None, updated_at: None, assignee: None, readiness: None, @@ -1448,6 +1449,7 @@ mod tests { Ok(current.clone()) } unused_backend_methods! { + fn edit_post_with_snapshots(&self, id: TicketIdOrSlug, edit: ticket::TicketPostEdit) -> ticket::Result<(ticket::Ticket, ticket::Ticket)>; fn list_workspace_tags(&self, request: ticket::TagListRequest) -> ticket::Result>; fn mutate_workspace_tag(&self, request: ticket::WorkspaceTagMutation) -> ticket::Result; fn set_ticket_tags(&self, id: TicketIdOrSlug, request: ticket::TicketTagSetRequest) -> ticket::Result>;