Make live sync configurable
This commit is contained in:
parent
5c91635ea2
commit
3eb6b623e7
6 changed files with 106 additions and 14 deletions
|
|
@ -204,9 +204,10 @@ Roadmap items should be actionable and checkable:
|
|||
`resource:ssh:revocations` for non-owner subjects. Authorized peers can pull
|
||||
SSH certificate-flow metadata with `ssh_cert.sync` on `resource:ssh:certs` and
|
||||
revocation metadata with `ssh_revocation.sync` on
|
||||
`resource:ssh:revocations`. The daemon live-syncs known peers every 30
|
||||
seconds using per-peer cursors. Conflicting records with already-known ids are
|
||||
rejected during sync import; this is pull-only metadata sync, not yet a
|
||||
`resource:ssh:revocations`. The daemon live-syncs known peers using per-peer
|
||||
cursors; `[sync] live_sync_enabled` and `live_sync_interval_ms` in
|
||||
`config.toml` control that loop. Conflicting records with already-known ids
|
||||
are rejected during sync import; this is pull-only metadata sync, not yet a
|
||||
signed CRDT/resource-log replication model.
|
||||
- cr-sqlite apply, iroh-docs, iroh-blobs provider/fetch, Automerge sync,
|
||||
broader auth enforcement, and Keyhive/BeeKEM-style authorization are future
|
||||
|
|
|
|||
10
README.md
10
README.md
|
|
@ -170,10 +170,12 @@ at the peer. `geth ssh revocation sync <node-id>` requires
|
|||
`ssh_revocation.sync` on `resource:ssh:revocations`. Both commands merge
|
||||
authorized peer metadata into the local store for offline listing and later
|
||||
approval/signing workflows. While the daemon is running, it also performs a
|
||||
background live-sync tick for known peers every 30 seconds. Live-sync stores
|
||||
per-peer high-water cursors in local metadata so repeated ticks request only
|
||||
newer SSH certificate-flow and revocation records. Sync import preserves local
|
||||
metadata by rejecting conflicting records with ids that already exist locally.
|
||||
background live-sync tick for known peers. The default interval is 30 seconds
|
||||
and can be changed in `config.toml` with `[sync] live_sync_enabled` and
|
||||
`live_sync_interval_ms`. Live-sync stores per-peer high-water cursors in local
|
||||
metadata so repeated ticks request only newer SSH certificate-flow and
|
||||
revocation records. Sync import preserves local metadata by rejecting
|
||||
conflicting records with ids that already exist locally.
|
||||
Before probing individual modules, the daemon asks the peer for an authorized
|
||||
sync-status summary over Iroh. The peer only returns stream watermarks for
|
||||
resources where the caller already has the matching capability, letting the
|
||||
|
|
|
|||
|
|
@ -13,6 +13,10 @@ local_discovery = true
|
|||
#
|
||||
# [iroh.relay_maps.home]
|
||||
# urls = ["https://relay.example.com"]
|
||||
#
|
||||
[sync]
|
||||
live_sync_enabled = true
|
||||
live_sync_interval_ms = 30000
|
||||
"#;
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
|
|
@ -116,11 +120,16 @@ pub enum ConfigError {
|
|||
EmptyRelayMap(String),
|
||||
#[error("invalid iroh relay URL in map `{map}`: {url}")]
|
||||
InvalidRelayUrl { map: String, url: String },
|
||||
#[error("invalid sync live_sync_enabled value: {0}")]
|
||||
InvalidLiveSyncEnabled(String),
|
||||
#[error("invalid sync live_sync_interval_ms value: {0}")]
|
||||
InvalidLiveSyncInterval(String),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq, Eq)]
|
||||
pub struct GethConfig {
|
||||
pub iroh: IrohConfig,
|
||||
pub sync: SyncConfig,
|
||||
}
|
||||
|
||||
impl GethConfig {
|
||||
|
|
@ -140,6 +149,7 @@ impl GethConfig {
|
|||
let document = text.parse::<toml_edit::DocumentMut>()?;
|
||||
let relay_maps = parse_relay_maps(&document)?;
|
||||
let local_discovery = parse_local_discovery(&document)?;
|
||||
let sync = parse_sync_config(&document)?;
|
||||
let relay_mode = match document.get("iroh").and_then(|iroh| iroh.get("relay_mode")) {
|
||||
Some(item) => {
|
||||
let value = item
|
||||
|
|
@ -156,6 +166,7 @@ impl GethConfig {
|
|||
relay_maps,
|
||||
local_discovery,
|
||||
},
|
||||
sync,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
@ -177,6 +188,21 @@ impl Default for IrohConfig {
|
|||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct SyncConfig {
|
||||
pub live_sync_enabled: bool,
|
||||
pub live_sync_interval_ms: u64,
|
||||
}
|
||||
|
||||
impl Default for SyncConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
live_sync_enabled: true,
|
||||
live_sync_interval_ms: 30_000,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct RelayMapConfig {
|
||||
pub urls: Vec<String>,
|
||||
|
|
@ -297,6 +323,37 @@ fn parse_local_discovery(document: &toml_edit::DocumentMut) -> Result<bool, Conf
|
|||
}
|
||||
}
|
||||
|
||||
fn parse_sync_config(document: &toml_edit::DocumentMut) -> Result<SyncConfig, ConfigError> {
|
||||
let live_sync_enabled = match document
|
||||
.get("sync")
|
||||
.and_then(|sync| sync.get("live_sync_enabled"))
|
||||
{
|
||||
Some(item) => item
|
||||
.as_bool()
|
||||
.ok_or_else(|| ConfigError::InvalidLiveSyncEnabled(item.to_string()))?,
|
||||
None => true,
|
||||
};
|
||||
let live_sync_interval_ms = match document
|
||||
.get("sync")
|
||||
.and_then(|sync| sync.get("live_sync_interval_ms"))
|
||||
{
|
||||
Some(item) => {
|
||||
let interval = item
|
||||
.as_integer()
|
||||
.ok_or_else(|| ConfigError::InvalidLiveSyncInterval(item.to_string()))?;
|
||||
if interval < 100 {
|
||||
return Err(ConfigError::InvalidLiveSyncInterval(interval.to_string()));
|
||||
}
|
||||
interval as u64
|
||||
}
|
||||
None => 30_000,
|
||||
};
|
||||
Ok(SyncConfig {
|
||||
live_sync_enabled,
|
||||
live_sync_interval_ms,
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
|
@ -306,6 +363,8 @@ mod tests {
|
|||
let config = GethConfig::parse(GethConfig::default_toml()).expect("parse config");
|
||||
assert_eq!(config.iroh.relay_mode, RelayMode::Default);
|
||||
assert!(config.iroh.local_discovery);
|
||||
assert!(config.sync.live_sync_enabled);
|
||||
assert_eq!(config.sync.live_sync_interval_ms, 30_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -381,6 +440,28 @@ mod tests {
|
|||
assert!(!config.iroh.local_discovery);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn config_parses_live_sync_settings() {
|
||||
let config =
|
||||
GethConfig::parse("[sync]\nlive_sync_enabled = false\nlive_sync_interval_ms = 250\n")
|
||||
.expect("parse");
|
||||
assert!(!config.sync.live_sync_enabled);
|
||||
assert_eq!(config.sync.live_sync_interval_ms, 250);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn config_rejects_invalid_live_sync_settings() {
|
||||
let error = GethConfig::parse("[sync]\nlive_sync_enabled = \"yes\"\n").expect_err("error");
|
||||
assert!(matches!(error, ConfigError::InvalidLiveSyncEnabled(_)));
|
||||
|
||||
let error = GethConfig::parse("[sync]\nlive_sync_interval_ms = 99\n").expect_err("error");
|
||||
assert!(matches!(error, ConfigError::InvalidLiveSyncInterval(_)));
|
||||
|
||||
let error =
|
||||
GethConfig::parse("[sync]\nlive_sync_interval_ms = \"fast\"\n").expect_err("error");
|
||||
assert!(matches!(error, ConfigError::InvalidLiveSyncInterval(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn config_rejects_non_bool_local_discovery() {
|
||||
let error = GethConfig::parse("[iroh]\nlocal_discovery = \"yes\"\n").expect_err("error");
|
||||
|
|
|
|||
|
|
@ -153,7 +153,6 @@ struct PipeRuntime {
|
|||
}
|
||||
|
||||
const PUBSUB_RING_LIMIT: usize = 256;
|
||||
const LIVE_SYNC_INTERVAL: Duration = Duration::from_secs(30);
|
||||
const PIPE_CONNECTION_RING_LIMIT: usize = 256;
|
||||
|
||||
#[derive(Debug, serde::Deserialize, serde::Serialize)]
|
||||
|
|
@ -205,7 +204,13 @@ pub async fn run_daemon(paths: GethPaths) -> Result<(), NodeError> {
|
|||
.clone()
|
||||
{
|
||||
spawn_iroh_control_accept_loop(node.clone(), endpoint);
|
||||
spawn_background_live_sync(node.clone());
|
||||
let config = GethConfig::load(&node.paths.config_file())?;
|
||||
if config.sync.live_sync_enabled {
|
||||
spawn_background_live_sync(
|
||||
node.clone(),
|
||||
Duration::from_millis(config.sync.live_sync_interval_ms),
|
||||
);
|
||||
}
|
||||
}
|
||||
let _peer_card_lan_discovery = start_peer_card_lan_discovery(&node).await;
|
||||
if Path::new(&paths.socket_path()).exists() {
|
||||
|
|
@ -2134,12 +2139,12 @@ fn spawn_iroh_control_accept_loop(node: LocalNode, endpoint: GethIrohEndpoint) {
|
|||
});
|
||||
}
|
||||
|
||||
fn spawn_background_live_sync(node: LocalNode) {
|
||||
fn spawn_background_live_sync(node: LocalNode, interval_duration: Duration) {
|
||||
tokio::spawn(async move {
|
||||
if let Err(error) = run_live_sync_once(&node).await {
|
||||
tracing::debug!(%error, "initial live sync tick failed");
|
||||
}
|
||||
let mut interval = tokio::time::interval(LIVE_SYNC_INTERVAL);
|
||||
let mut interval = tokio::time::interval(interval_duration);
|
||||
loop {
|
||||
interval.tick().await;
|
||||
if let Err(error) = run_live_sync_once(&node).await {
|
||||
|
|
|
|||
|
|
@ -91,11 +91,12 @@ as signed-list-ready records. The bootstrap can pull
|
|||
certificate-flow metadata over Iroh with `geth ssh cert sync <node-id>` when the
|
||||
peer grants `ssh_cert.sync` on `resource:ssh:certs`, and revocation metadata with
|
||||
`geth ssh revocation sync <node-id>` when the peer grants `ssh_revocation.sync`
|
||||
on `resource:ssh:revocations`. The daemon also runs a 30-second background
|
||||
on `resource:ssh:revocations`. The daemon also runs a configurable background
|
||||
live-sync tick for known peers and records per-peer high-water cursors in
|
||||
`module_state`, so repeated ticks request only records at or beyond the last
|
||||
remote cursor. Boundary duplicates are harmless because records are keyed by
|
||||
stable IDs and inserted with replace semantics.
|
||||
remote cursor. The default interval is 30 seconds and can be changed under
|
||||
`[sync]` in `config.toml`. Boundary duplicates are harmless because records are
|
||||
keyed by stable IDs and inserted with replace semantics.
|
||||
Before issuing per-module pulls, the daemon can request an authorized sync
|
||||
status summary over the same protected Iroh control ALPN. The serving peer
|
||||
validates endpoint/card binding and returns only watermarks for streams where
|
||||
|
|
|
|||
|
|
@ -254,6 +254,8 @@ resource-scoped capability decisions.
|
|||
`geth ssh revocation sync <node-id>`.
|
||||
- `[x]` The daemon background live-sync loop refreshes known peers without a
|
||||
manual command.
|
||||
- `[x]` Background live-sync can be disabled or retuned with `[sync]`
|
||||
`live_sync_enabled` and `live_sync_interval_ms`.
|
||||
- `[x]` SSH metadata live-sync stores per-peer high-water cursors in
|
||||
`module_state` and requests only records at or beyond the cursor.
|
||||
- `[x]` Local SSH cert request/read/approve/import commands can enforce
|
||||
|
|
|
|||
Loading…
Reference in a new issue