fix: own Runtime management authority stages and commit publication workflows

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);
+        .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<Json<WorkspaceRuntimeDetail>> {
+    run_host_workflow(api.host_executors.host()?, async move {
         let binding = api
             .store
             .update_workspace_runtime_binding_metadata(
-            &path.workspace_id,
+                &workspace_id,
-            &path.runtime_id,
+                &runtime_id,
-            display_name,
+                &display_name,
-            endpoint.as_str().trim_end_matches('/'),
+                &endpoint,
                 &Utc::now().to_rfc3339(),
             )
             .await?;
-    api.runtime_binding_expectations
+        publish_workspace_runtime_binding(&api, binding)?;
-        .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?,
+            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<WorkspaceRuntimeBinding> {
+    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<WorkspaceApi>,
     Extension(actor): Extension<RequestActor>,
@@ -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
+    require_active_runtime_registration_identity(&api).await?;
-        .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",
-        ));
-    }
     // 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
+    scan_runtime_registration_workers(&api, &runtime_id).await?;
-        .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()),
-    }
     let now = Utc::now().to_rfc3339();
     let record = WorkspaceRuntimeBinding {
         workspace_id: api.config.workspace_id.clone(),
@@ -29183,7 +29246,12 @@ async fn create_remote_runtime(
         updated_at: now,
         revoked_at: None,
     };
-    let (mutation, binding) = api
+    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,
@@ -29191,22 +29259,19 @@ async fn create_remote_runtime(
                 &actor.account_id,
             )
             .await?;
-    api.runtime_binding_expectations
+        publish_workspace_runtime_binding(&mutation_api, binding)?;
-        .write()
+        let resource = workspace_runtime_resources_response_on_host(
-        .unwrap_or_else(std::sync::PoisonError::into_inner)
+            &mutation_api,
-        .insert(
+            mutation_api.workspace_id(),
-            (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))
+    let ping = api
-        .await
+        .host_executors
-        .map_err(|_| Error::RuntimeOperationFailed {
+        .host()?
-            runtime_id: runtime_id.clone(),
+        .run(move || Ok::<_, ApiError>(runtime.ping(&ping_runtime_id)))
-            code: "runtime_connection_test_unavailable".to_string(),
+        .await?;
-            message: "Runtime connection test could not be completed".to_string(),
-        })?;
 
-    // The ping runs outside the store lock. A replaced/revoked connection must
+    // The ping runs outside the store lock. Recheck the entire live binding on
-    // not be reported as an observation of the earlier binding.
+    // the independent DB owner, never acquire its mutex on the Tokio waiter.
-    if !api.store.workspace_runtime_binding_matches(&binding)? {
+    let binding = require_current_runtime_test_binding(&api, binding).await?;
-        return Err(Error::RuntimeBindingConflict(
-            "Runtime connection changed while the connection test was running".to_string(),
-        )
-        .into());
-    }
     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<WorkspaceApi>,
 ) -> ApiResult<Json<WorkerLaunchOptionsResponse>> {
-    api.store
+    // Runtime summaries can call back into DB signing/authority. Profile config
-        .db_executor()
+    // 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<server_api::WorkerListRespon
 async fn workspace_runtime_resources_response(
     api: &WorkspaceApi,
     workspace_id: &str,
+) -> ApiResult<server_api::WorkspaceRuntimeListResponse> {
+    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<server_api::WorkspaceRuntimeListResponse> {
     let limit = api.config.max_records.min(200);
-    let runtime = api.runtime.clone();
+    let runtimes = api.runtime.list_runtimes(limit);
-    let runtimes = api
-        .host_executors
-        .host()?
-        .run(move || Ok::<_, ApiError>(runtime.list_runtimes(limit)))
-        .await?;
     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<WorkspaceRuntimeDetail> {
+    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<WorkspaceRuntimeDetail> {
     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()

crates/workspace-server/src/server_db_execution_tests.rs

@@ -1189,7 +1189,259 @@ 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,
+        "scan-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 { scan_runtime_registration_workers(&api, "scan-runtime").await },
+    ));
+    let request = observed(reply.entered.take().unwrap()).await.unwrap();
+    assert!(request.contains("/v1/workers"), "{request}");
+    assert!(
+        !reply.upstream_released.load(Ordering::SeqCst),
+        "registration scan blocked Tokio until watchdog bailout"
+    );
+    let store = fixture.store.clone();
+    observed(
+        fixture
+            .executor
+            .run(move || store.get_workspace_memory_settings(WORKSPACE)),
+    )
+    .await
+    .unwrap();
+    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);
+    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(
@@ -1202,7 +1454,15 @@ 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 {
+                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}");
@@ -1237,6 +1497,7 @@ async fn runtime_resource_summary_waits_off_tokio_and_db_and_retains_cancelled_h
         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() {