summaryrefslogtreecommitdiffstats
path: root/src/remake/mod.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake/mod.rs')
-rw-r--r--src/remake/mod.rs61
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>))