diff options
Diffstat (limited to 'src/remake/mod.rs')
| -rw-r--r-- | src/remake/mod.rs | 115 |
1 files changed, 90 insertions, 25 deletions
diff --git a/src/remake/mod.rs b/src/remake/mod.rs index ca7f5a0..9351cc5 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -5,7 +5,9 @@ //! where writes made after the base go): `compact`, `replace` and `import`. //! * `snapshot`: deterministic build of the next epoch's initial document. //! * `records`: the replicated remake records under `_continuum/remakes/<id>`. -//! * `coordinator`: proposal, vote count and decision on the node that was asked. +//! * `proposal`: writing a proposal (its checks, superseding expired remakes), and when +//! a node other than the coordinator may abort an expired one. +//! * `coordinator`: vote count and decision on the node that was asked. //! * `vote`: who votes, for which remake, and when a remake has its majority. //! * `content`: the new document of a `replace`/`import` remake, fetched by every voter. //! * `switch`: switching this node to a committed epoch (worker). @@ -20,12 +22,15 @@ //! depends on `AppState`, peers or sync internals. mod api; +#[cfg(test)] +mod bench; mod catchup; mod content; mod coordinator; mod delta; mod kind; mod local; +mod proposal; mod quarantine; mod records; pub(crate) mod snapshot; @@ -34,6 +39,7 @@ mod vote; #[cfg(test)] pub(crate) use api::testing; +pub(crate) use delta::Superseded; pub(crate) use kind::remake_genesis; pub(crate) use local::RemakeState; pub(crate) use switch::run_worker; @@ -42,7 +48,6 @@ use std::future::Future; use std::sync::Arc; use std::time::Duration; -use automerge::AutoCommit; use axum::{ Router, extract::DefaultBodyLimit, @@ -51,7 +56,8 @@ use axum::{ use tokio::sync::broadcast; use crate::document::replica::Epoch; -use crate::swarm::sync::ChangeSet; +use crate::document::store::{ChangeSet, StoreError}; +use crate::document::{Document, Transaction}; /// Remakes and their aftermath on this node, in numbers (`/admin/stats`). #[derive(Debug, Clone, Copy, Default)] @@ -71,7 +77,7 @@ pub(crate) struct Summary { } /// The `Summary` of `remake` with the records of `doc`, whose epoch is `epoch`. -pub(crate) fn summary(remake: &RemakeState, doc: &AutoCommit, epoch: Epoch) -> Summary { +pub(crate) fn summary(remake: &RemakeState, doc: &Document, epoch: Epoch) -> Summary { let local = records::EpochRecord::from(epoch); let mut summary = Summary { rejoining: remake.is_rejoining(), @@ -120,16 +126,24 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// Default basis for the voting period, and the grace margin after it. fn dormant_after(&self) -> Duration; + /// This node's clock against its peers' (`swarm::skew`): a coordinator judges the + /// remake's deadlines by it. + fn clock_skew(&self) -> impl Future<Output = crate::swarm::skew::Skew> + Send; + /// The electorate of a remake proposed now (`vote`): the members that said they are /// alive recently enough, read from `doc`. - fn electorate(&self, doc: &AutoCommit, now: crate::clock::UnixMillis) -> Vec<String>; + fn electorate( + &self, + doc: &impl crate::document::ReadDoc, + now: crate::clock::UnixMillis, + ) -> Vec<String>; /// Downloads a change from peers on the local epoch, in chunks from several at once, /// checked against its hash. fn fetch_change( &self, - hash: automerge::ChangeHash, - ) -> impl Future<Output = anyhow::Result<automerge::Change>> + Send; + hash: crate::document::ChangeHash, + ) -> impl Future<Output = anyhow::Result<crate::document::Change>> + Send; /// Downloads content of `size` bytes by its BLAKE3 hash from peers on the local /// epoch, checked against it. @@ -147,16 +161,44 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// Known (not forgotten) peers. fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send; - /// Stops serving requests (back to `Joining`, keeping the swarm ID), as while a - /// rejoin syncs a fresh document. Returns whether the node was serving. - fn suspend(&self) -> impl Future<Output = bool> + Send; + /// Starts a rejoin of `target`, under the document's write lock: the live document + /// becomes an empty one of `target`, written by `actor`, that takes no writes until + /// the rejoin ends. A ready node stops being ready (back to `Joining`, keeping the + /// swarm ID), and its document so far answers reads (marked stale) and takes its + /// writes meanwhile; a node that wasn't ready gets its document back. Afterwards + /// the node resets every peer's sync state and publishes the change. + fn begin_rejoin( + &self, + target: Epoch, + actor: crate::document::ActorId, + ) -> impl Future<Output = BegunRejoin> + Send; + + /// A copy of the document before the rejoin, with the writes it took since, if the + /// node keeps it ([`Host::begin_rejoin`]: it was ready). + fn rejoin_document(&self) -> impl Future<Output = Option<Document>> + Send; - /// Serves requests again after [`Host::suspend`]. - fn resume(&self) -> impl Future<Output = ()> + Send; + /// Ends a rejoin on its target, under the write lock: `f` writes into the live + /// document from the one before the rejoin (the node's, or `old` where it keeps + /// none), as many changes as it likes; then writes go to the live document again, + /// and a node that was ready is again. Returns what `f` returned and the document + /// before the rejoin. + fn finish_rejoin<T, F>( + &self, + old: Option<Document>, + f: F, + ) -> impl Future<Output = (T, Document)> + Send + where + F: FnOnce(&mut Document, &Document) -> T + Send + 'static, + T: Send + 'static; - /// Serves reads from this document (read-only, marked stale) while the node is not - /// ready; `None` ends it. - fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>); + /// Ends a rejoin by going back, under the write lock: the document before it (the + /// node's, with the writes it took meanwhile, or `old`) and `previous` are live + /// again, and a node that was ready is again. Afterwards as [`Host::begin_rejoin`]. + fn abandon_rejoin( + &self, + previous: Epoch, + old: Option<Document>, + ) -> impl Future<Output = ()> + Send; /// Whether the document caught up with a peer since the last reset. fn initial_sync_done(&self) -> bool; @@ -170,15 +212,27 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// Reads the document on the blocking pool. fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send where - F: FnOnce(&AutoCommit) -> T + Send + 'static, + F: FnOnce(&Document) -> T + Send + 'static, T: Send + 'static; - /// One committed change (rolled back on error), published and replicated. + /// One committed change (rolled back on error), published and replicated. While a + /// rejoin runs, into the document before it, or refused (`StoreError::Rejoining`) + /// by a node that keeps none (`AppState::mutate_doc`). 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, + F: FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static, T: Send + 'static, - E: Send + 'static; + E: From<StoreError> + Send + 'static; + + /// What a replay does with writes a newer one superseded (`--superseded-writes`). + fn superseded(&self) -> Superseded; + + /// Applies a change downloaded from peers (`fetch_change`), then publishes it like + /// a change received by sync. + fn apply_change( + &self, + change: crate::document::Change, + ) -> impl Future<Output = Result<(), String>> + Send; /// Replaces the document and its epoch atomically. Under the write lock, `f` gets /// the live document and returns the new one. Afterwards the node resets every @@ -187,12 +241,23 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { &self, epoch: Epoch, f: F, - ) -> impl Future<Output = Result<AutoCommit, E>> + Send + ) -> impl Future<Output = Result<Document, E>> + Send where - F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static, + F: FnOnce(&Document) -> Result<Document, E> + Send + 'static, E: Send + 'static; } +/// What [`Host::begin_rejoin`] left the rejoin with. +pub(crate) struct BegunRejoin { + /// The node was ready, and is again once the rejoin ends. + pub(crate) was_ready: bool, + /// The heads of the document before the rejoin as it began: what changes after + /// them, it took during the rejoin. + pub(crate) heads: Vec<crate::document::ChangeHash>, + /// The document before the rejoin, where the node doesn't keep it (it wasn't ready). + pub(crate) old: Option<Document>, +} + #[derive(Clone, Debug)] pub(crate) struct PeerSummary { pub(crate) id: String, @@ -222,12 +287,12 @@ impl PeerSummary { } } -/// Admin routes. Merged into the authenticated part of the node's router. -/// Largest remake request (bytes). The proposal carries the new document to every -/// node in one sync message, which the transport must deliver within its delivery -/// limit, so documents near this size are already a stretch. +/// Largest remake request (bytes). The coordinator holds the new document in memory +/// and every member downloads it in chunks before voting (`content`), so documents near +/// this size cost every node that much memory, and the swarm that many bytes per node. const REMAKE_REQUEST_LIMIT: usize = 64 * 1024 * 1024; +/// Admin routes. Merged into the authenticated part of the node's router. pub(crate) fn router<H: Host>() -> Router<H> { Router::new() .route( |