diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index b5d79eb8..f0dd332c 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -308,6 +308,7 @@ pub struct FsRuntimeStoreMigrationPlan { pub worker_count: usize, pub mapping_digest: String, pub mappings: Vec, + pub excluded_ephemeral_paths: Vec, } #[derive(Clone, Debug)] @@ -370,6 +371,7 @@ fn plan_v1_worker_identity( worker_count: 0, mapping_digest: legacy_worker_identity_mapping_digest(&[]), mappings: Vec::new(), + excluded_ephemeral_paths: Vec::new(), }; return Ok((plan, Vec::new())); } @@ -381,7 +383,7 @@ fn plan_v1_worker_identity( ), )); } - validate_runtime_tree_copyable(root)?; + let excluded_ephemeral_paths = runtime_tree_exclusions(root)?; let workers_dir = root.join(WORKERS_DIR); let mut entries = fs::read_dir(&workers_dir) @@ -476,6 +478,7 @@ fn plan_v1_worker_identity( worker_count: mappings.len(), mapping_digest: legacy_worker_identity_mapping_digest(&mappings), mappings, + excluded_ephemeral_paths, }; Ok((plan, planned)) } @@ -496,7 +499,54 @@ fn migration_sibling(root: &Path, suffix: &str) -> Result Ok(parent.join(format!(".{name}.{suffix}"))) } -fn validate_runtime_tree_copyable(source: &Path) -> Result<(), RuntimeError> { +fn runtime_ephemeral_socket(root: &Path, path: &Path) -> Result { + let relative = path.strip_prefix(root).map_err(|_| { + runtime_store_corrupt( + path, + format!("Runtime migration path escaped root {}", root.display()), + ) + })?; + let components = relative + .components() + .map(|component| component.as_os_str().to_string_lossy().into_owned()) + .collect::>(); + let known_path = components.len() == 5 + && components[0] == WORKERS_DIR + && components[1].parse::().is_ok() + && components[2] == "runs" + && components[3].parse::().is_ok() + && components[4] == "worker.sock"; + if !known_path { + return Ok(false); + } + #[cfg(unix)] + { + match std::os::unix::net::UnixStream::connect(path) { + Ok(_) => Err(runtime_store_corrupt( + path, + "Runtime migration found an active Worker socket; stop the legacy Runtime and Worker before migrating" + .to_string(), + )), + Err(error) + if matches!( + error.kind(), + std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound + ) => Ok(true), + Err(error) => Err(runtime_store_corrupt( + path, + format!("Runtime migration could not verify Worker socket liveness: {error}"), + )), + } + } + #[cfg(not(unix))] + Ok(false) +} + +fn collect_runtime_tree_exclusions( + root: &Path, + source: &Path, + excluded: &mut Vec, +) -> Result<(), RuntimeError> { let entries = fs::read_dir(source) .map_err(|error| runtime_io_error("read migration source", source, error))?; for entry in entries { @@ -507,18 +557,34 @@ fn validate_runtime_tree_copyable(source: &Path) -> Result<(), RuntimeError> { .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() { + collect_runtime_tree_exclusions(root, &source_path, excluded)?; + } else if file_type.is_file() { + } else if runtime_ephemeral_socket(root, &source_path)? { + excluded.push( + source_path + .strip_prefix(root) + .expect("validated Runtime migration path") + .to_string_lossy() + .into_owned(), + ); + } else { return Err(runtime_store_corrupt( &source_path, - "Runtime migration refuses symlinks and special files".to_string(), + "Runtime migration refuses unknown symlinks and special files".to_string(), )); } } Ok(()) } -fn copy_runtime_tree(source: &Path, target: &Path) -> Result<(), RuntimeError> { +fn runtime_tree_exclusions(root: &Path) -> Result, RuntimeError> { + let mut excluded = Vec::new(); + collect_runtime_tree_exclusions(root, root, &mut excluded)?; + excluded.sort(); + Ok(excluded) +} + +fn copy_runtime_tree(root: &Path, 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) @@ -533,14 +599,16 @@ fn copy_runtime_tree(source: &Path, target: &Path) -> Result<(), RuntimeError> { .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)?; + copy_runtime_tree(root, &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 if runtime_ephemeral_socket(root, &source_path)? { + continue; } else { return Err(runtime_store_corrupt( &source_path, - "Runtime migration refuses symlinks and special files".to_string(), + "Runtime migration refuses unknown symlinks and special files".to_string(), )); } } @@ -567,7 +635,7 @@ fn migrate_v1_worker_identity( ), )); } - if let Err(error) = copy_runtime_tree(root, &staging) { + if let Err(error) = copy_runtime_tree(root, root, &staging) { let _ = fs::remove_dir_all(&staging); return Err(error); } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 947f256d..f44492c5 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -4193,6 +4193,14 @@ mod tests { serde_json::to_vec_pretty(&runtime_json).unwrap(), ) .unwrap(); + #[cfg(unix)] + { + let run_dir = legacy_dir.join("runs").join("6"); + std::fs::create_dir_all(&run_dir).unwrap(); + let socket = + std::os::unix::net::UnixListener::bind(run_dir.join("worker.sock")).unwrap(); + drop(socket); + } let runtime_options = crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), @@ -4204,6 +4212,11 @@ mod tests { assert!(plan.migration_required); assert_eq!(plan.worker_count, 1); assert_eq!(plan.mappings[0].legacy_worker_id, 7); + #[cfg(unix)] + assert_eq!( + plan.excluded_ephemeral_paths, + vec!["workers/7/runs/6/worker.sock"] + ); assert_eq!( std::fs::read(&runtime_path).unwrap(), runtime_before_dry_run @@ -4215,7 +4228,10 @@ mod tests { let detail = restored.worker_detail(&WorkerRef::new(expected)).unwrap(); assert_eq!(detail.worker_id, expected); assert_eq!(detail.worker_ref.worker_id, expected); - assert!(root.join("workers").join(expected.to_string()).exists()); + let expected_worker_dir = root.join("workers").join(expected.to_string()); + assert!(expected_worker_dir.exists()); + #[cfg(unix)] + assert!(!expected_worker_dir.join("runs/6/worker.sock").exists()); assert!(!legacy_dir.exists()); let migrated_runtime: serde_json::Value = serde_json::from_slice(&std::fs::read(runtime_path).unwrap()).unwrap();