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

Live sync JSON documents over Iroh

Browse source
This commit is contained in:
Eric Wendland 2026-05-18 18:49:34 +02:00
commit 284207eb01
8 changed files with 370 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -272,6 +272,10 @@ pub async fn handle_request_async(
target,
node: Some(peer_node),
} => pipe_connect_to_peer(node, &peer_node, target).await,
ControlRequest::DocumentSync {
node: peer_node,
name,
} => document_sync_from_peer(node, &peer_node, &name).await,
other => handle_request(node, other),
}
}
@ -550,7 +554,8 @@ async fn peer_ping(node: &LocalNode, peer_node: &str) -> Result<ControlResponse,
| PeerControlResponse::SshRevocationSynced { .. }
| PeerControlResponse::KvSynced { .. }
| PeerControlResponse::PubsubPublished { .. }
| PeerControlResponse::PipeConnected { .. } => Err(NodeError::IrohPeer(
| PeerControlResponse::PipeConnected { .. }
| PeerControlResponse::DocumentSynced { .. } => Err(NodeError::IrohPeer(
"peer returned wrong response type to ping request".to_owned(),
)),
}
@ -660,7 +665,8 @@ async fn peer_auth_check(
| PeerControlResponse::SshRevocationSynced { .. }
| PeerControlResponse::KvSynced { .. }
| PeerControlResponse::PubsubPublished { .. }
| PeerControlResponse::PipeConnected { .. } => Err(NodeError::IrohPeer(
| PeerControlResponse::PipeConnected { .. }
| PeerControlResponse::DocumentSynced { .. } => Err(NodeError::IrohPeer(
"peer returned wrong response type to auth-check request".to_owned(),
)),
}
@ -799,7 +805,8 @@ async fn cas_fetch_from_peer(
| PeerControlResponse::SshRevocationSynced { .. }
| PeerControlResponse::KvSynced { .. }
| PeerControlResponse::PubsubPublished { .. }
| PeerControlResponse::PipeConnected { .. } => Err(NodeError::IrohPeer(
| PeerControlResponse::PipeConnected { .. }
| PeerControlResponse::DocumentSynced { .. } => Err(NodeError::IrohPeer(
"peer returned wrong response type to CAS fetch".to_owned(),
)),
}
@ -1123,6 +1130,88 @@ async fn pipe_connect_to_peer(
}
}
async fn document_sync_from_peer(
node: &LocalNode,
peer_node: &str,
name: &str,
) -> Result<ControlResponse, NodeError> {
geth_document::validate_document_name(name)
.map_err(|_| NodeError::InvalidDocumentName(name.to_owned()))?;
let stream = format!("document:{name}");
let since_ms =
load_live_sync_cursor(&Store::open(&node.paths.metadata_db())?, peer_node, &stream)?;
let response = request_peer_control(node, peer_node, "document-sync", |peer_card, nonce| {
PeerControlRequest::DocumentSync {
peer_card,
name: name.to_owned(),
since_ms,
nonce,
}
})
.await?;
match response {
PeerControlResponse::DocumentSynced {
node_id,
agent_id,
endpoint_id,
name: response_name,
state,
high_water_ms,
allowed,
reason,
note,
..
} if response_name == name => {
if !allowed {
return Ok(ControlResponse::DocumentSynced {
peer_node_id: node_id,
peer_agent_id: agent_id,
endpoint_id,
name: response_name,
updated: false,
allowed,
reason,
note,
});
}
let store = Store::open(&node.paths.metadata_db())?;
let mut updated = false;
if let Some(state) = state {
let local = ensure_local_document(&store, name)?;
if state.updated_at.0 >= local.updated_at_ms {
let normalized = geth_document::normalize_document_state(&state.state_json)?;
store.insert_document_resource(&StoredDocumentResource {
document_id: local.document_id,
resource_id: local.resource_id,
name: local.name,
state_json: normalized,
updated_at_ms: state.updated_at.0,
})?;
updated = true;
}
}
store_live_sync_cursor(&store, peer_node, &stream, high_water_ms)?;
Ok(ControlResponse::DocumentSynced {
peer_node_id: node_id,
peer_agent_id: agent_id,
endpoint_id,
name: response_name,
updated,
allowed,
reason,
note,
})
}
PeerControlResponse::DocumentSynced { .. } => Err(NodeError::IrohPeer(
"peer document sync response did not match request".to_owned(),
)),
PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
_ => Err(NodeError::IrohPeer(
"peer returned wrong response type to document sync".to_owned(),
)),
}
}
fn live_sync_cursor_key(peer_node: &str, stream: &str) -> String {
format!("live-sync:{peer_node}:{stream}")
}
@ -1227,6 +1316,10 @@ async fn request_peer_control(
| PeerControlResponse::PipeConnected {
nonce: response_nonce,
..
}
| PeerControlResponse::DocumentSynced {
nonce: response_nonce,
..
} if response_nonce == &nonce => Ok(response),
PeerControlResponse::Error { .. } => Ok(response),
_ => Err(NodeError::IrohPeer(format!(
@ -1275,6 +1368,7 @@ async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> {
}
let peers = Store::open(&node.paths.metadata_db())?.list_peer_cards()?;
let kv_stores = Store::open(&node.paths.metadata_db())?.list_kv_stores()?;
let documents = Store::open(&node.paths.metadata_db())?.list_document_resources()?;
for peer in peers {
if let Err(error) = ssh_cert_sync_from_peer(node, &peer.peer_id).await {
tracing::debug!(peer = %peer.peer_id, %error, "SSH cert live sync failed");
@ -1287,6 +1381,11 @@ async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> {
tracing::debug!(peer = %peer.peer_id, kv = %kv.name, %error, "KV live sync failed");
}
}
for document in &documents {
if let Err(error) = document_sync_from_peer(node, &peer.peer_id, &document.name).await {
tracing::debug!(peer = %peer.peer_id, document = %document.name, %error, "document live sync failed");
}
}
}
Ok(())
}
@ -1714,6 +1813,61 @@ async fn handle_iroh_control_connection(
note: "pipe connect authenticated endpoint/card binding and required pipe.connect on the remote pipe resource; byte streams are not implemented yet".to_owned(),
}
}
PeerControlRequest::DocumentSync {
peer_card,
name,
since_ms,
nonce,
} => {
geth_document::validate_document_name(&name)
.map_err(|_| NodeError::InvalidDocumentName(name.clone()))?;
peer_card.validate_candidate()?;
ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
let discovered = DiscoveredPeer::candidate(
peer_card.clone(),
UnixMillis(geth_store::now_ms()),
DiscoverySource::PeerExchange,
)?;
let store = Store::open(&node.paths.metadata_db())?;
store.upsert_peer_card(&StoredPeerCard {
peer_id: peer_card.node_id.to_string(),
card_json: serde_json::to_string(&peer_card)?,
updated_at_ms: discovered.discovered_at.0,
})?;
if let Some(document) = store.get_document_resource_by_name(&name)? {
let capability = "document.read".to_owned();
let high_water_ms = geth_store::now_ms();
let explanation = geth_auth::explain_auth_ops(
&load_auth_ops_for_resource(&store, &document.resource_id)?,
PrincipalId::new(peer_card.node_id.to_string()),
ResourceId::new(document.resource_id.clone()),
Capability::new(capability),
);
let state = if explanation.allowed && document.updated_at_ms >= since_ms {
Some(document_state_from_stored(&document))
} else {
None
};
PeerControlResponse::DocumentSynced {
node_id: node.node_id.clone(),
agent_id: node.agent_id.clone(),
endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
remote_endpoint_id,
name,
state,
high_water_ms,
allowed: explanation.allowed,
reason: explanation.reason,
evaluated_ops: explanation.evaluated_ops,
nonce,
note: "document sync authenticated endpoint/card binding and required document.read on the remote document resource; JSON LWW sync is a bootstrap before Automerge".to_owned(),
}
} else {
PeerControlResponse::Error {
message: format!("document not found: {name}"),
}
}
}
};
send.write_all(geth_control::encode_peer_response(&response)?.as_bytes())
.await
@ -1853,6 +2007,7 @@ pub fn handle_request(
ControlRequest::SshCertSync { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::SshRevocationSync { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::KvSync { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::DocumentSync { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::ResourceList => Ok(ControlResponse::ResourceList {
resources: store
.list_resources()?
@ -2906,6 +3061,21 @@ fn record_pipe_connection(
Ok(connection)
}
fn ensure_local_document(store: &Store, name: &str) -> Result<StoredDocumentResource, NodeError> {
if let Some(document) = store.get_document_resource_by_name(name)? {
return Ok(document);
}
let stored = StoredDocumentResource {
document_id: format!("document:{name}"),
resource_id: format!("resource:document:{name}"),
name: name.to_owned(),
state_json: "{}".to_owned(),
updated_at_ms: 0,
};
store.insert_document_resource(&stored)?;
Ok(stored)
}
fn document_resource_from_stored(stored: &StoredDocumentResource) -> DocumentResource {
DocumentResource {
id: stored.document_id.clone().into(),
@ -3433,6 +3603,21 @@ mod tests {
},
)
.expect("right pipe listen");
handle_request(
&right,
ControlRequest::DocumentCreate {
name: "notes".to_owned(),
},
)
.expect("right document create");
handle_request(
&right,
ControlRequest::DocumentSet {
name: "notes".to_owned(),
state_json: r#"{"title":"remote"}"#.to_owned(),
},
)
.expect("right document set");
let ping = handle_request_async(
&left,
@ -3581,6 +3766,29 @@ mod tests {
other => panic!("unexpected denied pipe connect response: {other:?}"),
}
let denied_document = handle_request_async(
&left,
ControlRequest::DocumentSync {
node: right_card.node_id.to_string(),
name: "notes".to_owned(),
},
)
.await
.expect("denied document sync");
match denied_document {
ControlResponse::DocumentSynced {
allowed,
updated,
reason,
..
} => {
assert!(!allowed);
assert!(!updated);
assert!(reason.contains("no active direct or group grant"));
}
other => panic!("unexpected denied document sync response: {other:?}"),
}
handle_request(
&right,
ControlRequest::AuthGrant {
@ -3621,6 +3829,16 @@ mod tests {
},
)
.expect("grant left pipe connect");
handle_request(
&right,
ControlRequest::AuthGrant {
subject: left.node_id.clone(),
resource: "resource:document:notes".to_owned(),
capability: "document.read".to_owned(),
grant_id: Some("grant:left-document-read".to_owned()),
},
)
.expect("grant left document read");
let allowed = handle_request_async(
&left,
@ -3784,6 +4002,44 @@ mod tests {
other => panic!("unexpected allowed pipe connect response: {other:?}"),
}
let document_sync = handle_request_async(
&left,
ControlRequest::DocumentSync {
node: right_card.node_id.to_string(),
name: "notes".to_owned(),
},
)
.await
.expect("allowed document sync");
match document_sync {
ControlResponse::DocumentSynced {
allowed,
updated,
reason,
note,
..
} => {
assert!(allowed);
assert!(updated);
assert!(reason.contains("direct grant"));
assert!(note.contains("JSON LWW"));
}
other => panic!("unexpected allowed document sync response: {other:?}"),
}
let synced_document = handle_request(
&left,
ControlRequest::DocumentGet {
name: "notes".to_owned(),
},
)
.expect("left document get synced");
match synced_document {
ControlResponse::DocumentGet { state } => {
assert_eq!(state.state_json, r#"{"title":"remote"}"#);
}
other => panic!("unexpected synced document get response: {other:?}"),
}
let denied_cert_sync = handle_request_async(
&left,
ControlRequest::SshCertSync {
@ -3943,6 +4199,14 @@ mod tests {
},
)
.expect("right second kv set");
handle_request(
&right,
ControlRequest::DocumentSet {
name: "notes".to_owned(),
state_json: r#"{"title":"live"}"#.to_owned(),
},
)
.expect("right live document set");
run_live_sync_once(&left)
.await
@ -3969,6 +4233,13 @@ mod tests {
.map(|entry| entry.value),
Some("synced".to_owned())
);
assert_eq!(
left_store
.get_document_resource_by_name("notes")
.expect("get live-synced document")
.map(|document| document.state_json),
Some(r#"{"title":"live"}"#.to_owned())
);
left_endpoint.shutdown().await;
right_endpoint.shutdown().await;