Merge remote-tracking branch 'origin/develop' into work/T-452-remove-repository-local-yoi

crates/worker-runtime/Cargo.toml

@@ -41,7 +41,7 @@ reqwest = { version = "0.13", optional = true, default-features = false, feature
 ring.workspace = true
 tar.workspace = true
 thiserror = { workspace = true }
-tokio = { workspace = true, features = ["net", "process", "rt", "sync", "time"] }
+tokio = { workspace = true, features = ["net", "process", "rt", "signal", "sync", "time"] }
 tracing.workspace = true
 tracing-subscriber.workspace = true
 toml.workspace = true

crates/worker-runtime/src/fs_store.rs

@@ -123,8 +123,8 @@ impl FsRuntimeStore {
         })?;
         let owner_lock = acquire_runtime_store_owner_lock(&root)?;
         recover_runtime_store_migration(&root)?;
-        let existed = root.exists();
-        if existed && !root.is_dir() {
+        let root_existed = root.exists();
+        if root_existed && !root.is_dir() {
             return Err(RuntimeError::StoreCorrupt {
                 operation: "open runtime store",
                 path: root,
@@ -132,17 +132,22 @@ impl FsRuntimeStore {
             });
         }
 
-        if !existed {
+        if !root_existed {
             fs::create_dir_all(&root).map_err(|source| RuntimeError::StoreIo {
                 operation: "create runtime store root",
                 path: root.clone(),
                 source,
             })?;
         }
+        let snapshot_exists = root.join(RUNTIME_FILE).is_file();
+        if root_existed && !snapshot_exists {
+            validate_fresh_runtime_store_layout(&root)?;
+        }
 
-        fs::create_dir_all(root.join(WORKERS_DIR)).map_err(|source| RuntimeError::StoreIo {
+        let workers_dir = root.join(WORKERS_DIR);
+        fs::create_dir_all(&workers_dir).map_err(|source| RuntimeError::StoreIo {
             operation: "create runtime store",
-            path: root.join(WORKERS_DIR),
+            path: workers_dir,
             source,
         })?;
         let legacy_events = root.join("events.jsonl");
@@ -158,14 +163,14 @@ impl FsRuntimeStore {
             }
         }
 
-        if existed {
+        if snapshot_exists {
             migrate_runtime_store(&root, runtime_id)?;
         }
         let store = Self {
             root,
             _owner_lock: Some(owner_lock),
         };
-        let state = if existed {
+        let state = if snapshot_exists {
             Some(store.load_runtime_state()?)
         } else {
             None
@@ -343,6 +348,38 @@ impl FsRuntimeStore {
     }
 }
 
+fn validate_fresh_runtime_store_layout(root: &Path) -> Result<(), RuntimeError> {
+    let entries = fs::read_dir(root).map_err(|source| RuntimeError::StoreIo {
+        operation: "inspect uninitialized runtime store",
+        path: root.to_path_buf(),
+        source,
+    })?;
+    for entry in entries {
+        let entry = entry.map_err(|source| RuntimeError::StoreIo {
+            operation: "inspect uninitialized runtime store",
+            path: root.to_path_buf(),
+            source,
+        })?;
+        let file_type = entry.file_type().map_err(|source| RuntimeError::StoreIo {
+            operation: "inspect uninitialized runtime store entry",
+            path: entry.path(),
+            source,
+        })?;
+        if entry.file_name() == "auth.toml" && file_type.is_file() {
+            continue;
+        }
+        return Err(RuntimeError::StoreCorrupt {
+            operation: "open runtime store",
+            path: root.to_path_buf(),
+            message: format!(
+                "unexpected persistence `{}` exists without a Runtime snapshot",
+                entry.file_name().to_string_lossy()
+            ),
+        });
+    }
+    Ok(())
+}
+
 fn acquire_runtime_store_owner_lock(
     root: &Path,
 ) -> Result<Arc<RuntimeStoreOwnerLock>, RuntimeError> {
@@ -2137,6 +2174,72 @@ mod tests {
         document["schema_version"].as_u64().unwrap()
     }
 
+    #[test]
+    fn auth_initialized_directory_without_snapshot_is_a_fresh_store() {
+        let parent = tempfile::tempdir().unwrap();
+        let root = parent.path().join("runtime");
+        fs::create_dir_all(&root).unwrap();
+        fs::write(root.join("auth.toml"), b"fixture auth material").unwrap();
+
+        let opened = FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap();
+
+        assert!(opened.state.is_none());
+        assert!(root.join(WORKERS_DIR).is_dir());
+        assert_eq!(
+            fs::read(root.join("auth.toml")).unwrap(),
+            b"fixture auth material"
+        );
+    }
+
+    #[test]
+    fn snapshotless_directory_rejects_non_auth_persistence() {
+        for unexpected in ["events.jsonl", "unknown.json"] {
+            let parent = tempfile::tempdir().unwrap();
+            let root = parent.path().join("runtime");
+            fs::create_dir_all(&root).unwrap();
+            fs::write(root.join("auth.toml"), b"fixture auth material").unwrap();
+            fs::write(root.join(unexpected), b"must remain").unwrap();
+
+            let error = FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap_err();
+
+            assert!(matches!(
+                error,
+                RuntimeError::StoreCorrupt { path, .. } if path == root
+            ));
+            assert_eq!(fs::read(root.join(unexpected)).unwrap(), b"must remain");
+            assert!(!root.join(WORKERS_DIR).exists());
+        }
+    }
+
+    #[test]
+    fn orphan_archive_without_runtime_snapshot_is_rejected() {
+        let parent = tempfile::tempdir().unwrap();
+        let root = parent.path().join("runtime");
+        fs::create_dir_all(root.join(ORPHANED_WORKERS_DIR).join("worker-a")).unwrap();
+
+        let error = FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap_err();
+
+        assert!(matches!(
+            error,
+            RuntimeError::StoreCorrupt { path, .. } if path == root
+        ));
+        assert!(root.join(ORPHANED_WORKERS_DIR).join("worker-a").is_dir());
+    }
+
+    #[test]
+    fn worker_records_without_runtime_snapshot_are_rejected() {
+        let parent = tempfile::tempdir().unwrap();
+        let root = parent.path().join("runtime");
+        fs::create_dir_all(root.join(WORKERS_DIR).join("orphan")).unwrap();
+
+        let error = FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap_err();
+
+        assert!(matches!(
+            error,
+            RuntimeError::StoreCorrupt { path, .. } if path == root
+        ));
+    }
+
     #[test]
     fn runtime_store_owner_lock_child_probe() {
         let Some(root) = std::env::var_os("YOI_TEST_RUNTIME_STORE_LOCK_ROOT") else {

crates/worker-runtime/src/http_server.rs

@@ -139,6 +139,22 @@ pub enum RuntimeHttpStoreSelection {
     },
 }
 
+pub async fn serve_runtime_http_with_shutdown<F>(
+    runtime: Runtime,
+    listener: TcpListener,
+    local_token: Option<String>,
+    shutdown: F,
+) -> Result<(), RuntimeHttpServerError>
+where
+    F: std::future::Future<Output = ()> + Send + 'static,
+{
+    let local_token = local_token.ok_or(RuntimeHttpServerError::AuthRequired)?;
+    axum::serve(listener, runtime_http_router(runtime, local_token))
+        .with_graceful_shutdown(shutdown)
+        .await?;
+    Ok(())
+}
+
 /// Serve an existing Runtime on a pre-bound listener.
 pub async fn serve_runtime_http(
     runtime: Runtime,
@@ -150,6 +166,25 @@ pub async fn serve_runtime_http(
     Ok(())
 }
 
+pub async fn serve_runtime_http_with_workspace_auth_shutdown<F>(
+    runtime: Runtime,
+    listener: TcpListener,
+    local_token: Option<String>,
+    workspace_auth: WorkspaceRuntimeHttpAuth,
+    shutdown: F,
+) -> Result<(), RuntimeHttpServerError>
+where
+    F: std::future::Future<Output = ()> + Send + 'static,
+{
+    axum::serve(
+        listener,
+        runtime_http_router_with_optional_auth(runtime, local_token, Some(workspace_auth)),
+    )
+    .with_graceful_shutdown(shutdown)
+    .await?;
+    Ok(())
+}
+
 pub async fn serve_runtime_http_with_workspace_auth(
     runtime: Runtime,
     listener: TcpListener,

crates/worker-runtime/src/main.rs

@@ -75,6 +75,23 @@ fn init_serve_tracing() {
         .try_init();
 }
 
+async fn shutdown_signal() {
+    #[cfg(unix)]
+    {
+        let mut terminate =
+            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+                .expect("install SIGTERM handler");
+        tokio::select! {
+            _ = tokio::signal::ctrl_c() => {},
+            _ = terminate.recv() => {},
+        }
+    }
+    #[cfg(not(unix))]
+    {
+        let _ = tokio::signal::ctrl_c().await;
+    }
+}
+
 fn run() -> Result<(), ProcessError> {
     let args = env::args().skip(1).collect::<Vec<_>>();
     if matches!(args.first().map(String::as_str), Some("migrate")) {
@@ -104,11 +121,12 @@ fn run() -> Result<(), ProcessError> {
             "yoi-runtime listening on {local_addr}; intended client is a trusted backend/proxy, not a browser"
         );
         let server = if let Some(workspace_auth) = workspace_http_auth {
-            worker_runtime::http_server::serve_runtime_http_with_workspace_auth(
+            worker_runtime::http_server::serve_runtime_http_with_workspace_auth_shutdown(
                 worker_runtime,
                 listener,
                 config.http.local_token,
                 workspace_auth,
+                shutdown_signal(),
             )
             .await
         } else {
@@ -117,10 +135,11 @@ fn run() -> Result<(), ProcessError> {
                     "Runtime HTTP server requires Workspace issuer auth or --local-token".to_owned(),
                 )
             })?;
-            worker_runtime::http_server::serve_runtime_http(
+            worker_runtime::http_server::serve_runtime_http_with_shutdown(
                 worker_runtime,
                 listener,
                 Some(local_token),
+                shutdown_signal(),
             )
             .await
         };