diff options
Diffstat (limited to 'src/state.rs')
| -rw-r--r-- | src/state.rs | 313 |
1 files changed, 243 insertions, 70 deletions
diff --git a/src/state.rs b/src/state.rs index a8b573b..63e6490 100644 --- a/src/state.rs +++ b/src/state.rs @@ -7,26 +7,22 @@ use std::sync::Arc; use std::time::Duration; use anyhow::{bail, Result}; -use automerge::{sync, AutoCommit}; +use automerge::{sync, AutoCommit, ChangeHash}; use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock}; use tokio::time::Instant; use tracing::{debug, error, warn}; -use crate::compaction::{Compaction, Host, PeerSummary}; -use crate::firewall::Firewall; -use crate::replica::{Epoch, Replica}; -use crate::sync::{ChangeSet, SyncEvent}; +use crate::document::replica::{Epoch, Replica}; +use crate::remake::{Host, PeerSummary, RemakeState}; +use crate::swarm::sync::{ChangeSet, SyncEvent}; use crate::transport; +use crate::transport::firewall::Firewall; #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum NodeLifecycle { - // TODO maybe rename discovered_cluster_id to cluster_id if that doesn't impair readability. - Joining { - discovered_cluster_id: Option<String>, - }, - Ready { - cluster_id: String, - }, + // 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(Debug, Clone, Default, PartialEq, Eq)] @@ -34,7 +30,7 @@ pub(crate) enum PeerMembership { #[default] Unknown, Member { - cluster_id: String, + swarm_id: String, }, } @@ -75,7 +71,7 @@ pub(crate) struct PeerState { pub(crate) sync_state: sync::State, pub(crate) hello_state: HelloState, - /// The peer sent HELLO; we still owe it a HELLO_REPLY (e.g. we had no cluster ID yet, + /// 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, @@ -105,8 +101,33 @@ pub(crate) struct PeerState { /// Consecutive sync exchanges that changed nothing (livelock detection). pub(crate) stalled_exchanges: u32, + + /// The peer proved the swarm key in its HELLO or HELLO_REPLY (always false and + /// irrelevant without a swarm key). + pub(crate) authenticated: bool, + + /// The answer we expect to our last sync message(s), if any. + pub(crate) sync_reply: Option<SyncReplyWait>, + /// How long to wait for it; doubles while answers stay out. + pub(crate) sync_reply_timeout: Duration, } +/// 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, + /// 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 (Mycelium gives up on a message after five minutes). +const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300); + impl Default for PeerState { fn default() -> Self { Self { @@ -124,6 +145,9 @@ impl Default for PeerState { last_sync_rx: None, last_resync_at: None, stalled_exchanges: 0, + authenticated: false, + sync_reply: None, + sync_reply_timeout: SYNC_REPLY_TIMEOUT, epoch: None, } } @@ -136,6 +160,56 @@ impl PeerState { 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; + } + + /// 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 { + // 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, + 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; + } + + /// 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) + { + return None; + } + let wait = self.sync_reply.take()?; + for hash in &wait.sent { + self.sync_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); + Some(wait.sent.len()) } fn quiet_for(&self, now: Instant) -> Duration { @@ -235,7 +309,7 @@ impl PeerState { } /// Heard from recently. A returning node counts as alive from its first message, - /// before the handshake finishes, so a cluster-wide forget can't make it flap. + /// 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 @@ -257,9 +331,9 @@ impl Timing { // Environment | Keepalive | Notes // --------------------------------------------+-----------+------------------------------------------------------------ // Test harness | 1-2 s | Dormancy after about 4–8 s and healing within seconds keep tests short. - // LAN or small cluster (up to ~50 nodes) | 15-30 s | Under 3.5 messages/s per node at 50 nodes and 30 s. Current setting. + // 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 cluster, which calls for partial views (SWIM-style random probing and syncing with a subset of peers). + // 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 { keepalive, @@ -281,8 +355,8 @@ pub(crate) struct AppState { pub(crate) lifecycle: Arc<RwLock<NodeLifecycle>>, - // Serializes cluster-ID establishment, peer membership, and readiness transitions. - //[concurrent]cluster_validation: Arc<Mutex<()>>, + // 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>, @@ -298,8 +372,16 @@ pub(crate) struct AppState { pub(crate) timing: Timing, - /// Per-node compaction state (prebuilt snapshots, switch status, archive). - pub(crate) compaction: Arc<Compaction>, + /// 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. + pub(crate) swarm_key: Option<Arc<crate::swarm::auth::SwarmKey>>, + + /// 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>>>, } impl AppState { @@ -317,94 +399,86 @@ impl AppState { matches!(*self.lifecycle.read().await, NodeLifecycle::Ready { .. }) } - pub(crate) async fn cluster_id(&self) -> Option<String> { + pub(crate) async fn swarm_id(&self) -> Option<String> { match &*self.lifecycle.read().await { - NodeLifecycle::Ready { cluster_id } => Some(cluster_id.clone()), + NodeLifecycle::Ready { swarm_id } => Some(swarm_id.clone()), NodeLifecycle::Joining { - discovered_cluster_id, - } => discovered_cluster_id.clone(), + discovered_swarm_id, + } => discovered_swarm_id.clone(), } } - // Record the expected cluster without making the node Ready. Bootstrap owns readiness. - pub(crate) async fn set_expected_cluster_id(&self, cluster_id: String) -> Result<()> { - //[concurrent]let _guard = self.cluster_validation.lock().await; + // 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_cluster_id: Some(existing), - } if existing != &cluster_id => { + discovered_swarm_id: Some(existing), + } if existing != &swarm_id => { bail!( - "Conflicting expected cluster IDs: expected {}, received {}", + "Conflicting expected swarm IDs: expected {}, received {}", existing, - cluster_id + swarm_id ); } NodeLifecycle::Joining { - discovered_cluster_id: None, + discovered_swarm_id: None, } => { let old_lifecycle = lifecycle.clone(); *lifecycle = NodeLifecycle::Joining { - discovered_cluster_id: Some(cluster_id), + discovered_swarm_id: Some(swarm_id), }; debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}"); Ok(()) } NodeLifecycle::Joining { - discovered_cluster_id: Some(_), + discovered_swarm_id: Some(_), } => { warn!("Node already Joining with the same identity"); Ok(()) } - NodeLifecycle::Ready { - cluster_id: existing, - } if existing == &cluster_id => { + NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => { warn!("Node already Ready with the same identity"); Ok(()) } - NodeLifecycle::Ready { - cluster_id: existing, - } => { + NodeLifecycle::Ready { swarm_id: existing } => { bail!( - "Cannot change established cluster ID: existing {}, received {}", + "Cannot change established swarm ID: existing {}, received {}", existing, - cluster_id + swarm_id ); } } } // Only bootstrap should call this, after validating document metadata. - pub(crate) async fn mark_ready(&self, cluster_id: String) -> Result<()> { - //[concurrent]let _guard = self.cluster_validation.lock().await; + 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_cluster_id: Some(expected), - } if expected != &cluster_id => { + discovered_swarm_id: Some(expected), + } if expected != &swarm_id => { bail!( - "Cannot mark node ready: expected cluster {}, received {}", + "Cannot mark node ready: expected swarm {}, received {}", expected, - cluster_id + swarm_id ); } NodeLifecycle::Joining { .. } => { let old_lifecycle = lifecycle.clone(); - *lifecycle = NodeLifecycle::Ready { cluster_id }; + *lifecycle = NodeLifecycle::Ready { swarm_id }; debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}"); } - NodeLifecycle::Ready { - cluster_id: existing, - } if existing == &cluster_id => { + NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => { debug!("Node already ready with the same identity"); } - NodeLifecycle::Ready { - cluster_id: existing, - } => { + NodeLifecycle::Ready { swarm_id: existing } => { bail!( - "Cannot change established cluster ID: existing {}, received {}", + "Cannot change established swarm ID: existing {}, received {}", existing, - cluster_id + swarm_id ); } } @@ -535,7 +609,7 @@ impl AppState { // 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::peer::local_keepalive(self).await; + let keepalive = crate::swarm::peer::local_keepalive(self).await; for (id, _) in &peers { if let Err(error) = self .transport @@ -552,6 +626,98 @@ impl AppState { } } +/// 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). That needs the + /// write lock, so it is opt-in. 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); + } + // 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(); + 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; + } +} + pub(crate) fn unix_millis() -> u64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -559,10 +725,10 @@ pub(crate) fn unix_millis() -> u64 { .unwrap_or(0) } -/// The node as seen by the compaction module. +/// The node as seen by the remake module. impl Host for AppState { - fn compaction(&self) -> &Compaction { - &self.compaction + fn remake(&self) -> &RemakeState { + &self.remake } fn node_id(&self) -> String { @@ -611,13 +777,13 @@ impl Host for AppState { let state = self.clone(); async move { let mut lifecycle = state.lifecycle.write().await; - let NodeLifecycle::Ready { cluster_id } = &*lifecycle else { + let NodeLifecycle::Ready { swarm_id } = &*lifecycle else { return false; }; - let cluster_id = cluster_id.clone(); - debug!(cluster_id = %cluster_id, "Node suspended (rejoining)"); + let swarm_id = swarm_id.clone(); + debug!(swarm_id = %swarm_id, "Node suspended (rejoining)"); *lifecycle = NodeLifecycle::Joining { - discovered_cluster_id: Some(cluster_id), + discovered_swarm_id: Some(swarm_id), }; true } @@ -628,16 +794,23 @@ impl Host for AppState { async move { let mut lifecycle = state.lifecycle.write().await; if let NodeLifecycle::Joining { - discovered_cluster_id: Some(cluster_id), + discovered_swarm_id: Some(swarm_id), } = &*lifecycle { - let cluster_id = cluster_id.clone(); - debug!(cluster_id = %cluster_id, "Node resumed"); - *lifecycle = NodeLifecycle::Ready { cluster_id }; + 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) |