diff options
Diffstat (limited to 'src/state.rs')
| -rw-r--r-- | src/state.rs | 1297 |
1 files changed, 0 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 - } - } -} |