eric/geth
Watch
1
0
Fork
You've already forked geth
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add live sync retry backoff

Browse source
This commit is contained in:
Eric Wendland 2026-07-05 22:57:52 +02:00
commit 2dce1f41cf
8 changed files with 225 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

11
README.md
View file

@ -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

7
crates/geth-cli/src/lib.rs
View file

@ -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
);

9
crates/geth-control/src/lib.rs
View file

@ -1698,6 +1698,12 @@ pub struct SyncStreamStatus {
pub last_error: Option<String>,
pub last_imported: usize,
pub last_rejected: usize,
#[serde(default)]
pub consecutive_failures: u32,
#[serde(default)]
pub retry_after_ms: Option<i64>,
#[serde(default)]
pub retry_in_ms: Option<i64>,
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(),
}],
}],

64
crates/geth-node/src/lib.rs
View file

@ -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 =

142
crates/geth-node/src/sync.rs
View file

@ -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<String>,
last_imported: usize,
last_rejected: usize,
#[serde(default)]
consecutive_failures: u32,
#[serde(default)]
retry_after_ms: Option<i64>,
}
#[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::<LiveSyncHealth>(&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<i64> {
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<bool, NodeError> {
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<ControlResponse, NodeError> {
let peers = store
.list_peer_cards()?
@ -143,6 +190,7 @@ pub(crate) fn sync_status_local(store: &Store) -> Result<ControlResponse, NodeEr
let status_state = sync_stream_state(&health, stale).to_owned();
let next_action =
sync_stream_next_action(&peer.peer_id, &stream, &health, stale);
let retry_in_ms = live_sync_retry_in_ms(&health, now_ms);
Ok(SyncStreamStatus {
stream,
cursor_ms,
@ -154,6 +202,9 @@ pub(crate) fn sync_status_local(store: &Store) -> Result<ControlResponse, NodeEr
last_error: health.last_error,
last_imported: health.last_imported,
last_rejected: health.last_rejected,
consecutive_failures: health.consecutive_failures,
retry_after_ms: health.retry_after_ms,
retry_in_ms,
next_action,
})
})
@ -211,7 +262,11 @@ pub(crate) fn should_live_sync_stream(
peer_node: &str,
stream: &str,
remote_watermarks: Option<&BTreeMap<String, i64>>,
mode: SyncRunMode,
) -> Result<bool, NodeError> {
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<SyncPeerRun, NodeError> {
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<SyncPeerRun, NodeError> {
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<String, i64>>,
mode: SyncRunMode,
streams: &mut Vec<SyncStreamRun>,
) -> 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<String, i64>>,
mode: SyncRunMode,
streams: &mut Vec<SyncStreamRun>,
) -> 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<String, i64>>,
mode: SyncRunMode,
streams: &mut Vec<SyncStreamRun>,
) -> 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<String, i64>>,
mode: SyncRunMode,
streams: &mut Vec<SyncStreamRun>,
) -> 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<String, i64>>,
mode: SyncRunMode,
streams: &mut Vec<SyncStreamRun>,
) -> 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<String, i64>>,
mode: SyncRunMode,
streams: &mut Vec<SyncStreamRun>,
) -> 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<String, i64>>,
mode: SyncRunMode,
streams: &mut Vec<SyncStreamRun>,
) -> 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<String, i64>>,
mode: SyncRunMode,
streams: &mut Vec<SyncStreamRun>,
) -> 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,

11
docs/architecture.md
View file

@ -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;

8
docs/production-readiness-roadmap.md
View file

@ -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

4
docs/roadmap.md
View file

@ -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.