diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 9622cfe..30bfdee 100644 --- a/crates/geth-node/src/lib.rs +++ b/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, - OverlayTunRuntime, PipeRuntime, PubsubGossipTopicRuntime, PubsubRuntime, + NodeRuntime, OverlayTunCounters, OverlayTunRuntime, PipeRuntime, PubsubGossipTopicRuntime, + 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 { - 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, -) -> Result { - 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 { - 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::(&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 { - 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::, NodeError>>()?; - streams.sort_by(|left, right| left.stream.cmp(&right.stream)); - Ok(SyncPeerStatus { - peer_node_id: peer.peer_id, - streams, - }) - }) - .collect::, 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 ` 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>, -) -> Result { - 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::(&file_root_cursor.state_json) + serde_json::from_str::(&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::(&db_cursor.state_json) + serde_json::from_str::(&db_cursor.state_json) .expect("parse db cursor") .cursor_ms, 8 diff --git a/crates/geth-node/src/runtime.rs b/crates/geth-node/src/runtime.rs index e71ff11..d84a8ff 100644 --- a/crates/geth-node/src/runtime.rs +++ b/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, } - -#[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, - pub(crate) last_error: Option, - 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, -} diff --git a/crates/geth-node/src/sync.rs b/crates/geth-node/src/sync.rs new file mode 100644 index 0000000..66c1643 --- /dev/null +++ b/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, + last_error: Option, + 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 { + 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, +) -> Result { + 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 { + 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::(&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 { + 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::, NodeError>>()?; + streams.sort_by(|left, right| left.stream.cmp(&right.stream)); + Ok(SyncPeerStatus { + peer_node_id: peer.peer_id, + streams, + }) + }) + .collect::, 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 ` 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>, +) -> Result { + 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) + } +} diff --git a/docs/production-readiness-roadmap.md b/docs/production-readiness-roadmap.md index e2ec6c3..8ea65d3 100644 --- a/docs/production-readiness-roadmap.md +++ b/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 - recording live in a sync-focused module. + - `[x]` Sync cursor keys, cursor persistence, stream health recording, and + 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.