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

Live-sync authorized file roots

Browse source
This commit is contained in:
Eric Wendland 2026-05-20 13:26:50 +02:00
commit bab13cc0a1
5 changed files with 128 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -917,6 +917,7 @@ 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 root = root
.map(|remote_root| {
let stored_name = remote_file_root_name(&node_id, &remote_root.name);
@ -960,9 +961,18 @@ async fn cas_root_sync_from_peer(
updated_at_ms: remote_root.updated_at_ms,
};
store.upsert_file_root(&stored)?;
imported_high_water = Some(remote_root.updated_at_ms);
Ok(file_root_from_stored(&stored))
})
.transpose()?;
if let Some(high_water_ms) = imported_high_water {
store_live_sync_cursor(
&store,
peer_node,
&format!("cas-tree:{response_name}"),
high_water_ms,
)?;
}
Ok(ControlResponse::CasRootSynced {
peer_node_id: node_id,
peer_agent_id: agent_id,
@ -1844,6 +1854,18 @@ fn sync_watermarks_for_peer(
}
}
}
for root in store.list_file_roots()? {
if root.path.starts_with("remote:") {
continue;
}
let resource = format!("resource:cas-tree:{}", root.name);
if can_sync_resource(store, &peer, &resource, "cas.fetch")? {
watermarks.push(SyncWatermark {
stream: format!("cas-tree:{}", root.name),
high_water: root.updated_at_ms,
});
}
}
watermarks.sort_by(|left, right| left.stream.cmp(&right.stream));
Ok(watermarks)
}
@ -2176,6 +2198,22 @@ async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> {
tracing::debug!(peer = %peer.peer_id, %error, "SSH revocation live sync failed");
}
}
if let Some(remote_watermarks) = remote_watermarks.as_ref() {
for stream in remote_watermarks
.keys()
.filter(|stream| stream.starts_with("cas-tree:"))
{
if should_live_sync_stream(&store, &peer.peer_id, stream, Some(remote_watermarks))?
{
let name = stream.trim_start_matches("cas-tree:");
if let Err(error) =
cas_root_sync_from_peer(node, &peer.peer_id, name, None).await
{
tracing::debug!(peer = %peer.peer_id, root = %name, %error, "file-root live sync failed");
}
}
}
}
for kv in &kv_stores {
let stream = format!("kv:{}", kv.name);
if should_live_sync_stream(&store, &peer.peer_id, &stream, remote_watermarks.as_ref())?
@ -5358,9 +5396,32 @@ mod tests {
path: db_path.display().to_string(),
})
.expect("insert db");
store
.upsert_file_root(&StoredFileRoot {
root_id: "file-root:shared".to_owned(),
resource_id: "resource:cas-tree:shared".to_owned(),
name: "shared".to_owned(),
path: dir.path().join("shared").display().to_string(),
latest_tree_hash: Some("abc".to_owned()),
latest_tree_json: None,
updated_at_ms: 123,
})
.expect("insert file root");
store
.upsert_file_root(&StoredFileRoot {
root_id: "file-root:remote:node:right:shared".to_owned(),
resource_id: "resource:cas-tree:shared".to_owned(),
name: "remote-node-right-shared".to_owned(),
path: "remote:node:right:shared".to_owned(),
latest_tree_hash: Some("def".to_owned()),
latest_tree_json: None,
updated_at_ms: 456,
})
.expect("insert imported remote file root");
grant_test_capability(&store, "node:left", "resource:kv:prefs", "kv.read");
grant_test_capability(&store, "node:left", "resource:db:notes", "db.sync");
grant_test_capability(&store, "node:left", "resource:cas-tree:shared", "cas.fetch");
let watermarks = sync_watermarks_for_peer(&store, "node:left").expect("watermarks");
@ -5372,6 +5433,15 @@ mod tests {
stream: "db:notes".to_owned(),
high_water: 7,
}));
assert!(watermarks.contains(&SyncWatermark {
stream: "cas-tree:shared".to_owned(),
high_water: 123,
}));
assert!(
!watermarks
.iter()
.any(|watermark| watermark.stream == "cas-tree:remote-node-right-shared")
);
assert!(
!watermarks
.iter()
@ -5475,7 +5545,7 @@ mod tests {
&right,
ControlRequest::CasRootAdd {
name: "shared".to_owned(),
path: right_file_root,
path: right_file_root.clone(),
},
)
.expect("right file root add");
@ -6588,6 +6658,19 @@ mod tests {
},
)
.expect("right live document set");
std::fs::write(right_file_root.join("live.txt"), b"live file root sync")
.expect("write live file");
let second_tree_hash = match handle_request(
&right,
ControlRequest::CasRootScan {
name: "shared".to_owned(),
},
)
.expect("right live file root scan")
{
ControlResponse::CasRootScanned { scan } => scan.tree.hash,
other => panic!("unexpected right live file root scan response: {other:?}"),
};
insert_mock_crsqlite_change(&right_db_path, 8, "live");
run_live_sync_once(&left)
@ -6622,6 +6705,34 @@ mod tests {
.map(|document| document.state_json),
Some(r#"{"title":"live"}"#.to_owned())
);
let live_remote_root = left_store
.list_file_roots()
.expect("list live-synced file roots")
.into_iter()
.find(|root| root.path == format!("remote:{}:shared", right.node_id))
.expect("live-synced remote root");
assert_eq!(
live_remote_root.latest_tree_hash.as_deref(),
Some(second_tree_hash.as_str())
);
assert!(
LocalCas::new(left.paths.cas_dir())
.has(&second_tree_hash)
.expect("left has live-synced tree object")
);
let file_root_cursor = left_store
.get_module_state(&live_sync_cursor_key(
right_card.node_id.as_str(),
"cas-tree:shared",
))
.expect("get live-synced file root cursor")
.expect("file root cursor exists");
assert_eq!(
serde_json::from_str::<LiveSyncCursor>(&file_root_cursor.state_json)
.expect("parse file root cursor")
.cursor_ms,
live_remote_root.updated_at_ms
);
let db_cursor = left_store
.get_module_state(&live_sync_cursor_key(
right_card.node_id.as_str(),