From 284207eb018fef22a22f72c652472a128e2cec6b Mon Sep 17 00:00:00 2001 From: Eric Wendland Date: Mon, 18 May 2026 18:49:34 +0200 Subject: [PATCH] Live sync JSON documents over Iroh --- AGENTS.md | 4 +- README.md | 7 +- crates/geth-cli/src/lib.rs | 23 +++ crates/geth-control/src/lib.rs | 34 ++++ crates/geth-node/src/lib.rs | 277 ++++++++++++++++++++++++++++++++- crates/geth-store/src/lib.rs | 18 +++ docs/architecture.md | 7 +- docs/roadmap.md | 7 +- 8 files changed, 370 insertions(+), 7 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index ffbb06b..a4b71ff 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -145,7 +145,9 @@ Roadmap items should be actionable and checkable: work. The auth evaluator already understands `kv.write_prefix:` grants for `kv.write_key:` requests. - Document resources can be registered locally and updated with validated local - JSON state. Automerge editing/state and sync are still roadmap work. + JSON state. `geth document sync ` and background live-sync can + pull authorized JSON last-writer-wins state over Iroh. Automerge editing/state + and true CRDT sync are still roadmap work. - Pubsub supports local daemon-lifetime publish/subscribe snapshots through a bounded in-memory ring buffer. Remote publish over the protected Iroh control ALPN requires `pubsub.publish` on `resource:pubsub:`. Iroh-gossip diff --git a/README.md b/README.md index 3fdffbd..b5be07d 100644 --- a/README.md +++ b/README.md @@ -105,7 +105,8 @@ The bootstrap implementation provides: - local SQLite-backed KV commands: `geth kv create/set/get`; `kv set` accepts `--subject ` to exercise local capability checks for non-local callers; `geth kv sync ` pulls authorized remote updates -- local JSON document commands: `geth document create/status/set/get` +- local JSON document commands: `geth document create/status/set/get`; `geth + document sync ` pulls authorized remote JSON state - local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub`; `geth pubsub pub --node ` publishes to an authorized peer - SSH certificate flow metadata: @@ -156,6 +157,10 @@ Remote pipe connect uses the same protected Iroh control path and requires `pipe.connect` on `resource:pipe:`. The current prototype records a remote connection attempt and whether a listener exists; byte streaming and forwarding are still future work. +Document sync is a bootstrap JSON last-writer-wins path before Automerge: +manual `geth document sync ` and background live-sync require +`document.read` on `resource:document:` and import only state that is not +older than the local document timestamp. Importing or pinging a peer card never grants capabilities by itself. When `[iroh].local_discovery = true`, the daemon also advertises and discovers signed peer cards on LAN using a geth-specific mDNS TXT payload. That payload is diff --git a/crates/geth-cli/src/lib.rs b/crates/geth-cli/src/lib.rs index f5695d4..2982698 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -359,6 +359,7 @@ pub enum DocumentCommand { Status { name: String }, Set { name: String, state_json: String }, Get { name: String }, + Sync { node: String, name: String }, } #[derive(Debug, Subcommand)] @@ -672,6 +673,7 @@ fn request_for_command(command: Command) -> Result { ControlRequest::DocumentSet { name, state_json } } DocumentCommand::Get { name } => ControlRequest::DocumentGet { name }, + DocumentCommand::Sync { node, name } => ControlRequest::DocumentSync { node, name }, }, Command::Ssh { command } => match command { SshCommand::Proxy { node } => ControlRequest::ModuleStub { @@ -1378,6 +1380,27 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { ControlResponse::DocumentGet { state } => { println!("{}", state.state_json); } + ControlResponse::DocumentSynced { + peer_node_id, + peer_agent_id, + endpoint_id, + name, + updated, + allowed, + reason, + note, + } => { + if allowed { + println!("synced document {name} from {peer_node_id}: updated={updated}"); + } else { + println!("document sync denied for {name} by {peer_node_id}"); + } + println!("agent: {peer_agent_id}"); + println!("endpoint: {endpoint_id}"); + println!("allowed: {allowed}"); + println!("reason: {reason}"); + println!("note: {note}"); + } ControlResponse::PubsubPublished { message } => { println!("published: {}", message.topic); println!("published_at_ms: {}", message.published_at.0); diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index 5c1f7a7..22609e2 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -213,6 +213,10 @@ pub enum ControlRequest { DocumentGet { name: String, }, + DocumentSync { + node: String, + name: String, + }, PubsubPub { topic: String, message: String, @@ -450,6 +454,16 @@ pub enum ControlResponse { DocumentGet { state: DocumentState, }, + DocumentSynced { + peer_node_id: String, + peer_agent_id: String, + endpoint_id: String, + name: String, + updated: bool, + allowed: bool, + reason: String, + note: String, + }, PubsubPublished { message: PubsubMessage, }, @@ -572,6 +586,12 @@ pub enum PeerControlRequest { target: String, nonce: String, }, + DocumentSync { + peer_card: PeerCard, + name: String, + since_ms: i64, + nonce: String, + }, } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] @@ -678,6 +698,20 @@ pub enum PeerControlResponse { nonce: String, note: String, }, + DocumentSynced { + node_id: String, + agent_id: String, + endpoint_id: String, + remote_endpoint_id: String, + name: String, + state: Option, + high_water_ms: i64, + allowed: bool, + reason: String, + evaluated_ops: usize, + nonce: String, + note: String, + }, Error { message: String, }, diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 9edeba4..7cbb6c4 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -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 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 { + 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 { + 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; diff --git a/crates/geth-store/src/lib.rs b/crates/geth-store/src/lib.rs index 6001116..ab3f1c9 100644 --- a/crates/geth-store/src/lib.rs +++ b/crates/geth-store/src/lib.rs @@ -445,6 +445,24 @@ impl Store { } } + pub fn list_document_resources(&self) -> Result, StoreError> { + let mut stmt = self.conn.prepare( + r#"SELECT document_id, resource_id, name, state_json, updated_at_ms + FROM document_resources ORDER BY name"#, + )?; + let rows = stmt.query_map([], |row| { + Ok(StoredDocumentResource { + document_id: row.get(0)?, + resource_id: row.get(1)?, + name: row.get(2)?, + state_json: row.get(3)?, + updated_at_ms: row.get(4)?, + }) + })?; + rows.collect::, _>>() + .map_err(StoreError::from) + } + pub fn upsert_file_root(&self, root: &StoredFileRoot) -> Result<(), StoreError> { self.conn.execute( r#"INSERT OR REPLACE INTO file_roots( diff --git a/docs/architecture.md b/docs/architecture.md index 0b9afd9..c8b81d6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -153,7 +153,12 @@ exposed to remote peers. `geth-document` currently registers local document resources and stores validated JSON state in the local SQLite metadata store through `document create/status/set/get`. This is a bootstrap editing surface, not yet -Automerge CRDT state. Automerge state encoding and sync are future work. +Automerge CRDT state. `geth document sync ` can pull remote JSON +state over the protected Iroh control ALPN when the peer grants `document.read` +on `resource:document:`. The daemon background live-sync loop runs the same +sync for local documents and known peers using per-peer/per-document cursors. +The import rule is last-writer-wins by document timestamp. Automerge state +encoding and sync are future work. `geth-pubsub` currently supports local publish/subscribe snapshots through the daemon control protocol. Messages live in a bounded in-memory ring buffer and diff --git a/docs/roadmap.md b/docs/roadmap.md index f406a99..4db9464 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -390,12 +390,17 @@ Automerge documents. state. - `[x]` Tests cover create, update, save, and reload of local JSON document state. + - `[x]` `geth document sync ` pulls authorized JSON state + from an imported peer over Iroh. + - `[x]` Background live-sync refreshes local JSON documents from known peers + using per-peer/per-document cursors. - `[ ]` Automerge document state is stored durably. - `[ ]` Tests cover Automerge create, update, save, and reload. - `[ ]` Automerge sync over Iroh. Acceptance criteria: - - Two local test nodes can synchronize document changes. + - `[x]` Bootstrap JSON last-writer-wins sync works across two local test + nodes over Iroh. - Resource authorization gates read/write sync. - Conflicts converge according to Automerge semantics.