diff options
Diffstat (limited to 'src/remake/mod.rs')
| -rw-r--r-- | src/remake/mod.rs | 61 |
1 files changed, 56 insertions, 5 deletions
diff --git a/src/remake/mod.rs b/src/remake/mod.rs index ef036a6..ca17956 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -2,10 +2,12 @@ //! Compaction is one: the new document has the old one's content without its history. //! //! * `kind`: what differs between kinds of remake (building the new document, and -//! where writes made after the base go): `compact` and `replace`. +//! 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, settle period and decision on the node that was asked. +//! * `coordinator`: proposal, 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). //! * `catchup`: rejoining a newer epoch from peers (late, new or failed nodes), then //! replaying or quarantining this node's own missing changes. @@ -19,6 +21,7 @@ mod api; mod catchup; +mod content; mod coordinator; mod delta; mod kind; @@ -27,6 +30,7 @@ mod quarantine; mod records; pub(crate) mod snapshot; mod switch; +mod vote; pub(crate) use local::RemakeState; pub(crate) use switch::run_worker; @@ -37,8 +41,9 @@ use std::time::Duration; use automerge::AutoCommit; use axum::{ - routing::{get, post}, Router, + extract::DefaultBodyLimit, + routing::{get, post}, }; use tokio::sync::broadcast; @@ -58,9 +63,28 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// The epoch of the local replica. fn current_epoch(&self) -> Epoch; - /// Lower bound and default basis for the settle period. + /// Default basis for the voting period, and the grace margin after it. fn dormant_after(&self) -> Duration; + /// 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_ms: u64) -> 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; + + /// Downloads content by its BLAKE3 hash from peers on the local epoch, checked + /// against it. + fn fetch_content(&self, hash: [u8; 32]) + -> impl Future<Output = anyhow::Result<Vec<u8>>> + Send; + + /// Starts sync with every peer (e.g. after holding it back). + fn request_sync(&self); + fn is_node_ready(&self) -> impl Future<Output = bool> + Send; /// Known (not forgotten) peers. @@ -117,18 +141,45 @@ pub(crate) struct PeerSummary { pub(crate) id: String, /// Not dormant. pub(crate) active: bool, + /// Completed the handshake and belongs to our swarm. A peer that merely sent us + /// something (a node of another swarm on the same network, a stranger before + /// admission) is not one, and nothing it advertises is followed. + pub(crate) member: bool, + /// What it last said about being ready; None before it said anything. + pub(crate) ready: Option<bool>, /// The epoch the peer last advertised. pub(crate) epoch: Option<Epoch>, } +impl PeerSummary { + /// A peer whose epoch can be followed and downloaded from: an active member that + /// holds the document it advertises. A rejoining peer advertises its target epoch + /// with an empty document and `ready: false`. + pub(crate) fn serves_epoch(&self, epoch: Epoch) -> bool { + self.serves() && self.epoch == Some(epoch) + } + + /// An active, ready member. + pub(crate) fn serves(&self) -> bool { + self.active && self.member && self.ready == Some(true) + } +} + /// 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 Mycelium must deliver within its deadline, so +/// documents near this size are already a stretch. +const REMAKE_REQUEST_LIMIT: usize = 64 * 1024 * 1024; + pub(crate) fn router<H: Host>() -> Router<H> { Router::new() .route( "/admin/remakes", get(api::list::<H>) .post(api::start::<H>) - .delete(api::abort_current::<H>), + .delete(api::abort_current::<H>) + // A replace or import request carries a whole document. + .layer(DefaultBodyLimit::max(REMAKE_REQUEST_LIMIT)), ) // The original endpoint, kept as a shortcut for a compacting remake. .route("/admin/compaction", post(api::start_compaction::<H>)) |
