diff options
Diffstat (limited to 'src/state.rs')
| -rw-r--r-- | src/state.rs | 655 |
1 files changed, 480 insertions, 175 deletions
diff --git a/src/state.rs b/src/state.rs index 1f3cf16..785eee8 100644 --- a/src/state.rs +++ b/src/state.rs @@ -12,9 +12,11 @@ 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; @@ -25,13 +27,32 @@ pub(crate) enum NodeLifecycle { 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 PeerMembership { +pub(crate) enum Standing { + /// Nothing proven yet. With a swarm key, everything but a handshake from it is + /// ignored (`swarm::peer::admit`). #[default] - Unknown, - Member { - swarm_id: String, - }, + 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)] @@ -47,72 +68,135 @@ pub(crate) enum Liveness { pub(crate) enum HelloState { #[default] NotSent, - /// A send is in flight. `next_delay` is how long to wait after it before retrying. - Sending { next_delay: Duration }, - /// HELLO was sent (or the send failed); no HELLO_REPLY yet. - AwaitingReply { - next_attempt: Instant, - next_delay: Duration, - }, + /// 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 PendingOutbound { - pub(crate) message: transport::EncodedMessage, - pub(crate) retry_delay: Duration, - pub(crate) next_attempt: Instant, +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, } -#[derive(Clone)] -pub(crate) struct PeerState { - pub(crate) membership: PeerMembership, - pub(crate) sync_state: sync::State, - pub(crate) hello_state: HelloState, +/// 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); - /// The peer sent HELLO; we still owe it a HELLO_REPLY (e.g. we had no swarm ID yet, - /// or sending failed). - pub(crate) reply_owed: 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); - /// Received from this peer, waiting to be processed. - pub(crate) pending_inbound_sync: Vec<transport::SyncEnvelope>, +#[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, - /// Generated for this peer, waiting for delivery. - pub(crate) pending_outbound_sync: Option<PendingOutbound>, + pub(crate) standing: Standing, + pub(crate) hello_state: HelloState, + /// Wait between unanswered HELLOs; reset whenever a new handshake starts. + pub(crate) hello_backoff: Backoff, - /// A sync pass was requested while an envelope was pending. Once that envelope - /// is delivered, the peer must be re-enqueued so the deferred changes go out. - pub(crate) needs_generate: bool, + /// 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. + /// 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. pub(crate) epoch: Option<Epoch>, - /// Consecutive sync exchanges that changed nothing (livelock detection). - pub(crate) stalled_exchanges: u32, + /// 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>, - /// The peer proved the swarm key in its HELLO or HELLO_REPLY (always false and - /// irrelevant without a swarm key). - pub(crate) authenticated: 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) sync_reply: Option<SyncReplyWait>, + pub(crate) reply: Option<SyncReplyWait>, /// How long to wait for it; doubles while answers stay out. - pub(crate) sync_reply_timeout: Duration, - - /// When sync with this peer last started from scratch. - pub(crate) sync_reset_at: Instant, + pub(crate) reply_backoff: Backoff, } /// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery: @@ -121,6 +205,9 @@ pub(crate) struct PeerState { #[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>, } @@ -131,110 +218,145 @@ const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10); /// Upper bound (Mycelium gives up on a message after five minutes). const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300); -/// How long the side that doesn't open a fresh sync waits for the other's opening -/// before opening itself (the other one may not have reset, or its message got lost). -const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3); - -impl Default for PeerState { - fn default() -> Self { +impl SyncSession { + pub(crate) fn new() -> Self { Self { - membership: PeerMembership::default(), - sync_state: sync::State::new(), - hello_state: HelloState::default(), - reply_owed: false, - pending_inbound_sync: Vec::new(), - pending_outbound_sync: None, - needs_generate: false, - liveness: Liveness::Active, - added_at: Instant::now(), - last_seen: None, - last_ping_at: None, - last_sync_rx: None, - last_resync_at: None, - stalled_exchanges: 0, - authenticated: false, - sync_reply: None, - sync_reply_timeout: SYNC_REPLY_TIMEOUT, - sync_reset_at: Instant::now(), - epoch: None, + state: sync::State::new(), + sending: None, + retry_at: None, + send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX), + reply: None, + reply_backoff: Backoff::new(SYNC_REPLY_TIMEOUT, SYNC_REPLY_TIMEOUT_MAX), } } -} -impl PeerState { - /// Start Automerge sync with this peer from scratch. Any envelope generated against - /// the old state is discarded; a stale RetryDelivery event is then a no-op. - pub(crate) fn reset_sync(&mut self) { - self.sync_state = sync::State::new(); - self.pending_outbound_sync = None; - self.needs_generate = false; - self.sync_reply = None; - self.sync_reply_timeout = SYNC_REPLY_TIMEOUT; - self.sync_reset_at = Instant::now(); + /// 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) } - /// After a reset, usually both sides do it and both open the sync at once. The - /// openings cross, and each side answers the other's: harmless between peers that - /// share history, but an empty peer answers with empty heads, which Automerge takes - /// for "lost everything", so the whole document goes out a second time. Only the - /// side with the smaller key opens; the other waits for its opening, up to - /// `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all. - pub(crate) fn opening_wait(&self, we_open: bool, now: Instant) -> Option<Duration> { - if we_open || self.sync_state.their_heads.is_some() { - return None; + /// 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); } - SYNC_OPENING_GRACE - .checked_sub(now.saturating_duration_since(self.sync_reset_at)) - .filter(|wait| !wait.is_zero()) + 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. - pub(crate) fn expect_sync_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) { - match &mut self.sync_reply { + /// `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.sync_reply = Some(SyncReplyWait { - deadline: now + self.sync_reply_timeout, + self.reply = Some(SyncReplyWait { + deadline: now + self.reply_backoff.current(), + started: (now, bytes), sent, }); } } } - /// Keepalives show both documents are the same: stop waiting for an answer. - pub(crate) fn sync_in_step(&mut self) { - self.sync_reply = None; - self.sync_reply_timeout = SYNC_REPLY_TIMEOUT; - } - - /// The peer answered. - pub(crate) fn sync_reply_received(&mut self) { - self.sync_reply = None; - self.sync_reply_timeout = SYNC_REPLY_TIMEOUT; + /// 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(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> { - if self.liveness != Liveness::Active - || self.hello_state != HelloState::Established - || self - .sync_reply - .as_ref() - .is_none_or(|wait| now < wait.deadline) - { + 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.sync_reply.take()?; + let wait = self.reply.take()?; for hash in &wait.sent { - self.sync_state.sent_hashes.remove(hash); + self.state.sent_hashes.remove(hash); } - self.sync_state.in_flight = false; - self.sync_reply_timeout = (self.sync_reply_timeout * 2).min(SYNC_REPLY_TIMEOUT_MAX); + self.state.in_flight = false; + self.reply_backoff.grow(); Some(wait.sent.len()) } +} + +impl PeerState { + pub(crate) fn new(leads: bool) -> Self { + Self { + leads, + standing: Standing::default(), + hello_state: HelloState::default(), + hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX), + sync: SyncSession::new(), + 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, + 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) { + self.sync = SyncSession::new(); + } + + /// 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)) @@ -243,11 +365,14 @@ impl PeerState { /// 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 matches!(self.liveness, Liveness::Dormant { .. }) { + 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 !matches!(self.hello_state, HelloState::Sending { .. }) { + if self.hello_state != HelloState::Sending { self.hello_state = HelloState::NotSent; } return true; @@ -255,6 +380,29 @@ impl PeerState { 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 { @@ -262,12 +410,13 @@ impl PeerState { } self.liveness = Liveness::Dormant { since: now }; + self.incidents.dormancies += 1; + self.unreachable_since.get_or_insert(now); self.reset_sync(); - if !matches!(self.hello_state, HelloState::Sending { .. }) { + if self.hello_state != HelloState::Sending { self.hello_state = HelloState::AwaitingReply { next_attempt: now + timing.dormant_probe, - next_delay: timing.dormant_probe, }; } true @@ -276,19 +425,14 @@ impl PeerState { /// 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 with the smaller key (`we_ping`). Each of its PINGs + /// 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, - we_ping: bool, - now: Instant, - ) -> bool { - let quiet_enough = if we_ping { + 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 @@ -307,43 +451,48 @@ impl PeerState { due } - /// Schedules a fresh handshake, which resets sync on both sides: the peer resets on - /// our HELLO, we reset on its reply. The peer worker sends the HELLO on its next tick. - /// Usable without awaiting (e.g. under a blocking lock). - pub(crate) fn request_resync(&mut self) { - if matches!(self.hello_state, HelloState::Sending { .. }) { - return; // a HELLO is already on its way - } - self.hello_state = HelloState::AwaitingReply { - next_attempt: Instant::now(), - next_delay: Duration::from_millis(500), - }; - } - /// Self-healing check on PING/PONG. Returns true if a re-handshake should start. pub(crate) fn needs_heal( &mut self, - local: &transport::HeadsDigest, - remote: &transport::HeadsDigest, + local: &transport::HeadsSummary, + remote: &transport::HeadsSummary, timing: &Timing, now: Instant, ) -> bool { - if local == remote - || self.liveness != Liveness::Active + 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 sync_idle = now.saturating_duration_since(self.last_sync_rx.unwrap_or(self.added_at)) - >= timing.resync_after; + 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 !(sync_idle && cooled_down) { + 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 @@ -366,9 +515,17 @@ pub(crate) struct Timing { 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, } 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 // --------------------------------------------+-----------+------------------------------------------------------------ @@ -380,8 +537,13 @@ impl Timing { Self { keepalive, dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row - resync_after: keepalive * 2, // resync_after should be >= 2* keepalive to guarantee at least one digest exchange while sync is idle + // >= 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)), } } } @@ -402,6 +564,10 @@ pub(crate) struct AppState { 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>, @@ -417,13 +583,62 @@ pub(crate) struct AppState { /// 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 anyone who - /// knows the swarm ID. + /// 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>>, + /// 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>>>, + /// 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>>>, +} + +/// 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, + /// 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 { @@ -431,6 +646,24 @@ impl AppState { 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 { @@ -539,10 +772,16 @@ impl AppState { Arc::clone( peers .entry(pk.to_owned()) - .or_insert_with(|| Arc::new(Mutex::new(PeerState::default()))), + .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))) + } + pub(crate) fn notify_store_changed(&self) { self.change_tx.send_modify(|version| { *version = version.wrapping_add(1); @@ -579,9 +818,14 @@ impl AppState { T: Send + 'static, { let doc = Arc::clone(&self.doc); - tokio::task::spawn_blocking(move || f(&mut doc.blocking_write().doc)) - .await - .expect("document write task panicked") + 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 @@ -604,13 +848,19 @@ impl AppState { match &result { // One API write = one Automerge change. Ok(_) => { - replica.doc.commit(); + // 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(unix_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 @@ -649,15 +899,31 @@ impl AppState { self.publish_change(); // Tell peers about the new epoch now instead of with the next keepalive (up to a - // keepalive interval later): sync with peers already on it resumes right away, - // and peers that never saw the commit start catching up. - let keepalive = crate::swarm::peer::local_keepalive(self).await; - for (id, _) in &peers { - if let Err(error) = self - .transport - .send_message(id, &transport::ParsedMessage::Ping(keepalive.clone())) - .await - { + // 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"); } } @@ -687,8 +953,8 @@ impl AppState { /// 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). That needs the - /// write lock, so it is opt-in. Stale reads have no heads. + /// 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, @@ -738,14 +1004,14 @@ impl AppState { if !with_heads { return (self.render(read, finish).await, None); } - // AutoCommit::get_heads takes &mut self (it closes a pending transaction). - let (data, mut heads) = self - .write_doc(move |doc| { - let heads = doc.get_heads(); - (read(doc), heads) - }) - .await; - heads.sort(); + // 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"); @@ -785,6 +1051,33 @@ impl Host for AppState { 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 } @@ -807,6 +1100,9 @@ impl Host for AppState { 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, }); } @@ -860,6 +1156,14 @@ impl Host for AppState { 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>> { @@ -903,6 +1207,7 @@ impl Host for AppState { let new_doc = f(&mut replica.doc)?; let old_doc = std::mem::replace(&mut replica.doc, new_doc); replica.epoch = epoch; + 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); |