Wire KV sync through iroh-docs

crates/geth-node/Cargo.toml

@@ -32,9 +32,12 @@ geth-ssh-identity = { path = "../geth-ssh-identity" }
 geth-ssh-proxy = { path = "../geth-ssh-proxy" }
 geth-store = { path = "../geth-store" }
 geth-types = { path = "../geth-types" }
+futures.workspace = true
 hex.workspace = true
 iroh.workspace = true
 iroh-blobs.workspace = true
+iroh-docs.workspace = true
+iroh-gossip.workspace = true
 swarm-discovery.workspace = true
 
 [dev-dependencies]

crates/geth-node/src/lib.rs

@@ -1,6 +1,7 @@
 pub mod service;
 
 use base64::Engine;
+use futures::StreamExt;
 use geth_auth::{AUTH_SIGNATURE_NAMESPACE, AuthExplanation, AuthOp, AuthOpKind, AuthOpSignature};
 use geth_cas::{
     BlobInfoSummary, CasTreeEntryKind, CasTreeObject, FileConflict, FileConflictKind,
@@ -52,6 +53,7 @@ use geth_types::{
     ResourceName, SshCertId, SshCertRequestId, UnixMillis, UserId,
 };
 use iroh::protocol::ProtocolHandler;
+use iroh_docs::api::protocol::{AddrInfoOptions, ShareMode};
 use std::collections::{BTreeMap, BTreeSet, VecDeque};
 use std::path::{Component, Path, PathBuf};
 use std::sync::{Arc, Mutex};
@@ -159,6 +161,8 @@ pub struct LocalNode {
     pub iroh_status: EndpointStatus,
     iroh_endpoint: Arc<Mutex<Option<GethIrohEndpoint>>>,
     iroh_blob_store: Arc<Mutex<Option<iroh_blobs::store::fs::FsStore>>>,
+    iroh_docs: Arc<Mutex<Option<iroh_docs::protocol::Docs>>>,
+    iroh_gossip: Arc<Mutex<Option<iroh_gossip::net::Gossip>>>,
     runtime: Arc<NodeRuntime>,
 }
 
@@ -199,6 +203,14 @@ struct LiveSyncHealth {
     last_rejected: usize,
 }
 
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+struct KvDocsState {
+    name: String,
+    namespace_id: String,
+    read_ticket: String,
+    updated_at_ms: i64,
+}
+
 struct PipeTcpConnectWire {
     peer_card: PeerCard,
     target_addr: String,
@@ -268,6 +280,8 @@ pub fn init_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
         iroh_status: EndpointStatus::scaffolded(),
         iroh_endpoint: Arc::new(Mutex::new(None)),
         iroh_blob_store: Arc::new(Mutex::new(None)),
+        iroh_docs: Arc::new(Mutex::new(None)),
+        iroh_gossip: Arc::new(Mutex::new(None)),
         runtime: Arc::new(NodeRuntime {
             pubsub: Mutex::new(PubsubRuntime::default()),
             pipes: Mutex::new(PipeRuntime::default()),
@@ -760,6 +774,17 @@ pub async fn handle_request_async(
                 ControlResponse::CasPrivateAdded { encrypted_hash, .. } => {
                     mirror_local_blob_to_iroh_blobs(node, encrypted_hash).await?;
                 }
+                ControlResponse::KvCreated { kv } => {
+                    mirror_kv_store_to_iroh_docs(node, kv.name.as_str()).await?;
+                }
+                ControlResponse::KvSet { entry } => {
+                    let name = entry
+                        .store
+                        .as_str()
+                        .strip_prefix("kv:")
+                        .unwrap_or(entry.store.as_str());
+                    mirror_kv_store_to_iroh_docs(node, name).await?;
+                }
                 _ => {}
             }
             Ok(response)
@@ -1849,6 +1874,7 @@ async fn kv_sync_from_peer(
             name: response_name,
             entries,
             high_water_ms,
+            docs_ticket,
             allowed,
             reason,
             note,
@@ -1866,9 +1892,13 @@ async fn kv_sync_from_peer(
                     note,
                 });
             }
+            let mut entries_imported = 0;
+            if let Some(docs_ticket) = docs_ticket {
+                entries_imported +=
+                    import_kv_entries_from_iroh_docs(node, name, &docs_ticket).await?;
+            }
             let store = Store::open(&node.paths.metadata_db())?;
             let kv = ensure_local_kv_store(&store, name)?;
-            let mut entries_imported = 0;
             for entry in entries {
                 geth_kv::validate_kv_key(&entry.key)
                     .map_err(|_| NodeError::InvalidKvKey(entry.key.clone()))?;
@@ -1886,6 +1916,8 @@ async fn kv_sync_from_peer(
                 }
             }
             store_live_sync_cursor(&store, peer_node, &stream, high_water_ms)?;
+            drop(store);
+            mirror_kv_store_to_iroh_docs(node, name).await?;
             Ok(ControlResponse::KvSynced {
                 peer_node_id: node_id,
                 peer_agent_id: agent_id,
@@ -1907,6 +1939,227 @@ async fn kv_sync_from_peer(
     }
 }
 
+fn iroh_docs(node: &LocalNode) -> Result<Option<iroh_docs::protocol::Docs>, NodeError> {
+    node.iroh_docs
+        .lock()
+        .map_err(|_| NodeError::RuntimeLockPoisoned)
+        .map(|docs| docs.clone())
+}
+
+fn iroh_gossip(node: &LocalNode) -> Result<Option<iroh_gossip::net::Gossip>, NodeError> {
+    node.iroh_gossip
+        .lock()
+        .map_err(|_| NodeError::RuntimeLockPoisoned)
+        .map(|gossip| gossip.clone())
+}
+
+fn kv_docs_state_key(name: &str) -> String {
+    format!("kv-doc:{name}")
+}
+
+fn kv_docs_meta_key() -> &'static str {
+    "__geth/kv/name"
+}
+
+async fn ensure_kv_docs_state(
+    node: &LocalNode,
+    name: &str,
+) -> Result<Option<(iroh_docs::api::Doc, KvDocsState)>, NodeError> {
+    let Some(docs) = iroh_docs(node)? else {
+        return Ok(None);
+    };
+    let stored_state = Store::open(&node.paths.metadata_db())?
+        .get_module_state(&kv_docs_state_key(name))?
+        .map(|stored| serde_json::from_str::<KvDocsState>(&stored.state_json))
+        .transpose()?;
+    if let Some(state) = stored_state {
+        let namespace = state
+            .namespace_id
+            .parse::<iroh_docs::NamespaceId>()
+            .map_err(|error| NodeError::IrohPeer(format!("invalid KV docs namespace: {error}")))?;
+        if let Some(doc) = docs
+            .open(namespace)
+            .await
+            .map_err(|error| NodeError::IrohPeer(format!("iroh-docs open failed: {error}")))?
+        {
+            return Ok(Some((doc, state)));
+        }
+    }
+
+    let doc = docs
+        .create()
+        .await
+        .map_err(|error| NodeError::IrohPeer(format!("iroh-docs create failed: {error}")))?;
+    let ticket = doc
+        .share(ShareMode::Read, AddrInfoOptions::RelayAndAddresses)
+        .await
+        .map_err(|error| NodeError::IrohPeer(format!("iroh-docs share failed: {error}")))?;
+    let state = KvDocsState {
+        name: name.to_owned(),
+        namespace_id: doc.id().to_string(),
+        read_ticket: ticket.to_string(),
+        updated_at_ms: geth_store::now_ms(),
+    };
+    let store = Store::open(&node.paths.metadata_db())?;
+    store.put_module_state(&StoredModuleState {
+        module: kv_docs_state_key(name),
+        state_json: serde_json::to_string(&state)?,
+        updated_at_ms: state.updated_at_ms,
+    })?;
+    Ok(Some((doc, state)))
+}
+
+async fn kv_docs_read_ticket(node: &LocalNode, name: &str) -> Result<Option<String>, NodeError> {
+    Ok(ensure_kv_docs_state(node, name)
+        .await?
+        .map(|(_, state)| state.read_ticket))
+}
+
+async fn mirror_kv_store_to_iroh_docs(node: &LocalNode, name: &str) -> Result<(), NodeError> {
+    let Some(docs) = iroh_docs(node)? else {
+        return Ok(());
+    };
+    let entries = {
+        let store = Store::open(&node.paths.metadata_db())?;
+        let kv = ensure_local_kv_store(&store, name)?;
+        store.list_kv_entries_since(&kv.kv_id, 0)?
+    };
+    let Some((doc, _state)) = ensure_kv_docs_state(node, name).await? else {
+        return Ok(());
+    };
+    let author = docs
+        .author_default()
+        .await
+        .map_err(|error| NodeError::IrohPeer(format!("iroh-docs author failed: {error}")))?;
+    doc.set_bytes(
+        author,
+        kv_docs_meta_key().as_bytes().to_vec(),
+        name.as_bytes().to_vec(),
+    )
+    .await
+    .map_err(|error| NodeError::IrohPeer(format!("iroh-docs set metadata failed: {error}")))?;
+    for entry in entries {
+        doc.set_bytes(author, entry.key.into_bytes(), entry.value.into_bytes())
+            .await
+            .map_err(|error| NodeError::IrohPeer(format!("iroh-docs set failed: {error}")))?;
+    }
+    Ok(())
+}
+
+async fn mirror_all_kv_stores_to_iroh_docs(node: &LocalNode) -> Result<usize, NodeError> {
+    if iroh_docs(node)?.is_none() {
+        return Ok(0);
+    }
+    let store = Store::open(&node.paths.metadata_db())?;
+    let mut mirrored = 0;
+    for kv in store.list_kv_stores()? {
+        mirror_kv_store_to_iroh_docs(node, &kv.name).await?;
+        mirrored += 1;
+    }
+    Ok(mirrored)
+}
+
+async fn import_kv_entries_from_iroh_docs(
+    node: &LocalNode,
+    name: &str,
+    ticket: &str,
+) -> Result<usize, NodeError> {
+    let Some(docs) = iroh_docs(node)? else {
+        return Ok(0);
+    };
+    let Some(blob_store) = iroh_blob_store(node)? else {
+        return Ok(0);
+    };
+    let ticket = ticket
+        .parse::<iroh_docs::DocTicket>()
+        .map_err(|error| NodeError::IrohPeer(format!("invalid iroh-docs ticket: {error}")))?;
+    let doc = docs
+        .import(ticket.clone())
+        .await
+        .map_err(|error| NodeError::IrohPeer(format!("iroh-docs import failed: {error}")))?;
+    let state = KvDocsState {
+        name: name.to_owned(),
+        namespace_id: doc.id().to_string(),
+        read_ticket: ticket.to_string(),
+        updated_at_ms: geth_store::now_ms(),
+    };
+    let store = Store::open(&node.paths.metadata_db())?;
+    store.put_module_state(&StoredModuleState {
+        module: kv_docs_state_key(name),
+        state_json: serde_json::to_string(&state)?,
+        updated_at_ms: state.updated_at_ms,
+    })?;
+
+    let mut imported = 0;
+    for _ in 0..5 {
+        imported += import_kv_doc_entries_once(node, name, &blob_store, &doc).await?;
+        if imported > 0 {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+    Ok(imported)
+}
+
+async fn import_kv_doc_entries_once(
+    node: &LocalNode,
+    name: &str,
+    blob_store: &iroh_blobs::store::fs::FsStore,
+    doc: &iroh_docs::api::Doc,
+) -> Result<usize, NodeError> {
+    let mut imported_entries = Vec::new();
+    let entries = doc
+        .get_many(iroh_docs::store::Query::single_latest_per_key())
+        .await
+        .map_err(|error| NodeError::IrohPeer(format!("iroh-docs query failed: {error}")))?;
+    futures::pin_mut!(entries);
+    while let Some(entry) = entries.next().await {
+        let entry = entry
+            .map_err(|error| NodeError::IrohPeer(format!("iroh-docs entry failed: {error}")))?;
+        let key = String::from_utf8(entry.key().to_vec())
+            .map_err(|error| NodeError::IrohPeer(format!("invalid KV docs key: {error}")))?;
+        if key == kv_docs_meta_key() {
+            continue;
+        }
+        geth_kv::validate_kv_key(&key).map_err(|_| NodeError::InvalidKvKey(key.clone()))?;
+        if entry.content_len() > 1024 * 1024 {
+            return Err(NodeError::IrohPeer(format!(
+                "KV docs value for {key} is too large: {} bytes",
+                entry.content_len()
+            )));
+        }
+        let value = blob_store
+            .get_bytes(entry.content_hash())
+            .await
+            .map_err(|error| {
+                NodeError::IrohPeer(format!("iroh-docs blob export failed: {error}"))
+            })?;
+        let value = String::from_utf8(value.to_vec())
+            .map_err(|error| NodeError::IrohPeer(format!("invalid KV docs value: {error}")))?;
+        let updated_at_ms = (entry.timestamp() / 1000) as i64;
+        imported_entries.push((key, value, updated_at_ms));
+    }
+
+    let store = Store::open(&node.paths.metadata_db())?;
+    let kv = ensure_local_kv_store(&store, name)?;
+    let mut imported = 0;
+    for (key, value, updated_at_ms) in imported_entries {
+        let should_import = store
+            .get_kv_entry(&kv.kv_id, &key)?
+            .is_none_or(|local| updated_at_ms >= local.updated_at_ms);
+        if should_import {
+            store.set_kv_entry(&StoredKvEntry {
+                kv_id: kv.kv_id.clone(),
+                key,
+                value,
+                updated_at_ms,
+            })?;
+            imported += 1;
+        }
+    }
+    Ok(imported)
+}
+
 async fn pubsub_publish_to_peer(
     node: &LocalNode,
     peer_node: &str,
@@ -4558,6 +4811,28 @@ async fn handle_iroh_control_connection(
             .await
             .map_err(|error| NodeError::IrohPeer(format!("iroh-blobs accept failed: {error}")));
     }
+    if alpn == display_alpn(iroh_docs::ALPN) {
+        let docs = iroh_docs(&node)?.ok_or_else(|| {
+            NodeError::IrohPeer(
+                "iroh-docs ALPN accepted but docs runtime is unavailable".to_owned(),
+            )
+        })?;
+        return docs
+            .accept(conn)
+            .await
+            .map_err(|error| NodeError::IrohPeer(format!("iroh-docs accept failed: {error}")));
+    }
+    if alpn == display_alpn(iroh_gossip::ALPN) {
+        let gossip = iroh_gossip(&node)?.ok_or_else(|| {
+            NodeError::IrohPeer(
+                "iroh-gossip ALPN accepted but gossip runtime is unavailable".to_owned(),
+            )
+        })?;
+        return gossip
+            .accept(conn)
+            .await
+            .map_err(|error| NodeError::IrohPeer(format!("iroh-gossip accept failed: {error}")));
+    }
     let (mut send, mut recv) = conn
         .accept_bi()
         .await
@@ -5094,6 +5369,13 @@ async fn handle_iroh_control_connection(
                 } else {
                     Vec::new()
                 };
+                let docs_ticket = if explanation.allowed {
+                    drop(store);
+                    mirror_kv_store_to_iroh_docs(&node, &name).await?;
+                    kv_docs_read_ticket(&node, &name).await?
+                } else {
+                    None
+                };
                 PeerControlResponse::KvSynced {
                     node_id: node.node_id.clone(),
                     agent_id: node.agent_id.clone(),
@@ -5102,6 +5384,7 @@ async fn handle_iroh_control_connection(
                     name,
                     entries,
                     high_water_ms,
+                    docs_ticket,
                     allowed: explanation.allowed,
                     reason: explanation.reason,
                     evaluated_ops: explanation.evaluated_ops,
@@ -7779,17 +8062,39 @@ pub fn handle_request(
 fn native_backend_statuses() -> Vec<NativeBackendStatus> {
     geth_iroh::native_iroh_libraries()
         .into_iter()
-        .map(|library| NativeBackendStatus {
-            module: library.module,
-            current_backend: "shared-iroh-endpoint-ready".to_owned(),
-            target_crate: library.crate_name,
-            target_version: library.crate_version,
-            status: "ready-to-wire".to_owned(),
-            blocker: format!(
-                "none; native ALPN {} is compiled against the daemon-owned iroh {} endpoint, but the module still needs its bootstrap control-path implementation replaced",
-                library.alpn,
-                geth_iroh::IROH_VERSION
-            ),
+        .map(|library| {
+            let (status, blocker) = match library.module.as_str() {
+                "cas" => (
+                    "wired",
+                    format!(
+                        "none; CAS payload fetches use native ALPN {} after geth control authorization preflight",
+                        library.alpn
+                    ),
+                ),
+                "kv" => (
+                    "wired",
+                    format!(
+                        "none; KV stores mirror into Iroh Documents over native ALPN {} after geth control authorization preflight",
+                        library.alpn
+                    ),
+                ),
+                _ => (
+                    "ready-to-wire",
+                    format!(
+                        "none; native ALPN {} is compiled against the daemon-owned iroh {} endpoint, but the module still needs its bootstrap control-path implementation replaced",
+                        library.alpn,
+                        geth_iroh::IROH_VERSION
+                    ),
+                ),
+            };
+            NativeBackendStatus {
+                module: library.module,
+                current_backend: "shared-iroh-endpoint-ready".to_owned(),
+                target_crate: library.crate_name,
+                target_version: library.crate_version,
+                status: status.to_owned(),
+                blocker,
+            }
         })
         .collect()
 }
@@ -9679,6 +9984,14 @@ async fn start_daemon_iroh_endpoint(
                     .map_err(|error| {
                         NodeError::IrohPeer(format!("failed to open iroh-blobs store: {error}"))
                     })?;
+            let gossip = iroh_gossip::net::Gossip::builder().spawn(endpoint.endpoint());
+            let blob_api: iroh_blobs::api::Store = blob_store.clone().into();
+            let docs = iroh_docs::protocol::Docs::persistent(node.paths.home().join("iroh-docs"))
+                .spawn(endpoint.endpoint(), blob_api, gossip.clone())
+                .await
+                .map_err(|error| {
+                    NodeError::IrohPeer(format!("failed to start iroh-docs: {error}"))
+                })?;
             let mirrored = mirror_local_cas_to_iroh_blobs(&node.paths, &blob_store).await?;
             tracing::info!(
                 mirrored_blobs = mirrored,
@@ -9693,6 +10006,19 @@ async fn start_daemon_iroh_endpoint(
                 .iroh_blob_store
                 .lock()
                 .map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(blob_store);
+            *node
+                .iroh_gossip
+                .lock()
+                .map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(gossip);
+            *node
+                .iroh_docs
+                .lock()
+                .map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(docs);
+            let mirrored_kv = mirror_all_kv_stores_to_iroh_docs(node).await?;
+            tracing::info!(
+                mirrored_kv_stores = mirrored_kv,
+                "native iroh-docs KV store is ready"
+            );
             Ok(Some(endpoint))
         }
         Err(error) => {
@@ -11510,6 +11836,16 @@ mod tests {
             }
             other => panic!("unexpected synced KV get response: {other:?}"),
         }
+        let left_kv_doc_state = Store::open(&left_paths.metadata_db())
+            .expect("open left store for kv doc state")
+            .get_module_state("kv-doc:prefs")
+            .expect("get left kv doc state")
+            .expect("left imported kv docs state");
+        let left_kv_doc_state: KvDocsState =
+            serde_json::from_str(&left_kv_doc_state.state_json).expect("parse kv docs state");
+        assert_eq!(left_kv_doc_state.name, "prefs");
+        assert!(!left_kv_doc_state.namespace_id.is_empty());
+        assert!(left_kv_doc_state.read_ticket.starts_with("doc"));
 
         let published = handle_request_async(
             &left,