From 925bce2a405b6d3255887fba88867ab8d23882dc Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 12 Oct 2026 11:24:12 +0900 Subject: [PATCH] fix: own Runtime management authority stages and commit publication workflows --- crates/workspace-server/src/server.rs | 259 +++++++++------ .../src/server_db_execution_tests.rs | 299 ++++++++++++++++-- 2 files changed, 451 insertions(+), 107 deletions(-) diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 2e7fb889a..becd6fd80 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -27608,30 +27608,44 @@ async fn scoped_update_remote_runtime( .as_deref() .map(str::trim) .filter(|value| !value.is_empty()) - .unwrap_or(&path.runtime_id); - let binding = api - .store - .update_workspace_runtime_binding_metadata( - &path.workspace_id, - &path.runtime_id, - display_name, - endpoint.as_str().trim_end_matches('/'), - &Utc::now().to_rfc3339(), - ) - .await?; - api.runtime_binding_expectations - .write() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .insert( - (path.workspace_id.clone(), path.runtime_id.clone()), - binding.clone(), - ); - api.runtime_subscription_broker - .unregister_runtime(&path.runtime_id); - api.register_workspace_runtime_binding(binding)?; - Ok(Json( - workspace_runtime_detail(&api, &path.workspace_id, &path.runtime_id).await?, - )) + .unwrap_or(&path.runtime_id) + .to_owned(); + update_remote_runtime_metadata_owned( + api, + path.workspace_id, + path.runtime_id, + display_name, + endpoint.as_str().trim_end_matches('/').to_owned(), + ) + .await +} + +// Post-auth mutation boundary: cancellation cannot split SQL completion from +// publication or discard the accepted workflow's request/resource authority. +async fn update_remote_runtime_metadata_owned( + api: WorkspaceApi, + workspace_id: String, + runtime_id: String, + display_name: String, + endpoint: String, +) -> ApiResult> { + run_host_workflow(api.host_executors.host()?, async move { + let binding = api + .store + .update_workspace_runtime_binding_metadata( + &workspace_id, + &runtime_id, + &display_name, + &endpoint, + &Utc::now().to_rfc3339(), + ) + .await?; + publish_workspace_runtime_binding(&api, binding)?; + Ok(Json( + workspace_runtime_detail_on_host(&api, &workspace_id, &runtime_id).await?, + )) + }) + .await } async fn scoped_reveal_runtime_trust_key( @@ -29116,6 +29130,75 @@ async fn list_workers( .await } +async fn require_active_runtime_registration_identity(api: &WorkspaceApi) -> ApiResult<()> { + let identities = api.signing_identities.clone(); + let workspace_id = api.config.workspace_id.clone(); + api.store + .db_executor() + .run(move || { + let identity = identities.get_validated(&workspace_id)?; + if identity.state != "active" { + return Err(settings_bad_request( + "workspace_signing_identity_unavailable", + "an active Workspace signing identity is required before registering a Runtime", + )); + } + Ok(()) + }) + .await +} + +async fn scan_runtime_registration_workers(api: &WorkspaceApi, runtime_id: &str) -> ApiResult<()> { + let runtime = api.runtime.clone(); + let runtime_id = runtime_id.to_owned(); + let limit = api.config.max_records.min(200); + api.host_executors.host()?.run(move || { + match runtime.list_workers_for_runtime(&runtime_id, limit) { + Ok(workers) if !workers.items.is_empty() => Err(Error::RuntimeBindingConflict(format!( + "Runtime `{runtime_id}` still has {} active worker(s) and cannot replace its connection", + workers.items.len() + )).into()), + Ok(_) | Err(RuntimeRegistryError::UnknownRuntime(_)) => Ok(()), + Err(error) => Err(error.into_error().into()), + } + }).await +} + +fn publish_workspace_runtime_binding( + api: &WorkspaceApi, + binding: WorkspaceRuntimeBinding, +) -> Result<()> { + api.runtime_binding_expectations + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert( + (binding.workspace_id.clone(), binding.runtime_id.clone()), + binding.clone(), + ); + api.runtime_subscription_broker + .unregister_runtime(&binding.runtime_id); + api.register_workspace_runtime_binding(binding) +} + +async fn require_current_runtime_test_binding( + api: &WorkspaceApi, + binding: WorkspaceRuntimeBinding, +) -> ApiResult { + let store = api.store.clone(); + api.store + .db_executor() + .run(move || { + if !store.workspace_runtime_binding_matches(&binding)? { + return Err(Error::RuntimeBindingConflict( + "Runtime connection changed while the connection test was running".to_string(), + ) + .into()); + } + Ok(binding) + }) + .await +} + async fn create_remote_runtime( State(api): State, Extension(actor): Extension, @@ -29136,33 +29219,13 @@ async fn create_remote_runtime( "the embedded Runtime is built in and cannot be managed as a remote Runtime", )); } - let identity = api - .signing_identities - .get_validated(&api.config.workspace_id)?; - if identity.state != "active" { - return Err(settings_bad_request( - "workspace_signing_identity_unavailable", - "an active Workspace signing identity is required before registering a Runtime", - )); - } + require_active_runtime_registration_identity(&api).await?; // Protect observed active Workers without removing the current connection. // This scan is not proof of idleness: an unavailable/auth-failing endpoint // must remain repairable by an owner using binding-ID CAS below. Replacement // neither stops nor deletes Workers; Runtime removal has separate preflight. // Rejected key/CAS mutations must leave the current registration usable. - match api - .runtime - .list_workers_for_runtime(&runtime_id, api.config.max_records.min(200)) - { - Ok(workers) if !workers.items.is_empty() => { - return Err(Error::RuntimeBindingConflict(format!( - "Runtime `{runtime_id}` still has {} active worker(s) and cannot replace its connection", - workers.items.len() - )).into()); - } - Ok(_) | Err(RuntimeRegistryError::UnknownRuntime(_)) => {} - Err(error) => return Err(error.into_error().into()), - } + scan_runtime_registration_workers(&api, &runtime_id).await?; let now = Utc::now().to_rfc3339(); let record = WorkspaceRuntimeBinding { workspace_id: api.config.workspace_id.clone(), @@ -29183,30 +29246,32 @@ async fn create_remote_runtime( updated_at: now, revoked_at: None, }; - let (mutation, binding) = api - .store - .put_workspace_runtime_binding_key( - record, - request.expected_binding_id.as_deref(), - &actor.account_id, + let mutation_api = api.clone(); + // One accepted owner keeps SQL commit, local registration and its response + // observation together even when the HTTP waiter disappears. Resource reads + // call the on-Host helper directly, never re-admit to the occupied Host lane. + let (mutation, resource) = run_host_workflow(api.host_executors.host()?, async move { + let (mutation, binding) = mutation_api + .store + .put_workspace_runtime_binding_key( + record, + request.expected_binding_id.as_deref(), + &actor.account_id, + ) + .await?; + publish_workspace_runtime_binding(&mutation_api, binding)?; + let resource = workspace_runtime_resources_response_on_host( + &mutation_api, + mutation_api.workspace_id(), ) - .await?; - api.runtime_binding_expectations - .write() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .insert( - (api.config.workspace_id.clone(), runtime_id.clone()), - binding.clone(), - ); - api.runtime_subscription_broker - .unregister_runtime(&runtime_id); - api.register_workspace_runtime_binding(binding)?; - let resource = workspace_runtime_resources_response(&api, &api.config.workspace_id) .await? .items .into_iter() .find(|resource| resource.runtime.runtime_id == runtime_id) .ok_or_else(|| Error::UnknownRuntime(runtime_id.clone()))?; + Ok::<_, ApiError>((mutation, resource)) + }) + .await?; let status = if mutation == WorkspaceRuntimeBindingMutation::Created { StatusCode::CREATED } else { @@ -29563,22 +29628,15 @@ async fn test_runtime_connection( let checked_at = Utc::now().to_rfc3339(); let runtime = api.runtime.clone(); let ping_runtime_id = runtime_id.clone(); - let ping = tokio::task::spawn_blocking(move || runtime.ping(&ping_runtime_id)) - .await - .map_err(|_| Error::RuntimeOperationFailed { - runtime_id: runtime_id.clone(), - code: "runtime_connection_test_unavailable".to_string(), - message: "Runtime connection test could not be completed".to_string(), - })?; + let ping = api + .host_executors + .host()? + .run(move || Ok::<_, ApiError>(runtime.ping(&ping_runtime_id))) + .await?; - // The ping runs outside the store lock. A replaced/revoked connection must - // not be reported as an observation of the earlier binding. - if !api.store.workspace_runtime_binding_matches(&binding)? { - return Err(Error::RuntimeBindingConflict( - "Runtime connection changed while the connection test was running".to_string(), - ) - .into()); - } + // The ping runs outside the store lock. Recheck the entire live binding on + // the independent DB owner, never acquire its mutex on the Tokio waiter. + let binding = require_current_runtime_test_binding(&api, binding).await?; let mut result = runtime_connection_test_response(api.workspace_id(), &runtime_id, checked_at, ping); // This result describes this attempt, not permission for a future request. @@ -29590,8 +29648,10 @@ async fn test_runtime_connection( async fn get_worker_launch_options( State(api): State, ) -> ApiResult> { - api.store - .db_executor() + // Runtime summaries can call back into DB signing/authority. Profile config + // reads are synchronous too, but this mixed helper must not occupy sole DB. + api.host_executors + .host()? .run(move || Ok(Json(worker_launch_options_response(&api)?))) .await } @@ -32161,14 +32221,23 @@ fn workers_response(api: WorkspaceApi) -> ApiResult ApiResult { + let api = api.clone(); + let workspace_id = workspace_id.to_owned(); + run_host_workflow(api.host_executors.host()?, async move { + workspace_runtime_resources_response_on_host(&api, &workspace_id).await + }) + .await +} + +// Already-owned Runtime workflows call this directly; re-admission to Host would +// wait behind the same parents. DB binding helpers still use their own DB owner. +async fn workspace_runtime_resources_response_on_host( + api: &WorkspaceApi, + workspace_id: &str, ) -> ApiResult { let limit = api.config.max_records.min(200); - let runtime = api.runtime.clone(); - let runtimes = api - .host_executors - .host()? - .run(move || Ok::<_, ApiError>(runtime.list_runtimes(limit))) - .await?; + let runtimes = api.runtime.list_runtimes(limit); let bindings = api .store .list_workspace_runtime_bindings(workspace_id, true) @@ -32267,12 +32336,26 @@ async fn workspace_runtime_detail( api: &WorkspaceApi, workspace_id: &str, runtime_id: &str, +) -> ApiResult { + let api = api.clone(); + let workspace_id = workspace_id.to_owned(); + let runtime_id = runtime_id.to_owned(); + run_host_workflow(api.host_executors.host()?, async move { + workspace_runtime_detail_on_host(&api, &workspace_id, &runtime_id).await + }) + .await +} + +async fn workspace_runtime_detail_on_host( + api: &WorkspaceApi, + workspace_id: &str, + runtime_id: &str, ) -> ApiResult { let binding = api .store .get_workspace_runtime_binding(workspace_id, runtime_id) .await?; - let mut resource = workspace_runtime_resources_response(api, workspace_id) + let mut resource = workspace_runtime_resources_response_on_host(api, workspace_id) .await? .items .into_iter() diff --git a/crates/workspace-server/src/server_db_execution_tests.rs b/crates/workspace-server/src/server_db_execution_tests.rs index cf6c84389..4d287776c 100644 --- a/crates/workspace-server/src/server_db_execution_tests.rs +++ b/crates/workspace-server/src/server_db_execution_tests.rs @@ -1189,12 +1189,12 @@ async fn subjektiv_runtime_waits_leave_db_and_control_free_and_cancel_retains_th } #[tokio::test(flavor = "current_thread")] -async fn runtime_resource_summary_waits_off_tokio_and_db_and_retains_cancelled_host_ownership() { +async fn runtime_registration_scan_keeps_sql_and_tokio_free_and_retains_cancelled_host_budget() { let fixture = Fixture::new().await; let mut reply = HeldRuntimeReply::new(); install_remote_fixture( &fixture, - "summary-runtime", + "scan-runtime", &format!("http://{}", reply.address), true, ); @@ -1202,42 +1202,303 @@ async fn runtime_resource_summary_waits_off_tokio_and_db_and_retains_cancelled_h let api = fixture.api.clone(); let task = tokio::spawn(crate::db_executor::with_operation_lifetime( Arc::new(CountJobLifetime(leases.clone())), - async move { workspace_runtime_resources_response(&api, WORKSPACE).await }, + async move { scan_runtime_registration_workers(&api, "scan-runtime").await }, )); let request = observed(reply.entered.take().unwrap()).await.unwrap(); - assert!(request.contains("/v1/runtime"), "{request}"); + assert!(request.contains("/v1/workers"), "{request}"); assert!( !reply.upstream_released.load(Ordering::SeqCst), - "upstream watchdog released before Tokio could observe the request" + "registration scan blocked Tokio until watchdog bailout" ); - let read_store = fixture.store.clone(); + let store = fixture.store.clone(); observed( fixture .executor - .run(move || read_store.get_workspace_memory_settings(WORKSPACE)), + .run(move || store.get_workspace_memory_settings(WORKSPACE)), ) .await .unwrap(); - assert_eq!( - observed(tokio::spawn(async { "control" })).await.unwrap(), - "control" - ); - observed(tokio::time::sleep(StdDuration::from_millis(1))).await; // timer contract only - assert!(!reply.upstream_released.load(Ordering::SeqCst)); - assert!(!task.is_finished()); + observed(tokio::time::sleep(StdDuration::from_millis(1))).await; + let mut queued_control = Box::pin(fixture.host_executor.run(|| Ok::<_, Error>(()))); + assert!(futures::poll!(&mut queued_control).is_pending()); + let rejected = scan_runtime_registration_workers(&fixture.api, "scan-runtime") + .await + .unwrap_err(); + response_json(rejected.into_response(), StatusCode::SERVICE_UNAVAILABLE).await; task.abort(); assert!(task.await.unwrap_err().is_cancelled()); - assert_eq!( - leases.load(Ordering::SeqCst), - 1, - "cancel released actual Runtime summary owner" - ); + assert_eq!(leases.load(Ordering::SeqCst), 1); + assert!(!reply.upstream_released.load(Ordering::SeqCst)); reply.finish(); + observed(queued_control).await.unwrap(); observed(fixture.host_executor.drain()).await.unwrap(); assert_eq!(leases.load(Ordering::SeqCst), 0); fixture.executor.drain().await.unwrap(); } +#[tokio::test(flavor = "current_thread")] +async fn runtime_management_authority_waits_off_tokio_and_retains_cancelled_db_admission() { + for stage in ["identity", "binding"] { + let fixture = Fixture::new().await; + let binding = + install_remote_fixture(&fixture, "authority-runtime", "http://127.0.0.1:9", true); + let (mut busy, waiting) = BusyObservation::install_store(fixture.store.clone()); + let mut lock = WriteLock::acquire(fixture.api.config.database_path.clone()); + let writer = fixture.store.clone(); + fixture + .executor + .dispatch(move || { + writer + .mark_api_token_used("missing-token", "2026-01-01T00:00:00Z") + .unwrap(); + }) + .unwrap(); + observed(waiting).await.unwrap(); + let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let api = fixture.api.clone(); + let expected = binding.clone(); + let mut authority = Box::pin(crate::db_executor::with_operation_lifetime( + Arc::new(CountJobLifetime(leases.clone())), + async move { + if stage == "identity" { + require_active_runtime_registration_identity(&api).await + } else { + require_current_runtime_test_binding(&api, expected) + .await + .map(|_| ()) + } + }, + )); + assert!( + futures::poll!(&mut authority).is_pending(), + "{stage}: authority must queue without blocking Tokio on the shared mutex" + ); + runtime_progress_while_locked(&lock).await; + let overload = if stage == "identity" { + require_active_runtime_registration_identity(&fixture.api) + .await + .unwrap_err() + } else { + require_current_runtime_test_binding(&fixture.api, binding.clone()) + .await + .unwrap_err() + }; + response_json(overload.into_response(), StatusCode::SERVICE_UNAVAILABLE).await; + drop(authority); + assert_eq!( + leases.load(Ordering::SeqCst), + 1, + "{stage}: cancelled DB authority lost its lease" + ); + lock.release(); + busy.resume(); + observed(fixture.executor.drain()).await.unwrap(); + assert_eq!(leases.load(Ordering::SeqCst), 0); + require_active_runtime_registration_identity(&fixture.api) + .await + .unwrap(); + assert_eq!( + require_current_runtime_test_binding(&fixture.api, binding.clone()) + .await + .unwrap() + .binding_id, + binding.binding_id + ); + fixture + .store + .revoke_workspace_runtime_binding( + WORKSPACE, + &binding.runtime_id, + "2026-01-02T00:00:00Z", + ) + .unwrap(); + let stale = require_current_runtime_test_binding(&fixture.api, binding) + .await + .unwrap_err(); + response_json(stale.into_response(), StatusCode::CONFLICT).await; + } +} + +#[tokio::test(flavor = "current_thread")] +async fn runtime_connection_post_ping_sql_waits_off_tokio_while_host_control_progresses() { + let fixture = Fixture::new().await; + let mut reply = HeldRuntimeReply::new(); + let binding = install_remote_fixture( + &fixture, + "ping-runtime", + &format!("http://{}", reply.address), + true, + ); + let mut test = Box::pin(test_runtime_connection( + State(fixture.api.clone()), + AxumPath(binding.runtime_id.clone()), + )); + assert!(futures::poll!(&mut test).is_pending()); + observed(fixture.executor.drain()).await.unwrap(); + assert!(futures::poll!(&mut test).is_pending()); + let request = observed(reply.entered.take().unwrap()).await.unwrap(); + assert!(request.contains("/ping"), "{request}"); + let (mut busy, waiting) = BusyObservation::install_store(fixture.store.clone()); + let mut lock = WriteLock::acquire(fixture.api.config.database_path.clone()); + let writer = fixture.store.clone(); + fixture + .executor + .dispatch(move || { + writer + .mark_api_token_used("missing-token", "2026-01-01T00:00:00Z") + .unwrap(); + }) + .unwrap(); + observed(waiting).await.unwrap(); + reply.finish(); + observed(fixture.host_executor.drain()).await.unwrap(); + assert!( + futures::poll!(&mut test).is_pending(), + "post-ping binding fence must await DB without a synchronous mutex wait" + ); + runtime_progress_while_locked(&lock).await; + assert_eq!( + observed(fixture.host_executor.run(|| Ok::<_, Error>("control"))) + .await + .unwrap(), + "control" + ); + lock.release(); + busy.resume(); + let Json(result) = observed(test).await.unwrap(); + assert_eq!(result.binding_id, binding.binding_id); + fixture.executor.drain().await.unwrap(); +} + +#[tokio::test(flavor = "current_thread")] +async fn cancelled_runtime_metadata_mutation_keeps_commit_and_registration_publication_owned() { + let fixture = Fixture::new().await; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + axum::serve( + listener, + Router::new().fallback(|| async { StatusCode::SERVICE_UNAVAILABLE }), + ) + .await + .unwrap(); + }); + install_remote_fixture(&fixture, "metadata-runtime", &endpoint, true); + let (mut busy, waiting) = BusyObservation::install_store(fixture.store.clone()); + let mut lock = WriteLock::acquire(fixture.api.config.database_path.clone()); + let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let mut update = Box::pin(crate::db_executor::with_operation_lifetime( + Arc::new(CountJobLifetime(leases.clone())), + update_remote_runtime_metadata_owned( + fixture.api.clone(), + WORKSPACE.into(), + "metadata-runtime".into(), + "updated metadata".into(), + endpoint, + ), + )); + assert!(futures::poll!(&mut update).is_pending()); + observed(waiting).await.unwrap(); // The real metadata SQL mutation is waiting. + runtime_progress_while_locked(&lock).await; + drop(update); + assert_eq!( + leases.load(Ordering::SeqCst), + 2, + "actual Host workflow and its accepted SQL both retain authority" + ); + lock.release(); + busy.resume(); + observed(fixture.host_executor.drain()).await.unwrap(); + observed(fixture.executor.drain()).await.unwrap(); + assert_eq!(leases.load(Ordering::SeqCst), 0); + let saved = fixture + .store + .get_workspace_runtime_binding(WORKSPACE, "metadata-runtime") + .unwrap() + .unwrap(); + assert_eq!(saved.display_name, "updated metadata"); + let published = fixture + .api + .runtime_binding_expectations + .read() + .unwrap() + .get(&(WORKSPACE.into(), "metadata-runtime".into())) + .cloned() + .unwrap(); + assert_eq!( + published, saved, + "SQL completion must publish even after waiter cancellation" + ); + fixture + .api + .runtime_subscription_broker + .unregister_runtime("metadata-runtime"); + fixture.api.worker_projection.shutdown_async().await; + fixture.executor.drain().await.unwrap(); + server.abort(); + let _ = server.await; +} + +#[tokio::test(flavor = "current_thread")] +async fn runtime_metadata_reads_wait_off_tokio_and_db_and_retain_cancelled_host_ownership() { + for read in ["resources", "launch_options"] { + let fixture = Fixture::new().await; + let mut reply = HeldRuntimeReply::new(); + install_remote_fixture( + &fixture, + "summary-runtime", + &format!("http://{}", reply.address), + true, + ); + let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let api = fixture.api.clone(); + let task = tokio::spawn(crate::db_executor::with_operation_lifetime( + Arc::new(CountJobLifetime(leases.clone())), + async move { + if read == "resources" { + workspace_runtime_resources_response(&api, WORKSPACE) + .await + .map(|_| ()) + } else { + get_worker_launch_options(State(api)).await.map(|_| ()) + } + }, + )); + let request = observed(reply.entered.take().unwrap()).await.unwrap(); + assert!(request.contains("/v1/runtime"), "{request}"); + assert!( + !reply.upstream_released.load(Ordering::SeqCst), + "upstream watchdog released before Tokio could observe the request" + ); + let read_store = fixture.store.clone(); + observed( + fixture + .executor + .run(move || read_store.get_workspace_memory_settings(WORKSPACE)), + ) + .await + .unwrap(); + assert_eq!( + observed(tokio::spawn(async { "control" })).await.unwrap(), + "control" + ); + observed(tokio::time::sleep(StdDuration::from_millis(1))).await; // timer contract only + assert!(!reply.upstream_released.load(Ordering::SeqCst)); + assert!(!task.is_finished()); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + assert_eq!( + leases.load(Ordering::SeqCst), + 1, + "cancel released actual Runtime summary owner" + ); + reply.finish(); + observed(fixture.host_executor.drain()).await.unwrap(); + assert_eq!(leases.load(Ordering::SeqCst), 0); + fixture.executor.drain().await.unwrap(); + } +} + #[tokio::test(flavor = "current_thread")] async fn worker_discovery_sql_waits_off_tokio_and_keeps_admission_until_actual_cleanup() { // Exercise the exact post-auth handler boundary: signed source verification