runtime: restore persisted workers without adapter panic

crates/worker-runtime/src/main.rs

@@ -18,7 +18,7 @@ use worker_runtime::auth::{
     RuntimeHttpAuthConfig, RuntimeIdentityMaterial, TrustedServerKey, decode_public_key,
 };
 use worker_runtime::error::RuntimeError;
-use worker_runtime::fs_store::{FsRuntimeStore, FsRuntimeStoreOptions};
+use worker_runtime::fs_store::FsRuntimeStoreOptions;
 use worker_runtime::http_server::{
     RuntimeHttpServerConfig, RuntimeHttpServerError, RuntimeHttpStoreSelection,
 };
@@ -112,12 +112,6 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
                 .map_err(ProcessError::Runtime)
         }
         RuntimeHttpStoreSelection::Fs { root } => {
-            FsRuntimeStore::migrate_legacy_worker_aggregates(
-                root,
-                fs_paths.worker_dir.join("sessions"),
-                fs_paths.worker_dir.join("metadata"),
-            )
-            .map_err(ProcessError::Runtime)?;
             let mut options = FsRuntimeStoreOptions::new(root.clone());
             options.display_name = config.http.display_name.clone();
             Runtime::with_fs_store_and_execution_backend(options, backend)

crates/worker-runtime/src/worker_backend.rs

@@ -1831,21 +1831,25 @@ mod tests {
 
         let before_restart =
             backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None);
+        let adapter = WorkerRuntimeExecutionBackend::new(FailingFactory).unwrap();
+        let (after_restore_kind, after_restore_workspace_id) = adapter
+            .run_on_adapter_runtime(async move {
                 let after_restore =
                     backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None);
+                let client = after_restore.client_handle();
+                Ok((
+                    client.kind().to_string(),
+                    client.workspace_id().map(str::to_string),
+                ))
+            })
+            .expect("restore must reconstruct its Workspace client inside the adapter Runtime");
 
         assert_eq!(
             before_restart.client_handle().kind(),
             "runtime-owned-workspace-client"
         );
-        assert_eq!(
+        assert_eq!(after_restore_kind, "runtime-owned-workspace-client");
-            after_restore.client_handle().kind(),
+        assert_eq!(after_restore_workspace_id.as_deref(), Some("workspace-a"));
-            "runtime-owned-workspace-client"
-        );
-        assert_eq!(
-            after_restore.client_handle().workspace_id(),
-            Some("workspace-a")
-        );
     }
 
     #[test]
@@ -2388,14 +2392,18 @@ mod tests {
             workspace_id: "workspace-restore".to_string(),
             base_url: "http://workspace.invalid".to_string(),
         });
+        let identity = RuntimeIdentityMaterial::generate("runtime-restore").unwrap();
         let controller = ProfileRuntimeWorkerFactory::new(root.path())
