diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/state.rs | 1297 | ||||
| -rw-r--r-- | src/state/host.rs | 276 | ||||
| -rw-r--r-- | src/state/mod.rs | 404 | ||||
| -rw-r--r-- | src/state/peer.rs | 352 | ||||
| -rw-r--r-- | src/state/read.rs | 100 | ||||
| -rw-r--r-- | src/state/session.rs | 155 | ||||
| -rw-r--r-- | src/state/timing.rs | 75 |
7 files changed, 1362 insertions, 1297 deletions
diff --git a/src/state.rs b/src/state.rs deleted file mode 100644 index 46ebfb9..0000000 --- a/src/state.rs +++ /dev/null @@ -1,1297 +0,0 @@ -//! Shared node state: the document, lifecycle, per-peer state and timing. - -use std::collections::HashMap; -use std::future::Future; -use std::sync::Arc; -use std::sync::atomic::AtomicBool; -use std::time::Duration; - -use anyhow::{Result, bail}; -use automerge::{AutoCommit, ChangeHash, sync}; -use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch}; -use tokio::time::Instant; -use tracing::{debug, error, warn}; - -use crate::backoff::Backoff; -use crate::document::replica::{Epoch, Replica}; -use crate::remake::{Host, PeerSummary, RemakeState}; -use crate::swarm::sync::{ChangeSet, SyncEvent}; -use crate::swarm::throughput::{SyncBytes, ThroughputTable}; -use crate::transport; -use crate::transport::firewall::Firewall; - -#[derive(Debug, Clone, PartialEq, Eq)] -pub(crate) enum NodeLifecycle { - // TODO maybe rename discovered_swarm_id to swarm_id if that doesn't impair readability. - Joining { discovered_swarm_id: Option<String> }, - Ready { swarm_id: String }, -} - -/// How far a peer got into the swarm, as far as we're concerned. -#[derive(Debug, Clone, Default, PartialEq, Eq)] -pub(crate) enum Standing { - /// Nothing proven yet. With a swarm key, everything but a handshake from it is - /// ignored (`swarm::peer::admit`). - #[default] - Stranger, - /// Proved the swarm key in a HELLO; no HELLO_REPLY yet. Only reached with a key. - Admitted, - /// Its HELLO_REPLY named our swarm (and, with a key, proved it): we sync with it. - Member { swarm_id: String }, -} - -impl Standing { - /// The swarm a member belongs to; None before its HELLO_REPLY. - pub(crate) fn swarm_id(&self) -> Option<&str> { - match self { - Self::Member { swarm_id } => Some(swarm_id), - Self::Stranger | Self::Admitted => None, - } - } - - /// Proved the swarm key (meaningful only with one). - pub(crate) fn is_admitted(&self) -> bool { - !matches!(self, Self::Stranger) - } -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(crate) enum Liveness { - Active, - /// No inbound message for Timing::dormant_after. Kept indefinitely, probed slowly. - Dormant { - since: Instant, - }, -} - -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub(crate) enum HelloState { - #[default] - NotSent, - /// A send is in flight. - Sending, - /// HELLO was sent (or the send failed); no HELLO_REPLY yet. Retried at - /// `next_attempt` (`PeerState::hello_backoff`, or `dormant_probe` while dormant). - AwaitingReply { next_attempt: Instant }, - /// HELLO_REPLY received. No more retries. - Established, -} - -/// A sync message on its way to our Mycelium (the HTTP request is in progress). -#[derive(Clone)] -pub(crate) struct Sending { - /// Changes it carries for the first time; taken back if the send fails. - pub(crate) sent: Vec<automerge::ChangeHash>, - /// A sync pass was requested meanwhile. Once this one is handed over, the peer must - /// be enqueued again so the deferred changes go out. - pub(crate) regenerate: bool, -} - -/// First wait for a HELLO_REPLY before sending the HELLO again; doubles while none -/// comes. -const HELLO_RETRY: Duration = Duration::from_secs(1); -const HELLO_RETRY_MAX: Duration = Duration::from_secs(30); - -/// First wait before generating again after our Mycelium refused a message; doubles -/// while it keeps refusing. -const SEND_RETRY: Duration = Duration::from_millis(500); -const SEND_RETRY_MAX: Duration = Duration::from_secs(30); - -#[derive(Clone)] -pub(crate) struct PeerState { - /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for - /// the pair, so computed once. - pub(crate) leads: bool, - - pub(crate) standing: Standing, - pub(crate) hello_state: HelloState, - /// Wait between unanswered HELLOs; reset whenever a new handshake starts. - pub(crate) hello_backoff: Backoff, - - /// Everything about the current Automerge sync with this peer; a reset replaces it. - pub(crate) sync: SyncSession, - - pub(crate) liveness: Liveness, - pub(crate) added_at: Instant, - pub(crate) last_seen: Option<Instant>, - pub(crate) last_ping_at: Option<Instant>, - - /// Last time a sync message from this peer was applied (survives resets: the - /// keepalive heal looks at it). - pub(crate) last_sync_rx: Option<Instant>, - - /// Last self-healing re-handshake (cooldown). - pub(crate) last_resync_at: Option<Instant>, - - /// Since when keepalives show the peer's heads differ from ours with no sync message - /// from it in between (None while they match). - pub(crate) diverged_since: Option<Instant>, - - /// Epoch the peer advertised in its most recent message (late ones aside: see - /// `epoch_seq`). - pub(crate) epoch: Option<Epoch>, - /// The peer's `epoch_seq` with `epoch`: how many times it changed epoch in its run. - /// An epoch with a lower one is older than what we have, delivered late. - pub(crate) epoch_seq: Option<u32>, - /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over. - pub(crate) run_id: Option<crate::transport::RunId>, - - /// Whether the peer was ready according to its most recent sync message or keepalive. - pub(crate) reports_ready: Option<bool>, - - /// Whether the peer said, in its most recent keepalive, that it would sync with us. - pub(crate) syncs_with_us: Option<bool>, - - /// How many heads its most recent keepalive listed (None if it sent a digest). - pub(crate) reported_heads: Option<usize>, - - /// HELLOs sent in the current handshake attempt without hearing anything back. - pub(crate) hello_attempts: u32, - /// Since when we consider the peer unreachable (three unanswered HELLOs, or - /// dormant). Cleared by any message from it; ignored after `dormant_probe`, so the - /// peer becomes a neighbour candidate again and gets tried once more. - pub(crate) unreachable_since: Option<Instant>, - - /// What went wrong with this peer so far (counted only, for /peers and /admin/stats). - pub(crate) incidents: Incidents, -} - -/// Trouble with a peer, counted since it was first listed. Lost keepalives and refused -/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING. -#[derive(Debug, Default, Clone, Copy, serde::Serialize, utoipa::ToSchema)] -#[serde(rename_all = "camelCase")] -pub(crate) struct Incidents { - /// Our sync messages left unanswered until the reply timeout (then sent again). - pub(crate) reply_timeouts: u64, - /// Re-handshakes because our documents differed while sync was idle. - pub(crate) heals: u64, - /// Times the peer went dormant. - pub(crate) dormancies: u64, - /// Time spent dormant, in milliseconds, up to its last return (the current - /// dormancy, if any, isn't included). - pub(crate) dormant_ms: u64, -} - -impl Incidents { - pub(crate) fn add(&mut self, other: &Incidents) { - self.reply_timeouts += other.reply_timeouts; - self.heals += other.heals; - self.dormancies += other.dormancies; - self.dormant_ms += other.dormant_ms; - } -} - -/// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart). -const UNREACHABLE_AFTER_HELLOS: u32 = 3; - -/// One Automerge sync with a peer, from a reset to the next. Everything a reset must -/// forget lives here, so a reset replaces it whole and can't miss a field. -#[derive(Clone)] -pub(crate) struct SyncSession { - pub(crate) state: sync::State, - - /// A message being handed to our Mycelium. - pub(crate) sending: Option<Sending>, - - /// After our Mycelium refused a message: no new one before this. The refused one is - /// not kept; the next one is generated afresh and carries everything pending. - pub(crate) retry_at: Option<Instant>, - /// The wait after the next refusal. - pub(crate) send_backoff: Backoff, - - /// The answer we expect to our last sync message(s), if any. - pub(crate) reply: Option<SyncReplyWait>, - /// How long to wait for it; doubles while answers stay out. - pub(crate) reply_backoff: Backoff, -} - -/// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery: -/// a lost message leaves both sides waiting (`in_flight`), and changes it carried are -/// never sent again (`sent_hashes`). We resend after a timeout instead. -#[derive(Clone, Debug)] -pub(crate) struct SyncReplyWait { - pub(crate) deadline: Instant, - /// When the first of the unanswered messages was generated, and the sync bytes - /// exchanged with the peer by then (`swarm::throughput`). - pub(crate) started: (Instant, SyncBytes), - /// Changes first sent in the unanswered messages. - pub(crate) sent: Vec<automerge::ChangeHash>, -} - -/// First wait for a sync answer; generous, since Mycelium delivers large messages in -/// many small packets. -const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10); -/// Upper bound, as a multiple of the first wait: 5 minutes (Mycelium gives up on a -/// message after five minutes). -const SYNC_REPLY_TIMEOUT_MAX_FACTOR: u32 = 30; -/// How long Mycelium tries to deliver a message before dropping it. -const MYCELIUM_DELIVERY_LIMIT: Duration = Duration::from_secs(300); - -impl SyncSession { - /// `reply_backoff`: waits for sync answers (`Timing::sync_reply_backoff`). - pub(crate) fn new(reply_backoff: Backoff) -> Self { - Self { - state: sync::State::new(), - sending: None, - retry_at: None, - send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX), - reply: None, - reply_backoff, - } - } - - /// Whether we leave the first message of this session to the peer. - /// - /// After a reset, both sides usually open at once, and the openings cross. Between - /// peers that share history that costs nothing. But a peer with an empty document - /// answers our opening with empty heads, which Automerge takes for "lost - /// everything", so the whole document would go out a second time. So a ready node - /// never opens towards a peer that isn't ready (one that is still catching up); that - /// peer opens itself. - pub(crate) fn leave_opening(&self, we_ready: bool, peer_ready: Option<bool>) -> bool { - self.state.their_heads.is_none() && we_ready && peer_ready == Some(false) - } - - /// Our Mycelium took the message. - pub(crate) fn send_succeeded(&mut self) -> bool { - self.retry_at = None; - self.send_backoff.reset(); - self.sending - .take() - .is_some_and(|sending| sending.regenerate) - } - - /// Our Mycelium refused the message: as far as the peer is concerned, it was never - /// sent. Takes back its changes (they go out with the next message), clears - /// `in_flight`, and returns how long to wait before generating again. None if a reset - /// already discarded what was being sent. - pub(crate) fn send_failed(&mut self, now: Instant) -> Option<Duration> { - let sending = self.sending.take()?; - for hash in &sending.sent { - self.state.sent_hashes.remove(hash); - } - if let Some(wait) = &mut self.reply { - wait.sent.retain(|hash| !sending.sent.contains(hash)); - } - self.state.in_flight = false; - let delay = self.send_backoff.next(); - self.retry_at = Some(now + delay); - Some(delay) - } - - /// A sync message went out; `sent` are the changes it carried for the first time. - /// `bytes` are the sync bytes exchanged with the peer before it. - pub(crate) fn expect_reply( - &mut self, - sent: Vec<automerge::ChangeHash>, - now: Instant, - bytes: SyncBytes, - ) { - match &mut self.reply { - // Still waiting for an earlier one: keep its deadline, remember these too. - Some(wait) => wait.sent.extend(sent), - None => { - self.reply = Some(SyncReplyWait { - deadline: now + self.reply_backoff.current(), - started: (now, bytes), - sent, - }); - } - } - } - - /// No answer is due any more: the peer answered, or keepalives show both documents - /// are the same. - pub(crate) fn reply_settled(&mut self) { - self.reply = None; - self.reply_backoff.reset(); - } - - /// If the answer is overdue: forgets that the unanswered changes were sent and that - /// a message is in flight, so the next sync pass sends them again. Returns how many - /// changes that concerns. - fn take_overdue_reply(&mut self, now: Instant) -> Option<usize> { - if self.reply.as_ref().is_none_or(|wait| now < wait.deadline) { - return None; - } - let wait = self.reply.take()?; - for hash in &wait.sent { - self.state.sent_hashes.remove(hash); - } - self.state.in_flight = false; - self.reply_backoff.grow(); - Some(wait.sent.len()) - } -} - -impl PeerState { - pub(crate) fn new(leads: bool, timing: &Timing) -> Self { - Self { - leads, - standing: Standing::default(), - hello_state: HelloState::default(), - hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX), - sync: SyncSession::new(timing.sync_reply_backoff()), - liveness: Liveness::Active, - added_at: Instant::now(), - last_seen: None, - last_ping_at: None, - last_sync_rx: None, - last_resync_at: None, - diverged_since: None, - epoch: None, - epoch_seq: None, - run_id: None, - reports_ready: None, - syncs_with_us: None, - reported_heads: None, - hello_attempts: 0, - unreachable_since: None, - incidents: Incidents::default(), - } - } - - /// Start Automerge sync with this peer from scratch. A message still being handed to - /// Mycelium was generated against the old state; its outcome is then ignored. - pub(crate) fn reset_sync(&mut self) { - // The same waits for answers, from the start. - let mut reply_backoff = self.sync.reply_backoff; - reply_backoff.reset(); - self.sync = SyncSession::new(reply_backoff); - } - - /// An overdue sync answer from an established, active peer (see - /// `SyncSession::take_overdue_reply`). - pub(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> { - if self.liveness != Liveness::Active || self.hello_state != HelloState::Established { - return None; - } - let overdue = self.sync.take_overdue_reply(now); - if overdue.is_some() { - self.incidents.reply_timeouts += 1; - } - overdue - } - - fn quiet_for(&self, now: Instant) -> Duration { - now.saturating_duration_since(self.last_seen.unwrap_or(self.added_at)) - } - - /// Records inbound activity. Returns true if the peer was dormant. - pub(crate) fn mark_seen(&mut self, now: Instant) -> bool { - self.last_seen = Some(now); - self.hello_attempts = 0; - self.unreachable_since = None; - - if let Liveness::Dormant { since } = self.liveness { - self.incidents.dormant_ms += now.saturating_duration_since(since).as_millis() as u64; - self.liveness = Liveness::Active; - // Force a fresh handshake; its reply resets sync on both sides. - if self.hello_state != HelloState::Sending { - self.hello_state = HelloState::NotSent; - } - return true; - } - false - } - - /// A HELLO went out (or failed to). After the third unanswered one, the peer counts - /// as unreachable. - pub(crate) fn hello_sent(&mut self, now: Instant) { - self.hello_attempts += 1; - if self.hello_attempts >= UNREACHABLE_AFTER_HELLOS { - self.unreachable_since.get_or_insert(now); - } - } - - /// Whether to leave the peer out when choosing neighbours (`swarm::topology`). - pub(crate) fn is_unreachable(&self, timing: &Timing, now: Instant) -> bool { - self.unreachable_since - .is_some_and(|since| now.saturating_duration_since(since) < timing.dormant_probe) - } - - /// No more HELLO retries (the peer isn't our neighbour any more). A HELLO in flight - /// finishes; a later handshake starts over. - pub(crate) fn stop_hello_retries(&mut self) { - if matches!(self.hello_state, HelloState::AwaitingReply { .. }) { - self.hello_state = HelloState::NotSent; - } - } - - /// Moves a silent peer to Dormant. Returns true on the transition. - pub(crate) fn check_dormancy(&mut self, timing: &Timing, now: Instant) -> bool { - if self.liveness != Liveness::Active || self.quiet_for(now) < timing.dormant_after { - return false; - } - - self.liveness = Liveness::Dormant { since: now }; - self.incidents.dormancies += 1; - self.unreachable_since.get_or_insert(now); - self.reset_sync(); - - if self.hello_state != HelloState::Sending { - self.hello_state = HelloState::AwaitingReply { - next_attempt: now + timing.dormant_probe, - }; - } - true - } - - /// True (and records the ping) if an established, quiet peer needs a keepalive. - /// - /// One PING and its PONG tell both sides that the other is alive, so only one side of - /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs - /// makes the other side's peer look recently heard from, so the other side never - /// pings while it does. If the pings stop, the other side pings once it has heard - /// nothing for half an interval more. Without that, both sides would get due at about - /// the same time (each one's quiet time starts with the same exchange) and their - /// PINGs would often cross. - pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool { - let quiet_enough = if self.leads { - timing.keepalive - } else { - timing.keepalive + timing.keepalive / 2 - }; - let due = self.liveness == Liveness::Active - && self.hello_state == HelloState::Established - && now.saturating_duration_since(self.added_at) >= timing.keepalive - && self.quiet_for(now) >= quiet_enough - && self - .last_ping_at - .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive); - - if due { - self.last_ping_at = Some(now); - } - due - } - - /// Self-healing check on PING/PONG. Returns true if a re-handshake should start. - pub(crate) fn needs_heal( - &mut self, - local: &transport::HeadsSummary, - remote: &transport::HeadsSummary, - timing: &Timing, - now: Instant, - ) -> bool { - if local == remote { - self.diverged_since = None; - return false; - } - - // How long the heads have differed without sync making progress. Measured from - // when we noticed, not from the last sync message: in a quiet swarm that one is - // long past, and the first keepalive after a lost message would heal at once, - // before the sender's reply timeout could resend it. Any sync message restarts - // the clock. - let noticed = match self.diverged_since { - Some(since) if self.last_sync_rx.is_none_or(|rx| rx <= since) => since, - _ => *self.diverged_since.insert(now), - }; - - // While an answer to our last sync message is due, a lost message is the reply - // timeout's to resend; healing is for when neither side has anything outstanding. - if self.liveness != Liveness::Active - || self.hello_state != HelloState::Established - || self.sync.reply.is_some() - { - return false; - } - - let persisted = now.saturating_duration_since(noticed) >= timing.resync_after; - let cooled_down = self - .last_resync_at - .is_none_or(|t| now.saturating_duration_since(t) >= timing.resync_after); - - if !(persisted && cooled_down) { - return false; - } - self.diverged_since = None; - self.incidents.heals += 1; - - self.last_resync_at = Some(now); - self.hello_state = HelloState::NotSent; // establish_peer resets sync on the reply - true - } - - /// Heard from recently. A returning node counts as alive from its first message, - /// before the handshake finishes, so a swarm-wide forget can't make it flap. - pub(crate) fn is_alive(&self, timing: &Timing, now: Instant) -> bool { - self.liveness == Liveness::Active - && self - .last_seen - .is_some_and(|t| now.saturating_duration_since(t) < timing.dormant_after) - } -} - -#[derive(Clone, Copy, Debug)] -pub(crate) struct Timing { - pub(crate) keepalive: Duration, - pub(crate) dormant_after: Duration, - pub(crate) resync_after: Duration, - pub(crate) dormant_probe: Duration, - /// While catching up: PING the source after this much silence (`swarm::source`). - pub(crate) catch_up_probe: Duration, - /// First wait for an answer to a sync message before sending again; doubles while - /// answers stay out (`SyncSession::reply_backoff`). - pub(crate) sync_reply_timeout: Duration, - /// How long the transport tries to deliver a message before dropping it (Mycelium: - /// five minutes): a peer silent for longer has nothing on the way to us any more. - pub(crate) delivery_limit: Duration, -} - -impl Timing { - /// How long a round-trip sample stays meaningful (`swarm::rtt`): a few keepalive - /// intervals, since nodes move. - pub(crate) fn rtt_fresh_for(&self) -> Duration { - self.keepalive * 5 - } - - // Recommended values - // Environment | Keepalive | Notes - // --------------------------------------------+-----------+------------------------------------------------------------ - // Test harness | 1-2 s | Dormancy after about 4–8 s and healing within seconds keep tests short. - // LAN or small swarm (up to ~50 nodes) | 15-30 s | Under 3.5 messages/s per node at 50 nodes and 30 s. Current setting. - // WAN, mobile or metered links, or 100+ nodes | 60-120 s | At 100 nodes and 30 s it’s about 6–7 messages/s per node. Fine, but rarely worth it on battery or metered links. - // Several hundred nodes or more | - | The full mesh becomes the problem in itself, not the interval. Message counts grow O(N²) across the swarm, which calls for partial views (SWIM-style random probing and syncing with a subset of peers). - pub(crate) fn from_keepalive(keepalive: Duration) -> Self { - Self::new(keepalive, SYNC_REPLY_TIMEOUT) - } - - /// [`from_keepalive`](Self::from_keepalive) with another first sync reply timeout: - /// for tests, whose messages don't travel for seconds. - pub(crate) fn new(keepalive: Duration, sync_reply_timeout: Duration) -> Self { - Self { - keepalive, - dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row - // >= 2 keepalives, so a second exchange confirms the difference; and >= twice - // the first sync reply timeout, so a lost message's sender resends it first. - resync_after: (keepalive * 2).max(sync_reply_timeout * 2), - dormant_probe: keepalive * 10, // >= 10*: dormant peers cost almost nothing, but still rejoin within minutes - // Seconds, not a keepalive interval: a dead source should cost little time. - // Only the source gets these, and only while catching up. - catch_up_probe: (keepalive / 30).clamp(Duration::from_secs(1), Duration::from_secs(5)), - sync_reply_timeout, - delivery_limit: MYCELIUM_DELIVERY_LIMIT, - } - } - - /// Waits for sync answers: `sync_reply_timeout`, doubling up to 30 times as much. - pub(crate) fn sync_reply_backoff(&self) -> Backoff { - Backoff::new( - self.sync_reply_timeout, - self.sync_reply_timeout * SYNC_REPLY_TIMEOUT_MAX_FACTOR, - ) - } -} - -#[derive(Clone)] -pub(crate) struct AppState { - /// The document and its epoch. Derefs to `AutoCommit`. - pub(crate) doc: Arc<RwLock<Replica>>, - - /// Copy of `doc.epoch` for cheap, lock-free reads (handshakes, keepalives, /peers). - /// Only ever changed while holding `doc`'s write lock, together with `doc.epoch`. - pub(crate) epoch_tx: watch::Sender<Epoch>, - - pub(crate) lifecycle: Arc<RwLock<NodeLifecycle>>, - - // Serializes swarm-ID establishment, peer membership, and readiness transitions. - //[concurrent]swarm_validation: Arc<Mutex<()>>, - pub(crate) peer_states: Arc<RwLock<HashMap<String, Arc<Mutex<PeerState>>>>>, - pub(crate) transport: Arc<dyn transport::Transport>, - pub(crate) firewall: Arc<Firewall>, - /// Messages and bytes exchanged, per peer (`transport::traffic`). - pub(crate) traffic: Arc<transport::traffic::Traffic>, - /// Node-wide figures for /admin/stats (`swarm::stats`). - pub(crate) stats: Arc<crate::swarm::stats::NodeStats>, - - pub(crate) sync_events: mpsc::UnboundedSender<SyncEvent>, - - // TODO see key_watchers.md ("Use a per-key notification registry instead of the global `watch<u64>`") - pub(crate) change_tx: watch::Sender<u64>, - - pub(crate) initial_sync_done: Arc<AtomicBool>, - - pub(crate) change_events: broadcast::Sender<Arc<ChangeSet>>, - - pub(crate) timing: Timing, - - /// Per-node remake state (prebuilt snapshots, switch status, archive). - pub(crate) remake: Arc<RemakeState>, - - /// Pre-shared key every member must prove in the handshake; None admits any node - /// that reaches us. - pub(crate) swarm_key: Option<Arc<crate::swarm::auth::SwarmKey>>, - - /// Random per run of this node; our HELLOs carry it and replies must echo it. - pub(crate) run_id: transport::RunId, - - /// Target number of neighbours (`--neighbours`); up to this many + 1 members the - /// swarm is a full mesh. - pub(crate) neighbours_k: usize, - /// Our alias (`--alias`), published in the member list (`swarm::members`). - pub(crate) alias: Option<String>, - /// Members alive within this window vote on remakes (`--electorate-window-days`). - pub(crate) electorate_window: Duration, - /// Chunked transfers served and fetched (`swarm::blob`). - pub(crate) blobs: Arc<crate::swarm::blob::Blobs>, - /// HTTP API users who may write protocol data through the store API - /// (`--protocol-writers`). - pub(crate) protocol_writers: Arc<std::collections::HashSet<String>>, - /// This node's own files, and how much it read and wrote them. - pub(crate) files: Arc<crate::files::Files>, - /// This node's document on disk (`--document-file`), if configured. - pub(crate) document_file: Option<Arc<crate::document::file::DocumentFile>>, - /// What this node remembers across restarts (`--state-file`), if configured. - pub(crate) state_file: Option<Arc<crate::swarm::state_file::StateFile>>, - /// The effective configuration with sources, imported values left out - /// (`config::Loaded::report`), for `GET /admin/config`. - pub(crate) config: Arc<serde_json::Value>, - /// Our own connectivity record and stability tier (`swarm::stability`). - pub(crate) stability: Arc<std::sync::Mutex<crate::swarm::stability::Stability>>, - /// Round-trip times to peers (`swarm::rtt`). - pub(crate) rtt: Arc<std::sync::Mutex<crate::swarm::rtt::RttTable>>, - /// Sync throughput with peers (`swarm::throughput`). - pub(crate) throughput: Arc<std::sync::Mutex<ThroughputTable>>, - /// The neighbour we are pulling changes from, and since when (`swarm::relay`). - pub(crate) pulling: Arc<std::sync::Mutex<Option<(String, Instant)>>>, - /// An announcement round is scheduled, for changes from these peers - /// (`swarm::relay`). - pub(crate) announce_pending: Arc<std::sync::Mutex<Option<std::collections::HashSet<String>>>>, - /// Our neighbours, once known (`swarm::topology`). - pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>, - /// Something the neighbours are computed from may have changed: the member list or a - /// peer's reachability (`swarm::topology::run_worker`). - pub(crate) topology_changed: Arc<tokio::sync::Notify>, - - /// While rejoining a newer epoch, the previous document, served read-only (and - /// marked stale) instead of answering 503. - pub(crate) stale: Arc<std::sync::RwLock<Option<StaleView>>>, - - /// While the initial sync isn't done: the one peer we catch up from. - pub(crate) catch_up_source: Arc<std::sync::Mutex<Option<CatchUpSource>>>, - /// Since when a closer peer has had a head start to become the source. - pub(crate) catch_up_preference: Arc<std::sync::Mutex<Option<Instant>>>, - /// A catch-up source given up for not answering our sync messages, and until when it - /// isn't chosen again (`swarm::source`). - pub(crate) catch_up_avoided: Arc<std::sync::Mutex<Option<(String, Instant)>>>, - /// The node's wall clock (`clock`). - pub(crate) clock: Arc<crate::clock::Clock>, -} - -/// The peer a catching-up node syncs with first (`swarm::source`). -#[derive(Clone, Debug)] -pub(crate) struct CatchUpSource { - pub(crate) peer: String, - /// When it was chosen or last sent us anything. - pub(crate) last_heard: Instant, - /// When it was chosen or last sent us a sync message: progress, not just liveness. - pub(crate) last_sync: Instant, - /// Our sync messages to it in a row whose answer didn't come in time. - pub(crate) reply_timeouts: u32, - /// When we last warned that catching up from it stalls. - pub(crate) last_warned: Option<Instant>, - /// Our last probe PING to it. - pub(crate) last_probe: Option<Instant>, - /// Its keepalives in a row saying it wouldn't sync with us. - pub(crate) refusals: u32, -} - -impl AppState { - pub(crate) fn local_id(&self) -> &str { - self.transport.local_id() - } - - /// Peers that can serve our epoch's data: active, ready members on our epoch - /// (`PeerSummary::serves_epoch`). - pub(crate) async fn active_peers_on_local_epoch(&self) -> Vec<String> { - let epoch = self.local_epoch(); - self.peer_summaries() - .await - .into_iter() - .filter(|peer| peer.serves_epoch(epoch)) - .map(|peer| peer.id) - .collect() - } - - pub(crate) fn lock_rtt(&self) -> std::sync::MutexGuard<'_, crate::swarm::rtt::RttTable> { - self.rtt - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - } - - /// The local epoch. Code that labels or applies sync messages must use the epoch - /// of the locked `Replica` instead, so label and document can't disagree. - pub(crate) fn local_epoch(&self) -> Epoch { - *self.epoch_tx.borrow() - } - - /// The local epoch and `Replica::epoch_seq`, read together. - pub(crate) async fn epoch_stamp(&self) -> (Epoch, u32) { - let replica = self.doc.read().await; - (replica.epoch, replica.epoch_seq) - } - - pub(crate) async fn is_ready(&self) -> bool { - matches!(*self.lifecycle.read().await, NodeLifecycle::Ready { .. }) - } - - pub(crate) async fn swarm_id(&self) -> Option<String> { - match &*self.lifecycle.read().await { - NodeLifecycle::Ready { swarm_id } => Some(swarm_id.clone()), - NodeLifecycle::Joining { - discovered_swarm_id, - } => discovered_swarm_id.clone(), - } - } - - // Record the expected swarm without making the node Ready. Bootstrap owns readiness. - pub(crate) async fn set_expected_swarm_id(&self, swarm_id: String) -> Result<()> { - //[concurrent]let _guard = self.swarm_validation.lock().await; - let mut lifecycle = self.lifecycle.write().await; - match &*lifecycle { - NodeLifecycle::Joining { - discovered_swarm_id: Some(existing), - } if existing != &swarm_id => { - bail!( - "Conflicting expected swarm IDs: expected {}, received {}", - existing, - swarm_id - ); - } - NodeLifecycle::Joining { - discovered_swarm_id: None, - } => { - let old_lifecycle = lifecycle.clone(); - *lifecycle = NodeLifecycle::Joining { - discovered_swarm_id: Some(swarm_id), - }; - debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}"); - Ok(()) - } - NodeLifecycle::Joining { - discovered_swarm_id: Some(_), - } => { - warn!("Node already Joining with the same identity"); - Ok(()) - } - NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => { - warn!("Node already Ready with the same identity"); - Ok(()) - } - NodeLifecycle::Ready { swarm_id: existing } => { - bail!( - "Cannot change established swarm ID: existing {}, received {}", - existing, - swarm_id - ); - } - } - } - - // Only bootstrap should call this, after validating document metadata. - pub(crate) async fn mark_ready(&self, swarm_id: String) -> Result<()> { - //[concurrent]let _guard = self.swarm_validation.lock().await; - let mut lifecycle = self.lifecycle.write().await; - match &*lifecycle { - NodeLifecycle::Joining { - discovered_swarm_id: Some(expected), - } if expected != &swarm_id => { - bail!( - "Cannot mark node ready: expected swarm {}, received {}", - expected, - swarm_id - ); - } - NodeLifecycle::Joining { .. } => { - let old_lifecycle = lifecycle.clone(); - *lifecycle = NodeLifecycle::Ready { swarm_id }; - debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}"); - } - NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => { - debug!("Node already ready with the same identity"); - } - NodeLifecycle::Ready { swarm_id: existing } => { - bail!( - "Cannot change established swarm ID: existing {}, received {}", - existing, - swarm_id - ); - } - } - Ok(()) - } - - /// Looks up a peer without creating it. Background work (sync, retries, - /// keepalives) must use this so a forgotten peer isn't resurrected. - pub(crate) async fn get_peer_state(&self, pk: &str) -> Option<Arc<Mutex<PeerState>>> { - self.peer_states.read().await.get(pk).cloned() - } - - pub(crate) async fn get_or_create_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> { - let mut peers = self.peer_states.write().await; - - Arc::clone( - peers - .entry(pk.to_owned()) - .or_insert_with(|| self.new_peer_state(pk)), - ) - } - - /// State for a peer we haven't known so far (not inserted). - pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> { - let leads = crate::swarm::pair::leads(self.local_id(), pk); - Arc::new(Mutex::new(PeerState::new(leads, &self.timing))) - } - - pub(crate) fn notify_store_changed(&self) { - self.change_tx.send_modify(|version| { - *version = version.wrapping_add(1); - }); - } - - /// Reads the document on the blocking pool; `finish` runs after the lock is released. - pub(crate) async fn render<T, R>( - &self, - read: impl FnOnce(&AutoCommit) -> T + Send + 'static, - finish: impl FnOnce(T) -> R + Send + 'static, - ) -> R - where - R: Send + 'static, - { - let doc = Arc::clone(&self.doc); - tokio::task::spawn_blocking(move || { - let data = { - let guard = doc.blocking_read(); - read(&guard.doc) - }; - finish(data) - }) - .await - .expect("document read task panicked") - } - - /// Mutates the document on the blocking pool. - pub(crate) async fn write_doc<T>( - &self, - f: impl FnOnce(&mut AutoCommit) -> T + Send + 'static, - ) -> T - where - T: Send + 'static, - { - let doc = Arc::clone(&self.doc); - tokio::task::spawn_blocking(move || { - let mut replica = doc.blocking_write(); - let out = f(&mut replica.doc); - replica.refresh_heads(); - out - }) - .await - .expect("document write task panicked") - } - - /// Mutates the document on the blocking pool as one Automerge change, then - /// publishes it. Write and notification run in the same blocking task, so a - /// cancelled request can't leave a committed write unannounced. - pub(crate) async fn mutate_doc<T, E>( - &self, - f: impl FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static, - ) -> Result<T, E> - where - T: Send + 'static, - E: Send + 'static, - { - let state = self.clone(); - - tokio::task::spawn_blocking(move || { - let result = { - let mut replica = state.doc.blocking_write(); - let result = f(&mut replica.doc); - match &result { - // One API write = one Automerge change. - Ok(_) => { - // Stamped with our clock (milliseconds), so receivers can tell - // how long the change took to reach them (`swarm::stats`). - replica.doc.commit_with( - automerge::transaction::CommitOptions::default() - .with_time(state.clock.now_millis() as i64), - ); - } - // Discard partial ops (e.g. old value deleted, new one not written). - Err(_) => { - replica.doc.rollback(); - } - } - replica.refresh_heads(); - result - }; // lock released before notifying - - if result.is_ok() { - state.publish_change(); - } - result - }) - .await - .expect("document write task panicked") - } - - pub(crate) fn publish_change(&self) { - self.notify_store_changed(); - if let Err(error) = self.sync_events.send(SyncEvent::DocumentChanged) { - // The write is committed; failing the request would misreport it. - error!(error = %error, "Sync worker is gone; change will not be replicated"); - } - } -} - -impl AppState { - /// After the document was replaced: every peer's sync state refers to the old - /// document, so start over with all of them. Peers still on the old epoch are - /// skipped by the sync worker until they switch too. - async fn after_epoch_switch(&self) { - let peers: Vec<(String, Arc<Mutex<PeerState>>)> = self - .peer_states - .read() - .await - .iter() - .map(|(id, peer)| (id.clone(), Arc::clone(peer))) - .collect(); - - // Sync states were reset under the document lock by the swap itself. - self.publish_change(); - - // Tell peers about the new epoch now instead of with the next keepalive (up to a - // keepalive interval later). Ready peers already on it learn it from our sync - // opening, which goes out right away; everyone else gets a PING, so peers that - // never saw the commit start catching up (we don't sync with them, so nothing - // else would tell them). - let local = self.local_epoch(); - let mut announce = Vec::new(); - for (id, peer_state_mutex) in &peers { - // Others learn it from their own neighbours. - if !crate::swarm::topology::is_neighbour(self, id) { - continue; - } - let mut peer_state = peer_state_mutex.lock().await; - // What it said about syncing with us was about our old epoch. - peer_state.syncs_with_us = None; - let gets_our_opening = peer_state.liveness == Liveness::Active - && matches!(peer_state.standing, Standing::Member { .. }) - && peer_state.epoch == Some(local) - && peer_state.reports_ready == Some(true); - if !gets_our_opening { - announce.push(id); - } - } - - for id in announce { - if let Err(error) = crate::swarm::peer::send_ping(self, id).await { - debug!(peer = %id, error = %format!("{error:#}"), "Epoch announcement failed"); - } - } - - for (id, _) in peers { - let _ = self.sync_events.send(SyncEvent::PeerNeedsSync(id)); - } - } -} - -/// A read-only view of the document the node served before it started rejoining. -pub(crate) struct StaleView { - pub(crate) epoch: Epoch, - pub(crate) doc: Arc<AutoCommit>, -} - -/// Where a read was served from. -pub(crate) enum ReadSource { - Live, - /// The pre-rejoin document of this epoch. - Stale(Epoch), -} - -impl AppState { - /// Renders a read for the store API: from the live document when the node is ready, - /// from the stale view while it rejoins, and otherwise `503`, or the live document - /// anyway with `unready_live`. - /// - /// With `with_heads`, a live read also returns the document's heads at the moment of - /// the read (for clients that base later requests on what they read). Stale reads - /// have no heads. - pub(crate) async fn render_read<T, R>( - &self, - unready_live: bool, - with_heads: bool, - read: impl FnOnce(&AutoCommit) -> T + Send + 'static, - finish: impl FnOnce(T) -> R + Send + 'static, - ) -> Result<(R, ReadSource, Option<Vec<ChangeHash>>), axum::http::StatusCode> - where - T: Send + 'static, - R: Send + 'static, - { - if self.is_ready().await { - let (rendered, heads) = self.render_live(with_heads, read, finish).await; - return Ok((rendered, ReadSource::Live, heads)); - } - - let stale = self - .stale - .read() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - .as_ref() - .map(|view| (view.epoch, Arc::clone(&view.doc))); - let Some((epoch, doc)) = stale else { - if unready_live { - let (rendered, heads) = self.render_live(with_heads, read, finish).await; - return Ok((rendered, ReadSource::Live, heads)); - } - return Err(axum::http::StatusCode::SERVICE_UNAVAILABLE); - }; - - let rendered = tokio::task::spawn_blocking(move || finish(read(&doc))) - .await - .expect("document read task panicked"); - Ok((rendered, ReadSource::Stale(epoch), None)) - } - - async fn render_live<T, R>( - &self, - with_heads: bool, - read: impl FnOnce(&AutoCommit) -> T + Send + 'static, - finish: impl FnOnce(T) -> R + Send + 'static, - ) -> (R, Option<Vec<ChangeHash>>) - where - T: Send + 'static, - R: Send + 'static, - { - if !with_heads { - return (self.render(read, finish).await, None); - } - // Heads and data under the same read lock (the heads are cached: `Replica`). - let doc = Arc::clone(&self.doc); - let (data, heads) = tokio::task::spawn_blocking(move || { - let replica = doc.blocking_read(); - (read(&replica.doc), replica.heads().to_vec()) - }) - .await - .expect("document read task panicked"); - let rendered = tokio::task::spawn_blocking(move || finish(data)) - .await - .expect("document read task panicked"); - (rendered, Some(heads)) - } - - fn set_stale(&self, view: Option<StaleView>) { - *self - .stale - .write() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = view; - } -} - -/// The node as seen by the remake module. -impl Host for AppState { - fn remake(&self) -> &RemakeState { - &self.remake - } - - fn clock(&self) -> Arc<crate::clock::Clock> { - Arc::clone(&self.clock) - } - - fn node_id(&self) -> String { - self.local_id().to_owned() - } - - fn current_epoch(&self) -> Epoch { - self.local_epoch() - } - - fn dormant_after(&self) -> Duration { - self.timing.dormant_after - } - - fn electorate(&self, doc: &AutoCommit, now_ms: u64) -> Vec<String> { - crate::swarm::members::electorate(doc, now_ms, self.electorate_window) - } - - fn fetch_change( - &self, - hash: ChangeHash, - ) -> impl Future<Output = Result<automerge::Change>> + Send { - let state = self.clone(); - async move { - let peers = state.active_peers_on_local_epoch().await; - crate::swarm::blob::fetch_change(&state, hash, peers).await - } - } - - fn fetch_content(&self, hash: [u8; 32]) -> impl Future<Output = Result<Vec<u8>>> + Send { - let state = self.clone(); - async move { - let peers = state.active_peers_on_local_epoch().await; - crate::swarm::blob::fetch_content(&state, hash, peers).await - } - } - - fn request_sync(&self) { - crate::swarm::source::resync_everyone(self); - } - - fn is_node_ready(&self) -> impl Future<Output = bool> + Send { - let state = self.clone(); - async move { state.is_ready().await } - } - - fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send { - let state = self.clone(); - async move { - let peers: Vec<(String, Arc<Mutex<PeerState>>)> = state - .peer_states - .read() - .await - .iter() - .map(|(id, peer)| (id.clone(), Arc::clone(peer))) - .collect(); - - let mut summaries = Vec::with_capacity(peers.len()); - for (id, peer) in peers { - let peer = peer.lock().await; - summaries.push(PeerSummary { - id, - active: peer.liveness == Liveness::Active, - member: matches!(peer.standing, Standing::Member { .. }) - && peer.hello_state == HelloState::Established, - ready: peer.reports_ready, - epoch: peer.epoch, - }); - } - summaries - } - } - - fn suspend(&self) -> impl Future<Output = bool> + Send { - let state = self.clone(); - async move { - let mut lifecycle = state.lifecycle.write().await; - let NodeLifecycle::Ready { swarm_id } = &*lifecycle else { - return false; - }; - let swarm_id = swarm_id.clone(); - debug!(swarm_id = %swarm_id, "Node suspended (rejoining)"); - *lifecycle = NodeLifecycle::Joining { - discovered_swarm_id: Some(swarm_id), - }; - true - } - } - - fn resume(&self) -> impl Future<Output = ()> + Send { - let state = self.clone(); - async move { - let mut lifecycle = state.lifecycle.write().await; - if let NodeLifecycle::Joining { - discovered_swarm_id: Some(swarm_id), - } = &*lifecycle - { - let swarm_id = swarm_id.clone(); - debug!(swarm_id = %swarm_id, "Node resumed"); - *lifecycle = NodeLifecycle::Ready { swarm_id }; - } - } - } - - fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>) { - self.set_stale(view.map(|(epoch, doc)| StaleView { - epoch, - doc: Arc::new(doc), - })); - } - - fn initial_sync_done(&self) -> bool { - self.initial_sync_done - .load(std::sync::atomic::Ordering::Acquire) - } - - fn reset_initial_sync(&self) { - self.initial_sync_done - .store(false, std::sync::atomic::Ordering::Release); - *self - .catch_up_source - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; - *self - .catch_up_preference - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; - } - - fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>> { - self.change_events.subscribe() - } - - fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send - where - F: FnOnce(&AutoCommit) -> T + Send + 'static, - T: Send + 'static, - { - let state = self.clone(); - async move { state.render(f, |value| value).await } - } - - fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send - where - F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static, - T: Send + 'static, - E: Send + 'static, - { - let state = self.clone(); - async move { state.mutate_doc(f).await } - } - - fn switch_epoch<E, F>( - &self, - epoch: Epoch, - f: F, - ) -> impl Future<Output = Result<AutoCommit, E>> + Send - where - F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static, - E: Send + 'static, - { - let state = self.clone(); - async move { - let swapped = { - let state = state.clone(); - tokio::task::spawn_blocking(move || -> Result<AutoCommit, E> { - let mut replica = state.doc.blocking_write(); - let new_doc = f(&mut replica.doc)?; - let old_doc = std::mem::replace(&mut replica.doc, new_doc); - replica.epoch = epoch; - replica.epoch_seq += 1; - replica.refresh_heads(); - // Updated under the same lock, so the cached epoch never disagrees - // with the document for anyone holding the lock. - state.epoch_tx.send_replace(epoch); - // Reset every peer's sync state before the lock is released, so no - // message is generated from or applied to a state that describes the - // old document. (Lock order doc -> peer, as in sync generation.) - let now = Instant::now(); - for peer in state.peer_states.blocking_read().values() { - let mut peer = peer.blocking_lock(); - peer.reset_sync(); - // A fresh start, not an idle sync: no heal re-handshake for it. - peer.last_sync_rx = Some(now); - } - Ok(old_doc) - }) - .await - .expect("epoch switch task panicked") - }; - - if swapped.is_ok() { - state.after_epoch_switch().await; - } - swapped - } - } -} diff --git a/src/state/host.rs b/src/state/host.rs new file mode 100644 index 0000000..863cc70 --- /dev/null +++ b/src/state/host.rs @@ -0,0 +1,276 @@ +//! The node as the remake module sees it (`remake::Host`), and what follows an epoch +//! switch. + +use std::future::Future; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Result; +use automerge::{AutoCommit, ChangeHash}; +use tokio::sync::{Mutex, broadcast}; +use tokio::time::Instant; +use tracing::debug; + +use super::{AppState, HelloState, Liveness, NodeLifecycle, PeerState, StaleView, Standing}; +use crate::document::replica::Epoch; +use crate::remake::{Host, PeerSummary, RemakeState}; +use crate::swarm::sync::{ChangeSet, SyncEvent}; + +impl AppState { + /// After the document was replaced: every peer's sync state refers to the old + /// document, so start over with all of them. Peers still on the old epoch are + /// skipped by the sync worker until they switch too. + async fn after_epoch_switch(&self) { + let peers: Vec<(String, Arc<Mutex<PeerState>>)> = self + .peer_states + .read() + .await + .iter() + .map(|(id, peer)| (id.clone(), Arc::clone(peer))) + .collect(); + + // Sync states were reset under the document lock by the swap itself. + self.publish_change(); + + // Tell peers about the new epoch now instead of with the next keepalive (up to a + // keepalive interval later). Ready peers already on it learn it from our sync + // opening, which goes out right away; everyone else gets a PING, so peers that + // never saw the commit start catching up (we don't sync with them, so nothing + // else would tell them). + let local = self.local_epoch(); + let mut announce = Vec::new(); + for (id, peer_state_mutex) in &peers { + // Others learn it from their own neighbours. + if !crate::swarm::topology::is_neighbour(self, id) { + continue; + } + let mut peer_state = peer_state_mutex.lock().await; + // What it said about syncing with us was about our old epoch. + peer_state.syncs_with_us = None; + let gets_our_opening = peer_state.liveness == Liveness::Active + && matches!(peer_state.standing, Standing::Member { .. }) + && peer_state.epoch == Some(local) + && peer_state.reports_ready == Some(true); + if !gets_our_opening { + announce.push(id); + } + } + + for id in announce { + if let Err(error) = crate::swarm::peer::send_ping(self, id).await { + debug!(peer = %id, error = %format!("{error:#}"), "Epoch announcement failed"); + } + } + + for (id, _) in peers { + let _ = self.sync_events.send(SyncEvent::PeerNeedsSync(id)); + } + } +} + +/// The node as seen by the remake module. +impl Host for AppState { + fn remake(&self) -> &RemakeState { + &self.remake + } + + fn clock(&self) -> Arc<crate::clock::Clock> { + Arc::clone(&self.clock) + } + + fn node_id(&self) -> String { + self.local_id().to_owned() + } + + fn current_epoch(&self) -> Epoch { + self.local_epoch() + } + + fn dormant_after(&self) -> Duration { + self.timing.dormant_after + } + + fn electorate(&self, doc: &AutoCommit, now_ms: u64) -> Vec<String> { + crate::swarm::members::electorate(doc, now_ms, self.electorate_window) + } + + fn fetch_change( + &self, + hash: ChangeHash, + ) -> impl Future<Output = Result<automerge::Change>> + Send { + let state = self.clone(); + async move { + let peers = state.active_peers_on_local_epoch().await; + crate::swarm::blob::fetch_change(&state, hash, peers).await + } + } + + fn fetch_content(&self, hash: [u8; 32]) -> impl Future<Output = Result<Vec<u8>>> + Send { + let state = self.clone(); + async move { + let peers = state.active_peers_on_local_epoch().await; + crate::swarm::blob::fetch_content(&state, hash, peers).await + } + } + + fn request_sync(&self) { + crate::swarm::source::resync_everyone(self); + } + + fn is_node_ready(&self) -> impl Future<Output = bool> + Send { + let state = self.clone(); + async move { state.is_ready().await } + } + + fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send { + let state = self.clone(); + async move { + let peers: Vec<(String, Arc<Mutex<PeerState>>)> = state + .peer_states + .read() + .await + .iter() + .map(|(id, peer)| (id.clone(), Arc::clone(peer))) + .collect(); + + let mut summaries = Vec::with_capacity(peers.len()); + for (id, peer) in peers { + let peer = peer.lock().await; + summaries.push(PeerSummary { + id, + active: peer.liveness == Liveness::Active, + member: matches!(peer.standing, Standing::Member { .. }) + && peer.hello_state == HelloState::Established, + ready: peer.reports_ready, + epoch: peer.epoch, + }); + } + summaries + } + } + + fn suspend(&self) -> impl Future<Output = bool> + Send { + let state = self.clone(); + async move { + let mut lifecycle = state.lifecycle.write().await; + let NodeLifecycle::Ready { swarm_id } = &*lifecycle else { + return false; + }; + let swarm_id = swarm_id.clone(); + debug!(swarm_id = %swarm_id, "Node suspended (rejoining)"); + *lifecycle = NodeLifecycle::Joining { + discovered_swarm_id: Some(swarm_id), + }; + true + } + } + + fn resume(&self) -> impl Future<Output = ()> + Send { + let state = self.clone(); + async move { + let mut lifecycle = state.lifecycle.write().await; + if let NodeLifecycle::Joining { + discovered_swarm_id: Some(swarm_id), + } = &*lifecycle + { + let swarm_id = swarm_id.clone(); + debug!(swarm_id = %swarm_id, "Node resumed"); + *lifecycle = NodeLifecycle::Ready { swarm_id }; + } + } + } + + fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>) { + self.set_stale(view.map(|(epoch, doc)| StaleView { + epoch, + doc: Arc::new(doc), + })); + } + + fn initial_sync_done(&self) -> bool { + self.initial_sync_done + .load(std::sync::atomic::Ordering::Acquire) + } + + fn reset_initial_sync(&self) { + self.initial_sync_done + .store(false, std::sync::atomic::Ordering::Release); + *self + .catch_up_source + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; + *self + .catch_up_preference + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; + } + + fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>> { + self.change_events.subscribe() + } + + fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send + where + F: FnOnce(&AutoCommit) -> T + Send + 'static, + T: Send + 'static, + { + let state = self.clone(); + async move { state.render(f, |value| value).await } + } + + fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send + where + F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static, + T: Send + 'static, + E: Send + 'static, + { + let state = self.clone(); + async move { state.mutate_doc(f).await } + } + + fn switch_epoch<E, F>( + &self, + epoch: Epoch, + f: F, + ) -> impl Future<Output = Result<AutoCommit, E>> + Send + where + F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static, + E: Send + 'static, + { + let state = self.clone(); + async move { + let swapped = { + let state = state.clone(); + tokio::task::spawn_blocking(move || -> Result<AutoCommit, E> { + let mut replica = state.doc.blocking_write(); + let new_doc = f(&mut replica.doc)?; + let old_doc = std::mem::replace(&mut replica.doc, new_doc); + replica.epoch = epoch; + replica.epoch_seq += 1; + replica.refresh_heads(); + // Updated under the same lock, so the cached epoch never disagrees + // with the document for anyone holding the lock. + state.epoch_tx.send_replace(epoch); + // Reset every peer's sync state before the lock is released, so no + // message is generated from or applied to a state that describes the + // old document. (Lock order doc -> peer, as in sync generation.) + let now = Instant::now(); + for peer in state.peer_states.blocking_read().values() { + let mut peer = peer.blocking_lock(); + peer.reset_sync(); + // A fresh start, not an idle sync: no heal re-handshake for it. + peer.last_sync_rx = Some(now); + } + Ok(old_doc) + }) + .await + .expect("epoch switch task panicked") + }; + + if swapped.is_ok() { + state.after_epoch_switch().await; + } + swapped + } + } +} diff --git a/src/state/mod.rs b/src/state/mod.rs new file mode 100644 index 0000000..c5a6885 --- /dev/null +++ b/src/state/mod.rs @@ -0,0 +1,404 @@ +//! Shared node state: the document, lifecycle, per-peer state and timing. +//! +//! * `peer`: what a node knows about each peer (`PeerState`): standing, liveness, +//! handshake, incidents. +//! * `session`: one Automerge sync with a peer (`SyncSession`). +//! * `timing`: the intervals the protocol runs on (`Timing`). +//! * `read`: serving reads, live or from the pre-rejoin copy. +//! * `host`: the node as the remake module sees it (`remake::Host`). + +mod host; +mod peer; +mod read; +mod session; +mod timing; + +pub(crate) use peer::{HelloState, Incidents, Liveness, PeerState, Standing}; +pub(crate) use read::{ReadSource, StaleView}; +pub(crate) use session::{Sending, SyncSession}; +pub(crate) use timing::Timing; + +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::time::Duration; + +use anyhow::{Result, bail}; +use automerge::AutoCommit; +use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch}; +use tokio::time::Instant; +use tracing::{debug, error, warn}; + +use crate::document::replica::{Epoch, Replica}; +use crate::remake::{Host, RemakeState}; +use crate::swarm::sync::{ChangeSet, SyncEvent}; +use crate::swarm::throughput::ThroughputTable; +use crate::transport; +use crate::transport::firewall::Firewall; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum NodeLifecycle { + // TODO maybe rename discovered_swarm_id to swarm_id if that doesn't impair readability. + Joining { discovered_swarm_id: Option<String> }, + Ready { swarm_id: String }, +} + +#[derive(Clone)] +pub(crate) struct AppState { + /// The document and its epoch. Derefs to `AutoCommit`. + pub(crate) doc: Arc<RwLock<Replica>>, + + /// Copy of `doc.epoch` for cheap, lock-free reads (handshakes, keepalives, /peers). + /// Only ever changed while holding `doc`'s write lock, together with `doc.epoch`. + pub(crate) epoch_tx: watch::Sender<Epoch>, + + pub(crate) lifecycle: Arc<RwLock<NodeLifecycle>>, + + // Serializes swarm-ID establishment, peer membership, and readiness transitions. + //[concurrent]swarm_validation: Arc<Mutex<()>>, + pub(crate) peer_states: Arc<RwLock<HashMap<String, Arc<Mutex<PeerState>>>>>, + pub(crate) transport: Arc<dyn transport::Transport>, + pub(crate) firewall: Arc<Firewall>, + /// Messages and bytes exchanged, per peer (`transport::traffic`). + pub(crate) traffic: Arc<transport::traffic::Traffic>, + /// Node-wide figures for /admin/stats (`swarm::stats`). + pub(crate) stats: Arc<crate::swarm::stats::NodeStats>, + + pub(crate) sync_events: mpsc::UnboundedSender<SyncEvent>, + + // TODO see key_watchers.md ("Use a per-key notification registry instead of the global `watch<u64>`") + pub(crate) change_tx: watch::Sender<u64>, + + pub(crate) initial_sync_done: Arc<AtomicBool>, + + pub(crate) change_events: broadcast::Sender<Arc<ChangeSet>>, + + pub(crate) timing: Timing, + + /// Per-node remake state (prebuilt snapshots, switch status, archive). + pub(crate) remake: Arc<RemakeState>, + + /// Pre-shared key every member must prove in the handshake; None admits any node + /// that reaches us. + pub(crate) swarm_key: Option<Arc<crate::swarm::auth::SwarmKey>>, + + /// Random per run of this node; our HELLOs carry it and replies must echo it. + pub(crate) run_id: transport::RunId, + + /// Target number of neighbours (`--neighbours`); up to this many + 1 members the + /// swarm is a full mesh. + pub(crate) neighbours_k: usize, + /// Our alias (`--alias`), published in the member list (`swarm::members`). + pub(crate) alias: Option<String>, + /// Members alive within this window vote on remakes (`--electorate-window-days`). + pub(crate) electorate_window: Duration, + /// Chunked transfers served and fetched (`swarm::blob`). + pub(crate) blobs: Arc<crate::swarm::blob::Blobs>, + /// HTTP API users who may write protocol data through the store API + /// (`--protocol-writers`). + pub(crate) protocol_writers: Arc<std::collections::HashSet<String>>, + /// This node's own files, and how much it read and wrote them. + pub(crate) files: Arc<crate::files::Files>, + /// This node's document on disk (`--document-file`), if configured. + pub(crate) document_file: Option<Arc<crate::document::file::DocumentFile>>, + /// What this node remembers across restarts (`--state-file`), if configured. + pub(crate) state_file: Option<Arc<crate::swarm::state_file::StateFile>>, + /// The effective configuration with sources, imported values left out + /// (`config::Loaded::report`), for `GET /admin/config`. + pub(crate) config: Arc<serde_json::Value>, + /// Our own connectivity record and stability tier (`swarm::stability`). + pub(crate) stability: Arc<std::sync::Mutex<crate::swarm::stability::Stability>>, + /// Round-trip times to peers (`swarm::rtt`). + pub(crate) rtt: Arc<std::sync::Mutex<crate::swarm::rtt::RttTable>>, + /// Sync throughput with peers (`swarm::throughput`). + pub(crate) throughput: Arc<std::sync::Mutex<ThroughputTable>>, + /// The neighbour we are pulling changes from, and since when (`swarm::relay`). + pub(crate) pulling: Arc<std::sync::Mutex<Option<(String, Instant)>>>, + /// An announcement round is scheduled, for changes from these peers + /// (`swarm::relay`). + pub(crate) announce_pending: Arc<std::sync::Mutex<Option<std::collections::HashSet<String>>>>, + /// Our neighbours, once known (`swarm::topology`). + pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>, + /// Something the neighbours are computed from may have changed: the member list or a + /// peer's reachability (`swarm::topology::run_worker`). + pub(crate) topology_changed: Arc<tokio::sync::Notify>, + + /// While rejoining a newer epoch, the previous document, served read-only (and + /// marked stale) instead of answering 503. + pub(crate) stale: Arc<std::sync::RwLock<Option<StaleView>>>, + + /// While the initial sync isn't done: the one peer we catch up from. + pub(crate) catch_up_source: Arc<std::sync::Mutex<Option<CatchUpSource>>>, + /// Since when a closer peer has had a head start to become the source. + pub(crate) catch_up_preference: Arc<std::sync::Mutex<Option<Instant>>>, + /// A catch-up source given up for not answering our sync messages, and until when it + /// isn't chosen again (`swarm::source`). + pub(crate) catch_up_avoided: Arc<std::sync::Mutex<Option<(String, Instant)>>>, + /// The node's wall clock (`clock`). + pub(crate) clock: Arc<crate::clock::Clock>, +} + +/// The peer a catching-up node syncs with first (`swarm::source`). +#[derive(Clone, Debug)] +pub(crate) struct CatchUpSource { + pub(crate) peer: String, + /// When it was chosen or last sent us anything. + pub(crate) last_heard: Instant, + /// When it was chosen or last sent us a sync message: progress, not just liveness. + pub(crate) last_sync: Instant, + /// Our sync messages to it in a row whose answer didn't come in time. + pub(crate) reply_timeouts: u32, + /// When we last warned that catching up from it stalls. + pub(crate) last_warned: Option<Instant>, + /// Our last probe PING to it. + pub(crate) last_probe: Option<Instant>, + /// Its keepalives in a row saying it wouldn't sync with us. + pub(crate) refusals: u32, +} + +impl AppState { + pub(crate) fn local_id(&self) -> &str { + self.transport.local_id() + } + + /// Peers that can serve our epoch's data: active, ready members on our epoch + /// (`PeerSummary::serves_epoch`). + pub(crate) async fn active_peers_on_local_epoch(&self) -> Vec<String> { + let epoch = self.local_epoch(); + self.peer_summaries() + .await + .into_iter() + .filter(|peer| peer.serves_epoch(epoch)) + .map(|peer| peer.id) + .collect() + } + + pub(crate) fn lock_rtt(&self) -> std::sync::MutexGuard<'_, crate::swarm::rtt::RttTable> { + self.rtt + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// The local epoch. Code that labels or applies sync messages must use the epoch + /// of the locked `Replica` instead, so label and document can't disagree. + pub(crate) fn local_epoch(&self) -> Epoch { + *self.epoch_tx.borrow() + } + + /// The local epoch and `Replica::epoch_seq`, read together. + pub(crate) async fn epoch_stamp(&self) -> (Epoch, u32) { + let replica = self.doc.read().await; + (replica.epoch, replica.epoch_seq) + } + + pub(crate) async fn is_ready(&self) -> bool { + matches!(*self.lifecycle.read().await, NodeLifecycle::Ready { .. }) + } + + pub(crate) async fn swarm_id(&self) -> Option<String> { + match &*self.lifecycle.read().await { + NodeLifecycle::Ready { swarm_id } => Some(swarm_id.clone()), + NodeLifecycle::Joining { + discovered_swarm_id, + } => discovered_swarm_id.clone(), + } + } + + // Record the expected swarm without making the node Ready. Bootstrap owns readiness. + pub(crate) async fn set_expected_swarm_id(&self, swarm_id: String) -> Result<()> { + //[concurrent]let _guard = self.swarm_validation.lock().await; + let mut lifecycle = self.lifecycle.write().await; + match &*lifecycle { + NodeLifecycle::Joining { + discovered_swarm_id: Some(existing), + } if existing != &swarm_id => { + bail!( + "Conflicting expected swarm IDs: expected {}, received {}", + existing, + swarm_id + ); + } + NodeLifecycle::Joining { + discovered_swarm_id: None, + } => { + let old_lifecycle = lifecycle.clone(); + *lifecycle = NodeLifecycle::Joining { + discovered_swarm_id: Some(swarm_id), + }; + debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}"); + Ok(()) + } + NodeLifecycle::Joining { + discovered_swarm_id: Some(_), + } => { + warn!("Node already Joining with the same identity"); + Ok(()) + } + NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => { + warn!("Node already Ready with the same identity"); + Ok(()) + } + NodeLifecycle::Ready { swarm_id: existing } => { + bail!( + "Cannot change established swarm ID: existing {}, received {}", + existing, + swarm_id + ); + } + } + } + + // Only bootstrap should call this, after validating document metadata. + pub(crate) async fn mark_ready(&self, swarm_id: String) -> Result<()> { + //[concurrent]let _guard = self.swarm_validation.lock().await; + let mut lifecycle = self.lifecycle.write().await; + match &*lifecycle { + NodeLifecycle::Joining { + discovered_swarm_id: Some(expected), + } if expected != &swarm_id => { + bail!( + "Cannot mark node ready: expected swarm {}, received {}", + expected, + swarm_id + ); + } + NodeLifecycle::Joining { .. } => { + let old_lifecycle = lifecycle.clone(); + *lifecycle = NodeLifecycle::Ready { swarm_id }; + debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}"); + } + NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => { + debug!("Node already ready with the same identity"); + } + NodeLifecycle::Ready { swarm_id: existing } => { + bail!( + "Cannot change established swarm ID: existing {}, received {}", + existing, + swarm_id + ); + } + } + Ok(()) + } + + /// Looks up a peer without creating it. Background work (sync, retries, + /// keepalives) must use this so a forgotten peer isn't resurrected. + pub(crate) async fn get_peer_state(&self, pk: &str) -> Option<Arc<Mutex<PeerState>>> { + self.peer_states.read().await.get(pk).cloned() + } + + pub(crate) async fn get_or_create_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> { + let mut peers = self.peer_states.write().await; + + Arc::clone( + peers + .entry(pk.to_owned()) + .or_insert_with(|| self.new_peer_state(pk)), + ) + } + + /// State for a peer we haven't known so far (not inserted). + pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> { + let leads = crate::swarm::pair::leads(self.local_id(), pk); + Arc::new(Mutex::new(PeerState::new(leads, &self.timing))) + } + + pub(crate) fn notify_store_changed(&self) { + self.change_tx.send_modify(|version| { + *version = version.wrapping_add(1); + }); + } + + /// Reads the document on the blocking pool; `finish` runs after the lock is released. + pub(crate) async fn render<T, R>( + &self, + read: impl FnOnce(&AutoCommit) -> T + Send + 'static, + finish: impl FnOnce(T) -> R + Send + 'static, + ) -> R + where + R: Send + 'static, + { + let doc = Arc::clone(&self.doc); + tokio::task::spawn_blocking(move || { + let data = { + let guard = doc.blocking_read(); + read(&guard.doc) + }; + finish(data) + }) + .await + .expect("document read task panicked") + } + + /// Mutates the document on the blocking pool. + pub(crate) async fn write_doc<T>( + &self, + f: impl FnOnce(&mut AutoCommit) -> T + Send + 'static, + ) -> T + where + T: Send + 'static, + { + let doc = Arc::clone(&self.doc); + tokio::task::spawn_blocking(move || { + let mut replica = doc.blocking_write(); + let out = f(&mut replica.doc); + replica.refresh_heads(); + out + }) + .await + .expect("document write task panicked") + } + + /// Mutates the document on the blocking pool as one Automerge change, then + /// publishes it. Write and notification run in the same blocking task, so a + /// cancelled request can't leave a committed write unannounced. + pub(crate) async fn mutate_doc<T, E>( + &self, + f: impl FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static, + ) -> Result<T, E> + where + T: Send + 'static, + E: Send + 'static, + { + let state = self.clone(); + + tokio::task::spawn_blocking(move || { + let result = { + let mut replica = state.doc.blocking_write(); + let result = f(&mut replica.doc); + match &result { + // One API write = one Automerge change. + Ok(_) => { + // Stamped with our clock (milliseconds), so receivers can tell + // how long the change took to reach them (`swarm::stats`). + replica.doc.commit_with( + automerge::transaction::CommitOptions::default() + .with_time(state.clock.now_millis() as i64), + ); + } + // Discard partial ops (e.g. old value deleted, new one not written). + Err(_) => { + replica.doc.rollback(); + } + } + replica.refresh_heads(); + result + }; // lock released before notifying + + if result.is_ok() { + state.publish_change(); + } + result + }) + .await + .expect("document write task panicked") + } + + pub(crate) fn publish_change(&self) { + self.notify_store_changed(); + if let Err(error) = self.sync_events.send(SyncEvent::DocumentChanged) { + // The write is committed; failing the request would misreport it. + error!(error = %error, "Sync worker is gone; change will not be replicated"); + } + } +} diff --git a/src/state/peer.rs b/src/state/peer.rs new file mode 100644 index 0000000..e96cd9c --- /dev/null +++ b/src/state/peer.rs @@ -0,0 +1,352 @@ +//! What a node knows about each peer: standing, liveness, handshake, incidents. + +use std::time::Duration; + +use tokio::time::Instant; + +use super::{SyncSession, Timing}; +use crate::backoff::Backoff; +use crate::document::replica::Epoch; +use crate::transport; + +/// How far a peer got into the swarm, as far as we're concerned. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub(crate) enum Standing { + /// Nothing proven yet. With a swarm key, everything but a handshake from it is + /// ignored (`swarm::peer::admit`). + #[default] + Stranger, + /// Proved the swarm key in a HELLO; no HELLO_REPLY yet. Only reached with a key. + Admitted, + /// Its HELLO_REPLY named our swarm (and, with a key, proved it): we sync with it. + Member { swarm_id: String }, +} + +impl Standing { + /// The swarm a member belongs to; None before its HELLO_REPLY. + pub(crate) fn swarm_id(&self) -> Option<&str> { + match self { + Self::Member { swarm_id } => Some(swarm_id), + Self::Stranger | Self::Admitted => None, + } + } + + /// Proved the swarm key (meaningful only with one). + pub(crate) fn is_admitted(&self) -> bool { + !matches!(self, Self::Stranger) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Liveness { + Active, + /// No inbound message for Timing::dormant_after. Kept indefinitely, probed slowly. + Dormant { + since: Instant, + }, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) enum HelloState { + #[default] + NotSent, + /// A send is in flight. + Sending, + /// HELLO was sent (or the send failed); no HELLO_REPLY yet. Retried at + /// `next_attempt` (`PeerState::hello_backoff`, or `dormant_probe` while dormant). + AwaitingReply { next_attempt: Instant }, + /// HELLO_REPLY received. No more retries. + Established, +} + +/// First wait for a HELLO_REPLY before sending the HELLO again; doubles while none +/// comes. +const HELLO_RETRY: Duration = Duration::from_secs(1); +const HELLO_RETRY_MAX: Duration = Duration::from_secs(30); + +#[derive(Clone)] +pub(crate) struct PeerState { + /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for + /// the pair, so computed once. + pub(crate) leads: bool, + + pub(crate) standing: Standing, + pub(crate) hello_state: HelloState, + /// Wait between unanswered HELLOs; reset whenever a new handshake starts. + pub(crate) hello_backoff: Backoff, + + /// Everything about the current Automerge sync with this peer; a reset replaces it. + pub(crate) sync: SyncSession, + + pub(crate) liveness: Liveness, + pub(crate) added_at: Instant, + pub(crate) last_seen: Option<Instant>, + pub(crate) last_ping_at: Option<Instant>, + + /// Last time a sync message from this peer was applied (survives resets: the + /// keepalive heal looks at it). + pub(crate) last_sync_rx: Option<Instant>, + + /// Last self-healing re-handshake (cooldown). + pub(crate) last_resync_at: Option<Instant>, + + /// Since when keepalives show the peer's heads differ from ours with no sync message + /// from it in between (None while they match). + pub(crate) diverged_since: Option<Instant>, + + /// Epoch the peer advertised in its most recent message (late ones aside: see + /// `epoch_seq`). + pub(crate) epoch: Option<Epoch>, + /// The peer's `epoch_seq` with `epoch`: how many times it changed epoch in its run. + /// An epoch with a lower one is older than what we have, delivered late. + pub(crate) epoch_seq: Option<u32>, + /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over. + pub(crate) run_id: Option<crate::transport::RunId>, + + /// Whether the peer was ready according to its most recent sync message or keepalive. + pub(crate) reports_ready: Option<bool>, + + /// Whether the peer said, in its most recent keepalive, that it would sync with us. + pub(crate) syncs_with_us: Option<bool>, + + /// How many heads its most recent keepalive listed (None if it sent a digest). + pub(crate) reported_heads: Option<usize>, + + /// HELLOs sent in the current handshake attempt without hearing anything back. + pub(crate) hello_attempts: u32, + /// Since when we consider the peer unreachable (three unanswered HELLOs, or + /// dormant). Cleared by any message from it; ignored after `dormant_probe`, so the + /// peer becomes a neighbour candidate again and gets tried once more. + pub(crate) unreachable_since: Option<Instant>, + + /// What went wrong with this peer so far (counted only, for /peers and /admin/stats). + pub(crate) incidents: Incidents, +} + +/// Trouble with a peer, counted since it was first listed. Lost keepalives and refused +/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING. +#[derive(Debug, Default, Clone, Copy, serde::Serialize, utoipa::ToSchema)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Incidents { + /// Our sync messages left unanswered until the reply timeout (then sent again). + pub(crate) reply_timeouts: u64, + /// Re-handshakes because our documents differed while sync was idle. + pub(crate) heals: u64, + /// Times the peer went dormant. + pub(crate) dormancies: u64, + /// Time spent dormant, in milliseconds, up to its last return (the current + /// dormancy, if any, isn't included). + pub(crate) dormant_ms: u64, +} + +impl Incidents { + pub(crate) fn add(&mut self, other: &Incidents) { + self.reply_timeouts += other.reply_timeouts; + self.heals += other.heals; + self.dormancies += other.dormancies; + self.dormant_ms += other.dormant_ms; + } +} + +/// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart). +const UNREACHABLE_AFTER_HELLOS: u32 = 3; + +impl PeerState { + pub(crate) fn new(leads: bool, timing: &Timing) -> Self { + Self { + leads, + standing: Standing::default(), + hello_state: HelloState::default(), + hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX), + sync: SyncSession::new(timing.sync_reply_backoff()), + liveness: Liveness::Active, + added_at: Instant::now(), + last_seen: None, + last_ping_at: None, + last_sync_rx: None, + last_resync_at: None, + diverged_since: None, + epoch: None, + epoch_seq: None, + run_id: None, + reports_ready: None, + syncs_with_us: None, + reported_heads: None, + hello_attempts: 0, + unreachable_since: None, + incidents: Incidents::default(), + } + } + + /// Start Automerge sync with this peer from scratch. A message still being handed to + /// Mycelium was generated against the old state; its outcome is then ignored. + pub(crate) fn reset_sync(&mut self) { + // The same waits for answers, from the start. + let mut reply_backoff = self.sync.reply_backoff; + reply_backoff.reset(); + self.sync = SyncSession::new(reply_backoff); + } + + /// An overdue sync answer from an established, active peer (see + /// `SyncSession::take_overdue_reply`). + pub(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> { + if self.liveness != Liveness::Active || self.hello_state != HelloState::Established { + return None; + } + let overdue = self.sync.take_overdue_reply(now); + if overdue.is_some() { + self.incidents.reply_timeouts += 1; + } + overdue + } + + fn quiet_for(&self, now: Instant) -> Duration { + now.saturating_duration_since(self.last_seen.unwrap_or(self.added_at)) + } + + /// Records inbound activity. Returns true if the peer was dormant. + pub(crate) fn mark_seen(&mut self, now: Instant) -> bool { + self.last_seen = Some(now); + self.hello_attempts = 0; + self.unreachable_since = None; + + if let Liveness::Dormant { since } = self.liveness { + self.incidents.dormant_ms += now.saturating_duration_since(since).as_millis() as u64; + self.liveness = Liveness::Active; + // Force a fresh handshake; its reply resets sync on both sides. + if self.hello_state != HelloState::Sending { + self.hello_state = HelloState::NotSent; + } + return true; + } + false + } + + /// A HELLO went out (or failed to). After the third unanswered one, the peer counts + /// as unreachable. + pub(crate) fn hello_sent(&mut self, now: Instant) { + self.hello_attempts += 1; + if self.hello_attempts >= UNREACHABLE_AFTER_HELLOS { + self.unreachable_since.get_or_insert(now); + } + } + + /// Whether to leave the peer out when choosing neighbours (`swarm::topology`). + pub(crate) fn is_unreachable(&self, timing: &Timing, now: Instant) -> bool { + self.unreachable_since + .is_some_and(|since| now.saturating_duration_since(since) < timing.dormant_probe) + } + + /// No more HELLO retries (the peer isn't our neighbour any more). A HELLO in flight + /// finishes; a later handshake starts over. + pub(crate) fn stop_hello_retries(&mut self) { + if matches!(self.hello_state, HelloState::AwaitingReply { .. }) { + self.hello_state = HelloState::NotSent; + } + } + + /// Moves a silent peer to Dormant. Returns true on the transition. + pub(crate) fn check_dormancy(&mut self, timing: &Timing, now: Instant) -> bool { + if self.liveness != Liveness::Active || self.quiet_for(now) < timing.dormant_after { + return false; + } + + self.liveness = Liveness::Dormant { since: now }; + self.incidents.dormancies += 1; + self.unreachable_since.get_or_insert(now); + self.reset_sync(); + + if self.hello_state != HelloState::Sending { + self.hello_state = HelloState::AwaitingReply { + next_attempt: now + timing.dormant_probe, + }; + } + true + } + + /// True (and records the ping) if an established, quiet peer needs a keepalive. + /// + /// One PING and its PONG tell both sides that the other is alive, so only one side of + /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs + /// makes the other side's peer look recently heard from, so the other side never + /// pings while it does. If the pings stop, the other side pings once it has heard + /// nothing for half an interval more. Without that, both sides would get due at about + /// the same time (each one's quiet time starts with the same exchange) and their + /// PINGs would often cross. + pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool { + let quiet_enough = if self.leads { + timing.keepalive + } else { + timing.keepalive + timing.keepalive / 2 + }; + let due = self.liveness == Liveness::Active + && self.hello_state == HelloState::Established + && now.saturating_duration_since(self.added_at) >= timing.keepalive + && self.quiet_for(now) >= quiet_enough + && self + .last_ping_at + .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive); + + if due { + self.last_ping_at = Some(now); + } + due + } + + /// Self-healing check on PING/PONG. Returns true if a re-handshake should start. + pub(crate) fn needs_heal( + &mut self, + local: &transport::HeadsSummary, + remote: &transport::HeadsSummary, + timing: &Timing, + now: Instant, + ) -> bool { + if local == remote { + self.diverged_since = None; + return false; + } + + // How long the heads have differed without sync making progress. Measured from + // when we noticed, not from the last sync message: in a quiet swarm that one is + // long past, and the first keepalive after a lost message would heal at once, + // before the sender's reply timeout could resend it. Any sync message restarts + // the clock. + let noticed = match self.diverged_since { + Some(since) if self.last_sync_rx.is_none_or(|rx| rx <= since) => since, + _ => *self.diverged_since.insert(now), + }; + + // While an answer to our last sync message is due, a lost message is the reply + // timeout's to resend; healing is for when neither side has anything outstanding. + if self.liveness != Liveness::Active + || self.hello_state != HelloState::Established + || self.sync.reply.is_some() + { + return false; + } + + let persisted = now.saturating_duration_since(noticed) >= timing.resync_after; + let cooled_down = self + .last_resync_at + .is_none_or(|t| now.saturating_duration_since(t) >= timing.resync_after); + + if !(persisted && cooled_down) { + return false; + } + self.diverged_since = None; + self.incidents.heals += 1; + + self.last_resync_at = Some(now); + self.hello_state = HelloState::NotSent; // establish_peer resets sync on the reply + true + } + + /// Heard from recently. A returning node counts as alive from its first message, + /// before the handshake finishes, so a swarm-wide forget can't make it flap. + pub(crate) fn is_alive(&self, timing: &Timing, now: Instant) -> bool { + self.liveness == Liveness::Active + && self + .last_seen + .is_some_and(|t| now.saturating_duration_since(t) < timing.dormant_after) + } +} diff --git a/src/state/read.rs b/src/state/read.rs new file mode 100644 index 0000000..b76582c --- /dev/null +++ b/src/state/read.rs @@ -0,0 +1,100 @@ +//! Serving reads: from the live document, or from the pre-rejoin copy while rejoining. + +use std::sync::Arc; + +use automerge::{AutoCommit, ChangeHash}; + +use super::AppState; +use crate::document::replica::Epoch; + +/// A read-only view of the document the node served before it started rejoining. +pub(crate) struct StaleView { + pub(crate) epoch: Epoch, + pub(crate) doc: Arc<AutoCommit>, +} + +/// Where a read was served from. +pub(crate) enum ReadSource { + Live, + /// The pre-rejoin document of this epoch. + Stale(Epoch), +} + +impl AppState { + /// Renders a read for the store API: from the live document when the node is ready, + /// from the stale view while it rejoins, and otherwise `503`, or the live document + /// anyway with `unready_live`. + /// + /// With `with_heads`, a live read also returns the document's heads at the moment of + /// the read (for clients that base later requests on what they read). Stale reads + /// have no heads. + pub(crate) async fn render_read<T, R>( + &self, + unready_live: bool, + with_heads: bool, + read: impl FnOnce(&AutoCommit) -> T + Send + 'static, + finish: impl FnOnce(T) -> R + Send + 'static, + ) -> Result<(R, ReadSource, Option<Vec<ChangeHash>>), axum::http::StatusCode> + where + T: Send + 'static, + R: Send + 'static, + { + if self.is_ready().await { + let (rendered, heads) = self.render_live(with_heads, read, finish).await; + return Ok((rendered, ReadSource::Live, heads)); + } + + let stale = self + .stale + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .as_ref() + .map(|view| (view.epoch, Arc::clone(&view.doc))); + let Some((epoch, doc)) = stale else { + if unready_live { + let (rendered, heads) = self.render_live(with_heads, read, finish).await; + return Ok((rendered, ReadSource::Live, heads)); + } + return Err(axum::http::StatusCode::SERVICE_UNAVAILABLE); + }; + + let rendered = tokio::task::spawn_blocking(move || finish(read(&doc))) + .await + .expect("document read task panicked"); + Ok((rendered, ReadSource::Stale(epoch), None)) + } + + async fn render_live<T, R>( + &self, + with_heads: bool, + read: impl FnOnce(&AutoCommit) -> T + Send + 'static, + finish: impl FnOnce(T) -> R + Send + 'static, + ) -> (R, Option<Vec<ChangeHash>>) + where + T: Send + 'static, + R: Send + 'static, + { + if !with_heads { + return (self.render(read, finish).await, None); + } + // Heads and data under the same read lock (the heads are cached: `Replica`). + let doc = Arc::clone(&self.doc); + let (data, heads) = tokio::task::spawn_blocking(move || { + let replica = doc.blocking_read(); + (read(&replica.doc), replica.heads().to_vec()) + }) + .await + .expect("document read task panicked"); + let rendered = tokio::task::spawn_blocking(move || finish(data)) + .await + .expect("document read task panicked"); + (rendered, Some(heads)) + } + + pub(super) fn set_stale(&self, view: Option<StaleView>) { + *self + .stale + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = view; + } +} diff --git a/src/state/session.rs b/src/state/session.rs new file mode 100644 index 0000000..94f9d71 --- /dev/null +++ b/src/state/session.rs @@ -0,0 +1,155 @@ +//! One Automerge sync with a peer, from a reset to the next. + +use std::time::Duration; + +use automerge::sync; +use tokio::time::Instant; + +use crate::backoff::Backoff; +use crate::swarm::throughput::SyncBytes; + +/// A sync message on its way to our Mycelium (the HTTP request is in progress). +#[derive(Clone)] +pub(crate) struct Sending { + /// Changes it carries for the first time; taken back if the send fails. + pub(crate) sent: Vec<automerge::ChangeHash>, + /// A sync pass was requested meanwhile. Once this one is handed over, the peer must + /// be enqueued again so the deferred changes go out. + pub(crate) regenerate: bool, +} + +/// First wait before generating again after our Mycelium refused a message; doubles +/// while it keeps refusing. +const SEND_RETRY: Duration = Duration::from_millis(500); +const SEND_RETRY_MAX: Duration = Duration::from_secs(30); + +/// One Automerge sync with a peer, from a reset to the next. Everything a reset must +/// forget lives here, so a reset replaces it whole and can't miss a field. +#[derive(Clone)] +pub(crate) struct SyncSession { + pub(crate) state: sync::State, + + /// A message being handed to our Mycelium. + pub(crate) sending: Option<Sending>, + + /// After our Mycelium refused a message: no new one before this. The refused one is + /// not kept; the next one is generated afresh and carries everything pending. + pub(crate) retry_at: Option<Instant>, + /// The wait after the next refusal. + pub(crate) send_backoff: Backoff, + + /// The answer we expect to our last sync message(s), if any. + pub(crate) reply: Option<SyncReplyWait>, + /// How long to wait for it; doubles while answers stay out. + pub(crate) reply_backoff: Backoff, +} + +/// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery: +/// a lost message leaves both sides waiting (`in_flight`), and changes it carried are +/// never sent again (`sent_hashes`). We resend after a timeout instead. +#[derive(Clone, Debug)] +pub(crate) struct SyncReplyWait { + pub(crate) deadline: Instant, + /// When the first of the unanswered messages was generated, and the sync bytes + /// exchanged with the peer by then (`swarm::throughput`). + pub(crate) started: (Instant, SyncBytes), + /// Changes first sent in the unanswered messages. + pub(crate) sent: Vec<automerge::ChangeHash>, +} + +impl SyncSession { + /// `reply_backoff`: waits for sync answers (`Timing::sync_reply_backoff`). + pub(crate) fn new(reply_backoff: Backoff) -> Self { + Self { + state: sync::State::new(), + sending: None, + retry_at: None, + send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX), + reply: None, + reply_backoff, + } + } + + /// Whether we leave the first message of this session to the peer. + /// + /// After a reset, both sides usually open at once, and the openings cross. Between + /// peers that share history that costs nothing. But a peer with an empty document + /// answers our opening with empty heads, which Automerge takes for "lost + /// everything", so the whole document would go out a second time. So a ready node + /// never opens towards a peer that isn't ready (one that is still catching up); that + /// peer opens itself. + pub(crate) fn leave_opening(&self, we_ready: bool, peer_ready: Option<bool>) -> bool { + self.state.their_heads.is_none() && we_ready && peer_ready == Some(false) + } + + /// Our Mycelium took the message. + pub(crate) fn send_succeeded(&mut self) -> bool { + self.retry_at = None; + self.send_backoff.reset(); + self.sending + .take() + .is_some_and(|sending| sending.regenerate) + } + + /// Our Mycelium refused the message: as far as the peer is concerned, it was never + /// sent. Takes back its changes (they go out with the next message), clears + /// `in_flight`, and returns how long to wait before generating again. None if a reset + /// already discarded what was being sent. + pub(crate) fn send_failed(&mut self, now: Instant) -> Option<Duration> { + let sending = self.sending.take()?; + for hash in &sending.sent { + self.state.sent_hashes.remove(hash); + } + if let Some(wait) = &mut self.reply { + wait.sent.retain(|hash| !sending.sent.contains(hash)); + } + self.state.in_flight = false; + let delay = self.send_backoff.next(); + self.retry_at = Some(now + delay); + Some(delay) + } + + /// A sync message went out; `sent` are the changes it carried for the first time. + /// `bytes` are the sync bytes exchanged with the peer before it. + pub(crate) fn expect_reply( + &mut self, + sent: Vec<automerge::ChangeHash>, + now: Instant, + bytes: SyncBytes, + ) { + match &mut self.reply { + // Still waiting for an earlier one: keep its deadline, remember these too. + Some(wait) => wait.sent.extend(sent), + None => { + self.reply = Some(SyncReplyWait { + deadline: now + self.reply_backoff.current(), + started: (now, bytes), + sent, + }); + } + } + } + + /// No answer is due any more: the peer answered, or keepalives show both documents + /// are the same. + pub(crate) fn reply_settled(&mut self) { + self.reply = None; + self.reply_backoff.reset(); + } + + /// If the answer is overdue: forgets that the unanswered changes were sent and that + /// a message is in flight, so the next sync pass sends them again. Returns how many + /// changes that concerns. + pub(super) fn take_overdue_reply(&mut self, now: Instant) -> Option<usize> { + if self.reply.as_ref().is_none_or(|wait| now < wait.deadline) { + return None; + } + let wait = self.reply.take()?; + for hash in &wait.sent { + self.state.sent_hashes.remove(hash); + } + self.state.in_flight = false; + self.reply_backoff.grow(); + Some(wait.sent.len()) + } +} diff --git a/src/state/timing.rs b/src/state/timing.rs new file mode 100644 index 0000000..406e207 --- /dev/null +++ b/src/state/timing.rs @@ -0,0 +1,75 @@ +//! The intervals the protocol runs on, derived from the keepalive. + +use std::time::Duration; + +use crate::backoff::Backoff; + +/// First wait for a sync answer; generous, since Mycelium delivers large messages in +/// many small packets. +const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10); +/// Upper bound, as a multiple of the first wait: 5 minutes (Mycelium gives up on a +/// message after five minutes). +const SYNC_REPLY_TIMEOUT_MAX_FACTOR: u32 = 30; +/// How long Mycelium tries to deliver a message before dropping it. +const MYCELIUM_DELIVERY_LIMIT: Duration = Duration::from_secs(300); + +#[derive(Clone, Copy, Debug)] +pub(crate) struct Timing { + pub(crate) keepalive: Duration, + pub(crate) dormant_after: Duration, + pub(crate) resync_after: Duration, + pub(crate) dormant_probe: Duration, + /// While catching up: PING the source after this much silence (`swarm::source`). + pub(crate) catch_up_probe: Duration, + /// First wait for an answer to a sync message before sending again; doubles while + /// answers stay out (`SyncSession::reply_backoff`). + pub(crate) sync_reply_timeout: Duration, + /// How long the transport tries to deliver a message before dropping it (Mycelium: + /// five minutes): a peer silent for longer has nothing on the way to us any more. + pub(crate) delivery_limit: Duration, +} + +impl Timing { + /// How long a round-trip sample stays meaningful (`swarm::rtt`): a few keepalive + /// intervals, since nodes move. + pub(crate) fn rtt_fresh_for(&self) -> Duration { + self.keepalive * 5 + } + + // Recommended values + // Environment | Keepalive | Notes + // --------------------------------------------+-----------+------------------------------------------------------------ + // Test harness | 1-2 s | Dormancy after about 4–8 s and healing within seconds keep tests short. + // LAN or small swarm (up to ~50 nodes) | 15-30 s | Under 3.5 messages/s per node at 50 nodes and 30 s. Current setting. + // WAN, mobile or metered links, or 100+ nodes | 60-120 s | At 100 nodes and 30 s it’s about 6–7 messages/s per node. Fine, but rarely worth it on battery or metered links. + // Several hundred nodes or more | - | The full mesh becomes the problem in itself, not the interval. Message counts grow O(N²) across the swarm, which calls for partial views (SWIM-style random probing and syncing with a subset of peers). + pub(crate) fn from_keepalive(keepalive: Duration) -> Self { + Self::new(keepalive, SYNC_REPLY_TIMEOUT) + } + + /// [`from_keepalive`](Self::from_keepalive) with another first sync reply timeout: + /// for tests, whose messages don't travel for seconds. + pub(crate) fn new(keepalive: Duration, sync_reply_timeout: Duration) -> Self { + Self { + keepalive, + dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row + // >= 2 keepalives, so a second exchange confirms the difference; and >= twice + // the first sync reply timeout, so a lost message's sender resends it first. + resync_after: (keepalive * 2).max(sync_reply_timeout * 2), + dormant_probe: keepalive * 10, // >= 10*: dormant peers cost almost nothing, but still rejoin within minutes + // Seconds, not a keepalive interval: a dead source should cost little time. + // Only the source gets these, and only while catching up. + catch_up_probe: (keepalive / 30).clamp(Duration::from_secs(1), Duration::from_secs(5)), + sync_reply_timeout, + delivery_limit: MYCELIUM_DELIVERY_LIMIT, + } + } + + /// Waits for sync answers: `sync_reply_timeout`, doubling up to 30 times as much. + pub(crate) fn sync_reply_backoff(&self) -> Backoff { + Backoff::new( + self.sync_reply_timeout, + self.sync_reply_timeout * SYNC_REPLY_TIMEOUT_MAX_FACTOR, + ) + } +} |