fix: exclude stale worker sockets from migration

2026-08-20 07:51:34 +09:00
parent 53ec914a52
commit 9d003a5c98
2 changed files with 94 additions and 10 deletions
@@ -308,6 +308,7 @@ pub struct FsRuntimeStoreMigrationPlan {
    pub worker_count: usize,
    pub mapping_digest: String,
    pub mappings: Vec<LegacyWorkerIdentityMapping>,
+    pub excluded_ephemeral_paths: Vec<String>,
 }

 #[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<PathBuf, RuntimeError>
    Ok(parent.join(format!(".{name}.{suffix}")))
 }

-fn validate_runtime_tree_copyable(source: &Path) -> Result<(), RuntimeError> {
+fn runtime_ephemeral_socket(root: &Path, path: &Path) -> Result<bool, RuntimeError> {
+    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::<Vec<_>>();
+    let known_path = components.len() == 5
+        && components[0] == WORKERS_DIR
+        && components[1].parse::<u64>().is_ok()
+        && components[2] == "runs"
+        && components[3].parse::<u64>().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<String>,
+) -> 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<Vec<String>, 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);
    }
@@ -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();