Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: make worker identity migrations dry-runnable

Browse Source
This commit is contained in:
Hare
2026-08-20 07:21:03 +09:00
parent de72afd9a1
commit d052cedc7d
8 changed files with 705 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1 -1
View File
@@ -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"
crates/worker-runtime/src/fs_store.rs
+320 -33
View File
@@ -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<FsRuntimeStoreMigrationPlan, RuntimeError> {
plan_v1_worker_identity(&options.root, &options.runtime_id).map(|(plan, _)| plan)
}
pub fn migrate(
options: &FsRuntimeStoreOptions,
) -> Result<FsRuntimeStoreMigrationPlan, RuntimeError> {
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<LegacyWorkerIdentityMapping>,
}
#[derive(Clone, Debug)]
struct PlannedRuntimeWorkerMigration {
mapping: LegacyWorkerIdentityMapping,
legacy_dir: PathBuf,
}
fn plan_v1_worker_identity(
root: &Path,
runtime_id: &str,
) -> Result<
(
FsRuntimeStoreMigrationPlan,
Vec<PlannedRuntimeWorkerMigration>,
),
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::<Result<Vec<_>, _>>()
.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::<u64>().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::<Result<Vec<_>, _>>()?;
})?
.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::<Vec<_>>();
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<PathBuf, RuntimeError> {
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::<Result<Vec<_>, _>>()
.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<FsRuntimeStoreMigrationPlan, RuntimeError> {
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<FsRuntimeStoreMigrationPlan, RuntimeError> {
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)]
crates/worker-runtime/src/identity.rs
+40 -2
View File
@@ -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)]
crates/worker-runtime/src/main.rs
+108 -1
View File
@@ -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::<Vec<_>>();
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<String>) -> 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<Runtime, ProcessError> {
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<String>) -> Result<(), ProcessErr
fn usage() -> &'static str {
r#"Usage: yoi-runtime [OPTIONS]
yoi-runtime migrate --dry-run [--runtime-id <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();
crates/worker-runtime/src/runtime.rs
+16 -4
View File
@@ -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);
crates/workspace-server/src/main.rs
+60 -3
View File
@@ -23,6 +23,7 @@ enum Command {
ConfigDiff(WorkspacePathOptions),
Identity(Vec<String>),
TrustRuntime(Vec<String>),
MigrateDryRun { database: Option<PathBuf> },
Skills(SkillsCommand),
Help,
}
@@ -85,6 +86,17 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
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<Command, CliError> {
"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<Command, CliError> {
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<Command, CliError> {
}
}
fn parse_migrate_command(args: &[String]) -> Result<Command, CliError> {
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<Command, CliError> {
let Some((subcommand, rest)) = args.split_first() else {
print_skills_help();
@@ -875,7 +914,8 @@ fn parse_listen(value: &str) -> Result<SocketAddr, CliError> {
fn print_help() {
println!(
"yoi-server\n\nUsage:\n yoi-server init [OPTIONS]\n yoi-server config <COMMAND> [OPTIONS]\n yoi-server identity init --server-id <SERVER_ID> [--replace]\n yoi-server identity show [--json]\n yoi-server trust-runtime add --runtime-id <RUNTIME_ID> --base-url <URL> --public-key <KEY> [--display-name <NAME>] [--replace]\n yoi-server trust-runtime list [--json] [--include-revoked]\n yoi-server trust-runtime revoke --runtime-id <RUNTIME_ID>\n yoi-server skills <COMMAND> [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 <COMMAND> [OPTIONS]\n yoi-server identity init --server-id <SERVER_ID> [--replace]\n yoi-server identity show [--json]\n yoi-server trust-runtime add --runtime-id <RUNTIME_ID> --base-url <URL> --public-key <KEY> [--display-name <NAME>] [--replace]\n yoi-server trust-runtime list [--json] [--include-revoked]\n yoi-server trust-runtime revoke --runtime-id <RUNTIME_ID>\n yoi-server skills <COMMAND> [OPTIONS]\n yoi-server migrate --dry-run [--database <PATH>]
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 <ADDR> 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 <PATH>]
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 <ADDR> 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()];
crates/workspace-server/src/retention.rs
+19
View File
@@ -166,6 +166,25 @@ pub enum WorkerRetentionError {
Invalid(String),
}
pub(crate) fn repair_worker_diagnostics_archive_table(conn: &Connection) -> crate::Result<bool> {
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 (
crates/workspace-server/src/store.rs
+141 -8
View File
@@ -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<LegacyWorkerIdentityMapping>,
pub repairs: Vec<String>,
}
#[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<Path>) -> Result<WorkspaceStoreMigrationPlan> {
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<Path>) -> Result<Self> {
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<Vec<LegacyWorkerIdentityMapping>> {
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();