refactor: extract live sync bookkeeping

crates/geth-node/src/lib.rs

@@ -1,5 +1,6 @@
 mod runtime;
 pub mod service;
+mod sync;
 
 use base64::Engine;
 use futures::StreamExt;
@@ -13,7 +14,7 @@ use geth_control::{
     CasBlob, CasProvider, ControlRequest, ControlResponse, KeychainStatusResponse,
     NativeBackendStatus, NodeIdResponse, OverlayPeer, OverlayWireRequest, OverlayWireResponse,
     PeerControlRequest, PeerControlResponse, PipeWireRequest, PipeWireResponse, StatusResponse,
-    SyncPeerRun, SyncPeerStatus, SyncStreamRun, SyncStreamStatus, SyncWatermark,
+    SyncPeerRun, SyncStreamRun, SyncWatermark,
 };
 use geth_crypto::AgentKey;
 use geth_db::DbResource;
@@ -56,13 +57,17 @@ use geth_types::{
 use iroh::protocol::ProtocolHandler;
 use iroh_docs::api::protocol::{AddrInfoOptions, ShareMode};
 use runtime::{
-    KvDocsState, LiveSyncCursor, LiveSyncHealth, NodeRuntime, OverlayTunCounters,
+    NodeRuntime, OverlayTunCounters, OverlayTunRuntime, PipeRuntime, PubsubGossipTopicRuntime,
-    OverlayTunRuntime, PipeRuntime, PubsubGossipTopicRuntime, PubsubRuntime,
+    PubsubRuntime,
 };
 use std::collections::{BTreeMap, BTreeSet};
 use std::path::{Component, Path, PathBuf};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
+use sync::{
+    KvDocsState, load_live_sync_cursor, record_live_sync_failure, record_live_sync_success,
+    should_live_sync_stream, store_live_sync_cursor, sync_status_local,
+};
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream};
 use tokio::sync::{mpsc, oneshot};
