From d052cedc7d386d0ffe5e5172ac3513430a2d5f14 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 20 Aug 2026 07:21:03 +0900 Subject: [PATCH] fix: make worker identity migrations dry-runnable --- Cargo.toml | 2 +- crates/worker-runtime/src/fs_store.rs | 353 ++++++++++++++++++++--- crates/worker-runtime/src/identity.rs | 42 ++- crates/worker-runtime/src/main.rs | 109 ++++++- crates/worker-runtime/src/runtime.rs | 20 +- crates/workspace-server/src/main.rs | 63 +++- crates/workspace-server/src/retention.rs | 19 ++ crates/workspace-server/src/store.rs | 149 +++++++++- 8 files changed, 705 insertions(+), 52 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index ad9a86a8..ea432561 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -109,7 +109,7 @@ serde = "1.0" serde_json = "1.0" serde_yaml = "0.9.34" tar = "0.4" -rusqlite = { version = "0.37", features = ["bundled"] } +rusqlite = { version = "0.37", features = ["backup", "bundled"] } ring = "0.17.14" sha2 = "0.11" tempfile = "3.27" diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index fc676607..b5d79eb8 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -2,7 +2,9 @@ use crate::catalog::{CreateWorkerRequest, WorkingDirectoryStatus}; use crate::config_bundle::ConfigBundle; use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic}; use crate::error::RuntimeError; -use crate::identity::{WorkerId, WorkerRef}; +use crate::identity::{ + LegacyWorkerIdentityMapping, WorkerId, WorkerRef, legacy_worker_identity_mapping_digest, +}; use crate::management::{RuntimeBackendKind, RuntimeStatus}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; @@ -52,6 +54,18 @@ pub struct FsRuntimeStore { } impl FsRuntimeStore { + pub fn migration_plan( + options: &FsRuntimeStoreOptions, + ) -> Result { + plan_v1_worker_identity(&options.root, &options.runtime_id).map(|(plan, _)| plan) + } + + pub fn migrate( + options: &FsRuntimeStoreOptions, + ) -> Result { + migrate_v1_worker_identity(&options.root, &options.runtime_id) + } + pub fn root(&self) -> &Path { &self.root } @@ -286,11 +300,36 @@ fn runtime_store_corrupt(path: &Path, message: String) -> RuntimeError { } } -fn migrate_v1_worker_identity(root: &Path, runtime_id: &str) -> Result<(), RuntimeError> { +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct FsRuntimeStoreMigrationPlan { + pub current_schema_version: u32, + pub target_schema_version: u32, + pub migration_required: bool, + pub worker_count: usize, + pub mapping_digest: String, + pub mappings: Vec, +} + +#[derive(Clone, Debug)] +struct PlannedRuntimeWorkerMigration { + mapping: LegacyWorkerIdentityMapping, + legacy_dir: PathBuf, +} + +fn plan_v1_worker_identity( + root: &Path, + runtime_id: &str, +) -> Result< + ( + FsRuntimeStoreMigrationPlan, + Vec, + ), + RuntimeError, +> { let runtime_path = root.join(RUNTIME_FILE); let bytes = fs::read(&runtime_path).map_err(|error| runtime_io_error("read", &runtime_path, error))?; - let mut document: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| { + let document: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| { runtime_store_corrupt( &runtime_path, format!("decode Runtime state {}: {error}", runtime_path.display()), @@ -305,10 +344,36 @@ fn migrate_v1_worker_identity(root: &Path, runtime_id: &str) -> Result<(), Runti "Runtime state is missing schema_version".to_string(), ) })?; - if schema_version == u64::from(SCHEMA_VERSION) { - return Ok(()); + let current_schema_version = u32::try_from(schema_version).map_err(|_| { + runtime_store_corrupt( + &runtime_path, + format!("Runtime store schema version {schema_version} is out of range"), + ) + })?; + let staging = migration_sibling(root, "schema-v2-staging")?; + let backup = migration_sibling(root, "schema-v1-backup")?; + if staging.exists() || backup.exists() { + return Err(runtime_store_corrupt( + root, + format!( + "unfinished Runtime migration artifact exists (staging={}, backup={})", + staging.display(), + backup.display() + ), + )); } - if schema_version != 1 { + if current_schema_version == SCHEMA_VERSION { + let plan = FsRuntimeStoreMigrationPlan { + current_schema_version, + target_schema_version: SCHEMA_VERSION, + migration_required: false, + worker_count: 0, + mapping_digest: legacy_worker_identity_mapping_digest(&[]), + mappings: Vec::new(), + }; + return Ok((plan, Vec::new())); + } + if current_schema_version != 1 { return Err(runtime_store_corrupt( &runtime_path, format!( @@ -316,30 +381,256 @@ fn migrate_v1_worker_identity(root: &Path, runtime_id: &str) -> Result<(), Runti ), )); } + validate_runtime_tree_copyable(root)?; - let legacy_ids = document - .get("workers") - .and_then(serde_json::Value::as_array) - .ok_or_else(|| { + let workers_dir = root.join(WORKERS_DIR); + let mut entries = fs::read_dir(&workers_dir) + .map_err(|error| runtime_io_error("read workers", &workers_dir, error))? + .collect::, _>>() + .map_err(|error| runtime_io_error("read workers", &workers_dir, error))?; + entries.sort_by_key(|entry| entry.file_name()); + let mut planned = Vec::with_capacity(entries.len()); + let mut target_ids = std::collections::BTreeSet::new(); + for entry in entries { + let legacy_dir = entry.path(); + if !legacy_dir.is_dir() { + return Err(runtime_store_corrupt( + &legacy_dir, + "legacy workers directory contains a non-directory entry".to_string(), + )); + } + let name = entry.file_name(); + let name = name.to_str().ok_or_else(|| { runtime_store_corrupt( - &runtime_path, - "Runtime state workers must be an array".to_string(), + &legacy_dir, + "legacy Worker directory is not UTF-8".to_string(), ) - })? - .iter() - .map(|value| { - value.as_u64().ok_or_else(|| { + })?; + let legacy_worker_id = name.parse::().map_err(|_| { + runtime_store_corrupt( + &legacy_dir, + format!("legacy Worker directory name must be numeric, found {name}"), + ) + })?; + let snapshot_path = legacy_dir.join(WORKER_FILE); + let snapshot: serde_json::Value = read_json(&snapshot_path, "read legacy worker snapshot")?; + let workspace_id = snapshot + .get("workspace_id") + .and_then(serde_json::Value::as_str) + .filter(|workspace_id| !workspace_id.is_empty()) + .ok_or_else(|| { runtime_store_corrupt( - &runtime_path, - "legacy Worker id must be unsigned".to_string(), + &snapshot_path, + "legacy Worker snapshot is missing workspace_id; unscoped Workers require an explicit migration disposition" + .to_string(), ) - }) - }) - .collect::, _>>()?; + })? + .to_string(); + let worker_id = WorkerId::from_legacy_binding(&workspace_id, runtime_id, legacy_worker_id); + if !target_ids.insert(worker_id) { + return Err(runtime_store_corrupt( + &snapshot_path, + format!("legacy Worker identity maps to duplicate target {worker_id}"), + )); + } + let target_dir = workers_dir.join(worker_id.to_string()); + if target_dir.exists() && target_dir != legacy_dir { + return Err(runtime_store_corrupt( + &target_dir, + format!("target Worker directory {worker_id} already exists"), + )); + } + let request = snapshot + .get("request") + .and_then(serde_json::Value::as_object) + .ok_or_else(|| { + runtime_store_corrupt( + &snapshot_path, + "Worker snapshot request must be an object".to_string(), + ) + })?; + if request.get("profile").is_none() { + return Err(runtime_store_corrupt( + &snapshot_path, + "Worker snapshot request is missing profile".to_string(), + )); + } + planned.push(PlannedRuntimeWorkerMigration { + mapping: LegacyWorkerIdentityMapping { + workspace_id, + runtime_id: runtime_id.to_string(), + legacy_worker_id, + worker_id, + }, + legacy_dir, + }); + } + let mappings = planned + .iter() + .map(|worker| worker.mapping.clone()) + .collect::>(); + let plan = FsRuntimeStoreMigrationPlan { + current_schema_version, + target_schema_version: SCHEMA_VERSION, + migration_required: true, + worker_count: mappings.len(), + mapping_digest: legacy_worker_identity_mapping_digest(&mappings), + mappings, + }; + Ok((plan, planned)) +} - let mut migrated_ids = Vec::with_capacity(legacy_ids.len()); - for legacy_id in legacy_ids { - let legacy_dir = root.join("workers").join(legacy_id.to_string()); +fn migration_sibling(root: &Path, suffix: &str) -> Result { + let parent = root.parent().ok_or_else(|| { + runtime_store_corrupt( + root, + "Runtime store root has no parent directory".to_string(), + ) + })?; + let name = root + .file_name() + .and_then(|name| name.to_str()) + .ok_or_else(|| { + runtime_store_corrupt(root, "Runtime store root name is not UTF-8".to_string()) + })?; + Ok(parent.join(format!(".{name}.{suffix}"))) +} + +fn validate_runtime_tree_copyable(source: &Path) -> Result<(), RuntimeError> { + let entries = fs::read_dir(source) + .map_err(|error| runtime_io_error("read migration source", source, error))?; + for entry in entries { + let entry = + entry.map_err(|error| runtime_io_error("read migration source", source, error))?; + let source_path = entry.path(); + let file_type = entry + .file_type() + .map_err(|error| runtime_io_error("inspect migration source", &source_path, error))?; + if file_type.is_dir() { + validate_runtime_tree_copyable(&source_path)?; + } else if !file_type.is_file() { + return Err(runtime_store_corrupt( + &source_path, + "Runtime migration refuses symlinks and special files".to_string(), + )); + } + } + Ok(()) +} + +fn copy_runtime_tree(source: &Path, target: &Path) -> Result<(), RuntimeError> { + fs::create_dir(target) + .map_err(|error| runtime_io_error("create migration staging", target, error))?; + let mut entries = fs::read_dir(source) + .map_err(|error| runtime_io_error("read migration source", source, error))? + .collect::, _>>() + .map_err(|error| runtime_io_error("read migration source", source, error))?; + entries.sort_by_key(|entry| entry.file_name()); + for entry in entries { + let source_path = entry.path(); + let target_path = target.join(entry.file_name()); + let file_type = entry + .file_type() + .map_err(|error| runtime_io_error("inspect migration source", &source_path, error))?; + if file_type.is_dir() { + copy_runtime_tree(&source_path, &target_path)?; + } else if file_type.is_file() { + fs::copy(&source_path, &target_path) + .map_err(|error| runtime_io_error("copy migration source", &source_path, error))?; + } else { + return Err(runtime_store_corrupt( + &source_path, + "Runtime migration refuses symlinks and special files".to_string(), + )); + } + } + Ok(()) +} + +fn migrate_v1_worker_identity( + root: &Path, + runtime_id: &str, +) -> Result { + let (plan, _) = plan_v1_worker_identity(root, runtime_id)?; + if !plan.migration_required { + return Ok(plan); + } + let staging = migration_sibling(root, "schema-v2-staging")?; + let backup = migration_sibling(root, "schema-v1-backup")?; + if staging.exists() || backup.exists() { + return Err(runtime_store_corrupt( + root, + format!( + "unfinished Runtime migration artifact exists (staging={}, backup={}); recover or remove it before retrying", + staging.display(), + backup.display() + ), + )); + } + if let Err(error) = copy_runtime_tree(root, &staging) { + let _ = fs::remove_dir_all(&staging); + return Err(error); + } + let staged_plan = match migrate_v1_worker_identity_in_place(&staging, runtime_id) { + Ok(plan) => plan, + Err(error) => { + let _ = fs::remove_dir_all(&staging); + return Err(error); + } + }; + let staged_store = FsRuntimeStore { + root: staging.clone(), + }; + if let Err(error) = staged_store.load_runtime_state() { + let _ = fs::remove_dir_all(&staging); + return Err(error); + } + fs::rename(root, &backup) + .map_err(|error| runtime_io_error("backup runtime store", root, error))?; + if let Err(error) = fs::rename(&staging, root) { + let rollback = fs::rename(&backup, root); + return match rollback { + Ok(()) => Err(runtime_io_error( + "activate migrated runtime store", + &staging, + error, + )), + Err(rollback_error) => Err(runtime_store_corrupt( + root, + format!( + "activate migrated Runtime store failed: {error}; rollback failed: {rollback_error}; backup remains at {}", + backup.display() + ), + )), + }; + } + fs::remove_dir_all(&backup) + .map_err(|error| runtime_io_error("remove runtime migration backup", &backup, error))?; + debug_assert_eq!(plan.mapping_digest, staged_plan.mapping_digest); + Ok(plan) +} + +fn migrate_v1_worker_identity_in_place( + root: &Path, + runtime_id: &str, +) -> Result { + let (plan, planned_workers) = plan_v1_worker_identity(root, runtime_id)?; + if !plan.migration_required { + return Ok(plan); + } + let runtime_path = root.join(RUNTIME_FILE); + let bytes = + fs::read(&runtime_path).map_err(|error| runtime_io_error("read", &runtime_path, error))?; + let mut document: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| { + runtime_store_corrupt( + &runtime_path, + format!("decode Runtime state {}: {error}", runtime_path.display()), + ) + })?; + + for planned_worker in &planned_workers { + let legacy_id = planned_worker.mapping.legacy_worker_id; + let legacy_dir = &planned_worker.legacy_dir; let legacy_snapshot_path = legacy_dir.join(WORKER_FILE); let bytes = fs::read(&legacy_snapshot_path) .map_err(|error| runtime_io_error("read", &legacy_snapshot_path, error))?; @@ -352,12 +643,8 @@ fn migrate_v1_worker_identity(root: &Path, runtime_id: &str) -> Result<(), Runti ), ) })?; - let workspace_id = snapshot - .get("workspace_id") - .and_then(serde_json::Value::as_str) - .unwrap_or("local") - .to_string(); - let worker_id = WorkerId::from_legacy_binding(&workspace_id, runtime_id, legacy_id); + let workspace_id = planned_worker.mapping.workspace_id.clone(); + let worker_id = planned_worker.mapping.worker_id; let worker_id_text = worker_id.to_string(); snapshot["schema_version"] = serde_json::Value::from(SCHEMA_VERSION); snapshot["worker_id"] = serde_json::Value::String(worker_id_text.clone()); @@ -394,19 +681,19 @@ fn migrate_v1_worker_identity(root: &Path, runtime_id: &str) -> Result<(), Runti &snapshot, "migrate Worker identity", )?; - migrated_ids.push(serde_json::Value::String(worker_id_text)); } document["schema_version"] = serde_json::Value::from(SCHEMA_VERSION); - document["workers"] = serde_json::Value::Array(migrated_ids); if let Some(object) = document.as_object_mut() { + object.remove("workers"); object.remove("next_worker_sequence"); } atomic_write_json( &runtime_path, &document, "migrate Runtime Worker identities", - ) + )?; + Ok(plan) } #[derive(Clone, Debug, Serialize, Deserialize)] diff --git a/crates/worker-runtime/src/identity.rs b/crates/worker-runtime/src/identity.rs index 78133811..c16fa7c3 100644 --- a/crates/worker-runtime/src/identity.rs +++ b/crates/worker-runtime/src/identity.rs @@ -1,4 +1,5 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; +use sha2::{Digest, Sha256}; use std::{fmt, str::FromStr}; use uuid::{Uuid, Version}; @@ -27,8 +28,6 @@ impl WorkerId { } pub fn from_legacy_binding(workspace_id: &str, runtime_id: &str, value: u64) -> Self { - use sha2::{Digest, Sha256}; - let mut hasher = Sha256::new(); hasher.update(b"yoi.workspace-worker-id.v1\0"); hasher.update(workspace_id.as_bytes()); @@ -101,6 +100,45 @@ impl fmt::Display for WorkerIdParseError { impl std::error::Error for WorkerIdParseError {} +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct LegacyWorkerIdentityMapping { + pub workspace_id: String, + pub runtime_id: String, + pub legacy_worker_id: u64, + pub worker_id: WorkerId, +} + +pub fn legacy_worker_identity_mapping_digest(mappings: &[LegacyWorkerIdentityMapping]) -> String { + let mut mappings = mappings.to_vec(); + mappings.sort_by(|left, right| { + ( + left.workspace_id.as_str(), + left.runtime_id.as_str(), + left.legacy_worker_id, + left.worker_id, + ) + .cmp(&( + right.workspace_id.as_str(), + right.runtime_id.as_str(), + right.legacy_worker_id, + right.worker_id, + )) + }); + let mut hasher = Sha256::new(); + hasher.update(b"yoi.workspace-worker-migration-plan.v1\0"); + for mapping in mappings { + hasher.update(mapping.workspace_id.as_bytes()); + hasher.update([0]); + hasher.update(mapping.runtime_id.as_bytes()); + hasher.update([0]); + hasher.update(mapping.legacy_worker_id.to_be_bytes()); + hasher.update(mapping.worker_id.to_string().as_bytes()); + hasher.update([b'\n']); + } + let digest = hasher.finalize(); + digest.iter().map(|byte| format!("{byte:02x}")).collect() +} + /// Runtime-local authority reference for Worker operations. The contained id is /// nevertheless the Workspace-owned stable identity; the Runtime does not mint it. #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index 117f0987..9abc266e 100644 --- a/crates/worker-runtime/src/main.rs +++ b/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::FsRuntimeStoreOptions; +use worker_runtime::fs_store::{FsRuntimeStore, FsRuntimeStoreOptions}; use worker_runtime::http_server::{ RuntimeHttpServerConfig, RuntimeHttpServerError, RuntimeHttpStoreSelection, }; @@ -44,6 +44,9 @@ fn main() -> ExitCode { fn run() -> Result<(), ProcessError> { let args = env::args().skip(1).collect::>(); + if matches!(args.first().map(String::as_str), Some("migrate")) { + return run_migration_command(args); + } if matches!( args.first().map(String::as_str), Some("identity" | "trust-server") @@ -78,6 +81,76 @@ fn run() -> Result<(), ProcessError> { Ok(()) } +fn run_migration_command(mut args: Vec) -> Result<(), ProcessError> { + args.remove(0); + let dry_run_index = args + .iter() + .position(|argument| argument == "--dry-run") + .ok_or_else(|| ProcessError::usage("migrate currently requires --dry-run".to_string()))?; + args.remove(dry_run_index); + let explicit_runtime_id = + if let Some(index) = args.iter().position(|argument| argument == "--runtime-id") { + if index + 1 >= args.len() { + return Err(ProcessError::usage( + "--runtime-id requires a value".to_string(), + )); + } + let runtime_id = args.remove(index + 1); + args.remove(index); + Some(runtime_id) + } else { + None + }; + let config = parse_args(args)?.ok_or_else(|| { + ProcessError::usage("migrate requires a Runtime store configuration".to_string()) + })?; + let root = match &config.http.store { + RuntimeHttpStoreSelection::Fs { root } => root.clone(), + RuntimeHttpStoreSelection::Memory => { + return Err(ProcessError::usage( + "migration dry-run requires the fs Runtime store".to_string(), + )); + } + _ => { + return Err(ProcessError::usage( + "unsupported Runtime catalog store selection".to_string(), + )); + } + }; + let persisted_runtime_id = read_runtime_auth_file(&runtime_auth_path(&config))? + .identity + .map(|identity| identity.identity_id); + let runtime_id = match (persisted_runtime_id, explicit_runtime_id) { + (Some(persisted), Some(explicit)) if persisted != explicit => { + return Err(ProcessError::usage(format!( + "--runtime-id {explicit} does not match persisted Runtime identity {persisted}" + ))); + } + (Some(persisted), _) => persisted, + (None, Some(explicit)) if !explicit.is_empty() => explicit, + (None, Some(_)) => { + return Err(ProcessError::usage( + "--runtime-id must not be empty".to_string(), + )); + } + (None, None) => { + return Err(ProcessError::usage( + "migration dry-run requires a persisted Runtime identity or explicit --runtime-id" + .to_string(), + )); + } + }; + let mut options = FsRuntimeStoreOptions::new(root).with_runtime_id(runtime_id); + options.display_name = config.http.display_name.clone(); + let plan = FsRuntimeStore::migration_plan(&options).map_err(ProcessError::Runtime)?; + println!( + "{}", + serde_json::to_string_pretty(&plan) + .map_err(|error| ProcessError::Auth(format!("encode migration plan: {error}")))? + ); + Ok(()) +} + fn build_runtime(config: &ProcessConfig) -> Result { let fs_paths = config.resolved_fs_paths(); let runtime_store_dir = match &config.http.store { @@ -791,6 +864,7 @@ fn run_trust_server_command(mut args: VecDeque) -> Result<(), ProcessErr fn usage() -> &'static str { r#"Usage: yoi-runtime [OPTIONS] + yoi-runtime migrate --dry-run [--runtime-id ] [OPTIONS] Starts a worker-backed Runtime REST command API for a trusted backend/proxy. Browsers must not connect to this Runtime process directly. @@ -911,6 +985,39 @@ mod tests { assert_eq!(paths.workdir_target, PathBuf::from("/tmp/yoi-workdirs")); } + #[test] + fn migration_dry_run_accepts_real_v1_document_without_workers_field() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + std::fs::create_dir_all(root.join("workers")).unwrap(); + std::fs::write( + root.join("runtime.json"), + serde_json::to_vec_pretty(&serde_json::json!({ + "schema_version": 1, + "assignments": [], + "execution": [], + "diagnostics": [] + })) + .unwrap(), + ) + .unwrap(); + let before = std::fs::read(root.join("runtime.json")).unwrap(); + run_migration_command(vec![ + "migrate".to_string(), + "--dry-run".to_string(), + "--runtime-id".to_string(), + "local".to_string(), + "--store".to_string(), + "fs".to_string(), + "--fs-root".to_string(), + temp.path().display().to_string(), + "--fs-runtime-dir".to_string(), + root.display().to_string(), + ]) + .unwrap(); + assert_eq!(std::fs::read(root.join("runtime.json")).unwrap(), before); + } + #[test] fn no_store_disables_runtime_catalog_persistence() { let config = parse_args(["--no-store"]).unwrap().unwrap(); diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index b45abbe7..947f256d 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -4186,7 +4186,7 @@ mod tests { let mut runtime_json: serde_json::Value = serde_json::from_slice(&std::fs::read(&runtime_path).unwrap()).unwrap(); runtime_json["schema_version"] = serde_json::json!(1); - runtime_json["workers"] = serde_json::json!([7]); + runtime_json["workers"] = serde_json::json!({"legacy": "ignored"}); runtime_json["next_worker_sequence"] = serde_json::json!(8); std::fs::write( &runtime_path, @@ -4194,12 +4194,23 @@ mod tests { ) .unwrap(); - let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { + let runtime_options = crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), runtime_id: runtime_id.to_string(), display_name: None, - }) - .unwrap(); + }; + let runtime_before_dry_run = std::fs::read(&runtime_path).unwrap(); + let plan = crate::fs_store::FsRuntimeStore::migration_plan(&runtime_options).unwrap(); + assert!(plan.migration_required); + assert_eq!(plan.worker_count, 1); + assert_eq!(plan.mappings[0].legacy_worker_id, 7); + assert_eq!( + std::fs::read(&runtime_path).unwrap(), + runtime_before_dry_run + ); + assert!(legacy_dir.exists()); + + let restored = Runtime::with_fs_store(runtime_options).unwrap(); let expected = WorkerId::from_legacy_binding("workspace-a", runtime_id, 7); let detail = restored.worker_detail(&WorkerRef::new(expected)).unwrap(); assert_eq!(detail.worker_id, expected); @@ -4209,6 +4220,7 @@ mod tests { let migrated_runtime: serde_json::Value = serde_json::from_slice(&std::fs::read(runtime_path).unwrap()).unwrap(); assert_eq!(migrated_runtime["schema_version"], serde_json::json!(2)); + assert!(migrated_runtime.get("workers").is_none()); assert!(migrated_runtime.get("next_worker_sequence").is_none()); drop(restored); diff --git a/crates/workspace-server/src/main.rs b/crates/workspace-server/src/main.rs index 7d09a1bd..77a59e4f 100644 --- a/crates/workspace-server/src/main.rs +++ b/crates/workspace-server/src/main.rs @@ -23,6 +23,7 @@ enum Command { ConfigDiff(WorkspacePathOptions), Identity(Vec), TrustRuntime(Vec), + MigrateDryRun { database: Option }, Skills(SkillsCommand), Help, } @@ -85,6 +86,17 @@ async fn run() -> Result<(), Box> { Command::ConfigDiff(options) => run_config_diff(options), Command::Identity(args) => run_identity_command(args), Command::TrustRuntime(args) => run_trust_runtime_command(args), + Command::MigrateDryRun { database } => { + let database = database.unwrap_or_else(ServerConfig::default_server_database_path); + let plan = SqliteWorkspaceStore::migration_plan(&database).map_err(|error| { + CliError(format!( + "migration dry-run failed for {}: {error}", + database.display() + )) + })?; + println!("{}", serde_json::to_string_pretty(&plan)?); + Ok(()) + } Command::Skills(command) => run_skills(command), Command::Help => Ok(()), } @@ -107,6 +119,7 @@ fn parse_command(args: &[String]) -> Result { "config" => parse_config_command(rest), "identity" => Ok(Command::Identity(rest.to_vec())), "trust-runtime" => Ok(Command::TrustRuntime(rest.to_vec())), + "migrate" => parse_migrate_command(rest), "skills" => parse_skills_command(rest), "serve" => { if rest.iter().any(|arg| arg == "--help" || arg == "-h") { @@ -120,7 +133,7 @@ fn parse_command(args: &[String]) -> Result { Ok(Command::Help) } other => Err(CliError(format!( - "unknown command `{other}`; expected `init`, `config`, `identity`, `trust-runtime`, `skills`, or `serve`" + "unknown command `{other}`; expected `init`, `config`, `identity`, `trust-runtime`, `migrate`, `skills`, or `serve`" ))), } } @@ -718,6 +731,32 @@ fn parse_config_command(args: &[String]) -> Result { } } +fn parse_migrate_command(args: &[String]) -> Result { + let mut dry_run = false; + let mut database = None; + let mut index = 0; + while index < args.len() { + match args[index].as_str() { + "--dry-run" => dry_run = true, + "--database" => { + index += 1; + database = + Some(PathBuf::from(args.get(index).ok_or_else(|| { + CliError("--database requires a path".to_string()) + })?)); + } + value => { + return Err(CliError(format!("unknown migrate option: {value}"))); + } + } + index += 1; + } + if !dry_run { + return Err(CliError("migrate currently requires --dry-run".to_string())); + } + Ok(Command::MigrateDryRun { database }) +} + fn parse_skills_command(args: &[String]) -> Result { let Some((subcommand, rest)) = args.split_first() else { print_skills_help(); @@ -875,7 +914,8 @@ fn parse_listen(value: &str) -> Result { fn print_help() { println!( - "yoi-server\n\nUsage:\n yoi-server init [OPTIONS]\n yoi-server config [OPTIONS]\n yoi-server identity init --server-id [--replace]\n yoi-server identity show [--json]\n yoi-server trust-runtime add --runtime-id --base-url --public-key [--display-name ] [--replace]\n yoi-server trust-runtime list [--json] [--include-revoked]\n yoi-server trust-runtime revoke --runtime-id \n yoi-server skills [OPTIONS]\n yoi-server serve [OPTIONS]\n\nOptions:\n -h, --help Print help" + "yoi-server\n\nUsage:\n yoi-server init [OPTIONS]\n yoi-server config [OPTIONS]\n yoi-server identity init --server-id [--replace]\n yoi-server identity show [--json]\n yoi-server trust-runtime add --runtime-id --base-url --public-key [--display-name ] [--replace]\n yoi-server trust-runtime list [--json] [--include-revoked]\n yoi-server trust-runtime revoke --runtime-id \n yoi-server skills [OPTIONS]\n yoi-server migrate --dry-run [--database ] + yoi-server serve [OPTIONS]\n\nOptions:\n -h, --help Print help" ); } @@ -899,7 +939,8 @@ fn print_skills_help() { fn print_serve_help() { println!( - "yoi-server serve\n\nUsage:\n yoi-server serve [OPTIONS]\n\nDescription:\n Serves the Workspace recorded in the Yoi server DB. Workspace records are stored in the XDG/Yoi data directory, and runtime sources are loaded from XDG runtimes.toml.\n\nOptions:\n --listen Listen address (default 127.0.0.1:8787)\n -h, --help Print help" + "yoi-server serve\n\nUsage:\n yoi-server migrate --dry-run [--database ] + yoi-server serve [OPTIONS]\n\nDescription:\n Serves the Workspace recorded in the Yoi server DB. Workspace records are stored in the XDG/Yoi data directory, and runtime sources are loaded from XDG runtimes.toml.\n\nOptions:\n --listen Listen address (default 127.0.0.1:8787)\n -h, --help Print help" ); } @@ -939,6 +980,22 @@ mod tests { assert_eq!(name, "debug-rust"); } + #[test] + fn parse_migrate_requires_dry_run_and_accepts_database_path() { + let error = parse_migrate_command(&[]).unwrap_err(); + assert_eq!(error.to_string(), "migrate currently requires --dry-run"); + let command = parse_migrate_command(&[ + "--dry-run".to_string(), + "--database".to_string(), + "/tmp/server.db".to_string(), + ]) + .unwrap(); + let Command::MigrateDryRun { database } = command else { + panic!("expected migration dry-run command"); + }; + assert_eq!(database, Some(PathBuf::from("/tmp/server.db"))); + } + #[test] fn parse_serve_accepts_listen_only() { let args = vec!["--listen".to_string(), "127.0.0.1:0".to_string()]; diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index f57787dd..3b34a6db 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -166,6 +166,25 @@ pub enum WorkerRetentionError { Invalid(String), } +pub(crate) fn repair_worker_diagnostics_archive_table(conn: &Connection) -> crate::Result { + let existed: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='worker_diagnostics_archives')", + [], + |row| row.get(0), + )?; + if !existed { + conn.execute_batch( + "CREATE TABLE worker_diagnostics_archives ( + operation_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL, + committed_at TEXT NOT NULL, expires_at TEXT NOT NULL, + FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id), + FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE);", + )?; + } + Ok(!existed) +} + pub(crate) fn create_worker_retention_tables(conn: &Connection) -> crate::Result<()> { conn.execute_batch(r#" CREATE TABLE workspace_worker_retention_policy_revisions ( diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index fd1c73ad..6fd3e48b 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -4,11 +4,15 @@ use std::time::Duration; use async_trait::async_trait; use flow::{CompiledFlowDefinition, FlowSourceKind, compile_flow_source}; -use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params}; +use rusqlite::{ + Connection, OpenFlags, OptionalExtension, TransactionBehavior, backup::Backup, params, +}; use serde::{Deserialize, Serialize}; use uuid::Uuid; -use worker_runtime::identity::{RuntimeWorkerRef, WorkerId}; +use worker_runtime::identity::{ + LegacyWorkerIdentityMapping, RuntimeWorkerRef, WorkerId, legacy_worker_identity_mapping_digest, +}; use crate::{Error, Result}; @@ -204,7 +208,7 @@ const MIGRATIONS: &[Migration] = &[ Migration { version: 37, name: "promote Workspace Worker UUIDv7 identity", - apply: promote_workspace_worker_uuid_identity, + apply: apply_workspace_worker_uuid_identity_migration, }, Migration { version: 38, @@ -219,6 +223,17 @@ struct Migration { apply: fn(&Connection) -> Result<()>, } +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct WorkspaceStoreMigrationPlan { + pub current_schema_version: i64, + pub target_schema_version: i64, + pub migration_required: bool, + pub worker_count: usize, + pub mapping_digest: String, + pub mappings: Vec, + pub repairs: Vec, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkspaceRecord { pub workspace_id: String, @@ -976,6 +991,66 @@ pub struct SqliteWorkspaceStore { } impl SqliteWorkspaceStore { + pub fn migration_plan(path: impl AsRef) -> Result { + let path = path.as_ref(); + let source = Connection::open_with_flags( + path, + OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI, + )?; + let current_schema_version = current_schema_version(&source)?; + let target_schema_version = MIGRATIONS + .last() + .map(|migration| i64::from(migration.version)) + .unwrap_or(current_schema_version); + let mut repairs = Vec::new(); + if current_schema_version < 37 && !table_exists(&source, "worker_diagnostics_archives")? { + repairs.push("create missing worker_diagnostics_archives table".to_string()); + } + + let mut candidate = Connection::open_in_memory()?; + { + let backup = Backup::new(&source, &mut candidate)?; + backup.run_to_completion(5, Duration::from_millis(10), None)?; + } + configure_sqlite(&candidate)?; + let mappings = if current_schema_version < 37 { + apply_migrations_through(&candidate, 36)?; + let tx = candidate.unchecked_transaction()?; + crate::retention::repair_worker_diagnostics_archive_table(&tx)?; + let mappings = promote_workspace_worker_uuid_identity(&tx)?; + tx.execute( + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (37, ?1)", + ["promote Workspace Worker UUIDv7 identity"], + )?; + tx.commit()?; + mappings + } else { + Vec::new() + }; + apply_migrations_through(&candidate, i64::MAX)?; + ticket::migrate_sqlite_ticket_schema(&candidate)?; + merge_request::migrate(&candidate).map_err(|error| Error::Store(error.to_string()))?; + validate_workspace_repository_references(&candidate)?; + let foreign_key_failures: i64 = + candidate.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| { + row.get(0) + })?; + if foreign_key_failures != 0 { + return Err(Error::Store(format!( + "migration dry-run found {foreign_key_failures} foreign key violation(s)" + ))); + } + Ok(WorkspaceStoreMigrationPlan { + current_schema_version, + target_schema_version, + migration_required: current_schema_version < target_schema_version, + worker_count: mappings.len(), + mapping_digest: legacy_worker_identity_mapping_digest(&mappings), + mappings, + repairs, + }) + } + pub fn open(path: impl AsRef) -> Result { let conn = Connection::open(path)?; Self::from_connection(conn) @@ -5310,7 +5385,13 @@ fn collect_legacy_text_worker_bindings( Ok(()) } -fn promote_workspace_worker_uuid_identity(conn: &Connection) -> Result<()> { +fn apply_workspace_worker_uuid_identity_migration(conn: &Connection) -> Result<()> { + promote_workspace_worker_uuid_identity(conn).map(|_| ()) +} + +fn promote_workspace_worker_uuid_identity( + conn: &Connection, +) -> Result> { conn.execute_batch( r#" PRAGMA defer_foreign_keys = ON; @@ -5367,7 +5448,9 @@ fn promote_workspace_worker_uuid_identity(conn: &Connection) -> Result<()> { collect_legacy_text_worker_bindings(conn, table, &mut legacy_workers)?; } - for (workspace_id, runtime_id, runtime_worker_id) in legacy_workers { + let mut mappings = Vec::with_capacity(legacy_workers.len()); + for (workspace_id, runtime_id, runtime_worker_id) in &legacy_workers { + let worker_id = WorkerId::from_legacy_binding(workspace_id, runtime_id, *runtime_worker_id); conn.execute( "INSERT INTO worker_identity_v37(\ workspace_id, runtime_id, runtime_worker_id, worker_id\ @@ -5376,10 +5459,15 @@ fn promote_workspace_worker_uuid_identity(conn: &Connection) -> Result<()> { workspace_id, runtime_id, runtime_worker_id, - WorkerId::from_legacy_binding(&workspace_id, &runtime_id, runtime_worker_id) - .to_string() + worker_id.to_string() ], )?; + mappings.push(LegacyWorkerIdentityMapping { + workspace_id: workspace_id.clone(), + runtime_id: runtime_id.clone(), + legacy_worker_id: *runtime_worker_id, + worker_id, + }); } conn.execute_batch( @@ -5581,7 +5669,7 @@ fn promote_workspace_worker_uuid_identity(conn: &Connection) -> Result<()> { DROP TABLE worker_identity_v37; "#, )?; - Ok(()) + Ok(mappings) } fn allocate_resource_human_key( @@ -5852,6 +5940,9 @@ pub(crate) fn apply_migrations_through(conn: &Connection, through_version: i64) i64::from(migration.version) > current && i64::from(migration.version) <= through_version }) { let tx = conn.unchecked_transaction()?; + if migration.version == 37 { + crate::retention::repair_worker_diagnostics_archive_table(&tx)?; + } (migration.apply)(&tx)?; tx.execute( "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", @@ -6369,6 +6460,48 @@ mod tests { .unwrap(); } + #[test] + fn migration_dry_run_repairs_missing_diagnostics_archive_without_mutating_source() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("server.db"); + { + let conn = Connection::open(&path).unwrap(); + configure_sqlite(&conn).unwrap(); + apply_migrations_through(&conn, 36).unwrap(); + conn.execute_batch( + "DROP TABLE worker_diagnostics_archives; + INSERT INTO workspaces(workspace_id, display_name, state, created_at, updated_at) + VALUES ('workspace-a', 'Workspace A', 'active', '1', '1'); + INSERT INTO worker_registry( + workspace_id, runtime_id, runtime_worker_id, display_name, + retention_state, created_at, updated_at + ) VALUES ('workspace-a', 'runtime-a', 7, 'Worker 7', 'normal', '1', '1');", + ) + .unwrap(); + } + let before = std::fs::read(&path).unwrap(); + let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap(); + assert_eq!(plan.current_schema_version, 36); + assert_eq!(plan.target_schema_version, 38); + assert!(plan.migration_required); + assert_eq!(plan.worker_count, 1); + assert_eq!(plan.mappings[0].legacy_worker_id, 7); + assert_eq!( + plan.repairs, + vec!["create missing worker_diagnostics_archives table"] + ); + assert_eq!(std::fs::read(&path).unwrap(), before); + + let store = SqliteWorkspaceStore::open(&path).unwrap(); + store + .with_conn(|conn| { + assert!(table_exists(conn, "worker_diagnostics_archives")?); + assert_eq!(current_schema_version(conn)?, 38); + Ok(()) + }) + .unwrap(); + } + #[test] fn v38_backfills_workspace_scoped_objective_and_worker_human_keys() { let conn = Connection::open_in_memory().unwrap();