From 2dce1f41cf843dac3158a17f0697ed674bf8998b Mon Sep 17 00:00:00 2001 From: Eric Wendland Date: Sun, 5 Jul 2026 22:57:52 +0200 Subject: [PATCH] feat: add live sync retry backoff --- README.md | 11 ++- crates/geth-cli/src/lib.rs | 7 +- crates/geth-control/src/lib.rs | 9 ++ crates/geth-node/src/lib.rs | 64 +++++++++++- crates/geth-node/src/sync.rs | 142 ++++++++++++++++++++++++--- docs/architecture.md | 11 ++- docs/production-readiness-roadmap.md | 8 +- docs/roadmap.md | 4 +- 8 files changed, 225 insertions(+), 31 deletions(-) diff --git a/README.md b/README.md index f484ee9..5a4eacc 100644 --- a/README.md +++ b/README.md @@ -584,10 +584,13 @@ The daemon also runs best-effort live sync for imported peers. `geth sync now [node]` triggers the same sync pass immediately, and `geth sync status` reports the last local attempt, success, cursor, import count, rejection count, and error per peer stream. `geth sync status --json` also includes per-stream -`state`, `stale`, `stale_after_ms`, and `next_action` fields so smoke tests can -fail on stale or failed streams. Keychain and auth sync now use per-peer -high-water cursors, while receivers still verify every imported signed operation -before it can affect the reduced keychain or authorization views. +`state`, `stale`, `stale_after_ms`, `consecutive_failures`, `retry_after_ms`, +`retry_in_ms`, and `next_action` fields so smoke tests can fail on stale or +failed streams. Background live-sync backs off failed streams, while +`geth sync now [node]` is an immediate operator retry. Keychain and auth sync +now use per-peer high-water cursors, while receivers still verify every imported +signed operation before it can affect the reduced keychain or authorization +views. ## Two-Machine Smoke Test diff --git a/crates/geth-cli/src/lib.rs b/crates/geth-cli/src/lib.rs index 74f269f..b77f664 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -3334,7 +3334,7 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { } for stream in peer.streams { println!( - " {}\tstate={}\tstale={}\tcursor={}\tlast_attempt={}\tlast_success={}\timported={}\trejected={}\terror={}\tnext={}", + " {}\tstate={}\tstale={}\tcursor={}\tlast_attempt={}\tlast_success={}\timported={}\trejected={}\tfailures={}\tretry_in_ms={}\terror={}\tnext={}", stream.stream, stream.state, stream.stale, @@ -3349,6 +3349,11 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { .unwrap_or_else(|| "never".to_owned()), stream.last_imported, stream.last_rejected, + stream.consecutive_failures, + stream + .retry_in_ms + .map(|value| value.to_string()) + .unwrap_or_else(|| "-".to_owned()), stream.last_error.unwrap_or_else(|| "-".to_owned()), stream.next_action ); diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index f565a30..eabdfa8 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -1698,6 +1698,12 @@ pub struct SyncStreamStatus { pub last_error: Option, pub last_imported: usize, pub last_rejected: usize, + #[serde(default)] + pub consecutive_failures: u32, + #[serde(default)] + pub retry_after_ms: Option, + #[serde(default)] + pub retry_in_ms: Option, pub next_action: String, } @@ -2719,6 +2725,9 @@ mod tests { last_error: None, last_imported: 2, last_rejected: 0, + consecutive_failures: 0, + retry_after_ms: None, + retry_in_ms: None, next_action: "none".to_owned(), }], }], diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index fe0967b..bb0e74e 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -11415,7 +11415,11 @@ fn ssh_revocation_from_stored( #[cfg(test)] mod tests { use super::*; - use crate::sync::{record_live_sync_success, run_live_sync_once}; + use crate::sync::{ + SyncRunMode, record_live_sync_failure, record_live_sync_success, run_live_sync_once, + should_live_sync_stream, + }; + use std::collections::BTreeMap; use tokio::io::AsyncReadExt; fn skip_iroh_integration_tests() -> bool { @@ -12022,6 +12026,64 @@ mod tests { assert_eq!(peers[0].streams[0].next_action, "none"); } + #[test] + fn live_sync_failures_back_off_background_streams() { + let store = Store::open_memory().expect("open"); + store + .upsert_peer_card(&StoredPeerCard { + peer_id: "node:left".to_owned(), + card_json: "{}".to_owned(), + updated_at_ms: 1, + }) + .expect("insert peer"); + record_live_sync_failure( + &store, + "node:left", + "kv:prefs", + &NodeError::IrohPeer("peer unavailable".to_owned()), + ) + .expect("record failure"); + + let response = sync_status_local(&store).expect("sync status"); + let ControlResponse::SyncStatus { peers, .. } = response else { + panic!("unexpected sync status response"); + }; + assert_eq!(peers.len(), 1); + let stream = &peers[0].streams[0]; + assert_eq!(stream.stream, "kv:prefs"); + assert_eq!(stream.state, "failed"); + assert_eq!(stream.consecutive_failures, 1); + assert!(stream.retry_after_ms.is_some()); + assert!( + stream + .retry_in_ms + .is_some_and(|retry_in_ms| retry_in_ms > 0) + ); + assert!(stream.next_action.contains("geth sync now node:left")); + + let watermarks = BTreeMap::from([("kv:prefs".to_owned(), 42)]); + assert!( + !should_live_sync_stream( + &store, + "node:left", + "kv:prefs", + Some(&watermarks), + SyncRunMode::Background, + ) + .expect("background decision") + ); + assert!( + should_live_sync_stream( + &store, + "node:left", + "kv:prefs", + Some(&watermarks), + SyncRunMode::Manual, + ) + .expect("manual decision") + ); + } + #[test] fn operator_errors_include_recovery_commands() { let peer_error = diff --git a/crates/geth-node/src/sync.rs b/crates/geth-node/src/sync.rs index 82710fe..648bbf6 100644 --- a/crates/geth-node/src/sync.rs +++ b/crates/geth-node/src/sync.rs @@ -11,6 +11,8 @@ use geth_store::{Store, StoredModuleState}; use std::collections::BTreeMap; pub(crate) const LIVE_SYNC_STALE_AFTER_MS: i64 = 120_000; +const LIVE_SYNC_RETRY_BASE_MS: i64 = 1_000; +const LIVE_SYNC_RETRY_MAX_MS: i64 = 60_000; #[derive(Debug, serde::Deserialize, serde::Serialize)] pub(crate) struct LiveSyncCursor { @@ -24,6 +26,10 @@ struct LiveSyncHealth { last_error: Option, last_imported: usize, last_rejected: usize, + #[serde(default)] + consecutive_failures: u32, + #[serde(default)] + retry_after_ms: Option, } #[derive(Debug, serde::Deserialize, serde::Serialize)] @@ -34,6 +40,12 @@ pub(crate) struct KvDocsState { pub(crate) updated_at_ms: i64, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum SyncRunMode { + Background, + Manual, +} + pub(crate) fn live_sync_cursor_key(peer_node: &str, stream: &str) -> String { format!("live-sync:{peer_node}:{stream}") } @@ -93,6 +105,8 @@ pub(crate) fn record_live_sync_success( last_error: None, last_imported: imported, last_rejected: rejected, + consecutive_failures: 0, + retry_after_ms: None, })?, updated_at_ms: now, })?; @@ -109,20 +123,53 @@ pub(crate) fn record_live_sync_failure( let previous = store .get_module_state(&live_sync_status_key(peer_node, stream))? .and_then(|state| serde_json::from_str::(&state.state_json).ok()); + let last_success_ms = previous.as_ref().and_then(|health| health.last_success_ms); + let consecutive_failures = previous + .as_ref() + .map_or(1, |health| health.consecutive_failures.saturating_add(1)); + let retry_after_ms = now.saturating_add(live_sync_retry_delay_ms(consecutive_failures)); 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_success_ms, last_error: Some(error.to_string()), last_imported: 0, last_rejected: 0, + consecutive_failures, + retry_after_ms: Some(retry_after_ms), })?, updated_at_ms: now, })?; load_live_sync_cursor(store, peer_node, stream) } +fn live_sync_retry_delay_ms(consecutive_failures: u32) -> i64 { + let exponent = consecutive_failures.saturating_sub(1).min(6); + LIVE_SYNC_RETRY_BASE_MS + .saturating_mul(1_i64 << exponent) + .min(LIVE_SYNC_RETRY_MAX_MS) +} + +fn live_sync_retry_in_ms(health: &LiveSyncHealth, now_ms: i64) -> Option { + health + .retry_after_ms + .map(|retry_after_ms| retry_after_ms.saturating_sub(now_ms)) +} + +pub(crate) fn live_sync_stream_in_backoff( + store: &Store, + peer_node: &str, + stream: &str, +) -> Result { + let now = geth_store::now_ms(); + let Some(state) = store.get_module_state(&live_sync_status_key(peer_node, stream))? else { + return Ok(false); + }; + let health: LiveSyncHealth = serde_json::from_str(&state.state_json)?; + Ok(live_sync_retry_in_ms(&health, now).is_some_and(|retry_in_ms| retry_in_ms > 0)) +} + pub(crate) fn sync_status_local(store: &Store) -> Result { let peers = store .list_peer_cards()? @@ -143,6 +190,7 @@ pub(crate) fn sync_status_local(store: &Store) -> Result Result>, + mode: SyncRunMode, ) -> Result { + if mode == SyncRunMode::Background && live_sync_stream_in_backoff(store, peer_node, stream)? { + return Ok(false); + } let Some(remote_watermarks) = remote_watermarks else { return Ok(true); }; @@ -232,12 +287,26 @@ pub(crate) fn should_live_sync_stream( pub(crate) async fn run_sync_for_peer( node: &LocalNode, peer_node: &str, +) -> Result { + run_sync_for_peer_with_mode(node, peer_node, SyncRunMode::Manual).await +} + +async fn run_sync_for_peer_with_mode( + node: &LocalNode, + peer_node: &str, + mode: SyncRunMode, ) -> Result { let kv_stores = Store::open(&node.paths.metadata_db())?.list_kv_stores()?; let documents = Store::open(&node.paths.metadata_db())?.list_document_resources()?; let dbs = Store::open(&node.paths.metadata_db())?.list_db_resources()?; let mut streams = Vec::new(); - let remote_watermarks = + let store = Store::open(&node.paths.metadata_db())?; + let remote_watermarks = if mode == SyncRunMode::Background + && live_sync_stream_in_backoff(&store, peer_node, "sync-status")? + { + push_skipped_sync_stream(&store, peer_node, "sync-status", &mut streams)?; + None + } else { match super::sync_status_from_peer(node, peer_node) .await .map(|watermarks| { @@ -275,12 +344,41 @@ pub(crate) async fn run_sync_for_peer( }); None } - }; + } + }; - sync_keychain_stream(node, peer_node, remote_watermarks.as_ref(), &mut streams).await?; - sync_auth_stream(node, peer_node, remote_watermarks.as_ref(), &mut streams).await?; - sync_ssh_cert_stream(node, peer_node, remote_watermarks.as_ref(), &mut streams).await?; - sync_ssh_revocation_stream(node, peer_node, remote_watermarks.as_ref(), &mut streams).await?; + sync_keychain_stream( + node, + peer_node, + remote_watermarks.as_ref(), + mode, + &mut streams, + ) + .await?; + sync_auth_stream( + node, + peer_node, + remote_watermarks.as_ref(), + mode, + &mut streams, + ) + .await?; + sync_ssh_cert_stream( + node, + peer_node, + remote_watermarks.as_ref(), + mode, + &mut streams, + ) + .await?; + sync_ssh_revocation_stream( + node, + peer_node, + remote_watermarks.as_ref(), + mode, + &mut streams, + ) + .await?; if let Some(remote_watermarks) = remote_watermarks.as_ref() { for stream in remote_watermarks @@ -292,6 +390,7 @@ pub(crate) async fn run_sync_for_peer( peer_node, stream, Some(remote_watermarks), + mode, &mut streams, ) .await?; @@ -305,6 +404,7 @@ pub(crate) async fn run_sync_for_peer( &kv.name, &stream, remote_watermarks.as_ref(), + mode, &mut streams, ) .await?; @@ -317,6 +417,7 @@ pub(crate) async fn run_sync_for_peer( &document.name, &stream, remote_watermarks.as_ref(), + mode, &mut streams, ) .await?; @@ -329,6 +430,7 @@ pub(crate) async fn run_sync_for_peer( &db.name, &stream, remote_watermarks.as_ref(), + mode, &mut streams, ) .await?; @@ -412,11 +514,12 @@ async fn sync_keychain_stream( node: &LocalNode, peer_node: &str, remote_watermarks: Option<&BTreeMap>, + mode: SyncRunMode, streams: &mut Vec, ) -> Result<(), NodeError> { let stream = "keychain"; let store = Store::open(&node.paths.metadata_db())?; - if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks)? { + if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks, mode)? { return push_skipped_sync_stream(&store, peer_node, stream, streams); } let since_ms = load_live_sync_cursor(&store, peer_node, stream)?; @@ -463,11 +566,12 @@ async fn sync_auth_stream( node: &LocalNode, peer_node: &str, remote_watermarks: Option<&BTreeMap>, + mode: SyncRunMode, streams: &mut Vec, ) -> Result<(), NodeError> { let stream = "auth"; let store = Store::open(&node.paths.metadata_db())?; - if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks)? { + if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks, mode)? { return push_skipped_sync_stream(&store, peer_node, stream, streams); } let since_ms = load_live_sync_cursor(&store, peer_node, stream)?; @@ -514,11 +618,12 @@ async fn sync_ssh_cert_stream( node: &LocalNode, peer_node: &str, remote_watermarks: Option<&BTreeMap>, + mode: SyncRunMode, streams: &mut Vec, ) -> Result<(), NodeError> { let stream = "ssh-certs"; let store = Store::open(&node.paths.metadata_db())?; - if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks)? { + if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks, mode)? { return push_skipped_sync_stream(&store, peer_node, stream, streams); } match super::ssh_cert_sync_from_peer(node, peer_node, None).await { @@ -562,11 +667,12 @@ async fn sync_ssh_revocation_stream( node: &LocalNode, peer_node: &str, remote_watermarks: Option<&BTreeMap>, + mode: SyncRunMode, streams: &mut Vec, ) -> Result<(), NodeError> { let stream = "ssh-revocations"; let store = Store::open(&node.paths.metadata_db())?; - if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks)? { + if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks, mode)? { return push_skipped_sync_stream(&store, peer_node, stream, streams); } match super::ssh_revocation_sync_from_peer(node, peer_node, None).await { @@ -606,10 +712,11 @@ async fn sync_cas_tree_stream( peer_node: &str, stream: &str, remote_watermarks: Option<&BTreeMap>, + mode: SyncRunMode, streams: &mut Vec, ) -> Result<(), NodeError> { let store = Store::open(&node.paths.metadata_db())?; - if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks)? { + if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks, mode)? { return push_skipped_sync_stream(&store, peer_node, stream, streams); } let name = stream.trim_start_matches("cas-tree:"); @@ -657,10 +764,11 @@ async fn sync_kv_stream( name: &str, stream: &str, remote_watermarks: Option<&BTreeMap>, + mode: SyncRunMode, streams: &mut Vec, ) -> Result<(), NodeError> { let store = Store::open(&node.paths.metadata_db())?; - if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks)? { + if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks, mode)? { return push_skipped_sync_stream(&store, peer_node, stream, streams); } match super::kv_sync_from_peer(node, peer_node, name, None).await { @@ -705,10 +813,11 @@ async fn sync_document_stream( name: &str, stream: &str, remote_watermarks: Option<&BTreeMap>, + mode: SyncRunMode, streams: &mut Vec, ) -> Result<(), NodeError> { let store = Store::open(&node.paths.metadata_db())?; - if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks)? { + if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks, mode)? { return push_skipped_sync_stream(&store, peer_node, stream, streams); } match super::document_sync_from_peer(node, peer_node, name, None).await { @@ -751,10 +860,11 @@ async fn sync_db_stream( name: &str, stream: &str, remote_watermarks: Option<&BTreeMap>, + mode: SyncRunMode, streams: &mut Vec, ) -> Result<(), NodeError> { let store = Store::open(&node.paths.metadata_db())?; - if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks)? { + if !should_live_sync_stream(&store, peer_node, stream, remote_watermarks, mode)? { return push_skipped_sync_stream(&store, peer_node, stream, streams); } match super::db_sync_from_peer(node, peer_node, name, 100, None).await { @@ -806,7 +916,7 @@ pub(crate) async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError } let peers = Store::open(&node.paths.metadata_db())?.list_peer_cards()?; for peer in peers { - let run = run_sync_for_peer(node, &peer.peer_id).await?; + let run = run_sync_for_peer_with_mode(node, &peer.peer_id, SyncRunMode::Background).await?; for stream in &run.streams { tracing::debug!( peer = %peer.peer_id, diff --git a/docs/architecture.md b/docs/architecture.md index 5005453..84947b6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -150,10 +150,13 @@ still verified against trusted-admin OpenSSH signatures before import. Operators can run `geth sync now [node]` to trigger the same best-effort pass immediately and `geth sync status` to inspect locally recorded last-attempt, last-success, cursor, import/rejection counts, and errors for each peer stream. -JSON status also includes `state`, `stale`, `stale_after_ms`, and `next_action` -fields so scripts can fail on unhealthy streams. Common daemon errors include a -`next:` recovery line for missing peer cards, missing grants, missing resources, -unavailable endpoints, and missing DB/KV/document registrations. +JSON status also includes `state`, `stale`, `stale_after_ms`, +`consecutive_failures`, `retry_after_ms`, `retry_in_ms`, and `next_action` +fields so scripts can fail on unhealthy streams. Background live-sync backs off +failed streams without blocking explicit `geth sync now [node]` retries. Common +daemon errors include a `next:` recovery line for missing peer cards, missing +grants, missing resources, unavailable endpoints, and missing DB/KV/document +registrations. Host-opening paths are intentionally narrow. TCP pipe forwarding accepts only explicit loopback socket addresses on both the local listener and remote target; diff --git a/docs/production-readiness-roadmap.md b/docs/production-readiness-roadmap.md index 554e8c3..c71c034 100644 --- a/docs/production-readiness-roadmap.md +++ b/docs/production-readiness-roadmap.md @@ -228,11 +228,11 @@ Goal: make convergence and failure behavior predictable enough for automation. - `[x]` Document merge semantics are documented and tested. - `[x]` DB change application limits are documented and tested. -- `[ ]` Add live-sync retry and backoff policy. +- `[x]` Add live-sync retry and backoff policy. Acceptance criteria: - - `[ ]` Failed peer streams back off without starving healthy streams. - - `[ ]` Retry state appears in `geth sync status --json`. - - `[ ]` Operators can trigger immediate retry with `geth sync now`. + - `[x]` Failed peer streams back off without starving healthy streams. + - `[x]` Retry state appears in `geth sync status --json`. + - `[x]` Operators can trigger immediate retry with `geth sync now`. ## Phase 7: Script And Automation UX diff --git a/docs/roadmap.md b/docs/roadmap.md index d8b9634..58276f2 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -389,8 +389,10 @@ geth-to-geth connections without granting trust from discovery alone. watermark has not advanced. - `[x]` `geth sync status` reports last local attempt, success, cursor, import/rejection counts, and error for each recorded peer stream. + - `[x]` Failed background live-sync streams record consecutive failures and + retry-after metadata. - `[x]` `geth sync now [node]` triggers the same best-effort sync pass that - background live sync uses. + background live sync uses, bypassing stream backoff for operator retries. - `[x]` Tests verify unauthorized streams are omitted from summary output. - `[x]` Tests verify persisted stream health is exposed in local sync status.