@@ -186,8 +191,6 @@ pub struct LocalNode {
 const PUBSUB_RING_LIMIT: usize = 256;
 const PIPE_CONNECTION_RING_LIMIT: usize = 256;
 const PIPE_MESSAGE_RING_LIMIT: usize = 1024;
-const LIVE_SYNC_STALE_AFTER_MS: i64 = 120_000;
-
 struct PipeTcpConnectWire {
     peer_card: PeerCard,
     target_addr: String,
@@ -3709,197 +3712,6 @@ async fn db_sync_from_peer(
     }
 }
 
-fn live_sync_cursor_key(peer_node: &str, stream: &str) -> String {
-    format!("live-sync:{peer_node}:{stream}")
-}
-
-fn live_sync_status_prefix(peer_node: &str) -> String {
-    format!("live-sync-status:{peer_node}:")
-}
-
-fn live_sync_status_key(peer_node: &str, stream: &str) -> String {
-    format!("{}{}", live_sync_status_prefix(peer_node), stream)
-}
-
-fn load_live_sync_cursor(store: &Store, peer_node: &str, stream: &str) -> Result<i64, NodeError> {
-    let key = live_sync_cursor_key(peer_node, stream);
-    let Some(state) = store.get_module_state(&key)? else {
-        return Ok(0);
-    };
-    let cursor: LiveSyncCursor = serde_json::from_str(&state.state_json)?;
-    Ok(cursor.cursor_ms)
-}
-
-fn store_live_sync_cursor(
-    store: &Store,
-    peer_node: &str,
-    stream: &str,
-    cursor_ms: i64,
-) -> Result<(), NodeError> {
-    store.put_module_state(&StoredModuleState {
-        module: live_sync_cursor_key(peer_node, stream),
-        state_json: serde_json::to_string(&LiveSyncCursor { cursor_ms })?,
-        updated_at_ms: geth_store::now_ms(),
-    })?;
-    Ok(())
-}
-
-fn record_live_sync_success(
-    store: &Store,
-    peer_node: &str,
-    stream: &str,
-    imported: usize,
-    rejected: usize,
-    cursor_ms: Option<i64>,
-) -> Result<i64, NodeError> {
-    let now = geth_store::now_ms();
-    if let Some(cursor_ms) = cursor_ms {
-        store_live_sync_cursor(store, peer_node, stream, cursor_ms)?;
-    }
-    store.put_module_state(&StoredModuleState {
-        module: live_sync_status_key(peer_node, stream),
-        state_json: serde_json::to_string(&LiveSyncHealth {
-            last_attempt_ms: now,
-            last_success_ms: Some(now),
-            last_error: None,
-            last_imported: imported,
-            last_rejected: rejected,
-        })?,
-        updated_at_ms: now,
-    })?;
-    load_live_sync_cursor(store, peer_node, stream)
-}
-
-fn record_live_sync_failure(
-    store: &Store,
-    peer_node: &str,
-    stream: &str,
-    error: &NodeError,
-) -> Result<i64, NodeError> {
-    let now = geth_store::now_ms();
-    let previous = store
-        .get_module_state(&live_sync_status_key(peer_node, stream))?
-        .and_then(|state| serde_json::from_str::<LiveSyncHealth>(&state.state_json).ok());
-    store.put_module_state(&StoredModuleState {
-        module: live_sync_status_key(peer_node, stream),
-        state_json: serde_json::to_string(&LiveSyncHealth {
-            last_attempt_ms: now,
-            last_success_ms: previous.and_then(|health| health.last_success_ms),
-            last_error: Some(error.to_string()),
-            last_imported: 0,
-            last_rejected: 0,
-        })?,
-        updated_at_ms: now,
-    })?;
-    load_live_sync_cursor(store, peer_node, stream)
-}
-
-fn sync_status_local(store: &Store) -> Result<ControlResponse, NodeError> {
-    let peers = store
-        .list_peer_cards()?
-        .into_iter()
-        .map(|peer| {
-            let prefix = live_sync_status_prefix(&peer.peer_id);
-            let now_ms = geth_store::now_ms();
-            let mut streams = store
-                .list_module_states_with_prefix(&prefix)?
-                .into_iter()
-                .map(|state| {
-                    let stream = state.module.trim_start_matches(&prefix).to_owned();
-                    let health: LiveSyncHealth = serde_json::from_str(&state.state_json)?;
-                    let cursor_ms = load_live_sync_cursor(store, &peer.peer_id, &stream)?;
-                    let stale = health.last_success_ms.is_some_and(|last_success_ms| {
-                        now_ms.saturating_sub(last_success_ms) > LIVE_SYNC_STALE_AFTER_MS
-                    });
-                    let status_state = sync_stream_state(&health, stale).to_owned();
-                    let next_action =
-                        sync_stream_next_action(&peer.peer_id, &stream, &health, stale);
-                    Ok(SyncStreamStatus {
-                        stream,
-                        cursor_ms,
-                        state: status_state,
-                        stale,
-                        stale_after_ms: LIVE_SYNC_STALE_AFTER_MS,
-                        last_attempt_ms: Some(health.last_attempt_ms),
-                        last_success_ms: health.last_success_ms,
-                        last_error: health.last_error,
-                        last_imported: health.last_imported,
-                        last_rejected: health.last_rejected,
-                        next_action,
-                    })
-                })
-                .collect::<Result<Vec<_>, NodeError>>()?;
-            streams.sort_by(|left, right| left.stream.cmp(&right.stream));
-            Ok(SyncPeerStatus {
-                peer_node_id: peer.peer_id,
-                streams,
-            })
-        })
-        .collect::<Result<Vec<_>, NodeError>>()?;
-    Ok(ControlResponse::SyncStatus {
-        peers,
-        note: "sync status is local daemon health for best-effort live sync; signed logs remain the durable source of truth".to_owned(),
-    })
-}
-
-fn sync_stream_state(health: &LiveSyncHealth, stale: bool) -> &'static str {
-    if health.last_error.is_some() {
-        "failed"
-    } else if health.last_success_ms.is_none() {
-        "never-succeeded"
-    } else if stale {
-        "stale"
-    } else {
-        "ok"
-    }
-}
-
-fn sync_stream_next_action(
-    peer_node: &str,
-    stream: &str,
-    health: &LiveSyncHealth,
-    stale: bool,
-) -> String {
-    if health.last_error.is_some() {
-        format!(
-            "run `geth sync now {peer_node}`; if it still fails, run `geth auth explain <subject> <resource> <capability>` for stream {stream}"
-        )
-    } else if health.last_success_ms.is_none() {
-        format!(
-            "run `geth sync now {peer_node}` to establish the first successful sync for {stream}"
-        )
-    } else if stale {
-        format!(
-            "run `geth sync now {peer_node}` and check `geth peer ping {peer_node}` if the stream remains stale"
-        )
-    } else {
-        "none".to_owned()
-    }
-}
-
-fn should_live_sync_stream(
-    store: &Store,
-    peer_node: &str,
-    stream: &str,
-    remote_watermarks: Option<&BTreeMap<String, i64>>,
-) -> Result<bool, NodeError> {
-    let Some(remote_watermarks) = remote_watermarks else {
-        return Ok(true);
-    };
-    let Some(remote_high_water) = remote_watermarks.get(stream) else {
-        return Ok(false);
-    };
-    if *remote_high_water == 0 {
-        return Ok(false);
-    }
-    let local_cursor = load_live_sync_cursor(store, peer_node, stream)?;
-    if stream.starts_with("db:") {
-        Ok(*remote_high_water > local_cursor)
-    } else {
-        Ok(*remote_high_water >= local_cursor)
-    }
-}
-
 fn sync_watermarks_for_peer(
     store: &Store,
     peer_node: &str,
@@ -13036,7 +12848,10 @@ mod tests {
         assert_eq!(peers[0].streams[0].cursor_ms, 42);
         assert_eq!(peers[0].streams[0].state, "ok");
         assert!(!peers[0].streams[0].stale);
-        assert_eq!(peers[0].streams[0].stale_after_ms, LIVE_SYNC_STALE_AFTER_MS);
+        assert_eq!(
+            peers[0].streams[0].stale_after_ms,
+            sync::LIVE_SYNC_STALE_AFTER_MS
+        );
         assert_eq!(peers[0].streams[0].last_imported, 3);
         assert_eq!(peers[0].streams[0].last_rejected, 1);
         assert!(peers[0].streams[0].last_error.is_none());
@@ -14784,27 +14599,27 @@ mod tests {
                 .expect("left has live-synced tree object")
         );
         let file_root_cursor = left_store
-            .get_module_state(&live_sync_cursor_key(
+            .get_module_state(&sync::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)
+            serde_json::from_str::<sync::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(
+            .get_module_state(&sync::live_sync_cursor_key(
                 right_card.node_id.as_str(),
                 "db:notes",
             ))
             .expect("get live-synced db cursor")
             .expect("db cursor exists");
         assert_eq!(
-            serde_json::from_str::<LiveSyncCursor>(&db_cursor.state_json)
+            serde_json::from_str::<sync::LiveSyncCursor>(&db_cursor.state_json)
                 .expect("parse db cursor")
                 .cursor_ms,
             8

crates/geth-node/src/runtime.rs

@@ -58,25 +58,3 @@ pub(crate) struct OverlayTunCounters {
     pub(crate) packets_to_peers: u64,
     pub(crate) last_error: Option<String>,
 }
-
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
-pub(crate) struct LiveSyncCursor {
-    pub(crate) cursor_ms: i64,
-}
-
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
-pub(crate) struct LiveSyncHealth {
-    pub(crate) last_attempt_ms: i64,
-    pub(crate) last_success_ms: Option<i64>,
-    pub(crate) last_error: Option<String>,
-    pub(crate) last_imported: usize,
-    pub(crate) last_rejected: usize,
-}
-
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
-pub(crate) struct KvDocsState {
-    pub(crate) name: String,
-    pub(crate) namespace_id: String,
-    pub(crate) read_ticket: String,
-    pub(crate) updated_at_ms: i64,
-}

crates/geth-node/src/sync.rs

@@ -0,0 +1,230 @@
+//! Live-sync cursor and health bookkeeping.
+//!
+//! This module owns the daemon-local metadata keys and status reduction for
+//! best-effort live sync. Resource-specific pull logic still lives with the
+//! module handlers; this layer records whether streams are healthy and whether
+//! remote watermarks warrant another pull.
+
+use crate::NodeError;
+use geth_control::{ControlResponse, SyncPeerStatus, SyncStreamStatus};
+use geth_store::{Store, StoredModuleState};
+use std::collections::BTreeMap;
+
+pub(crate) const LIVE_SYNC_STALE_AFTER_MS: i64 = 120_000;
+
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+pub(crate) struct LiveSyncCursor {
+    pub(crate) cursor_ms: i64,
+}
+
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+struct LiveSyncHealth {
+    last_attempt_ms: i64,
+    last_success_ms: Option<i64>,
+    last_error: Option<String>,
+    last_imported: usize,
+    last_rejected: usize,
+}
+
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+pub(crate) struct KvDocsState {
+    pub(crate) name: String,
+    pub(crate) namespace_id: String,
+    pub(crate) read_ticket: String,
+    pub(crate) updated_at_ms: i64,
+}
+
+pub(crate) fn live_sync_cursor_key(peer_node: &str, stream: &str) -> String {
+    format!("live-sync:{peer_node}:{stream}")
+}
+
+fn live_sync_status_prefix(peer_node: &str) -> String {
+    format!("live-sync-status:{peer_node}:")
+}
+
+fn live_sync_status_key(peer_node: &str, stream: &str) -> String {
+    format!("{}{}", live_sync_status_prefix(peer_node), stream)
+}
+
+pub(crate) fn load_live_sync_cursor(
+    store: &Store,
+    peer_node: &str,
+    stream: &str,
+) -> Result<i64, NodeError> {
+    let key = live_sync_cursor_key(peer_node, stream);
+    let Some(state) = store.get_module_state(&key)? else {
+        return Ok(0);
+    };
+    let cursor: LiveSyncCursor = serde_json::from_str(&state.state_json)?;
+    Ok(cursor.cursor_ms)
+}
+
+pub(crate) fn store_live_sync_cursor(
+    store: &Store,
+    peer_node: &str,
+    stream: &str,
+    cursor_ms: i64,
+) -> Result<(), NodeError> {
+    store.put_module_state(&StoredModuleState {
+        module: live_sync_cursor_key(peer_node, stream),
+        state_json: serde_json::to_string(&LiveSyncCursor { cursor_ms })?,
+        updated_at_ms: geth_store::now_ms(),
+    })?;
+    Ok(())
+}
+
+pub(crate) fn record_live_sync_success(
+    store: &Store,
+    peer_node: &str,
+    stream: &str,
+    imported: usize,
+    rejected: usize,
+    cursor_ms: Option<i64>,
+) -> Result<i64, NodeError> {
+    let now = geth_store::now_ms();
+    if let Some(cursor_ms) = cursor_ms {
+        store_live_sync_cursor(store, peer_node, stream, cursor_ms)?;
+    }
+    store.put_module_state(&StoredModuleState {
+        module: live_sync_status_key(peer_node, stream),
+        state_json: serde_json::to_string(&LiveSyncHealth {
+            last_attempt_ms: now,
+            last_success_ms: Some(now),
+            last_error: None,
+            last_imported: imported,
+            last_rejected: rejected,
+        })?,
+        updated_at_ms: now,
+    })?;
+    load_live_sync_cursor(store, peer_node, stream)
+}
+
+pub(crate) fn record_live_sync_failure(
+    store: &Store,
+    peer_node: &str,
+    stream: &str,
+    error: &NodeError,
+) -> Result<i64, NodeError> {
+    let now = geth_store::now_ms();
+    let previous = store
+        .get_module_state(&live_sync_status_key(peer_node, stream))?
+        .and_then(|state| serde_json::from_str::<LiveSyncHealth>(&state.state_json).ok());
+    store.put_module_state(&StoredModuleState {
+        module: live_sync_status_key(peer_node, stream),
+        state_json: serde_json::to_string(&LiveSyncHealth {
+            last_attempt_ms: now,
+            last_success_ms: previous.and_then(|health| health.last_success_ms),
+            last_error: Some(error.to_string()),
+            last_imported: 0,
+            last_rejected: 0,
+        })?,
+        updated_at_ms: now,
+    })?;
+    load_live_sync_cursor(store, peer_node, stream)
+}
+
+pub(crate) fn sync_status_local(store: &Store) -> Result<ControlResponse, NodeError> {
+    let peers = store
+        .list_peer_cards()?
+        .into_iter()
+        .map(|peer| {
+            let prefix = live_sync_status_prefix(&peer.peer_id);
+            let now_ms = geth_store::now_ms();
+            let mut streams = store
+                .list_module_states_with_prefix(&prefix)?
+                .into_iter()
+                .map(|state| {
+                    let stream = state.module.trim_start_matches(&prefix).to_owned();
+                    let health: LiveSyncHealth = serde_json::from_str(&state.state_json)?;
+                    let cursor_ms = load_live_sync_cursor(store, &peer.peer_id, &stream)?;
+                    let stale = health.last_success_ms.is_some_and(|last_success_ms| {
+                        now_ms.saturating_sub(last_success_ms) > LIVE_SYNC_STALE_AFTER_MS
+                    });
+                    let status_state = sync_stream_state(&health, stale).to_owned();
+                    let next_action =
+                        sync_stream_next_action(&peer.peer_id, &stream, &health, stale);
+                    Ok(SyncStreamStatus {
+                        stream,
+                        cursor_ms,
+                        state: status_state,
+                        stale,
+                        stale_after_ms: LIVE_SYNC_STALE_AFTER_MS,
+                        last_attempt_ms: Some(health.last_attempt_ms),
+                        last_success_ms: health.last_success_ms,
+                        last_error: health.last_error,
+                        last_imported: health.last_imported,
+                        last_rejected: health.last_rejected,
+                        next_action,
+                    })
+                })
+                .collect::<Result<Vec<_>, NodeError>>()?;
+            streams.sort_by(|left, right| left.stream.cmp(&right.stream));
+            Ok(SyncPeerStatus {
+                peer_node_id: peer.peer_id,
+                streams,
+            })
+        })
+        .collect::<Result<Vec<_>, NodeError>>()?;
+    Ok(ControlResponse::SyncStatus {
+        peers,
+        note: "sync status is local daemon health for best-effort live sync; signed logs remain the durable source of truth".to_owned(),
+    })
+}
+
+fn sync_stream_state(health: &LiveSyncHealth, stale: bool) -> &'static str {
+    if health.last_error.is_some() {
+        "failed"
+    } else if health.last_success_ms.is_none() {
+        "never-succeeded"
+    } else if stale {
+        "stale"
+    } else {
+        "ok"
+    }
+}
+
+fn sync_stream_next_action(
+    peer_node: &str,
+    stream: &str,
+    health: &LiveSyncHealth,
+    stale: bool,
+) -> String {
+    if health.last_error.is_some() {
+        format!(
+            "run `geth sync now {peer_node}`; if it still fails, run `geth auth explain <subject> <resource> <capability>` for stream {stream}"
+        )
+    } else if health.last_success_ms.is_none() {
+        format!(
+            "run `geth sync now {peer_node}` to establish the first successful sync for {stream}"
+        )
+    } else if stale {
+        format!(
+            "run `geth sync now {peer_node}` and check `geth peer ping {peer_node}` if the stream remains stale"
+        )
+    } else {
+        "none".to_owned()
+    }
+}
+
+pub(crate) fn should_live_sync_stream(
+    store: &Store,
+    peer_node: &str,
+    stream: &str,
+    remote_watermarks: Option<&BTreeMap<String, i64>>,
+) -> Result<bool, NodeError> {
+    let Some(remote_watermarks) = remote_watermarks else {
+        return Ok(true);
+    };
+    let Some(remote_high_water) = remote_watermarks.get(stream) else {
+        return Ok(false);
+    };
+    if *remote_high_water == 0 {
+        return Ok(false);
+    }
+    let local_cursor = load_live_sync_cursor(store, peer_node, stream)?;
+    if stream.starts_with("db:") {
+        Ok(*remote_high_water > local_cursor)
+    } else {
+        Ok(*remote_high_water >= local_cursor)
+    }
+}

docs/production-readiness-roadmap.md

@@ -67,10 +67,12 @@ behavior.
     points.
   - `[ ]` No generic `geth-common` crate is introduced.
 
-- `[ ]` Extract live-sync engine.
+- `[~]` Extract live-sync engine.
   Acceptance criteria:
-  - `[ ]` Sync stream selection, watermarks, cursors, run results, and health
+  - `[x]` Sync cursor keys, cursor persistence, stream health recording, and
-    recording live in a sync-focused module.
+    local sync-status reduction live in a sync-focused module.
+  - `[ ]` Sync stream selection, watermarks, and run result helpers live in a
+    sync-focused module.
   - `[ ]` Per-module sync handlers have consistent interfaces.
   - `[ ]` `geth sync status --json` output remains stable.