Add three-way file root apply

crates/geth-node/src/lib.rs

@@ -7917,6 +7917,44 @@ fn apply_file_root_tree(
         std::fs::create_dir_all(&target)?;
     }
 
+    if let Some((local_root, base_tree, local_tree, local_tree_hash)) =
+        load_apply_three_way_state(store, &cas, &target)?
+    {
+        return apply_file_root_tree_three_way(
+            store,
+            &source_root,
+            &local_root,
+            &cas,
+            target,
+            dry_run,
+            &tree_hash,
+            &tree,
+            &base_tree,
+            &local_tree,
+            local_tree_hash.as_ref(),
+        );
+    }
+
+    apply_file_root_tree_conservative(
+        store,
+        &source_root,
+        &cas,
+        target,
+        dry_run,
+        &tree_hash,
+        &tree,
+    )
+}
+
+fn apply_file_root_tree_conservative(
+    store: &Store,
+    source_root: &StoredFileRoot,
+    cas: &LocalCas,
+    target: PathBuf,
+    dry_run: bool,
+    tree_hash: &BlobHash,
+    tree: &CasTreeObject,
+) -> Result<ControlResponse, NodeError> {
     let mut files_written = 0;
     let mut dirs_created = 0;
     let mut conflicts = Vec::new();
@@ -7928,9 +7966,9 @@ fn apply_file_root_tree(
                     if !out.is_dir() {
                         conflicts.push(record_apply_conflict(
                             store,
-                            &source_root,
+                            source_root,
                             &entry.path,
-                            &tree_hash,
+                            tree_hash,
                             "remote directory conflicts with a local non-directory path",
                             dry_run,
                         )?);
@@ -7946,9 +7984,9 @@ fn apply_file_root_tree(
                 let Some(blob) = &entry.blob else {
                     conflicts.push(record_apply_conflict(
                         store,
-                        &source_root,
+                        source_root,
                         &entry.path,
-                        &tree_hash,
+                        tree_hash,
                         "remote file entry is missing a CAS blob hash",
                         dry_run,
                     )?);
@@ -7958,9 +7996,9 @@ fn apply_file_root_tree(
                     if !out.is_file() {
                         conflicts.push(record_apply_conflict(
                             store,
-                            &source_root,
+                            source_root,
                             &entry.path,
-                            &tree_hash,
+                            tree_hash,
                             "remote file conflicts with a local non-file path",
                             dry_run,
                         )?);
@@ -7969,9 +8007,9 @@ fn apply_file_root_tree(
                     if hash_path(&out)? != *blob {
                         conflicts.push(record_apply_conflict(
                             store,
-                            &source_root,
+                            source_root,
                             &entry.path,
-                            &tree_hash,
+                            tree_hash,
                             "local file differs from remote CAS tree; refusing to overwrite",
                             dry_run,
                         )?);
@@ -7989,7 +8027,7 @@ fn apply_file_root_tree(
     }
 
     Ok(ControlResponse::CasRootApplied {
-        source: source.to_owned(),
+        source: source_root.name.clone(),
         target,
         files_written,
         dirs_created,
@@ -7999,6 +8037,325 @@ fn apply_file_root_tree(
     })
 }
 
+type ApplyThreeWayState = (
+    StoredFileRoot,
+    CasTreeObject,
+    CasTreeObject,
+    Option<BlobHash>,
+);
+
+fn load_apply_three_way_state(
+    store: &Store,
+    cas: &LocalCas,
+    target: &Path,
+) -> Result<Option<ApplyThreeWayState>, NodeError> {
+    if !target.is_dir() {
+        return Ok(None);
+    }
+    let target = std::fs::canonicalize(target)?;
+    let Some(local_root) = store
+        .list_file_roots()?
+        .into_iter()
+        .filter(|root| !root.path.starts_with("remote:"))
+        .find(|root| {
+            std::fs::canonicalize(&root.path)
+                .map(|path| path == target)
+                .unwrap_or(false)
+        })
+    else {
+        return Ok(None);
+    };
+    let Some(base_tree_json) = local_root.latest_tree_json.as_deref() else {
+        return Ok(None);
+    };
+    let base_tree = serde_json::from_str(base_tree_json)?;
+    let local_scan = cas.add_tree_path(&target)?;
+    Ok(Some((
+        local_root,
+        base_tree,
+        local_scan.tree,
+        Some(local_scan.object.hash),
+    )))
+}
+
+#[allow(clippy::too_many_arguments)]
+fn apply_file_root_tree_three_way(
+    store: &Store,
+    source_root: &StoredFileRoot,
+    local_root: &StoredFileRoot,
+    cas: &LocalCas,
+    target: PathBuf,
+    dry_run: bool,
+    remote_tree_hash: &BlobHash,
+    remote_tree: &CasTreeObject,
+    base_tree: &CasTreeObject,
+    local_tree: &CasTreeObject,
+    local_tree_hash: Option<&BlobHash>,
+) -> Result<ControlResponse, NodeError> {
+    let mut files_written = 0;
+    let mut dirs_created = 0;
+    let mut conflicts = record_three_way_apply_conflicts(
+        store,
+        source_root,
+        local_root,
+        base_tree,
+        local_tree_hash,
+        remote_tree_hash,
+        remote_tree,
+        local_tree,
+        dry_run,
+    )?;
+    let conflict_paths = conflicts
+        .iter()
+        .map(|conflict| conflict.path.clone())
+        .collect::<BTreeSet<_>>();
+    let base_entries = owned_tree_entry_map(base_tree);
+    let local_entries = owned_tree_entry_map(local_tree);
+    let remote_entries = owned_tree_entry_map(remote_tree);
+    let mut consumed_base_paths = BTreeSet::new();
+    let mut consumed_remote_paths = BTreeSet::new();
+
+    for (from, to) in remote_rename_pairs(&base_entries, &remote_entries) {
+        if conflict_paths.contains(&from) || conflict_paths.contains(&to) {
+            continue;
+        }
+        let Some(base_entry) = base_entries.get(&from) else {
+            continue;
+        };
+        if local_entries.get(&from) != Some(base_entry) {
+            continue;
+        }
+        let Some(remote_entry) = remote_entries.get(&to) else {
+            continue;
+        };
+        let from_path = safe_tree_output_path(&target, &from)?;
+        let to_path = safe_tree_output_path(&target, &to)?;
+        if to_path.exists() {
+            if path_matches_entry(cas, &to_path, remote_entry)? {
+                consumed_base_paths.insert(from);
+                consumed_remote_paths.insert(to);
+            } else {
+                conflicts.push(record_apply_conflict_with_trees(
+                    store,
+                    source_root,
+                    &to,
+                    local_root.latest_tree_hash.as_deref(),
+                    local_tree_hash,
+                    remote_tree_hash,
+                    "remote rename target conflicts with an existing local path",
+                    dry_run,
+                )?);
+            }
+            continue;
+        }
+        if !from_path.exists() {
+            continue;
+        }
+        if !dry_run {
+            if let Some(parent) = to_path.parent() {
+                std::fs::create_dir_all(parent)?;
+            }
+            std::fs::rename(&from_path, &to_path)?;
+            set_executable_bit(&to_path, remote_entry.executable)?;
+        }
+        consumed_base_paths.insert(from);
+        consumed_remote_paths.insert(to);
+    }
+
+    let mut remote_paths = remote_entries.keys().cloned().collect::<Vec<_>>();
+    remote_paths.sort_by(|left, right| {
+        kind_order_for_apply(remote_entries.get(left))
+            .cmp(&kind_order_for_apply(remote_entries.get(right)))
+            .then_with(|| left.cmp(right))
+    });
+    for path in remote_paths {
+        if consumed_remote_paths.contains(&path) || conflict_paths.contains(&path) {
+            continue;
+        }
+        let Some(remote_entry) = remote_entries.get(&path) else {
+            continue;
+        };
+        let base_entry = base_entries.get(&path);
+        let local_entry = local_entries.get(&path);
+        if local_entry == Some(remote_entry) || base_entry == Some(remote_entry) {
+            continue;
+        }
+        if base_entry.is_some() && local_entry != base_entry {
+            continue;
+        }
+        if base_entry.is_none() && local_entry.is_some() {
+            conflicts.push(record_apply_conflict_with_trees(
+                store,
+                source_root,
+                &path,
+                local_root.latest_tree_hash.as_deref(),
+                local_tree_hash,
+                remote_tree_hash,
+                "remote create conflicts with an existing local path",
+                dry_run,
+            )?);
+            continue;
+        }
+        match remote_entry.kind {
+            CasTreeEntryKind::Directory => {
+                let out = safe_tree_output_path(&target, &path)?;
+                if !out.exists() {
+                    dirs_created += 1;
+                    if !dry_run {
+                        std::fs::create_dir_all(out)?;
+                    }
+                }
+            }
+            CasTreeEntryKind::File => {
+                let Some(blob) = &remote_entry.blob else {
+                    conflicts.push(record_apply_conflict_with_trees(
+                        store,
+                        source_root,
+                        &path,
+                        local_root.latest_tree_hash.as_deref(),
+                        local_tree_hash,
+                        remote_tree_hash,
+                        "remote file entry is missing a CAS blob hash",
+                        dry_run,
+                    )?);
+                    continue;
+                };
+                let out = safe_tree_output_path(&target, &path)?;
+                files_written += 1;
+                if !dry_run {
+                    let bytes = cas.read_bytes(blob)?;
+                    write_file_atomic(&out, &bytes)?;
+                    set_executable_bit(&out, remote_entry.executable)?;
+                }
+            }
+        }
+    }
+
+    let mut deleted_paths = base_entries
+        .keys()
+        .filter(|path| {
+            !remote_entries.contains_key(*path)
+                && !consumed_base_paths.contains(*path)
+                && !conflict_paths.contains(*path)
+        })
+        .cloned()
+        .collect::<Vec<_>>();
+    deleted_paths.sort_by(|left, right| right.cmp(left));
+    for path in deleted_paths {
+        let Some(base_entry) = base_entries.get(&path) else {
+            continue;
+        };
+        if local_entries.get(&path) != Some(base_entry) {
+            continue;
+        }
+        let out = safe_tree_output_path(&target, &path)?;
+        if dry_run || !out.exists() {
+            continue;
+        }
+        match base_entry.kind {
+            CasTreeEntryKind::Directory => {
+                let _ = std::fs::remove_dir(&out);
+            }
+            CasTreeEntryKind::File => {
+                std::fs::remove_file(&out)?;
+            }
+        }
+    }
+
+    Ok(ControlResponse::CasRootApplied {
+        source: source_root.name.clone(),
+        target,
+        files_written,
+        dirs_created,
+        conflicts,
+        dry_run,
+        note: "three-way file-root apply updated only paths where local state still matched the recorded base; ambiguous edits remain durable conflicts".to_owned(),
+    })
+}
+
+#[allow(clippy::too_many_arguments)]
+fn record_three_way_apply_conflicts(
+    store: &Store,
+    source_root: &StoredFileRoot,
+    local_root: &StoredFileRoot,
+    base_tree: &CasTreeObject,
+    local_tree_hash: Option<&BlobHash>,
+    remote_tree_hash: &BlobHash,
+    remote_tree: &CasTreeObject,
+    local_tree: &CasTreeObject,
+    dry_run: bool,
+) -> Result<Vec<FileConflict>, NodeError> {
+    geth_cas::detect_tree_conflicts(base_tree, local_tree, remote_tree)
+        .into_iter()
+        .map(|conflict| {
+            record_apply_conflict_with_trees(
+                store,
+                source_root,
+                &conflict.path,
+                local_root.latest_tree_hash.as_deref(),
+                local_tree_hash,
+                remote_tree_hash,
+                &conflict.detail,
+                dry_run,
+            )
+        })
+        .collect()
+}
+
+fn owned_tree_entry_map(tree: &CasTreeObject) -> BTreeMap<String, geth_cas::CasTreeEntry> {
+    tree.entries
+        .iter()
+        .map(|entry| (entry.path.clone(), entry.clone()))
+        .collect()
+}
+
+fn remote_rename_pairs(
+    base_entries: &BTreeMap<String, geth_cas::CasTreeEntry>,
+    remote_entries: &BTreeMap<String, geth_cas::CasTreeEntry>,
+) -> Vec<(String, String)> {
+    let mut pairs = Vec::new();
+    for (base_path, base_entry) in base_entries {
+        let Some(base_blob) = &base_entry.blob else {
+            continue;
+        };
+        if remote_entries.contains_key(base_path) {
+            continue;
+        }
+        if let Some((remote_path, _)) = remote_entries.iter().find(|(remote_path, remote_entry)| {
+            !base_entries.contains_key(*remote_path)
+                && remote_entry.blob.as_ref() == Some(base_blob)
+        }) {
+            pairs.push((base_path.clone(), remote_path.clone()));
+        }
+    }
+    pairs.sort();
+    pairs
+}
+
+fn path_matches_entry(
+    cas: &LocalCas,
+    path: &Path,
+    entry: &geth_cas::CasTreeEntry,
+) -> Result<bool, NodeError> {
+    match entry.kind {
+        CasTreeEntryKind::Directory => Ok(path.is_dir()),
+        CasTreeEntryKind::File => {
+            let Some(blob) = &entry.blob else {
+                return Ok(false);
+            };
+            Ok(path.is_file() && hash_path(path)? == *blob && cas.has(blob)?)
+        }
+    }
+}
+
+fn kind_order_for_apply(entry: Option<&geth_cas::CasTreeEntry>) -> u8 {
+    match entry.map(|entry| &entry.kind) {
+        Some(CasTreeEntryKind::Directory) => 0,
+        Some(CasTreeEntryKind::File) => 1,
+        None => 2,
+    }
+}
+
 fn safe_tree_output_path(root: &Path, relative: &str) -> Result<PathBuf, NodeError> {
     let relative_path = Path::new(relative);
     if relative_path.is_absolute() {
@@ -8052,6 +8409,45 @@ fn record_apply_conflict(
     file_conflict_from_stored(stored)
 }
 
+#[allow(clippy::too_many_arguments)]
+fn record_apply_conflict_with_trees(
+    store: &Store,
+    root: &StoredFileRoot,
+    path: &str,
+    base_tree: Option<&str>,
+    local_tree: Option<&BlobHash>,
+    remote_tree: &BlobHash,
+    detail: &str,
+    dry_run: bool,
+) -> Result<FileConflict, NodeError> {
+    let created_at = geth_store::now_ms();
+    let stored = StoredFileConflict {
+        conflict_id: generated_file_conflict_id(
+            &root.name,
+            path,
+            FileConflictKind::ConcurrentEdit.as_str(),
+            created_at,
+        ),
+        root_name: root.name.clone(),
+        resource_id: root.resource_id.clone(),
+        path: path.to_owned(),
+        kind: FileConflictKind::ConcurrentEdit.as_str().to_owned(),
+        status: FileConflictStatus::Open.as_str().to_owned(),
+        base_tree_hash: base_tree.map(ToOwned::to_owned),
+        local_tree_hash: local_tree.map(ToString::to_string),
+        remote_tree_hash: Some(remote_tree.to_string()),
+        detail: detail.to_owned(),
+        resolution: None,
+        resolution_note: None,
+        created_at_ms: created_at,
+        resolved_at_ms: None,
+    };
+    if !dry_run {
+        store.upsert_file_conflict(&stored)?;
+    }
+    file_conflict_from_stored(stored)
+}
+
 fn record_sync_tree_conflicts(
     store: &Store,
     local_root: &StoredFileRoot,

crates/geth/tests/bootstrap.rs

@@ -1651,6 +1651,174 @@ fn cas_file_root_apply_writes_missing_files_and_records_conflicts() {
     }
 }
 
+#[test]
+fn cas_file_root_apply_three_way_updates_deletes_and_renames_safe_changes() {
+    let home = tempfile::tempdir().expect("tempdir");
+    let paths = geth_config::GethPaths::from_home(home.path());
+    let node = geth_node::init_node(&paths).expect("init node");
+    let target_path = home.path().join("target-root");
+    std::fs::create_dir_all(&target_path).expect("create target root");
+    std::fs::write(target_path.join("update.txt"), b"base").expect("write base update");
+    std::fs::write(target_path.join("delete.txt"), b"base delete").expect("write base delete");
+    std::fs::write(target_path.join("rename.txt"), b"base rename").expect("write base rename");
+
+    geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasRootAdd {
+            name: "shared".to_owned(),
+            path: target_path.clone(),
+        },
+    )
+    .expect("add local root");
+    geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasRootScan {
+            name: "shared".to_owned(),
+        },
+    )
+    .expect("scan local base");
+
+    let source_path = home.path().join("source-root");
+    std::fs::create_dir_all(&source_path).expect("create source root");
+    std::fs::write(source_path.join("update.txt"), b"remote").expect("write remote update");
+    std::fs::write(source_path.join("rename-new.txt"), b"base rename")
+        .expect("write remote rename");
+    std::fs::write(source_path.join("created.txt"), b"created").expect("write remote create");
+    geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasRootAdd {
+            name: "remote-node-peer-shared".to_owned(),
+            path: source_path,
+        },
+    )
+    .expect("add remote root");
+    geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasRootScan {
+            name: "remote-node-peer-shared".to_owned(),
+        },
+    )
+    .expect("scan remote root");
+
+    let response = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasRootApply {
+            source: "remote-node-peer-shared".to_owned(),
+            target: target_path.clone(),
+            dry_run: false,
+        },
+    )
+    .expect("three-way apply");
+    match response {
+        geth_control::ControlResponse::CasRootApplied {
+            files_written,
+            conflicts,
+            note,
+            ..
+        } => {
+            assert_eq!(files_written, 2);
+            assert!(conflicts.is_empty());
+            assert!(note.contains("three-way"));
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+
+    assert_eq!(
+        std::fs::read(target_path.join("update.txt")).expect("read updated"),
+        b"remote"
+    );
+    assert!(
+        !target_path.join("delete.txt").exists(),
+        "safe remote delete should remove unchanged local base file"
+    );
+    assert!(
+        !target_path.join("rename.txt").exists(),
+        "safe remote rename should remove old path"
+    );
+    assert_eq!(
+        std::fs::read(target_path.join("rename-new.txt")).expect("read renamed"),
+        b"base rename"
+    );
+    assert_eq!(
+        std::fs::read(target_path.join("created.txt")).expect("read created"),
+        b"created"
+    );
+}
+
+#[test]
+fn cas_file_root_apply_three_way_keeps_ambiguous_changes_as_conflicts() {
+    let home = tempfile::tempdir().expect("tempdir");
+    let paths = geth_config::GethPaths::from_home(home.path());
+    let node = geth_node::init_node(&paths).expect("init node");
+    let target_path = home.path().join("target-root");
+    std::fs::create_dir_all(&target_path).expect("create target root");
+    std::fs::write(target_path.join("note.txt"), b"base").expect("write base");
+
+    geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasRootAdd {
+            name: "shared".to_owned(),
+            path: target_path.clone(),
+        },
+    )
+    .expect("add local root");
+    geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasRootScan {
+            name: "shared".to_owned(),
+        },
+    )
+    .expect("scan local base");
+    std::fs::write(target_path.join("note.txt"), b"local edit").expect("write local edit");
+
+    let source_path = home.path().join("source-root");
+    std::fs::create_dir_all(&source_path).expect("create source root");
+    std::fs::write(source_path.join("note.txt"), b"remote edit").expect("write remote edit");
+    geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasRootAdd {
+            name: "remote-node-peer-shared".to_owned(),
+            path: source_path,
+        },
+    )
+    .expect("add remote root");
+    geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasRootScan {
+            name: "remote-node-peer-shared".to_owned(),
+        },
+    )
+    .expect("scan remote root");
+
+    let response = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasRootApply {
+            source: "remote-node-peer-shared".to_owned(),
+            target: target_path.clone(),
+            dry_run: false,
+        },
+    )
+    .expect("three-way conflict apply");
+    match response {
+        geth_control::ControlResponse::CasRootApplied { conflicts, .. } => {
+            assert_eq!(conflicts.len(), 1);
+            assert_eq!(conflicts[0].path, "note.txt");
+            assert_eq!(
+                conflicts[0].kind,
+                geth_cas::FileConflictKind::ConcurrentEdit
+            );
+            assert!(conflicts[0].base_tree.is_some());
+            assert!(conflicts[0].local_tree.is_some());
+            assert!(conflicts[0].remote_tree.is_some());
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+    assert_eq!(
+        std::fs::read(target_path.join("note.txt")).expect("read local after conflict"),
+        b"local edit"
+    );
+}
+
 #[test]
 fn cas_file_conflict_record_list_and_resolve() {
     let home = tempfile::tempdir().expect("tempdir");