eric/geth
Watch
1
0
Fork
You've already forked geth
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Record file root sync conflicts

Browse source
This commit is contained in:
Eric Wendland 2026-05-21 01:40:50 +02:00
commit 475c6a4a2a
8 changed files with 324 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

164
crates/geth-node/src/lib.rs
View file

@ -1184,6 +1184,7 @@ async fn cas_root_sync_from_peer(
name: response_name,
root: None,
tree_bytes_imported: false,
sync_conflicts: Vec::new(),
allowed,
reason,
note,
@ -1192,11 +1193,13 @@ async fn cas_root_sync_from_peer(
let store = Store::open(&node.paths.metadata_db())?;
let mut tree_bytes_imported = false;
let mut imported_high_water = None;
let mut sync_conflicts = Vec::new();
let root = root
.map(|remote_root| {
let stored_name = remote_file_root_name(&node_id, &remote_root.name);
let stored_path = format!("remote:{node_id}:{}", remote_root.name);
if let Some(existing) = store.get_file_root_by_name(&stored_name)? {
let existing_remote_root = store.get_file_root_by_name(&stored_name)?;
if let Some(existing) = &existing_remote_root {
if existing.path != stored_path {
return Err(NodeError::IrohPeer(format!(
"refusing to overwrite existing file root {} at {}",
@ -1204,6 +1207,12 @@ async fn cas_root_sync_from_peer(
)));
}
}
let base_tree = existing_remote_root
.as_ref()
.and_then(|existing| existing.latest_tree_json.as_deref())
.map(serde_json::from_str)
.transpose()?;
let mut remote_tree_json = None;
if let (Some(hash), Some(content)) = (
remote_root.latest_tree.clone(),
tree_content_base64.as_ref(),
@ -1224,6 +1233,25 @@ async fn cas_root_sync_from_peer(
&info.path.to_string_lossy(),
)?;
tree_bytes_imported = true;
let remote_tree: CasTreeObject = geth_codec::decode_canonical(&decoded)?;
remote_tree_json = Some(serde_json::to_string(&remote_tree)?);
if let (Some(base_tree), Some(local_root)) =
(&base_tree, store.get_file_root_by_name(&remote_root.name)?)
{
if local_root.latest_tree_json.is_some() {
sync_conflicts.extend(record_sync_tree_conflicts(
&store,
&local_root,
existing_remote_root
.as_ref()
.and_then(|existing| existing.latest_tree_hash.as_deref()),
base_tree,
local_root.latest_tree_hash.as_deref(),
remote_root.latest_tree.as_ref(),
&remote_tree,
)?);
}
}
}
let stored = StoredFileRoot {
root_id: format!("file-root:remote:{node_id}:{}", remote_root.name),
@ -1231,7 +1259,7 @@ async fn cas_root_sync_from_peer(
name: stored_name,
path: stored_path,
latest_tree_hash: remote_root.latest_tree.as_ref().map(ToString::to_string),
latest_tree_json: None,
latest_tree_json: remote_tree_json,
updated_at_ms: remote_root.updated_at_ms,
};
store.upsert_file_root(&stored)?;
@ -1254,6 +1282,7 @@ async fn cas_root_sync_from_peer(
name: response_name,
root,
tree_bytes_imported,
sync_conflicts,
allowed,
reason,
note,
@ -6253,6 +6282,56 @@ fn record_apply_conflict(
file_conflict_from_stored(stored)
}
fn record_sync_tree_conflicts(
store: &Store,
local_root: &StoredFileRoot,
base_tree_hash: Option<&str>,
base_tree: &CasTreeObject,
local_tree_hash: Option<&str>,
remote_tree_hash: Option<&BlobHash>,
remote_tree: &CasTreeObject,
) -> Result<Vec<FileConflict>, NodeError> {
let Some(local_tree_json) = local_root.latest_tree_json.as_deref() else {
return Ok(Vec::new());
};
let local_tree: CasTreeObject = serde_json::from_str(local_tree_json)?;
let tree_conflicts = geth_cas::detect_tree_conflicts(base_tree, &local_tree, remote_tree);
let created_at = geth_store::now_ms();
tree_conflicts
.into_iter()
.enumerate()
.map(|(index, conflict)| {
let created_at = created_at + index as i64;
let stored = StoredFileConflict {
conflict_id: generated_file_conflict_id(
&local_root.name,
&conflict.path,
conflict.kind.as_str(),
created_at,
),
root_name: local_root.name.clone(),
resource_id: local_root.resource_id.clone(),
path: conflict.path,
kind: conflict.kind.as_str().to_owned(),
status: FileConflictStatus::Open.as_str().to_owned(),
base_tree_hash: base_tree_hash.map(ToOwned::to_owned),
local_tree_hash: local_tree_hash.map(ToOwned::to_owned),
remote_tree_hash: remote_tree_hash.map(ToString::to_string),
detail: format!(
"{}; detected while syncing remote file-root metadata",
conflict.detail
),
resolution: None,
resolution_note: None,
created_at_ms: created_at,
resolved_at_ms: None,
};
store.upsert_file_conflict(&stored)?;
file_conflict_from_stored(stored)
})
.collect()
}
fn write_file_atomic(path: &Path, bytes: &[u8]) -> Result<(), NodeError> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
@ -7330,6 +7409,70 @@ mod tests {
);
}
#[test]
fn file_root_sync_records_automatic_three_tree_conflicts() {
let store = Store::open_memory().expect("open");
let base_tree = test_tree(vec![
test_tree_file("edit.txt", "01"),
test_tree_file("delete-edit.txt", "02"),
test_tree_file("rename.txt", "03"),
]);
let local_tree = test_tree(vec![
test_tree_file("edit.txt", "04"),
test_tree_file("delete-edit.txt", "05"),
test_tree_file("local-rename.txt", "03"),
]);
let remote_tree = test_tree(vec![
test_tree_file("edit.txt", "06"),
test_tree_file("remote-rename.txt", "03"),
]);
let local_root = StoredFileRoot {
root_id: "file-root:shared".to_owned(),
resource_id: "resource:cas-tree:shared".to_owned(),
name: "shared".to_owned(),
path: "/tmp/shared".to_owned(),
latest_tree_hash: Some("local-tree".to_owned()),
latest_tree_json: Some(serde_json::to_string(&local_tree).expect("local tree json")),
updated_at_ms: 2,
};
store.upsert_file_root(&local_root).expect("upsert root");
let conflicts = record_sync_tree_conflicts(
&store,
&local_root,
Some("base-tree"),
&base_tree,
local_root.latest_tree_hash.as_deref(),
Some(&BlobHash::new("remote-tree".to_owned())),
&remote_tree,
)
.expect("record conflicts");
assert_eq!(conflicts.len(), 3);
assert!(conflicts.iter().any(|conflict| {
conflict.path == "edit.txt" && conflict.kind == FileConflictKind::ConcurrentEdit
}));
assert!(conflicts.iter().any(|conflict| {
conflict.path == "delete-edit.txt" && conflict.kind == FileConflictKind::DeleteEdit
}));
assert!(conflicts.iter().any(|conflict| {
conflict.path == "rename.txt" && conflict.kind == FileConflictKind::Rename
}));
let stored = store
.list_file_conflicts(Some("shared"))
.expect("list conflicts");
assert_eq!(stored.len(), 3);
assert!(stored.iter().all(|conflict| {
conflict.base_tree_hash.as_deref() == Some("base-tree")
&& conflict.local_tree_hash.as_deref() == Some("local-tree")
&& conflict.remote_tree_hash.as_deref() == Some("remote-tree")
&& conflict
.detail
.contains("syncing remote file-root metadata")
}));
}
#[test]
fn lan_discovery_address_selection_uses_iroh_direct_addresses() {
let key = AgentKey::generate();
@ -8754,4 +8897,21 @@ mod tests {
left_endpoint.shutdown().await;
right_endpoint.shutdown().await;
}
fn test_tree(entries: Vec<geth_cas::CasTreeEntry>) -> CasTreeObject {
CasTreeObject {
version: geth_cas::CAS_TREE_OBJECT_VERSION,
entries,
}
}
fn test_tree_file(path: &str, byte: &str) -> geth_cas::CasTreeEntry {
geth_cas::CasTreeEntry {
path: path.to_owned(),
kind: CasTreeEntryKind::File,
blob: Some(BlobHash::new(byte.repeat(32))),
size_bytes: 1,
executable: false,
}
}
}