-            .with_runtime_id("runtime-restore")
+            .with_remote_worker_mutation_identity(identity)
             .with_runtime_store_dir(&runtime_store_dir)
             .restore_controller(WorkerExecutionRestoreRequest {
                 worker_ref: worker_ref.clone(),
                 run_generation: 1,
                 request,
-                workspace_scope: None,
+                workspace_scope: Some(crate::runtime::RuntimeWorkspaceScope::new(
+                    "workspace-restore",
+                    "server-main",
+                )),
                 context: test_execution_context(worker_ref),
                 previous_working_directory: None,
                 working_directory: None,

crates/worker-runtime/src/worker_source.rs

@@ -133,7 +133,6 @@ pub trait EmbeddedWorkerMutationDispatcher: Send + Sync {
 enum RuntimeWorkerMutationTransport {
     Remote {
         base_url: String,
-        client: reqwest::blocking::Client,
     },
     Embedded {
         dispatcher: Arc<dyn EmbeddedWorkerMutationDispatcher>,
@@ -161,7 +160,6 @@ impl RuntimeWorkerMutationForwarder {
             source_worker_id: source_worker_id.into(),
             transport: RuntimeWorkerMutationTransport::Remote {
                 base_url: base_url.into().trim_end_matches('/').to_string(),
-                client: reqwest::blocking::Client::new(),
             },
         }
     }
@@ -199,33 +197,17 @@ impl RuntimeWorkerMutationForwarder {
         )?;
         match (&self.transport, proof) {
             (
-                RuntimeWorkerMutationTransport::Remote { base_url, client },
+                RuntimeWorkerMutationTransport::Remote { base_url },
                 RuntimeOwnedWorkerMutationProof::Remote(token),
-            ) => {
+            ) => execute_remote_worker_remove_http(RemoteWorkerRemoveHttpRequest {
-                let url = format!(
+                base_url: base_url.clone(),
-                    "{base_url}/api/w/{}/workers/remove",
+                workspace_id: self.scope.workspace_id.clone(),
-                    self.scope.workspace_id
+                token,
-                );
+                target_runtime_id: target_runtime_id.to_string(),
-                let body = serde_json::json!({
+                target_worker_id: target_worker_id.to_string(),
-                    "target_runtime_id": target_runtime_id,
+                expected_worker_revision: expected_worker_revision.to_string(),
-                    "target_worker_id": target_worker_id,
+                reason: reason.to_string(),
-                    "expected_worker_revision": expected_worker_revision,
+            }),
-                    "reason": reason,
-                });
-                let response = client
-                    .post(url)
-                    .header(crate::auth::WORKER_MUTATION_SOURCE_PROOF_HEADER, token)
-                    .json(&body)
-                    .send()
-                    .map_err(|error| {
-                        RuntimeWorkerMutationForwardError::Transport(error.to_string())
-                    })?;
-                let status = response.status().as_u16();
-                let body = response.text().map_err(|error| {
-                    RuntimeWorkerMutationForwardError::Transport(error.to_string())
-                })?;
-                Ok(WorkspaceResponse { status, body })
-            }
             (
                 RuntimeWorkerMutationTransport::Embedded { dispatcher },
                 RuntimeOwnedWorkerMutationProof::InProcess(claims),
@@ -241,6 +223,69 @@ impl RuntimeWorkerMutationForwarder {
     }
 }
 
+struct RemoteWorkerRemoveHttpRequest {
+    base_url: String,
+    workspace_id: String,
+    token: String,
+    target_runtime_id: String,
+    target_worker_id: String,
+    expected_worker_revision: String,
+    reason: String,
+}
+
+fn execute_remote_worker_remove_http(
+    request: RemoteWorkerRemoveHttpRequest,
+) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
+    if tokio::runtime::Handle::try_current().is_ok() {
+        return std::thread::Builder::new()
+            .name("yoi-worker-mutation-http".to_string())
+            .spawn(move || execute_remote_worker_remove_http_blocking(request))
+            .map_err(|error| {
+                RuntimeWorkerMutationForwardError::Transport(format!(
+                    "failed to start Worker mutation HTTP thread: {error}"
+                ))
+            })?
+            .join()
+            .map_err(|_| {
+                RuntimeWorkerMutationForwardError::Transport(
+                    "Worker mutation HTTP thread panicked".to_string(),
+                )
+            })?;
+    }
+
+    execute_remote_worker_remove_http_blocking(request)
+}
+
+fn execute_remote_worker_remove_http_blocking(
+    request: RemoteWorkerRemoveHttpRequest,
+) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
+    let url = format!(
+        "{}/api/w/{}/workers/remove",
+        request.base_url, request.workspace_id
+    );
+    let body = serde_json::json!({
+        "target_runtime_id": request.target_runtime_id,
+        "target_worker_id": request.target_worker_id,
+        "expected_worker_revision": request.expected_worker_revision,
+        "reason": request.reason,
+    });
+    let client = reqwest::blocking::Client::new();
+    let response = client
+        .post(url)
+        .header(
+            crate::auth::WORKER_MUTATION_SOURCE_PROOF_HEADER,
+            request.token,
+        )
+        .json(&body)
+        .send()
+        .map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
+    let status = response.status().as_u16();
+    let body = response
+        .text()
+        .map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
+    Ok(WorkspaceResponse { status, body })
+}
+
 pub struct RuntimeOwnedWorkspaceClient {
     workspace_id: String,
     base_url: String,
@@ -489,8 +534,8 @@ mod tests {
         );
     }
 
-    #[test]
+    #[tokio::test(flavor = "multi_thread")]
-    fn remote_forwarder_stamps_signed_proof_inside_runtime_before_http_delivery() {
+    async fn remote_forwarder_is_safe_in_async_runtime_and_stamps_signed_proof() {
         use std::io::{Read, Write};
         use std::net::TcpListener;
         use std::sync::Mutex;

crates/workspace-server/src/server.rs

@@ -709,23 +709,6 @@ impl WorkspaceApi {
             .await?;
         import_configured_repositories(store.as_ref(), &config)?;
         config.repositories = load_configured_repositories_from_store(store.as_ref(), &config)?;
-        let default_embedded_root =
-            ServerConfig::default_embedded_runtime_store_root(&config.workspace_id);
-        if config.embedded_runtime_store_root == default_embedded_root
-            && let (Some(legacy_sessions), Some(data_dir)) =
-                (manifest::paths::sessions_dir(), manifest::paths::data_dir())
-        {
-            worker_runtime::fs_store::FsRuntimeStore::migrate_legacy_worker_aggregates(
-                &config.embedded_runtime_store_root,
-                legacy_sessions,
-                data_dir.join("workers"),
-            )
-            .map_err(|error| {
-                crate::Error::Store(format!(
-                    "failed to migrate embedded Runtime Worker aggregates: {error}"
-                ))
-            })?;
-        }
         let embedded_runtime = EmbeddedWorkerRuntime::new_fs_store_with_execution_backend(
             config.workspace_id.clone(),
             config.embedded_runtime_store_root.clone(),