diff options
34 files changed, 2386 insertions, 337 deletions
@@ -9,7 +9,6 @@ tokio-console = ["dep:console-subscriber"] [dependencies] anyhow = "1.0" argon2 = { version = "0.6", features = ["getrandom"] } -async-nats = "0.50" async-stream = "0.3" async-trait = "0.1" automerge = "0.12" @@ -18,9 +17,8 @@ console-subscriber = { version = "0.5.0", optional = true } base64 = "0.23" blake3 = "1.8" bytes = "1.12" -clap = { version = "4.6", features = ["derive"] } +clap = "4.6" dashmap = "6.2" -futures-util = "0.3" getrandom = "0.4" libc = "0.2" password-hash = "0.6" @@ -30,12 +28,9 @@ serde = { version = "1.0", features = ["derive"] } serde_bytes = "0.11" serde_json = "1.0" sha2 = "0.11" -sysinfo = "0.39" thiserror = "2.0" toml = "1.1" tokio = { version = "1.53", features = ["full"] } -tokio-tungstenite = { version = "0.30", features = ["native-tls"] } -tower = "0.5" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } url = "2.5" diff --git a/src/api/auth.rs b/src/api/auth.rs index 70b1af5..de4220c 100644 --- a/src/api/auth.rs +++ b/src/api/auth.rs @@ -72,14 +72,18 @@ fn parse_basic(headers: &HeaderMap) -> Option<(String, String)> { Some((user.to_owned(), pass.to_owned())) } -async fn authenticate(state: &AuthState, headers: &HeaderMap) -> bool { - let Some((user, pass)) = parse_basic(headers) else { - return false; - }; - let Some(hash) = state.users.get(&user).cloned() else { - return false; - }; +/// The user a request authenticated as (a request extension, set by [`basic_auth`]). +#[derive(Clone, Debug)] +pub(crate) struct AuthenticatedUser(pub(crate) String); +/// The user, if the credentials are valid. +async fn authenticate(state: &AuthState, headers: &HeaderMap) -> Option<String> { + let (user, pass) = parse_basic(headers)?; + let hash = state.users.get(&user).cloned()?; + verify(state, &user, pass, hash).await.then_some(user) +} + +async fn verify(state: &AuthState, user: &str, pass: String, hash: String) -> bool { let key = *blake3::hash(format!("{user}\0{pass}\0{hash}").as_bytes()).as_bytes(); // Fast path. The shard guard lives only until the end of this `match`; @@ -124,10 +128,10 @@ async fn authenticate(state: &AuthState, headers: &HeaderMap) -> bool { pub async fn basic_auth( State(state): State<AuthState>, headers: HeaderMap, - request: Request<Body>, + mut request: Request<Body>, next: axum::middleware::Next, ) -> Response { - if !authenticate(&state, &headers).await { + let Some(user) = authenticate(&state, &headers).await else { sleep(Duration::from_millis(1500)).await; return ( StatusCode::UNAUTHORIZED, @@ -135,7 +139,8 @@ pub async fn basic_auth( "Unauthorized", ) .into_response(); - } + }; + request.extensions_mut().insert(AuthenticatedUser(user)); next.run(request).await } diff --git a/src/api/mod.rs b/src/api/mod.rs index ce1f6ae..1fba48b 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -24,7 +24,7 @@ use tracing::{error, warn}; use utoipa::OpenApi; use utoipa_swagger_ui::SwaggerUi; -use crate::api::auth::{AuthState, basic_auth}; +use crate::api::auth::{AuthState, AuthenticatedUser, basic_auth}; use crate::document::store::StoreError; use crate::state::AppState; @@ -104,6 +104,31 @@ fn parse_key_path(key: &str) -> Result<Vec<String>, StatusCode> { Ok(segments) } +/// A path for a write. Protocol data (`_continuum`: remake records, members, forget +/// records, epoch metadata) is what every node acts on, so a write there acts on the +/// whole swarm: only users listed in `--protocol-writers` may, and each such write is +/// logged. Epoch metadata (and `_continuum` as a whole, which contains it) is never +/// writable: the protocol reads it at the snapshot change, so an edit only misleads +/// whoever reads it later. `403` otherwise. +fn parse_write_path( + state: &AppState, + user: &AuthenticatedUser, + key: &str, + method: &str, +) -> Result<Vec<String>, StatusCode> { + let path = parse_key_path(key)?; + if path.first().map(String::as_str) != Some(crate::remake::snapshot::META_ROOT) { + return Ok(path); + } + let epochs = path.len() < 2 || path[1] == crate::remake::snapshot::EPOCHS; + if epochs || !state.protocol_writers.contains(&user.0) { + warn!(user = %user.0, method, path = %key, "Refused a write to protocol data"); + return Err(StatusCode::FORBIDDEN); + } + warn!(user = %user.0, method, path = %key, "Protocol data written through the store API; this acts on the whole swarm"); + Ok(path) +} + fn response_for_value( value: &JsonValue, accept: Option<&HeaderValue>, diff --git a/src/api/peers.rs b/src/api/peers.rs index 7caaaf3..919f187 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -182,7 +182,7 @@ pub(crate) async fn list_peers( } peers.sort_by(|a, b| a.public_key.cmp(&b.public_key)); - let local_heads = state.doc.write().await.doc.get_heads().len(); + let local_heads = state.doc.read().await.heads().len(); Ok(Json(PeerListResponse { local_alias: state.alias.clone(), local_epoch: local_epoch.into(), diff --git a/src/api/stats.rs b/src/api/stats.rs index bc925ff..838cc0f 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -134,8 +134,8 @@ async fn collect(state: &AppState) -> StatsResponse { let node = state.stats.snapshot(); let (heads, tier) = { - let mut replica = state.doc.write().await; - let heads = replica.doc.get_heads().len(); + let replica = state.doc.read().await; + let heads = replica.heads().len(); (heads, members::tier(&replica.doc, state.local_id())) }; let saved = state.stats.saved_size(); diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index d29207c..f68ca85 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -130,6 +130,7 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou (direction, "sync", by.sync), (direction, "ping", by.ping), (direction, "pong", by.pong), + (direction, "blob", by.blob), ] }; let mut series = Vec::new(); diff --git a/src/api/store.rs b/src/api/store.rs index d6ff306..aa9465c 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -4,6 +4,7 @@ use std::io::Error as IoError; use automerge::ChangeHash; use axum::{ + Extension, body::{Body, Bytes}, extract::{Path, Query, State}, http::{ @@ -17,7 +18,8 @@ use serde_json::Value as JsonValue; use tokio::sync::broadcast; use tracing::{debug, error, instrument}; -use super::{parse_key_path, response_for_value, store_status}; +use super::auth::AuthenticatedUser; +use super::{parse_key_path, parse_write_path, response_for_value, store_status}; use crate::document::store::{self, StoreError}; use crate::state::{AppState, ReadSource}; use crate::swarm::sync::paths_related; @@ -194,6 +196,7 @@ fn mark_source( responses( (status = 200, description = "Value written successfully"), (status = 400, description = "Invalid path, malformed JSON, invalid UTF-8, or unsupported JSON value"), + (status = 403, description = "Protocol data (_continuum) and the user isn't a protocol writer, or epoch metadata"), (status = 409, description = "Path conflicts with an existing non-map value"), (status = 415, description = "Unsupported Content-Type"), (status = 500, description = "Internal error or synchronization notification failure"), @@ -202,6 +205,7 @@ fn mark_source( )] pub(crate) async fn put_store_entry( State(state): State<AppState>, + Extension(user): Extension<AuthenticatedUser>, Path(key): Path<String>, headers: HeaderMap, body: Bytes, @@ -210,7 +214,7 @@ pub(crate) async fn put_store_entry( return Err(StatusCode::SERVICE_UNAVAILABLE); } - let path = parse_key_path(&key)?; + let path = parse_write_path(&state, &user, &key, "PUT")?; let content_type = headers .get(CONTENT_TYPE) @@ -251,6 +255,7 @@ pub(crate) async fn put_store_entry( responses( (status = 204, description = "Value deleted successfully"), (status = 400, description = "Invalid store path"), + (status = 403, description = "Protocol data (_continuum) and the user isn't a protocol writer, or epoch metadata"), (status = 404, description = "Value does not exist"), (status = 500, description = "Deletion or synchronization notification failed"), (status = 503, description = "Node is not ready") @@ -259,10 +264,11 @@ pub(crate) async fn put_store_entry( #[instrument(skip(state), fields(path, prune = query.prune))] pub(crate) async fn delete_store_entry( State(state): State<AppState>, + Extension(user): Extension<AuthenticatedUser>, Path(key): Path<String>, Query(query): Query<DeleteQuery>, ) -> Result<StatusCode, StatusCode> { - let path = parse_key_path(&key)?; + let path = parse_write_path(&state, &user, &key, "DELETE")?; tracing::Span::current().record("path", path.join("/").as_str()); diff --git a/src/config/mod.rs b/src/config/mod.rs index 4f6c5e0..0fee76e 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -82,6 +82,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ "Users of the HTTP API (htpasswd file with Argon2 hashes). Required.", ), option( + "protocol-writers", + Kind::List, + "USER,[USER...]", + None, + "HTTP API users (from htpasswd) who may write protocol data under /store/_continuum (remake records, members, forget records), e.g. to repair a stuck remake. Such writes act on the whole swarm and are logged. Epoch metadata is never writable.", + ), + option( "debug-api", Kind::Flag, "BOOL", @@ -100,7 +107,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ Kind::Text, "PATH", None, - "Pre-shared swarm key: only nodes that know it may join (raw bytes, at least 16; e.g. `head -c 32 /dev/urandom`). Without it, anyone who knows the swarm ID can.", + "Pre-shared swarm key: only nodes that know it may join (raw bytes, at least 16; e.g. `head -c 32 /dev/urandom`). Without it, any node that can reach a member may join.", ), option( "alias", @@ -131,6 +138,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ "Members that renewed their alive stamp within this many days vote on remakes (a majority of them must agree). Matters on the node that coordinates a remake.", ), option( + "blob-chunk-kib", + Kind::Integer, + "KIB", + Some("64"), + "Chunk size when serving large transfers to other nodes (e.g. an epoch's snapshot to a rejoining node), in KiB (1 to 4096).", + ), + option( "mycelium-rest-url", Kind::Text, "SCHEME://HOSTNAME:PORT", @@ -194,6 +208,8 @@ pub(crate) struct Settings { pub(crate) bind_addr: SocketAddr, pub(crate) htpasswd: String, #[serde(default)] + pub(crate) protocol_writers: Vec<String>, + #[serde(default)] pub(crate) debug_api: bool, #[serde(default)] pub(crate) seeds: Vec<String>, @@ -205,6 +221,8 @@ pub(crate) struct Settings { pub(crate) neighbours: usize, #[serde(default = "default_electorate_window_days")] pub(crate) electorate_window_days: u64, + #[serde(default = "default_blob_chunk_kib")] + pub(crate) blob_chunk_kib: u32, pub(crate) mycelium_rest_url: Option<String>, pub(crate) mycelium_key_file: Option<PathBuf>, pub(crate) mycelium_network_key_file: Option<PathBuf>, @@ -227,6 +245,9 @@ fn default_neighbours() -> usize { fn default_electorate_window_days() -> u64 { 7 } +fn default_blob_chunk_kib() -> u32 { + 64 +} fn default_mycelium_peers() -> Vec<String> { vec!["tls://127.0.0.1:9651".to_owned()] } @@ -274,6 +295,9 @@ impl Settings { if self.keepalive_secs == 0 { bail!("keepalive-secs must be at least 1"); } + if !(1..=4096).contains(&self.blob_chunk_kib) { + bail!("blob-chunk-kib must be between 1 and 4096"); + } if self.electorate_window_days == 0 { bail!("electorate-window-days must be at least 1"); } diff --git a/src/document/replica.rs b/src/document/replica.rs index af807ea..e24f93c 100644 --- a/src/document/replica.rs +++ b/src/document/replica.rs @@ -9,7 +9,7 @@ use std::fmt; use std::ops::{Deref, DerefMut}; -use automerge::AutoCommit; +use automerge::{AutoCommit, ChangeHash}; /// One generation of the swarm document. /// @@ -77,14 +77,39 @@ impl fmt::Debug for Epoch { /// /// Derefs to the document, so code that only needs the `AutoCommit` keeps working /// with a guard of `Replica`. +/// +/// The heads are cached: `AutoCommit::get_heads` takes `&mut self` (it closes a pending +/// transaction), so without the cache every keepalive and status read would need the +/// write lock. Every path that changes the document under the write lock calls +/// [`Replica::refresh_heads`] before releasing it (`AppState::write_doc`, `mutate_doc`, +/// `switch_epoch`, the sync apply); readers take the read lock and use +/// [`Replica::heads`]. pub struct Replica { pub epoch: Epoch, pub doc: AutoCommit, + heads: Vec<ChangeHash>, } impl Replica { pub fn new(epoch: Epoch, doc: AutoCommit) -> Self { - Self { epoch, doc } + let mut replica = Self { + epoch, + doc, + heads: Vec::new(), + }; + replica.refresh_heads(); + replica + } + + /// The document's heads as of the last refresh, sorted. + pub fn heads(&self) -> &[ChangeHash] { + &self.heads + } + + /// Call after changing the document (under the write lock). + pub fn refresh_heads(&mut self) { + self.heads = self.doc.get_heads(); + self.heads.sort_unstable(); } } diff --git a/src/document/store.rs b/src/document/store.rs index 9e18693..573b811 100644 --- a/src/document/store.rs +++ b/src/document/store.rs @@ -243,6 +243,29 @@ pub fn keys_at(doc: &AutoCommit, path: &[String]) -> Result<Vec<String>, StoreEr Ok(merged_keys(doc, &maps).into_iter().collect()) } +/// Every concurrent value at `path`, newest first; empty if there is none. Where +/// `read_path` picks one (or merges concurrent maps into one), this keeps them apart. +pub(crate) fn read_candidates( + doc: &AutoCommit, + path: &[String], +) -> Result<Vec<JsonValue>, StoreError> { + let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; + let mut maps = vec![automerge::ROOT]; + for segment in parents { + maps = match resolve(doc, &maps, segment, None)? { + Some(Resolved::Maps(next)) => next, + _ => return Ok(Vec::new()), + }; + } + candidates(doc, &maps, leaf)? + .into_iter() + .map(|(value, id)| match value { + Value::Scalar(scalar) => scalar_to_json(scalar.into_owned()), + Value::Object(_) => object_to_json(doc, &id, None), + }) + .collect() +} + /// Writes (replaces) the value at `path`, creating intermediate maps. pub fn write_path( doc: &mut AutoCommit, diff --git a/src/main.rs b/src/main.rs index 5337f90..217a9a4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -139,9 +139,12 @@ async fn main() -> Result<()> { neighbours_k: args.neighbours, alias: args.alias.clone(), electorate_window: Duration::from_secs(args.electorate_window_days * 24 * 3600), + blobs: Arc::new(swarm::blob::Blobs::new(args.blob_chunk_kib * 1024)), + protocol_writers: Arc::new(args.protocol_writers.iter().cloned().collect()), config: Arc::new(config.report()), neighbours: Default::default(), pulling: Default::default(), + announce_pending: Default::default(), rtt: Default::default(), throughput: Default::default(), stability: Default::default(), diff --git a/src/remake/api.rs b/src/remake/api.rs index de8166c..826c81e 100644 --- a/src/remake/api.rs +++ b/src/remake/api.rs @@ -12,9 +12,10 @@ //! * `GET /admin/epochs` shows the local epoch, the archived previous ones and whether //! the node is catching up with a newer epoch from a peer. //! * `DELETE /admin/remakes/{id}` asks the coordinator to abort. Only possible while -//! proposed; `409` once committed (a switch can't be undone) or already over. -//! `DELETE /admin/remakes` does the same for the remake of the local epoch that -//! is in progress, without having to know its ID. +//! undecided; `409` once committed (a switch can't be undone) or aborted. An expired +//! remake is aborted right away by the node asked: its coordinator is gone, and +//! can't commit it any more anyway (`coordinator`). `DELETE /admin/remakes` does the +//! same for the undecided remake of the local epoch, without having to know its ID. //! //! Request body for POST (all optional): //! `{ "kind": "compact", "spec": null, "baseHeads": ["<hex>", …], @@ -22,15 +23,19 @@ //! "archiveRetention": {"epochs": 2, "seconds": 2592000}, //! "settleSecs": 480, "switchDelaySecs": 60 }` //! -//! `?force=true` accepts a `switchDelaySecs` below one keepalive interval: for tests -//! and debugging, not for real swarms. +//! `?force=` overrides checks: `true` all of them, or a comma-separated list of +//! * `switchDelaySecs`: accept one below a keepalive interval (for tests and debugging, +//! not for real swarms); +//! * `supersede`: abort expired remakes of this epoch and propose in their place. +//! Without it, an expired remake still holding votes answers `409`. //! //! `kind` is `compact` (no spec), `replace`, whose spec is //! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }` //! (see `kind::Replace`), or `import`, whose spec is //! `{ "document": "<base64 of an Automerge document>", "rules": […] }` (see -//! `kind::Import`; `GET /admin/document` exports one). Reports leave the document out -//! (`contentBytes`). Requests may be up to 64 MiB. +//! `kind::Import`; `GET /admin/document` exports one). The document stays on the +//! coordinator; the proposal refers to it by hash (`proposal.content`), and every voter +//! downloads it (`content`). Requests may be up to 64 MiB. //! //! An import can also be uploaded as is, with `Content-Type: application/octet-stream`; //! then `rules` (JSON), `baseHeads` (comma-separated), `settleSecs` and @@ -38,6 +43,7 @@ //! their paths: //! `curl --data-binary @backup.automerge -H 'Content-Type: application/octet-stream' --url-query 'rules=[{"from":"","to":""}]' …/admin/remakes` +use std::collections::HashSet; use std::time::Duration; use automerge::ChangeHash; @@ -61,7 +67,8 @@ use uuid::Uuid; use super::coordinator; use super::local::LocalStatus; use super::records::{ - self, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, Vote, unix_millis, + self, AbortRequest, ContentRef, Decision, EpochRecord, Phase, Policy, Proposal, Record, + Retention, Vote, unix_millis, }; use super::{Host, kind, vote}; use crate::document::store::StoreError; @@ -77,9 +84,8 @@ pub(crate) struct WaitQuery { pub(crate) struct StartQuery { #[serde(default)] wait: bool, - /// Allows a switch delay shorter than one keepalive interval. - #[serde(default)] - force: bool, + /// The checks to override (`Force`). + force: Option<String>, // Only for an import uploaded as `application/octet-stream`, which has no JSON // body to carry them: /// The rules, as JSON (as in the spec). @@ -149,10 +155,57 @@ struct StartRequest { /// How long votes are collected; twice the dormancy threshold if absent. settle_secs: Option<u64>, /// How long nodes wait after seeing the commit before switching; one keepalive - /// interval if absent. Shorter only with `?force=true`. + /// interval if absent. Shorter only with `?force=switchDelaySecs`. switch_delay_secs: Option<u64>, } +/// The checks `?force=` overrides: `true` all of them, or a comma-separated list. +#[derive(Debug, Default, Clone, Copy)] +struct Force { + switch_delay: bool, + supersede: bool, +} + +impl Force { + fn parse(value: Option<&str>) -> Result<Self, String> { + let mut force = Force::default(); + match value.map(str::trim) { + None | Some("" | "false") => {} + Some("true") => { + force = Force { + switch_delay: true, + supersede: true, + }; + } + Some(list) => { + for name in list.split(',').map(str::trim) { + match name { + "switchDelaySecs" => force.switch_delay = true, + "supersede" => force.supersede = true, + other => { + return Err(format!( + "unknown force {other:?}: use true, or a list of \ + switchDelaySecs and supersede" + )); + } + } + } + } + } + Ok(force) + } +} + +/// An expired remake of the epoch a proposal would supersede. +#[derive(Debug)] +struct ExpiredRemake { + id: String, + coordinator: String, + /// When it expired (Unix milliseconds). + since: u64, + tally: vote::Tally, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct Report { @@ -160,10 +213,6 @@ struct Report { phase: Phase, #[serde(flatten)] record: Record, - /// Size of the kind's build input (e.g. a replacement document), which reports - /// leave out. - #[serde(skip_serializing_if = "Option::is_none")] - content_bytes: Option<usize>, /// What happened to this remake on the node answering the request. #[serde(skip_serializing_if = "Option::is_none")] local: Option<LocalStatus>, @@ -173,6 +222,12 @@ struct Report { enum StartError { #[error("remake {0} is already open for this epoch")] Open(String), + /// Expired remakes still hold votes; superseding them takes `force=supersede`. + #[error("expired remakes of this epoch still hold votes")] + Expired(Vec<ExpiredRemake>), + /// The proposer votes for its own remake, so it must be free to. + #[error("this node can't vote for a new remake: {0}")] + VoteHeld(String), #[error("base head {0} is not in this node's document")] UnknownBase(String), #[error(transparent)] @@ -195,12 +250,10 @@ fn grace<H: Host>(host: &H) -> u64 { host.dormant_after().as_millis() as u64 } -fn report<H: Host>(host: &H, id: String, mut record: Record) -> Report { - let content = std::mem::take(&mut record.proposal.content); +fn report<H: Host>(host: &H, id: String, record: Record) -> Report { Report { phase: record.phase(unix_millis(), grace(host)), local: host.remake().status(&id), - content_bytes: (!content.is_null()).then(|| content.to_string().len()), id, record, } @@ -279,6 +332,10 @@ async fn start_kind<H: Host>( if !host.is_node_ready().await { return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); } + let force = match Force::parse(query.force.as_deref()) { + Ok(force) => force, + Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None), + }; let request: StartRequest = if is_octet_stream(headers) { match query.import_request(&body) { @@ -369,7 +426,7 @@ async fn start_kind<H: Host>( let keepalive = dormant_after / 4; let switch_delay = request.switch_delay_secs.map(Duration::from_secs); match switch_delay { - Some(delay) if delay < keepalive && query.force => { + Some(delay) if delay < keepalive && force.switch_delay => { warn!( switch_delay_secs = delay.as_secs(), keepalive_secs = keepalive.as_secs(), @@ -380,7 +437,7 @@ async fn start_kind<H: Host>( return error_response( StatusCode::BAD_REQUEST, format!( - "switchDelaySecs must be at least {} (one keepalive interval), so the commit reaches every node before they switch; add ?force=true to override", + "switchDelaySecs must be at least {} (one keepalive interval), so the commit reaches every node before they switch; add ?force=switchDelaySecs to override", keepalive.as_secs() ), None, @@ -394,7 +451,7 @@ async fn start_kind<H: Host>( let id = Uuid::new_v4().to_string(); let grace = grace(&host); - let proposal = Proposal { + let mut proposal = Proposal { coordinator: host.node_id(), from_epoch: host.current_epoch().into(), base_heads: Vec::new(), @@ -414,9 +471,17 @@ async fn start_kind<H: Host>( electorate: Vec::new(), quorum: 0, kind: kind.name().to_owned(), - content: prepared.content, + content: None, translation: prepared.translation, }; + // The content stays here, referred to by hash; voters fetch it from us. + if let Some(bytes) = prepared.content { + let reference = ContentRef::of(&bytes); + if let Some(hash) = reference.hash() { + host.remake().put_content(hash, std::sync::Arc::new(bytes)); + } + proposal.content = Some(reference); + } // From here on, nothing depends on the client: writing the proposal and starting the // coordinator run in their own task, so a client that disconnects mid-request can't @@ -426,6 +491,7 @@ async fn start_kind<H: Host>( id.clone(), proposal, base, + force.supersede, now, grace, )); @@ -450,6 +516,38 @@ async fn start_kind<H: Host>( Some(json!({ "remake": open })), ); } + Err(StartError::Expired(expired)) => { + let active: HashSet<String> = host + .peer_summaries() + .await + .into_iter() + .filter(|peer| peer.active) + .map(|peer| peer.id) + .collect(); + let me = host.node_id(); + let expired: Vec<JsonValue> = expired + .iter() + .map(|remake| { + json!({ + "id": remake.id, + "coordinator": remake.coordinator, + "coordinatorActive": remake.coordinator == me + || active.contains(&remake.coordinator), + "expiredAt": remake.since, + "tally": remake.tally.to_string(), + }) + }) + .collect(); + return error_response( + StatusCode::CONFLICT, + "an expired remake of this epoch still holds votes: abort it, or add \ + ?force=supersede to abort it and propose in its place", + Some(json!({ "expired": expired, "force": "supersede" })), + ); + } + Err(error @ StartError::VoteHeld(_)) => { + return error_response(StatusCode::CONFLICT, error.to_string(), None); + } Err(error @ StartError::UnknownBase(_)) => { return error_response(StatusCode::BAD_REQUEST, error.to_string(), None); } @@ -492,19 +590,69 @@ async fn propose<H: Host>( id: String, mut proposal: Proposal, base: Option<Vec<ChangeHash>>, + supersede: bool, now: u64, grace: u64, ) -> Result<Record, StartError> { let record_id = id.clone(); let electorate_host = host.clone(); let me = host.node_id(); + let started = host.remake().started_at() + grace; let record = host .commit_doc(move |doc| -> Result<Record, StartError> { - if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| { + let mut all = records::load_all(doc); + if let Some((open, _)) = all.iter().find(|(_, record)| { record.proposal.from_epoch == proposal.from_epoch && record.phase(now, grace).is_open() }) { - return Err(StartError::Open(open)); + return Err(StartError::Open(open.clone())); + } + let mut register = records::load_register(doc); + + // Expired remakes hold their votes; their coordinators can't commit them + // any more, so they may be aborted here. + let expired: Vec<&String> = all + .iter() + .filter(|(_, record)| { + record.proposal.from_epoch == proposal.from_epoch + && record.phase(now, grace) == Phase::Expired + }) + .map(|(id, _)| id) + .collect(); + if !expired.is_empty() { + if !supersede { + let details = all + .iter() + .filter(|(id, _)| expired.contains(&id)) + .map(|(id, record)| ExpiredRemake { + id: id.clone(), + coordinator: record.proposal.coordinator.clone(), + since: record.proposal.settle_until.saturating_add(grace), + tally: vote::tally(id, record, &all, ®ister), + }) + .collect(); + return Err(StartError::Expired(details)); + } + let decision = Decision::Aborted { + reason: format!("superseded by remake {record_id}"), + decided_at: now, + }; + for id in expired { + records::write_decision(doc, id, &decision)?; + info!(remake = %id, by = %record_id, "Superseded expired remake"); + } + all = records::load_all(doc); + register = records::load_register(doc); + } + + match vote::freedom(&me, &proposal.from_epoch, &all, ®ister, started) { + vote::Freedom::Any => {} + vote::Freedom::Only(id) => { + return Err(StartError::VoteHeld(format!( + "it may have voted for remake {id} before its start" + ))); + } + vote::Freedom::Held(reason) => return Err(StartError::VoteHeld(reason)), } // The base is the requested heads, or everything before the proposal (the @@ -534,12 +682,16 @@ async fn propose<H: Host>( proposal.electorate = electorate; records::write_proposal(doc, &record_id, &proposal)?; - let own_vote = Vote { at: now }; - records::write_vote(doc, &record_id, &me, &own_vote)?; + let own_vote = Vote { + remake: record_id.clone(), + at: now, + }; + records::write_vote(doc, &me, &own_vote)?; Ok(Record { proposal, decision: None, abort_requested: None, + contested: false, votes: [(me, own_vote)].into(), }) }) @@ -585,22 +737,18 @@ pub(crate) async fn abort<H: Host>( } /// `DELETE /admin/remakes`: aborts whichever remake of the local epoch is still -/// proposed. There is at most one, unless two coordinators raced (then both abort anyway -/// and this picks the oldest). +/// undecided (proposed or expired), the oldest if there are several. pub(crate) async fn abort_current<H: Host>( State(host): State<H>, Query(query): Query<WaitQuery>, ) -> Response { let local = EpochRecord::from(host.current_epoch()); - let now = unix_millis(); - let grace = grace(&host); let open = host .read_doc(move |doc| { records::load_all(doc) .into_iter() .filter(|(_, record)| { - record.proposal.from_epoch == local - && record.phase(now, grace) == Phase::Proposed + record.proposal.from_epoch == local && record.decision.is_none() }) .min_by_key(|(_, record)| record.proposal.proposed_at) .map(|(id, _)| id) @@ -624,6 +772,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { match current.phase { Phase::Proposed => {} + Phase::Expired => return abort_expired(&host, &id).await, Phase::Committed => { return error_response( StatusCode::CONFLICT, @@ -631,7 +780,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { Some(json!({ "remake": current })), ); } - Phase::Aborted | Phase::Expired => { + Phase::Aborted => { return error_response( StatusCode::CONFLICT, "already over", @@ -641,7 +790,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { } // The coordinator applies the request on its next check (within about a second). - // If it is gone, the proposal expires anyway. + // If it is gone, the proposal expires, and can then be aborted right away. if current.record.abort_requested.is_none() { let request = AbortRequest { by: host.node_id(), @@ -674,6 +823,47 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { } } +/// Aborts expired remake `id` from this node: its coordinator is gone, and can't commit +/// it any more (`coordinator`). Answers with the final report. +async fn abort_expired<H: Host>(host: &H, id: &str) -> Response { + let record_id = id.to_owned(); + let decision = Decision::Aborted { + reason: format!( + "aborted by operator via {} after it expired", + host.node_id() + ), + decided_at: unix_millis(), + }; + let grace = grace(host); + let written = host + .commit_doc(move |doc| -> Result<bool, StoreError> { + let expired = records::load(doc, &record_id) + .is_some_and(|record| record.phase(unix_millis(), grace) == Phase::Expired); + if expired { + records::write_decision(doc, &record_id, &decision)?; + } + Ok(expired) + }) + .await; + match written { + Ok(true) => info!(remake = %id, "Aborted expired remake"), + // Decided meanwhile: the report says how. + Ok(false) => {} + Err(error) => { + error!(remake = %id, error = %format!("{error:#}"), "Failed to abort expired remake"); + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + "failed to record the abort", + None, + ); + } + } + match load_report(host, id).await { + Some(report) => Json(report).into_response(), + None => error_response(StatusCode::NOT_FOUND, "no such remake", None), + } +} + /// `GET /admin/epochs`: the local epoch, the previous epochs this node still keeps, and /// an ongoing catch-up with a peer's newer epoch (`null` if none). pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response { diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index db25251..cad5a22 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -2,13 +2,17 @@ //! //! Peers on another epoch don't sync, so a node that missed a switch (new, late, //! returning, or one whose own build failed) can't follow through ordinary sync. Once -//! an active peer advertises a newer generation, the node +//! active, ready members advertise a newer generation (two of them for a node that is +//! ready itself, one for a node that never was; at most `MAX_GENERATION_GAP` ahead), +//! the node //! //! 1. stops being ready and keeps its document aside; reads are answered from a copy //! of it, marked stale (`X-Continuum-Stale`), writes get `503`; //! 2. switches to an empty document labelled with the newer epoch: an empty history -//! conflicts with nothing, so ordinary sync fills it from peers on that epoch, -//! exactly like a new node joining; +//! conflicts with nothing. It downloads the epoch's snapshot change in chunks from +//! several peers on that epoch at once (`swarm::blob`; sync waits meanwhile), then +//! ordinary sync adds the changes made since, exactly like a new node joining. If +//! the download fails, sync transfers the snapshot too, in one message; //! 3. once caught up with a peer, checks that the history starts with the epoch's //! snapshot change; //! 4. decides about its own changes the swarm never saw (none, for a new node): @@ -23,13 +27,13 @@ //! If no peer completes a sync in time, or the check fails, the node restores its old //! document and epoch and tries again later. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::convert::Infallible; use std::time::Duration; use automerge::{AutoCommit, ChangeHash}; use tokio::time::Instant; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; use super::Host; use super::delta::{self, ReplayStats}; @@ -44,12 +48,23 @@ use crate::document::replica::Epoch; const RETRY: Duration = Duration::from_secs(30); /// Lower bound for how long a rejoin may take to complete a sync. const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60); +/// A peer more than this many generations ahead is lying or broken (every generation +/// is a committed remake); its epoch is ignored instead of followed. +const MAX_GENERATION_GAP: u64 = 1000; +/// Peers that must advertise a newer epoch before a ready node leaves its document for +/// it (fewer if it has fewer active members). A node that was never ready has nothing +/// to lose and follows the first peer it sees. +const CORROBORATION: usize = 2; /// Called by the switch worker when there is nothing to switch to locally, and on every /// tick while a rejoin is in progress. pub(super) async fn check<H: Host>(host: &H) { if let Some((target, started)) = host.remake().rejoin_progress() { let timeout = host.dormant_after().max(MIN_REJOIN_TIMEOUT); + if host.remake().fetching_snapshot() { + // The download has its own stall timeout; the clock restarts after it. + return; + } if host.initial_sync_done() { finish(host).await; } else if started.elapsed() > timeout { @@ -65,27 +80,62 @@ pub(super) async fn check<H: Host>(host: &H) { } let local = host.current_epoch(); - let mut ahead: Vec<(String, Epoch)> = host - .peer_summaries() - .await + let peers = host.peer_summaries().await; + let active_members = peers + .iter() + .filter(|peer| peer.active && peer.member) + .count(); + // Only ready members count: a rejoining peer advertises an epoch it doesn't hold + // yet, and a node of another swarm (or a stranger) advertises one that isn't ours. + let mut ahead: Vec<(String, Epoch)> = peers .into_iter() - .filter(|peer| peer.active) + .filter(|peer| peer.serves()) .filter_map(|peer| peer.epoch.map(|epoch| (peer.id, epoch))) .filter(|(_, epoch)| epoch.generation > local.generation) .collect(); ahead.sort_by(|a, b| a.0.cmp(&b.0)); + ahead.retain(|(peer, epoch)| { + let plausible = epoch.generation - local.generation <= MAX_GENERATION_GAP; + if !plausible { + warn!(peer = %peer, epoch = %epoch, local = %local, "Peer advertises an implausible epoch; ignoring it"); + } + plausible + }); - let Some(generation) = ahead.iter().map(|(_, epoch)| epoch.generation).max() else { + // One peer's word is not enough for a ready node to give up its document: a buggy + // or hostile member could otherwise keep the swarm rejoining an epoch nobody + // serves. A newer epoch is followed once enough distinct peers advertise it. + let required = if host.is_node_ready().await { + active_members.clamp(1, CORROBORATION) + } else { + 1 + }; + let mut advertised: HashMap<Epoch, usize> = HashMap::new(); + for (_, epoch) in &ahead { + *advertised.entry(*epoch).or_default() += 1; + } + let corroborated: Vec<Epoch> = advertised + .into_iter() + .filter(|(_, count)| *count >= required) + .map(|(epoch, _)| epoch) + .collect(); + + let Some(generation) = corroborated.iter().map(|epoch| epoch.generation).max() else { if host.remake().catch_up().is_some() { host.remake().set_catch_up(None); } + if !ahead.is_empty() { + debug!( + peers = ahead.len(), + required, "A newer epoch is advertised, but not by enough peers yet" + ); + } return; }; - let mut targets: Vec<Epoch> = ahead - .iter() - .filter(|(_, epoch)| epoch.generation == generation) - .map(|(_, epoch)| *epoch) + let mut targets: Vec<Epoch> = corroborated + .into_iter() + .filter(|epoch| epoch.generation == generation) .collect(); targets.sort_by_key(|epoch| epoch.id); targets.dedup(); @@ -147,6 +197,45 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) { started: Instant::now(), was_ready, }); + + // Sync waits until the snapshot is here (or its download failed). + host.remake().set_fetching_snapshot(Some(target)); + tokio::spawn(fetch_snapshot(host.clone(), target)); +} + +/// Downloads the snapshot change of `target` and applies it, then lets sync go on. +async fn fetch_snapshot<H: Host>(host: H, target: Epoch) { + let outcome = host.fetch_change(ChangeHash(target.id)).await; + let still_rejoining = host + .remake() + .rejoin_progress() + .is_some_and(|(rejoining, _)| rejoining == target) + && host.current_epoch() == target; + if still_rejoining { + match outcome { + Ok(change) => { + let applied = host + .commit_doc(move |doc| doc.apply_changes([change])) + .await; + match applied { + Ok(()) => { + info!(epoch = %target, "Applied the epoch's snapshot; syncing the changes made since") + } + Err(error) => { + warn!(epoch = %target, error = %error, "Couldn't apply the downloaded snapshot; syncing it instead") + } + } + } + Err(error) => warn!( + epoch = %target, + error = %format!("{error:#}"), + "Couldn't download the epoch's snapshot; syncing it instead" + ), + } + host.remake().restart_rejoin_clock(target); + } + host.remake().set_fetching_snapshot(None); + host.request_sync(); } /// What to do with this node's own changes the new epoch lacks. @@ -465,7 +554,7 @@ mod tests { policy, proposed_at: 0, kind: "compact", - content: &JsonValue::Null, + content: None, translation: &JsonValue::Null, }, ) diff --git a/src/remake/content.rs b/src/remake/content.rs new file mode 100644 index 0000000..e5d78c5 --- /dev/null +++ b/src/remake/content.rs @@ -0,0 +1,95 @@ +//! Content of remakes: the new document of a `replace` or `import` remake. +//! +//! The proposal records only a reference (BLAKE3 hash and size). The coordinator keeps +//! the bytes in memory; every member fetches them before voting (in chunks, from +//! several peers at once: `swarm::blob`) and serves them in turn. A committed remake +//! therefore has its content on a majority of the electorate. Nodes that switch need +//! it to build the snapshot; nodes that rejoin later don't (they download the snapshot +//! itself). +//! +//! Content no open remake refers to is dropped an hour after it was stored. + +use std::collections::HashSet; +use std::sync::Arc; +use std::time::Duration; + +use tracing::{info, warn}; + +use super::Host; +use super::records::{ContentRef, Record}; + +/// The content `reference` names, if this node has it; otherwise starts fetching it in +/// the background (once) and returns None. +pub(super) fn available_or_fetch<H: Host>( + host: &H, + reference: &ContentRef, +) -> Option<Arc<Vec<u8>>> { + let hash = reference.hash()?; + if let Some(bytes) = host.remake().content(&hash) { + return Some(bytes); + } + if host.remake().start_content_fetch(hash) { + let host = host.clone(); + let reference = reference.clone(); + tokio::spawn(async move { + fetch(&host, hash, &reference).await; + host.remake().end_content_fetch(&hash); + }); + } + None +} + +/// The content `reference` names: from this node, or fetched now. Waits for a download +/// already running. +pub(super) async fn ensure<H: Host>( + host: &H, + reference: &ContentRef, +) -> Result<Arc<Vec<u8>>, String> { + let hash = reference + .hash() + .ok_or_else(|| format!("unreadable content hash {:?}", reference.blake3))?; + loop { + if let Some(bytes) = host.remake().content(&hash) { + return Ok(bytes); + } + if host.remake().start_content_fetch(hash) { + let fetched = fetch(host, hash, reference).await; + host.remake().end_content_fetch(&hash); + return fetched + .ok_or_else(|| format!("couldn't download content {}", reference.blake3)); + } + // Someone else is downloading it. + while host.remake().content_fetch_running(&hash) { + tokio::time::sleep(Duration::from_millis(500)).await; + } + } +} + +async fn fetch<H: Host>(host: &H, hash: [u8; 32], reference: &ContentRef) -> Option<Arc<Vec<u8>>> { + match host.fetch_content(hash).await { + Ok(bytes) if bytes.len() as u64 == reference.size => { + info!(content = %reference.blake3, bytes = bytes.len(), "Downloaded remake content"); + let bytes = Arc::new(bytes); + host.remake().put_content(hash, Arc::clone(&bytes)); + Some(bytes) + } + Ok(bytes) => { + warn!(content = %reference.blake3, bytes = bytes.len(), expected = reference.size, "Remake content has the wrong size"); + None + } + Err(error) => { + warn!(content = %reference.blake3, error = %format!("{error:#}"), "Couldn't download remake content"); + None + } + } +} + +/// Drops content no open remake in `records` refers to (after its grace period). +pub(super) fn prune<H: Host>(host: &H, records: &[(String, Record)], now: u64, grace: u64) { + let referenced: HashSet<[u8; 32]> = records + .iter() + .filter(|(_, record)| record.phase(now, grace).is_open()) + .filter_map(|(_, record)| record.proposal.content.as_ref()?.hash()) + .collect(); + host.remake().prune_contents(&referenced); +} diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs index 5885884..9bcf8ba 100644 --- a/src/remake/coordinator.rs +++ b/src/remake/coordinator.rs @@ -6,48 +6,85 @@ //! * an operator requested an abort, //! * the local epoch changed, //! * a competing remake of the same epoch committed, -//! * a majority is out of reach (too many members voted for a competing remake), or -//! * the voting period (`settleSecs`) passed without a majority. +//! * the voting period (`settleSecs`) passed without a majority, or +//! * a majority is out of reach (too many members voted for a competing remake). +//! +//! The decision is written in the same step that checks the records once more, so a +//! commit never overrides an abort request or a competitor's commit that arrived while +//! the snapshot was being built. Nor is a commit written once the remake has expired: +//! from then on, any node may abort it (`api`), and the two mustn't race. //! //! Committing builds the snapshot at the base heads and records the new epoch, so //! every node can verify its own build against it. //! -//! The task lives only in memory. If the coordinator dies, nobody decides and the -//! proposal expires (its votes become void). +//! The task lives only in memory. A coordinator that restarts takes its undecided +//! remakes up again once ready (`resume`); past the voting period, that means aborting +//! them. If it is gone for good, nobody decides and the proposal expires; its votes +//! stay cast (`vote`). use std::time::Duration; use tracing::{info, warn}; use super::Host; -use super::records::{self, Decision, Record}; +use super::content; +use super::records::{self, Decision, EpochRecord, Phase, Record}; use super::snapshot; use super::vote; use crate::document::store::StoreError; const TICK: Duration = Duration::from_secs(1); +/// Coordinates remake `id` until it is decided; returns at once if this node already +/// does. pub(crate) async fn run<H: Host>(host: H, id: String) { + if !host.remake().start_coordinating(&id) { + return; + } info!(remake = %id, "Coordinating remake"); loop { tokio::time::sleep(TICK).await; match step(&host, &id).await { - Ok(true) => return, + Ok(true) => break, Ok(false) => {} Err(error) => { warn!(remake = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying"); } } } + host.remake().end_coordinating(&id); +} + +/// Takes up coordinating this node's undecided remakes of `epoch` among `all` that it +/// doesn't coordinate (it restarted since proposing them). Called by the remake worker +/// once the node is ready, so its records are complete. +pub(super) fn resume<H: Host>(host: &H, epoch: &EpochRecord, all: &[(String, Record)]) { + let me = host.node_id(); + for (id, record) in all { + if record.decision.is_none() + && record.proposal.coordinator == me + && &record.proposal.from_epoch == epoch + && !host.remake().is_coordinating(id) + { + info!(remake = %id, "Resuming coordination of own remake"); + tokio::spawn(run(host.clone(), id.clone())); + } + } } /// One evaluation. Returns true once the coordinator is done. async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> { let wanted = id.to_owned(); - let (record, all) = host - .read_doc(move |doc| (records::load(doc, &wanted), records::load_all(doc))) + let (record, all, register) = host + .read_doc(move |doc| { + ( + records::load(doc, &wanted), + records::load_all(doc), + records::load_register(doc), + ) + }) .await; let Some(record) = record else { @@ -58,15 +95,26 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> { return Ok(true); } - let now = records::unix_millis(); - let grace = host.dormant_after().as_millis() as u64; - - if let Some(reason) = abort_reason(host, id, &record, &all, now, grace) { + if EpochRecord::from(host.current_epoch()) != record.proposal.from_epoch { + decide_abort(host, id, "local epoch changed during the vote".to_owned()).await?; + return Ok(true); + } + if let Some(reason) = obstacle(id, &record, &all) { decide_abort(host, id, reason).await?; return Ok(true); } - let tally = vote::tally(id, &record, &all, now, grace); + let tally = vote::tally(id, &record, &all, ®ister); + // The deadline first: a majority counted after it is too late. + if records::unix_millis() >= record.proposal.settle_until { + decide_abort( + host, + id, + format!("no majority in the voting period: {tally}"), + ) + .await?; + return Ok(true); + } if tally.reached() { info!(remake = %id, votes = %tally, "Majority reached"); commit(host, id, &record).await?; @@ -76,49 +124,24 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> { decide_abort(host, id, format!("a majority is out of reach: {tally}")).await?; return Ok(true); } - if now >= record.proposal.settle_until { - decide_abort( - host, - id, - format!("no majority in the voting period: {tally}"), - ) - .await?; - return Ok(true); - } Ok(false) } -fn abort_reason<H: Host>( - host: &H, - id: &str, - record: &Record, - all: &[(String, Record)], - now: u64, - grace: u64, -) -> Option<String> { - let proposal = &record.proposal; - +/// Why remake `id` must be aborted whatever its votes, if it must. +fn obstacle(id: &str, record: &Record, all: &[(String, Record)]) -> Option<String> { if let Some(request) = &record.abort_requested { return Some(format!("aborted by operator via {}", request.by)); } - - if records::EpochRecord::from(host.current_epoch()) != proposal.from_epoch { - return Some("local epoch changed during the vote".to_owned()); - } - - if let Some((other, _)) = all.iter().find(|(other, candidate)| { + let (other, _) = all.iter().find(|(other, candidate)| { other != id - && candidate.proposal.from_epoch == proposal.from_epoch - && candidate.phase(now, grace) == records::Phase::Committed - }) { - return Some(format!("competing remake {other} committed")); - } - - None + && candidate.proposal.from_epoch == record.proposal.from_epoch + && candidate.committed_epoch().is_some() + })?; + Some(format!("competing remake {other} committed")) } async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> { - let Some(request) = record.snapshot_request(id) else { + let Some(mut request) = record.snapshot_request(id) else { return decide_abort( host, id, @@ -126,6 +149,12 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor ) .await; }; + if let Some(reference) = &record.proposal.content { + match content::ensure(host, reference).await { + Ok(bytes) => request.content = Some(bytes), + Err(reason) => return decide_abort(host, id, reason).await, + } + } // Clone under the read lock, build without any lock. let copy = host.read_doc(|doc| doc.clone()).await; @@ -144,30 +173,74 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor snapshot_bytes, decided_at: records::unix_millis(), }; - write_decision(host, id, decision).await?; - - // The local switch reuses it instead of building again. - host.remake().store_prebuilt(id, snapshot); - - info!(remake = %id, to_epoch = %to_epoch, snapshot_bytes, "Remake committed"); + match decide(host, id, decision).await? { + Some(Decision::Committed { .. }) => { + // The local switch reuses it instead of building again. + host.remake().store_prebuilt(id, snapshot); + info!(remake = %id, to_epoch = %to_epoch, snapshot_bytes, "Remake committed"); + } + Some(Decision::Aborted { reason, .. }) => { + warn!(remake = %id, reason = %reason, "Remake aborted while building its snapshot"); + } + None => {} + } Ok(()) } async fn decide_abort<H: Host>(host: &H, id: &str, reason: String) -> Result<(), StoreError> { - warn!(remake = %id, reason = %reason, "Remake aborted"); - write_decision( - host, - id, - Decision::Aborted { - reason, - decided_at: records::unix_millis(), - }, - ) - .await + let decision = Decision::Aborted { + reason: reason.clone(), + decided_at: records::unix_millis(), + }; + if decide(host, id, decision).await?.is_some() { + warn!(remake = %id, reason = %reason, "Remake aborted"); + } + Ok(()) } -async fn write_decision<H: Host>(host: &H, id: &str, decision: Decision) -> Result<(), StoreError> { +/// Writes `decision`, checked again under the write lock against the records as they +/// are now (building the snapshot takes a while): a commit becomes an abort if the +/// operator asked for one, a competing remake committed, the remake expired, or the +/// majority is gone. +/// Returns what was written; nothing if the remake is decided already or gone. +async fn decide<H: Host>( + host: &H, + id: &str, + decision: Decision, +) -> Result<Option<Decision>, StoreError> { let id = id.to_owned(); - host.commit_doc(move |doc| records::write_decision(doc, &id, &decision)) - .await + let grace = host.dormant_after().as_millis() as u64; + host.commit_doc(move |doc| -> Result<Option<Decision>, StoreError> { + let all = records::load_all(doc); + let Some((_, record)) = all.iter().find(|(other, _)| *other == id) else { + return Ok(None); + }; + if record.decision.is_some() { + return Ok(None); + } + let decision = match decision { + Decision::Committed { decided_at, .. } => { + let register = records::load_register(doc); + let tally = vote::tally(&id, record, &all, ®ister); + let expired = record.phase(records::unix_millis(), grace) == Phase::Expired; + match obstacle(&id, record, &all) { + Some(reason) => Decision::Aborted { reason, decided_at }, + // Others may abort it by now (`api`): too late to commit. + None if expired => Decision::Aborted { + reason: format!("expired while building the snapshot: {tally}"), + decided_at, + }, + None if !tally.reached() => Decision::Aborted { + reason: format!("the majority was lost: {tally}"), + decided_at, + }, + None => decision, + } + } + aborted @ Decision::Aborted { .. } => aborted, + }; + records::write_decision(doc, &id, &decision)?; + Ok(Some(decision)) + }) + .await } diff --git a/src/remake/delta.rs b/src/remake/delta.rs index d318c00..c8f6b7c 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -27,7 +27,7 @@ use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop}; use tracing::warn; use super::kind::{Translation, Translator}; -use super::snapshot::{EPOCHS, META_ROOT, REMAKES}; +use super::snapshot::{EPOCHS, META_ROOT, REMAKES, VOTES}; use crate::document::store::{self, StoreError}; #[derive(Debug, Default, Clone)] @@ -135,7 +135,10 @@ fn log_failure(path: &[String], error: &StoreError) { } fn is_protocol_path(path: &[String]) -> bool { - matches!(path, [root, part, ..] if root == META_ROOT && (part == REMAKES || part == EPOCHS)) + let [root, part, ..] = path else { + return false; + }; + root == META_ROOT && [REMAKES, VOTES, EPOCHS].contains(&part.as_str()) } /// True for the root and for `_continuum` itself. diff --git a/src/remake/kind.rs b/src/remake/kind.rs index 685fa28..8131f7f 100644 --- a/src/remake/kind.rs +++ b/src/remake/kind.rs @@ -28,11 +28,12 @@ use crate::document::store::{self, StoreError}; /// Name of the kind used when a request doesn't name one. pub(crate) const DEFAULT_KIND: &str = "compact"; -/// What the coordinator stores in the proposal for every node. +/// What the coordinator prepares for every node. #[derive(Debug, Default)] pub(crate) struct Prepared { - /// Input to [`RemakeKind::build`]; replicated with the proposal (old epoch only). - pub(crate) content: JsonValue, + /// Input to [`RemakeKind::build`], if the kind takes any. The proposal refers to it + /// by hash; voters fetch it from the coordinator and each other (`content`). + pub(crate) content: Option<Vec<u8>>, /// Input to [`RemakeKind::translate`]; also kept in the new epoch's metadata, so /// nodes that rejoin later can translate their writes. pub(crate) translation: JsonValue, @@ -64,7 +65,7 @@ pub(crate) trait RemakeKind: Send + Sync + 'static { fn build( &self, old_at_base: &AutoCommit, - content: &JsonValue, + content: Option<&[u8]>, target: &mut AutoCommit, ) -> Result<(), StoreError>; @@ -107,7 +108,7 @@ impl RemakeKind for Compact { fn build( &self, old_at_base: &AutoCommit, - _content: &JsonValue, + _content: Option<&[u8]>, target: &mut AutoCommit, ) -> Result<(), StoreError> { store::materialize_into(old_at_base, target, &is_reserved) @@ -199,8 +200,10 @@ impl RemakeKind for Replace { )); } + let content = + serde_json::to_vec(&canonical(spec.document)).map_err(|error| error.to_string())?; Ok(Prepared { - content: canonical(spec.document), + content: Some(content), translation: prepare_rules(spec.rules)?, }) } @@ -208,12 +211,15 @@ impl RemakeKind for Replace { fn build( &self, _old_at_base: &AutoCommit, - content: &JsonValue, + content: Option<&[u8]>, target: &mut AutoCommit, ) -> Result<(), StoreError> { - // Canonical again: the content went through the old epoch's document, and the - // operations (hence the epoch ID) must not depend on any key order. - let JsonValue::Object(document) = canonical(content.clone()) else { + let content = content.ok_or_else(|| anyhow::anyhow!("replace without content"))?; + let document: JsonValue = serde_json::from_slice(content) + .map_err(|error| anyhow::anyhow!("replacement document: {error}"))?; + // Canonical again: the operations (hence the epoch ID) must not depend on any + // key order. + let JsonValue::Object(document) = canonical(document) else { return Ok(()); // propose only accepts objects }; for (key, value) in document { @@ -300,11 +306,8 @@ struct ImportSpec { rules: Vec<RuleSpec>, } -/// The document in `content` (base64 of a saved Automerge document). -fn load_content(content: &JsonValue) -> Result<AutoCommit, String> { - let JsonValue::String(encoded) = content else { - return Err("the document must be a base64 string".to_owned()); - }; +/// The document in a request (base64 of a saved Automerge document). +fn load_encoded(encoded: &str) -> Result<AutoCommit, String> { let bytes = BASE64 .decode(encoded.trim()) .map_err(|error| format!("the document is not valid base64: {error}"))?; @@ -320,17 +323,17 @@ impl RemakeKind for Import { fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> { let spec: ImportSpec = serde_json::from_value(spec.clone()).map_err(|error| error.to_string())?; - let imported = load_content(&JsonValue::String(spec.document))?; + let imported = load_encoded(&spec.document)?; // Keep only what goes into the new epoch: the merged view without protocol data - // and history. Usually much smaller than what was uploaded, and it is what gets - // replicated with the proposal. + // and history. Usually much smaller than what was uploaded, and it is what every + // voter downloads. let mut content = AutoCommit::new(); store::materialize_into(&imported, &mut content, &is_reserved) .map_err(|error| format!("the document can't be read: {error}"))?; Ok(Prepared { - content: JsonValue::String(BASE64.encode(content.save())), + content: Some(content.save()), translation: prepare_rules(spec.rules)?, }) } @@ -338,10 +341,11 @@ impl RemakeKind for Import { fn build( &self, _old_at_base: &AutoCommit, - content: &JsonValue, + content: Option<&[u8]>, target: &mut AutoCommit, ) -> Result<(), StoreError> { - let imported = load_content(content).map_err(|reason| anyhow::anyhow!(reason))?; + let content = content.ok_or_else(|| anyhow::anyhow!("import without content"))?; + let imported = AutoCommit::load(content)?; // Deterministic: the same bytes give the same view, copied in a fixed order. store::materialize_into(&imported, target, &is_reserved) } @@ -412,7 +416,7 @@ mod tests { fn build( &self, _old: &AutoCommit, - _content: &JsonValue, + _content: Option<&[u8]>, _target: &mut AutoCommit, ) -> Result<(), StoreError> { Ok(()) @@ -559,7 +563,7 @@ mod tests { let build = || { let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..])); Import - .build(&AutoCommit::new(), &prepared.content, &mut target) + .build(&AutoCommit::new(), prepared.content.as_deref(), &mut target) .unwrap(); target.commit_with(automerge::transaction::CommitOptions::default().with_time(0)); target @@ -591,7 +595,7 @@ mod tests { let prepared = replace(json!({ "document": document })).unwrap(); let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..])); Replace - .build(&AutoCommit::new(), &prepared.content, &mut target) + .build(&AutoCommit::new(), prepared.content.as_deref(), &mut target) .unwrap(); target.commit_with(automerge::transaction::CommitOptions::default().with_time(0)); ( diff --git a/src/remake/local.rs b/src/remake/local.rs index d2ea2c2..04ac0b0 100644 --- a/src/remake/local.rs +++ b/src/remake/local.rs @@ -1,12 +1,13 @@ //! Per-node, in-memory remake state: snapshots built ahead of the switch, switch -//! progress for reports, a rejoin in progress, quarantined changes, and archived -//! documents of previous epochs. +//! progress for reports, a rejoin in progress, quarantined changes, archived documents +//! of previous epochs, and the content of remakes (`content`). //! //! Nothing here is replicated or durable. After a restart the node simply has no //! archive or quarantine, and rebuilds snapshots when needed. -use std::collections::HashMap; -use std::sync::{Mutex, MutexGuard}; +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; use automerge::AutoCommit; use serde::Serialize; @@ -16,12 +17,23 @@ use super::records::{Retention, unix_millis}; use super::snapshot::Snapshot; use crate::document::replica::Epoch; -#[derive(Default)] pub(crate) struct RemakeState { + /// When this node started (Unix milliseconds): votes it cast before may be lost + /// (`vote`). + started_at: u64, // A std mutex: every access is short and never spans an await. inner: Mutex<Inner>, } +impl Default for RemakeState { + fn default() -> Self { + Self { + started_at: unix_millis(), + inner: Mutex::default(), + } + } +} + #[derive(Default)] struct Inner { /// Built by the coordinator at commit time; consumed by the local switch. @@ -34,9 +46,21 @@ struct Inner { catch_up: Option<CatchUp>, /// A rejoin in progress: the node syncs an empty document of a newer epoch. rejoin: Option<Rejoin>, + /// While the rejoin downloads the target epoch's snapshot (`catchup`), sync waits. + fetching_snapshot: Option<Epoch>, quarantine: Vec<Quarantined>, + /// Content of remakes, by BLAKE3 hash, and when it was stored (served to peers). + contents: HashMap<[u8; 32], (Arc<Vec<u8>>, Instant)>, + /// Content being downloaded. + content_fetches: HashSet<[u8; 32]>, + /// Remakes this node coordinates now (`coordinator`). + coordinating: HashSet<String>, } +/// Content no remake refers to any more is kept this long after it was stored, for +/// nodes that switch late. +const CONTENT_GRACE: Duration = Duration::from_secs(3600); + /// Catching up with a newer epoch: attempts and the last failure. #[derive(Clone, Debug)] pub(crate) struct CatchUp { @@ -119,6 +143,23 @@ pub(crate) struct ArchivedEpoch { } impl RemakeState { + pub(crate) fn started_at(&self) -> u64 { + self.started_at + } + + /// Whether this node starts coordinating `id` now (false if it already does). + pub(crate) fn start_coordinating(&self, id: &str) -> bool { + self.lock().coordinating.insert(id.to_owned()) + } + + pub(crate) fn end_coordinating(&self, id: &str) { + self.lock().coordinating.remove(id); + } + + pub(crate) fn is_coordinating(&self, id: &str) -> bool { + self.lock().coordinating.contains(id) + } + fn lock(&self) -> MutexGuard<'_, Inner> { self.inner .lock() @@ -211,6 +252,54 @@ impl RemakeState { self.lock().rejoin.take() } + pub(crate) fn content(&self, hash: &[u8; 32]) -> Option<Arc<Vec<u8>>> { + self.lock() + .contents + .get(hash) + .map(|(bytes, _)| Arc::clone(bytes)) + } + + pub(crate) fn put_content(&self, hash: [u8; 32], bytes: Arc<Vec<u8>>) { + self.lock().contents.insert(hash, (bytes, Instant::now())); + } + + /// Marks a download of `hash` as started; false if one already is. + pub(crate) fn start_content_fetch(&self, hash: [u8; 32]) -> bool { + self.lock().content_fetches.insert(hash) + } + + pub(crate) fn end_content_fetch(&self, hash: &[u8; 32]) { + self.lock().content_fetches.remove(hash); + } + + pub(crate) fn content_fetch_running(&self, hash: &[u8; 32]) -> bool { + self.lock().content_fetches.contains(hash) + } + + /// Drops content that no remake in `referenced` needs, once its grace has passed. + pub(crate) fn prune_contents(&self, referenced: &HashSet<[u8; 32]>) { + self.lock().contents.retain(|hash, (_, stored)| { + referenced.contains(hash) || stored.elapsed() < CONTENT_GRACE + }); + } + + /// Whether sync should wait: a rejoin is downloading its epoch's snapshot. + pub(crate) fn fetching_snapshot(&self) -> bool { + self.lock().fetching_snapshot.is_some() + } + + pub(crate) fn set_fetching_snapshot(&self, target: Option<Epoch>) { + self.lock().fetching_snapshot = target; + } + + /// The rejoin of `target`, if still in progress, gets its full time for the sync + /// that follows the download. + pub(crate) fn restart_rejoin_clock(&self, target: Epoch) { + if let Some(rejoin) = self.lock().rejoin.as_mut().filter(|r| r.target == target) { + rejoin.started = Instant::now(); + } + } + pub(crate) fn add_quarantine(&self, quarantined: Quarantined) { self.lock().quarantine.push(quarantined); } diff --git a/src/remake/mod.rs b/src/remake/mod.rs index 0895931..ca17956 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -7,6 +7,7 @@ //! * `records`: the replicated remake records under `_continuum/remakes/<id>`. //! * `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. @@ -20,6 +21,7 @@ mod api; mod catchup; +mod content; mod coordinator; mod delta; mod kind; @@ -68,6 +70,21 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// 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. @@ -124,10 +141,30 @@ 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 diff --git a/src/remake/records.rs b/src/remake/records.rs index ed018ca..cd3e5be 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -7,7 +7,16 @@ //! * `decision`: written once, by the coordinator only (committed or aborted). //! * `abortRequested`: written by any node on an operator's request. It is an input to //! the coordinator's decision, not a decision itself. -//! * `votes/<pk>`: each written by that member only (`vote`). +//! +//! Once a remake has expired, its coordinator no longer commits it, and any node may +//! abort it on an operator's request (`api`). Should clocks disagree by more than the +//! grace margin, a commit and such an abort could still be written concurrently. Such +//! a record is *contested*: it reads as committed, and nodes don't switch to it on +//! their own. +//! +//! Votes are kept apart, in the votes register `_continuum/votes/<pk>`: one entry per +//! member for the whole epoch, written by that member only (`vote`). Reading a record +//! fills in the votes that name it. //! //! A proposal whose coordinator never decides (crash, partition) is reported as //! `expired` once its settle period plus a grace margin has passed. Expiry is derived, @@ -24,7 +33,7 @@ use serde_json::Value as JsonValue; use tracing::warn; use super::kind::DEFAULT_KIND; -use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest}; +use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest, VOTES}; use crate::document::replica::Epoch; use crate::document::store::{self, StoreError}; @@ -137,14 +146,37 @@ pub(crate) struct Proposal { /// The remake kind (`compact` for records written before kinds existed). #[serde(default = "default_kind")] pub(crate) kind: String, - /// The kind's build input. - #[serde(default, skip_serializing_if = "JsonValue::is_null")] - pub(crate) content: JsonValue, + /// The kind's build input, if it has one: held by the coordinator and fetched by + /// every voter (`content`), not stored in the record. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) content: Option<ContentRef>, /// The kind's translation input. #[serde(default, skip_serializing_if = "JsonValue::is_null")] pub(crate) translation: JsonValue, } +/// Content of a remake, by reference. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ContentRef { + /// BLAKE3 of the bytes, hex. + pub(crate) blake3: String, + pub(crate) size: u64, +} + +impl ContentRef { + pub(crate) fn of(bytes: &[u8]) -> Self { + Self { + blake3: blake3::hash(bytes).to_hex().to_string(), + size: bytes.len() as u64, + } + } + + pub(crate) fn hash(&self) -> Option<[u8; 32]> { + decode_hex32(&self.blake3) + } +} + fn default_kind() -> String { DEFAULT_KIND.to_owned() } @@ -178,13 +210,18 @@ pub(crate) struct AbortRequest { pub(crate) at: u64, } -/// A member's vote for a remake. -#[derive(Clone, Debug, Serialize, Deserialize)] +/// A member's vote: its entry in the votes register. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct Vote { + pub(crate) remake: String, pub(crate) at: u64, } +/// The votes register: every concurrent value of each member's entry, by member public +/// key. Two values mean the member voted again without seeing its first vote (`vote`). +pub(crate) type Register = BTreeMap<String, Vec<Vote>>; + #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct Record { @@ -193,8 +230,17 @@ pub(crate) struct Record { pub(crate) decision: Option<Decision>, #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) abort_requested: Option<AbortRequest>, - /// By member public key. - #[serde(default)] + /// Decisions written concurrently disagree; `decision` is then the commit. Not + /// stored: found when the record is read. + #[serde( + default, + skip_deserializing, + skip_serializing_if = "std::ops::Not::not" + )] + pub(crate) contested: bool, + /// The register's votes naming this remake, by member public key. Not stored in + /// the record: filled in when it is read (`load`). + #[serde(default, skip_deserializing)] pub(crate) votes: BTreeMap<String, Vote>, } @@ -229,8 +275,8 @@ impl Record { } } - /// Everything needed to (re)build this remake's snapshot; None if the proposal - /// is unreadable. + /// Everything needed to (re)build this remake's snapshot but the content (which the + /// caller adds, `content::ensure`); None if the proposal is unreadable. pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> { Some(SnapshotRequest { remake_id: id.to_owned(), @@ -239,7 +285,7 @@ impl Record { policy: serde_json::to_value(&self.proposal.policy).ok()?, proposed_at: self.proposal.proposed_at, kind: self.proposal.kind.clone(), - content: self.proposal.content.clone(), + content: None, translation: self.proposal.translation.clone(), }) } @@ -269,8 +315,37 @@ pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> { path } -/// The record with this ID, if it exists and is readable. +/// The record with this ID, if it exists and is readable, with its votes. pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> { + let record = load_record(doc, id)?; + Some(with_votes(id, record, &load_register(doc))) +} + +/// All readable records, with their votes. +pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> { + let register = load_register(doc); + store::keys_at(doc, &records_dir()) + .unwrap_or_default() + .into_iter() + .filter_map(|id| { + let record = load_record(doc, &id)?; + Some((id.clone(), with_votes(&id, record, ®ister))) + }) + .collect() +} + +fn with_votes(id: &str, mut record: Record, register: &Register) -> Record { + record.votes = register + .iter() + .filter_map(|(voter, votes)| { + let vote = votes.iter().find(|vote| vote.remake == id)?; + Some((voter.clone(), vote.clone())) + }) + .collect(); + record +} + +fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> { let value = match store::read_path(doc, &record_path(id, "")) { Ok(Some(value)) => value, Ok(None) => return None, @@ -279,21 +354,73 @@ pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> { return None; } }; - match serde_json::from_value(value) { - Ok(record) => Some(record), + let mut record: Record = match serde_json::from_value(value) { + Ok(record) => record, Err(error) => { warn!(remake = %id, error = %error, "Skipping unreadable remake record"); - None + return None; } + }; + if let Some(commit) = contested_decision(doc, id) { + record.decision = Some(commit); + record.contested = true; } + Some(record) } -/// All readable records. -pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> { - store::keys_at(doc, &records_dir()) +/// A commit among concurrent decisions of remake `id` that disagree, if they do. +fn contested_decision(doc: &AutoCommit, id: &str) -> Option<Decision> { + let decisions: Vec<Decision> = store::read_candidates(doc, &record_path(id, "decision")) + .ok()? + .into_iter() + .filter_map(|value| serde_json::from_value(value).ok()) + .collect(); + let committed: Vec<&EpochRecord> = decisions + .iter() + .filter_map(|decision| match decision { + Decision::Committed { to_epoch, .. } => Some(to_epoch), + Decision::Aborted { .. } => None, + }) + .collect(); + let agree = committed.is_empty() + || (committed.len() == decisions.len() && committed.iter().all(|e| *e == committed[0])); + if agree { + return None; + } + decisions + .into_iter() + .find(|decision| matches!(decision, Decision::Committed { .. })) +} + +fn register_path(voter: &str) -> Vec<String> { + vec![META_ROOT.to_owned(), VOTES.to_owned(), voter.to_owned()] +} + +/// The votes register; unreadable entries are skipped with a warning. +pub(crate) fn load_register(doc: &AutoCommit) -> Register { + store::keys_at(doc, &[META_ROOT.to_owned(), VOTES.to_owned()]) .unwrap_or_default() .into_iter() - .filter_map(|id| load(doc, &id).map(|record| (id, record))) + .filter_map(|voter| { + let values = match store::read_candidates(doc, ®ister_path(&voter)) { + Ok(values) => values, + Err(error) => { + warn!(voter = %voter, error = %error, "Failed to read vote"); + return None; + } + }; + let votes: Vec<Vote> = values + .into_iter() + .filter_map(|value| match serde_json::from_value(value) { + Ok(vote) => Some(vote), + Err(error) => { + warn!(voter = %voter, error = %error, "Skipping unreadable vote"); + None + } + }) + .collect(); + (!votes.is_empty()).then_some((voter, votes)) + }) .collect() } @@ -331,16 +458,10 @@ pub(crate) fn write_abort_request( write_part(doc, id, "abortRequested", request) } -pub(crate) fn write_vote( - doc: &mut AutoCommit, - id: &str, - voter: &str, - vote: &Vote, -) -> Result<(), StoreError> { +/// Writes `voter`'s entry in the votes register (replacing its earlier one). +pub(crate) fn write_vote(doc: &mut AutoCommit, voter: &str, vote: &Vote) -> Result<(), StoreError> { let json = serde_json::to_value(vote).map_err(anyhow::Error::from)?; - let mut path = record_path(id, "votes"); - path.push(voter.to_owned()); - store::write_path(doc, &path, &json) + store::write_path(doc, ®ister_path(voter), &json) } pub(crate) fn decode_hex32(hex: &str) -> Option<[u8; 32]> { @@ -390,7 +511,7 @@ mod tests { electorate: vec!["node-a".into(), "node-b".into()], quorum: 2, kind: DEFAULT_KIND.to_owned(), - content: JsonValue::Null, + content: None, translation: JsonValue::Null, } } @@ -418,7 +539,12 @@ mod tests { decided_at: 11_000, }; write_decision(&mut doc, "c-1", &decision).unwrap(); - write_vote(&mut doc, "c-1", "node-b", &Vote { at: 7 }).unwrap(); + let vote = |remake: &str| Vote { + remake: remake.into(), + at: 7, + }; + write_vote(&mut doc, "node-b", &vote("c-1")).unwrap(); + write_vote(&mut doc, "node-c", &vote("other")).unwrap(); doc.commit(); let all = load_all(&doc); @@ -434,11 +560,41 @@ mod tests { } #[test] + fn concurrent_votes_are_kept_apart() { + let mut a = AutoCommit::new(); + let vote = |remake: &str, at| Vote { + remake: remake.into(), + at, + }; + write_vote(&mut a, "node-a", &vote("x", 1)).unwrap(); + a.commit(); + let mut b = a.fork(); + // node-b votes in both, never seeing its other vote (a lost vote). + write_vote(&mut a, "node-b", &vote("x", 2)).unwrap(); + write_vote(&mut b, "node-b", &vote("y", 3)).unwrap(); + a.commit(); + b.commit(); + a.merge(&mut b).unwrap(); + + let register = load_register(&a); + assert_eq!(register["node-a"], [vote("x", 1)]); + let mut twice = register["node-b"].clone(); + twice.sort_by_key(|vote| vote.at); + assert_eq!(twice, [vote("x", 2), vote("y", 3)]); + + // Seen, a vote is replaced. + write_vote(&mut a, "node-b", &vote("z", 4)).unwrap(); + a.commit(); + assert_eq!(load_register(&a)["node-b"], [vote("z", 4)]); + } + + #[test] fn phases() { let mut record = Record { proposal: proposal(10_000), decision: None, abort_requested: None, + contested: false, votes: BTreeMap::new(), }; assert_eq!(record.phase(10_000, 500), Phase::Proposed); @@ -452,6 +608,45 @@ mod tests { } #[test] + fn disagreeing_decisions_are_contested() { + let mut base = AutoCommit::new(); + write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap(); + base.commit(); + let (mut a, mut b, mut c) = (base.fork(), base.fork(), base.fork()); + let committed = Decision::Committed { + to_epoch: EpochRecord { + generation: 1, + id: "cd".repeat(32), + }, + snapshot_bytes: 1, + decided_at: 2, + }; + write_decision(&mut a, "c-1", &committed).unwrap(); + write_decision( + &mut b, + "c-1", + &Decision::Aborted { + reason: "test".into(), + decided_at: 3, + }, + ) + .unwrap(); + // The same decision, concurrently: they agree. + write_decision(&mut c, "c-1", &committed).unwrap(); + a.commit(); + b.commit(); + c.commit(); + let mut agreeing = a.fork(); + agreeing.merge(&mut c).unwrap(); + assert!(!load(&agreeing, "c-1").unwrap().contested); + + a.merge(&mut b).unwrap(); + let record = load(&a, "c-1").unwrap(); + assert!(record.contested); + assert_eq!(record.phase(0, 0), Phase::Committed); + } + + #[test] fn unreadable_records_are_skipped() { let mut doc = AutoCommit::new(); store::write_path( diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs index efe7e37..e50a2a2 100644 --- a/src/remake/snapshot.rs +++ b/src/remake/snapshot.rs @@ -35,6 +35,8 @@ use crate::document::store::{self, StoreError}; pub(crate) const META_ROOT: &str = "_continuum"; pub(crate) const REMAKES: &str = "remakes"; +/// The votes register (`records`). +pub(crate) const VOTES: &str = "votes"; pub(crate) const EPOCHS: &str = "epochs"; /// Everything that determines a snapshot besides the base content. @@ -51,8 +53,8 @@ pub(crate) struct SnapshotParams<'a> { pub(crate) proposed_at: u64, /// Name of the remake kind, which builds the user content. pub(crate) kind: &'a str, - /// The kind's build input, from the proposal. - pub(crate) content: &'a JsonValue, + /// The kind's build input (the proposal's content). + pub(crate) content: Option<&'a [u8]>, /// The kind's translation input; recorded in the epoch metadata. pub(crate) translation: &'a JsonValue, } @@ -92,7 +94,7 @@ pub(crate) struct SnapshotRequest { pub(crate) policy: JsonValue, pub(crate) proposed_at: u64, pub(crate) kind: String, - pub(crate) content: JsonValue, + pub(crate) content: Option<std::sync::Arc<Vec<u8>>>, pub(crate) translation: JsonValue, } @@ -105,7 +107,7 @@ impl SnapshotRequest { policy: &self.policy, proposed_at: self.proposed_at, kind: &self.kind, - content: &self.content, + content: self.content.as_deref().map(Vec::as_slice), translation: &self.translation, } } @@ -265,15 +267,18 @@ pub(crate) fn build( kind.build(&base, params.content, &mut snapshot)?; // Protocol data carries forward, whatever the kind: forget records, earlier epochs' - // metadata (rejoining nodes need it). Remake records describe the old epoch's - // protocol and stay behind. + // metadata (rejoining nodes need it). Remake records and the votes register describe + // the old epoch's protocol and stay behind. match store::delete_path(&mut snapshot, &[META_ROOT.to_owned()], false) { Ok(()) | Err(StoreError::NotFound) => {} Err(error) => return Err(error.into()), } let mut protocol_keys = store::keys_at(&base, &[META_ROOT.to_owned()])?; protocol_keys.sort(); - for key in protocol_keys.into_iter().filter(|key| key != REMAKES) { + for key in protocol_keys + .into_iter() + .filter(|key| key != REMAKES && key != VOTES) + { store::copy_path(&base, &mut snapshot, &[META_ROOT.to_owned(), key])?; } @@ -367,7 +372,7 @@ mod tests { policy, proposed_at: 0, kind: "compact", - content: &NULL, + content: None, translation: &NULL, } } @@ -526,6 +531,12 @@ mod tests { .unwrap(); store::write_path( &mut a, + &[META_ROOT.into(), VOTES.into(), "pk".into()], + &json!({ "remake": "c-1", "at": 1 }), + ) + .unwrap(); + store::write_path( + &mut a, &[META_ROOT.into(), "forgotten".into(), "pk".into()], &json!({ "at": 1 }), ) @@ -540,6 +551,7 @@ mod tests { .unwrap() .unwrap(); assert!(meta.get(REMAKES).is_none()); + assert!(meta.get(VOTES).is_none()); assert_eq!(meta["forgotten"]["pk"]["at"], json!(1)); let epoch = &meta[EPOCHS]["1"]; diff --git a/src/remake/switch.rs b/src/remake/switch.rs index e9f49ee..cf5ea75 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -26,6 +26,8 @@ use tracing::{error, info, warn}; use super::Host; use super::catchup; +use super::content; +use super::coordinator; use super::delta::{self, ReplayStats}; use super::kind::Translator; use super::local::LocalStatus; @@ -82,7 +84,16 @@ async fn check<H: Host>(host: &H) { vote::cast(host).await; let all = host.read_doc(records::load_all).await; + content::prune( + host, + &all, + unix_millis(), + host.dormant_after().as_millis() as u64, + ); let local_record = EpochRecord::from(local); + if host.is_node_ready().await { + coordinator::resume(host, &local_record, &all); + } let committed: Vec<(String, Record)> = all .into_iter() @@ -94,6 +105,10 @@ async fn check<H: Host>(host: &H) { let (id, record) = match committed.as_slice() { // Nothing committed here; maybe peers moved on without us. [] => return catchup::check(host).await, + [(id, record)] if record.contested => { + error!(remake = %id, "Decisions on this remake disagree; not switching on my own"); + return catchup::check(host).await; + } [(id, record)] => (id.clone(), record.clone()), several => { // Majority voting prevents this; never guess which one is right. Peers @@ -157,10 +172,16 @@ async fn switch<H: Host>( id: &str, record: &Record, ) -> Result<(ReplayStats, Option<String>), String> { - let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id)) + let (Some(expected), Some(mut request)) = + (record.committed_epoch(), record.snapshot_request(id)) else { return Err("unreadable remake record".to_owned()); }; + // Voters fetched it already; a node that didn't vote fetches it now. Without it the + // switch fails, and the node rejoins (downloading the snapshot instead). + if let Some(reference) = &record.proposal.content { + request.content = Some(content::ensure(host, reference).await?); + } let translator = request .translator() .ok_or_else(|| format!("unknown remake kind {:?}", request.kind))?; diff --git a/src/remake/vote.rs b/src/remake/vote.rs index d73326d..16de87a 100644 --- a/src/remake/vote.rs +++ b/src/remake/vote.rs @@ -3,25 +3,32 @@ //! //! The electorate is fixed in the proposal: the members that renewed their alive stamp //! within the coordinator's `--electorate-window-days` (`swarm::members::electorate`), -//! the coordinator always included. Each member votes for at most one open remake of -//! its epoch: the first it sees (the oldest proposal, if it sees several at once). Two -//! competing remakes, even in different partitions, can't both reach a majority, so at -//! most one commits. A vote for a remake that aborted or expired is void; the member -//! may vote again. +//! the coordinator always included. //! -//! Votes live in the record (`votes/<pk>`, written only by that member). A node votes -//! once it is ready, on its current epoch, and only if it is in the electorate. +//! Votes live in the votes register (`_continuum/votes/<pk>`, written only by that +//! member): one entry per member for the whole epoch (the snapshot leaves it behind). +//! A member votes for at most one remake of its epoch, the first it sees (the oldest +//! proposal, if it sees several at once), and votes again only once that remake is +//! aborted. Neither expiry nor a deleted record frees a vote. Two competing remakes, +//! even in different partitions, can't both reach a majority, so at most one commits. //! -//! Gap: votes are not persisted locally. A node whose vote reached only the coordinator -//! and that then restarts and resyncs from nodes that never saw the vote may vote again -//! for a competitor. +//! A node votes once it is ready, on its current epoch, only if it is in the electorate, +//! and only once it holds the remake's content (`content`), so a committed remake has +//! its content on a majority. +//! +//! A vote can be lost: a node whose vote reached only the coordinator restarts and +//! resyncs from nodes that never saw it. Should it vote again for a competitor, its +//! entry ends up with two concurrent values, and it counts for no remake. To avoid +//! that, a node doesn't vote while more than one undecided remake it could have voted +//! for is older than its start, and with one, votes only for that one. use std::collections::HashSet; use tracing::{info, warn}; use super::Host; -use super::records::{self, EpochRecord, Phase, Record, Vote, unix_millis}; +use super::content; +use super::records::{self, Decision, EpochRecord, Phase, Record, Register, Vote, unix_millis}; use crate::document::store::StoreError; /// A majority of `electorate` members. @@ -36,8 +43,10 @@ pub(crate) struct Tally { pub(crate) quorum: usize, /// Voted for it. pub(crate) yes: usize, - /// Voted for a competing open remake. + /// Voted for a competing remake that isn't aborted. pub(crate) elsewhere: usize, + /// Voted for two remakes: counts for neither. + pub(crate) twice: usize, /// Not voted (yet). pub(crate) pending: usize, } @@ -57,14 +66,51 @@ impl std::fmt::Display for Tally { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!( f, - "{} of {} voted for it, {} for a competing remake, {} not yet; {} needed", - self.yes, self.electorate, self.elsewhere, self.pending, self.quorum - ) + "{} of {} voted for it, {} for a competing remake, ", + self.yes, self.electorate, self.elsewhere + )?; + if self.twice > 0 { + write!(f, "{} for two, ", self.twice)?; + } + write!(f, "{} not yet; {} needed", self.pending, self.quorum) + } +} + +/// What a member's entry in the register amounts to. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Held<'a> { + /// No vote, or only for aborted remakes. + Free, + For(&'a str), + /// For two remakes, neither aborted. + Twice, +} + +/// What `votes` (one member's entry) amount to among `all` records. +pub(crate) fn held<'a>(votes: &'a [Vote], all: &[(String, Record)]) -> Held<'a> { + let aborted = |remake: &str| { + all.iter().any(|(id, record)| { + id == remake && matches!(record.decision, Some(Decision::Aborted { .. })) + }) + }; + let mut kept = votes + .iter() + .map(|vote| vote.remake.as_str()) + .filter(|remake| !aborted(remake)); + let Some(first) = kept.next() else { + return Held::Free; + }; + if kept.all(|other| other == first) { + Held::For(first) + } else { + Held::Twice } } -fn open(record: &Record, now: u64, grace: u64) -> bool { - record.phase(now, grace).is_open() +fn held_by<'a>(member: &str, register: &'a Register, all: &[(String, Record)]) -> Held<'a> { + register + .get(member) + .map_or(Held::Free, |votes| held(votes, all)) } /// The tally of remake `id` among `all` records. @@ -72,8 +118,7 @@ pub(crate) fn tally( id: &str, record: &Record, all: &[(String, Record)], - now: u64, - grace: u64, + register: &Register, ) -> Tally { let electorate: HashSet<&str> = record .proposal @@ -81,51 +126,95 @@ pub(crate) fn tally( .iter() .map(String::as_str) .collect(); - let yes: HashSet<&str> = record - .votes - .keys() - .map(String::as_str) - .filter(|voter| electorate.contains(voter)) - .collect(); - let elsewhere: HashSet<&str> = all - .iter() - .filter(|(other, candidate)| { - other != id - && candidate.proposal.from_epoch == record.proposal.from_epoch - && open(candidate, now, grace) - }) - .flat_map(|(_, candidate)| candidate.votes.keys().map(String::as_str)) - .filter(|voter| electorate.contains(voter) && !yes.contains(voter)) - .collect(); + let (mut yes, mut elsewhere, mut twice) = (0, 0, 0); + for member in &electorate { + match held_by(member, register, all) { + Held::For(remake) if remake == id => yes += 1, + Held::For(_) => elsewhere += 1, + Held::Twice => twice += 1, + Held::Free => {} + } + } Tally { electorate: electorate.len(), quorum: record.proposal.quorum, - yes: yes.len(), - elsewhere: elsewhere.len(), - pending: electorate.len() - yes.len() - elsewhere.len(), + yes, + elsewhere, + twice, + pending: electorate.len() - yes - elsewhere - twice, } } -/// The remake `me` should vote for now, if any: none if it already voted for an open -/// remake of `epoch`; otherwise the oldest proposed one whose electorate includes it. +/// What `me` may vote for in `epoch`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum Freedom { + Any, + /// Only this remake: the one it may have voted for before its start. + Only(String), + /// Nothing, for this reason. + Held(String), +} + +/// What `me` may vote for in `epoch`. Remakes proposed before `started` (this node's +/// start, plus a margin for clocks) may hold a vote of ours that we lost in a restart. +pub(crate) fn freedom( + me: &str, + epoch: &EpochRecord, + all: &[(String, Record)], + register: &Register, + started: u64, +) -> Freedom { + match held_by(me, register, all) { + Held::Free => {} + Held::For(remake) => { + return Freedom::Held(format!("this node voted for remake {remake}")); + } + Held::Twice => { + return Freedom::Held("this node voted for two remakes".to_owned()); + } + } + let mut older = all + .iter() + .filter(|(_, record)| { + &record.proposal.from_epoch == epoch + && record.decision.is_none() + && record.proposal.proposed_at <= started + && record.proposal.electorate.iter().any(|member| member == me) + }) + .map(|(id, _)| id); + match (older.next(), older.next()) { + (None, _) => Freedom::Any, + (Some(id), None) => Freedom::Only(id.clone()), + (Some(_), Some(_)) => Freedom::Held( + "several undecided remakes are older than this node's start; it may have voted \ + for one before" + .to_owned(), + ), + } +} + +/// The remake `me` should vote for now, if any: the oldest proposed one of `epoch` +/// whose electorate includes it, as far as it is free to vote (`freedom`). pub(crate) fn choice( me: &str, epoch: &EpochRecord, all: &[(String, Record)], + register: &Register, + started: u64, now: u64, grace: u64, ) -> Option<String> { - let of_epoch = || { - all.iter() - .filter(move |(_, record)| &record.proposal.from_epoch == epoch) + let only = match freedom(me, epoch, all, register, started) { + Freedom::Any => None, + Freedom::Only(id) => Some(id), + Freedom::Held(_) => return None, }; - if of_epoch().any(|(_, record)| open(record, now, grace) && record.votes.contains_key(me)) { - return None; - } - of_epoch() - .filter(|(_, record)| { - record.phase(now, grace) == Phase::Proposed + all.iter() + .filter(|(id, record)| { + &record.proposal.from_epoch == epoch + && record.phase(now, grace) == Phase::Proposed && record.proposal.electorate.iter().any(|member| member == me) + && only.as_ref().is_none_or(|only| only == id) }) .min_by(|(a_id, a), (b_id, b)| { (a.proposal.proposed_at, a_id).cmp(&(b.proposal.proposed_at, b_id)) @@ -142,23 +231,38 @@ pub(super) async fn cast<H: Host>(host: &H) { let me = host.node_id(); let epoch = EpochRecord::from(host.current_epoch()); let grace = host.dormant_after().as_millis() as u64; + let started = host.remake().started_at() + grace; // Usually there is nothing to vote on: look first, without writing. let (voter, wanted) = (me.clone(), epoch.clone()); let due = host .read_doc(move |doc| { - choice( + let all = records::load_all(doc); + let register = records::load_register(doc); + let id = choice( &voter, &wanted, - &records::load_all(doc), + &all, + ®ister, + started, unix_millis(), grace, - ) - .is_some() + )?; + let content = all + .into_iter() + .find(|(other, _)| *other == id) + .and_then(|(_, record)| record.proposal.content); + Some((id, content)) }) .await; - if !due { + let Some((chosen, content)) = due else { return; + }; + // Only with the content in hand: a committed remake must have it on a majority. + if let Some(reference) = &content { + if content::available_or_fetch(host, reference).is_none() { + return; + } } // Decided again under the write lock, so two passes can't both vote. @@ -167,10 +271,17 @@ pub(super) async fn cast<H: Host>(host: &H) { .commit_doc(move |doc| -> Result<Option<String>, StoreError> { let now = unix_millis(); let all = records::load_all(doc); - let Some(id) = choice(&voter, &epoch, &all, now, grace) else { + let register = records::load_register(doc); + let Some(id) = choice(&voter, &epoch, &all, ®ister, started, now, grace) + .filter(|id| *id == chosen) + else { return Ok(None); }; - records::write_vote(doc, &id, &voter, &Vote { at: now })?; + let vote = Vote { + remake: id.clone(), + at: now, + }; + records::write_vote(doc, &voter, &vote)?; Ok(Some(id)) }) .await; @@ -185,11 +296,10 @@ pub(super) async fn cast<H: Host>(host: &H) { mod tests { use super::*; use crate::document::replica::Epoch; - use crate::remake::records::{Decision, Policy, Proposal}; + use crate::remake::records::{Policy, Proposal}; use serde_json::Value as JsonValue; - use std::collections::BTreeMap; - fn record(proposed_at: u64, electorate: &[&str], votes: &[&str]) -> Record { + fn record(proposed_at: u64, electorate: &[&str]) -> Record { Record { proposal: Proposal { coordinator: electorate[0].to_owned(), @@ -206,19 +316,48 @@ mod tests { electorate: electorate.iter().map(|m| (*m).to_owned()).collect(), quorum: quorum(electorate.len()), kind: "compact".to_owned(), - content: JsonValue::Null, + content: None, translation: JsonValue::Null, }, decision: None, abort_requested: None, - votes: votes - .iter() - .map(|v| ((*v).to_owned(), Vote { at: proposed_at })) - .collect::<BTreeMap<_, _>>(), + contested: false, + votes: Default::default(), } } + /// A register from (voter, remakes voted for) pairs. + fn register(votes: &[(&str, &[&str])]) -> Register { + votes + .iter() + .map(|(voter, remakes)| { + let votes = remakes + .iter() + .map(|remake| Vote { + remake: (*remake).to_owned(), + at: 0, + }) + .collect(); + ((*voter).to_owned(), votes) + }) + .collect() + } + + fn aborted(mut record: Record) -> Record { + record.decision = Some(Decision::Aborted { + reason: "test".into(), + decided_at: 150, + }); + record + } + const ALL: [&str; 5] = ["a", "b", "c", "d", "e"]; + /// Long before any proposal: no restart concerns. + const STARTED: u64 = 0; + + fn genesis() -> EpochRecord { + Epoch::GENESIS.into() + } #[test] fn majorities() { @@ -232,60 +371,145 @@ mod tests { #[test] fn competing_remakes_split_the_vote() { let all = vec![ - ("x".to_owned(), record(100, &ALL, &["a", "b"])), - ("y".to_owned(), record(200, &ALL, &["c", "z"])), // z isn't in the electorate + ("x".to_owned(), record(100, &ALL)), + ("y".to_owned(), record(200, &ALL)), ]; - let x = tally("x", &all[0].1, &all, 150, 0); + // z isn't in the electorate. + let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["y"]), ("z", &["y"])]); + let x = tally("x", &all[0].1, &all, &votes); assert_eq!((x.yes, x.elsewhere, x.pending, x.quorum), (2, 1, 2, 3)); assert!(!x.reached() && !x.out_of_reach()); - let y = tally("y", &all[1].1, &all, 150, 0); + let y = tally("y", &all[1].1, &all, &votes); assert_eq!((y.yes, y.elsewhere, y.pending), (1, 2, 2)); assert!(!y.out_of_reach()); // d votes for x: x has its majority, y can't get one any more. - let all = vec![ - ("x".to_owned(), record(100, &ALL, &["a", "b", "d"])), - ("y".to_owned(), record(200, &ALL, &["c"])), - ]; - assert!(tally("x", &all[0].1, &all, 150, 0).reached()); - assert!(tally("y", &all[1].1, &all, 150, 0).out_of_reach()); + let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["y"]), ("d", &["x"])]); + assert!(tally("x", &all[0].1, &all, &votes).reached()); + assert!(tally("y", &all[1].1, &all, &votes).out_of_reach()); } #[test] fn votes_for_aborted_remakes_are_void() { - let mut aborted = record(100, &ALL, &["a", "b", "c"]); - aborted.decision = Some(Decision::Aborted { - reason: "test".into(), - decided_at: 150, - }); let all = vec![ - ("x".to_owned(), aborted), - ("y".to_owned(), record(200, &ALL, &["d"])), + ("x".to_owned(), aborted(record(100, &ALL))), + ("y".to_owned(), record(200, &ALL)), ]; - let y = tally("y", &all[1].1, &all, 250, 0); + let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["x"]), ("d", &["y"])]); + let y = tally("y", &all[1].1, &all, &votes); assert_eq!((y.yes, y.elsewhere, y.pending), (1, 0, 4)); // c may vote again. assert_eq!( - choice("c", &Epoch::GENESIS.into(), &all, 250, 0).as_deref(), + choice("c", &genesis(), &all, &votes, STARTED, 250, 0).as_deref(), Some("y") ); } #[test] + fn expiry_frees_no_vote() { + // x's coordinator is gone: x expired, but a, b and c still hold their votes. + let all = vec![ + ("x".to_owned(), record(100, &ALL)), + ("y".to_owned(), record(5_000, &ALL)), + ]; + let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["x"])]); + assert_eq!(all[0].1.phase(5_000, 0), Phase::Expired); + assert_eq!( + choice("a", &genesis(), &all, &votes, STARTED, 5_000, 0), + None + ); + let y = tally("y", &all[1].1, &all, &votes); + assert_eq!((y.yes, y.elsewhere, y.pending), (0, 3, 2)); + assert!(y.out_of_reach()); + } + + #[test] + fn voting_twice_counts_for_neither() { + let all = vec![ + ("x".to_owned(), record(100, &ALL)), + ("y".to_owned(), record(200, &ALL)), + ]; + let votes = register(&[("a", &["x", "y"]), ("b", &["x"]), ("c", &["x", "x"])]); + let x = tally("x", &all[0].1, &all, &votes); + assert_eq!((x.yes, x.elsewhere, x.twice, x.pending), (2, 0, 1, 2)); + assert_eq!( + x.to_string(), + "2 of 5 voted for it, 0 for a competing remake, 1 for two, 2 not yet; 3 needed" + ); + assert!(matches!( + freedom("a", &genesis(), &all, &votes, STARTED), + Freedom::Held(_) + )); + + // Once y is aborted, a's vote for x stands. + let all = vec![ + ("x".to_owned(), record(100, &ALL)), + ("y".to_owned(), aborted(record(200, &ALL))), + ]; + assert_eq!(tally("x", &all[0].1, &all, &votes).yes, 3); + } + + #[test] fn one_vote_per_epoch_for_the_oldest() { let all = vec![ - ("y".to_owned(), record(200, &ALL, &[])), - ("x".to_owned(), record(100, &ALL, &[])), + ("y".to_owned(), record(200, &ALL)), + ("x".to_owned(), record(100, &ALL)), ]; - let epoch: EpochRecord = Epoch::GENESIS.into(); - assert_eq!(choice("a", &epoch, &all, 150, 0).as_deref(), Some("x")); - assert_eq!(choice("outsider", &epoch, &all, 150, 0), None); + let none = register(&[]); + assert_eq!( + choice("a", &genesis(), &all, &none, STARTED, 150, 0).as_deref(), + Some("x") + ); + assert_eq!( + choice("outsider", &genesis(), &all, &none, STARTED, 150, 0), + None + ); + + let voted = register(&[("a", &["y"])]); + assert_eq!(choice("a", &genesis(), &all, &voted, STARTED, 150, 0), None); + } - let voted = vec![ - ("y".to_owned(), record(200, &ALL, &["a"])), - ("x".to_owned(), record(100, &ALL, &[])), + #[test] + fn after_a_start_only_the_remake_voted_for_before() { + let started = 150; + // Proposed before the start: a may have voted for x then. + let all = vec![ + ("x".to_owned(), record(100, &ALL)), + ("y".to_owned(), record(200, &ALL)), + ]; + let none = register(&[]); + assert_eq!( + freedom("a", &genesis(), &all, &none, started), + Freedom::Only("x".to_owned()) + ); + assert_eq!( + choice("a", &genesis(), &all, &none, started, 250, 1_000).as_deref(), + Some("x") + ); + + // Two of them: a may have voted for either, so for none. + let all = vec![ + ("w".to_owned(), record(50, &ALL)), + ("x".to_owned(), record(100, &ALL)), + ("y".to_owned(), record(200, &ALL)), + ]; + assert!(matches!( + freedom("a", &genesis(), &all, &none, started), + Freedom::Held(_) + )); + assert_eq!( + choice("a", &genesis(), &all, &none, started, 250, 1_000), + None + ); + + // Once they are decided, a is free again; it couldn't vote for an outsider's. + let all = vec![ + ("w".to_owned(), aborted(record(50, &ALL))), + ("x".to_owned(), aborted(record(100, &ALL))), + ("y".to_owned(), record(200, &ALL)), + ("z".to_owned(), record(120, &["b", "c"])), ]; - assert_eq!(choice("a", &epoch, &voted, 150, 0), None); + assert_eq!(freedom("a", &genesis(), &all, &none, started), Freedom::Any); } } diff --git a/src/state.rs b/src/state.rs index df053a6..785eee8 100644 --- a/src/state.rs +++ b/src/state.rs @@ -583,8 +583,8 @@ 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. @@ -597,6 +597,11 @@ pub(crate) struct AppState { 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>, @@ -608,6 +613,9 @@ pub(crate) struct AppState { 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>>>, @@ -638,6 +646,18 @@ 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() @@ -798,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 @@ -835,6 +860,7 @@ impl AppState { replica.doc.rollback(); } } + replica.refresh_heads(); result }; // lock released before notifying @@ -927,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, @@ -978,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"); @@ -1029,6 +1055,29 @@ impl Host for AppState { 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 } @@ -1051,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, }); } @@ -1155,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); diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs new file mode 100644 index 0000000..8c891ae --- /dev/null +++ b/src/swarm/blob.rs @@ -0,0 +1,508 @@ +//! Blobs: content-addressed bytes fetched in chunks from several peers at once. +//! +//! Large transfers can't go in one Mycelium message (it must arrive within Mycelium's +//! five-minute deadline), and a blob every node holds identically can come from many +//! of them. A blob is an Automerge change by its hash (e.g. an epoch's snapshot change, +//! which every node on that epoch holds byte for byte), or content by its BLAKE3 hash +//! (e.g. the new document of a `replace` or `import` remake). +//! +//! * Serving: a request without chunks gets the manifest (size, chunk size, a BLAKE3 +//! hash per chunk); one with chunks gets each chunk as a message of its own. A node +//! that doesn't hold the blob answers `Missing`. The bytes of the blob served last +//! are kept, so serving a download doesn't extract them for every chunk. Serving runs +//! in tasks of its own, at most [`MAX_SERVING`] chunk sends at a time. +//! * Fetching: the manifest from the first peer that has it (closest first), then the +//! chunks, a few in flight per peer from up to [`MAX_SOURCES`] peers. Every chunk is +//! checked against the manifest; a peer that sends a bad one is dropped, a chunk +//! that doesn't arrive in time is asked of another peer. The caller checks the whole +//! (for a change: its hash). +//! +//! The chunk size is the serving node's (`--blob-chunk-kib`); the manifest carries it. + +use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use anyhow::{Result, anyhow, bail}; +use automerge::ChangeHash; +use tokio::sync::{Semaphore, mpsc}; +use tokio::time::Instant; +use tracing::{debug, info, warn}; + +use crate::state::AppState; +use crate::transport::{BlobBody, BlobId, BlobMessage, ParsedMessage}; + +/// Chunk sends in progress at once, over all downloads served. +const MAX_SERVING: usize = 8; +/// Peers a download fetches chunks from at once. +const MAX_SOURCES: usize = 4; +/// Chunks in flight per peer. +const WINDOW: usize = 4; +/// Largest blob a download accepts. +const MAX_SIZE: u64 = 4 << 30; +/// Largest chunk a download accepts. +const MAX_CHUNK: u32 = 4 << 20; +/// How long to wait for a manifest from one peer. +const MANIFEST_TIMEOUT: Duration = Duration::from_secs(15); +/// How long to wait for a chunk before asking another peer (raised by the peer's +/// round-trip time). +const CHUNK_TIMEOUT: Duration = Duration::from_secs(60); +/// A download gives up after this long without any chunk arriving. +const STALL_TIMEOUT: Duration = Duration::from_secs(180); +/// Bad chunks after which a peer is no longer asked. +const MAX_BAD: u32 = 2; + +/// Per node: what is being served and fetched. +pub(crate) struct Blobs { + chunk_size: u32, + serving: Arc<Semaphore>, + /// The blob served last. + cache: Mutex<Option<(BlobId, Arc<Prepared>)>>, + /// Answers for downloads in progress, by blob. + fetches: Mutex<HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>>, +} + +struct Prepared { + bytes: Arc<Vec<u8>>, + hashes: Vec<[u8; 32]>, +} + +impl Blobs { + pub(crate) fn new(chunk_size: u32) -> Self { + Self { + chunk_size: chunk_size.max(1), + serving: Arc::new(Semaphore::new(MAX_SERVING)), + cache: Mutex::new(None), + fetches: Mutex::new(HashMap::new()), + } + } + + fn fetches( + &self, + ) -> MutexGuard<'_, HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>> { + self.fetches + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn cached(&self, id: BlobId) -> Option<Arc<Prepared>> { + self.cache + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .as_ref() + .filter(|(cached, _)| *cached == id) + .map(|(_, prepared)| Arc::clone(prepared)) + } + + fn remember(&self, id: BlobId, prepared: Arc<Prepared>) { + *self + .cache + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some((id, prepared)); + } +} + +fn chunk_hashes(bytes: &[u8], chunk_size: u32) -> Vec<[u8; 32]> { + bytes + .chunks(chunk_size as usize) + .map(|chunk| *blake3::hash(chunk).as_bytes()) + .collect() +} + +fn chunk_count(size: u64, chunk_size: u32) -> u64 { + size.div_ceil(u64::from(chunk_size)) +} + +/// A blob message from `peer`. +pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) { + match message.body { + BlobBody::Request { chunks } => { + let state = state.clone(); + let peer = peer.to_owned(); + // Off the inbound loop: answering may send many large messages. + tokio::spawn(async move { serve(&state, &peer, message.id, chunks).await }); + } + body => { + let sender = state.blobs.fetches().get(&message.id).cloned(); + match sender { + Some(sender) => { + let _ = sender.send((peer.to_owned(), body)); + } + None => { + debug!(peer = %peer, blob = %message.id, "Blob answer for no download in progress") + } + } + } + } +} + +async fn prepare(state: &AppState, id: BlobId) -> Option<Arc<Prepared>> { + if let Some(prepared) = state.blobs.cached(id) { + return Some(prepared); + } + let bytes = match id { + BlobId::Change(hash) => { + let doc = Arc::clone(&state.doc); + tokio::task::spawn_blocking(move || { + // get_change_by_hash needs the document mutably. + let mut replica = doc.blocking_write(); + replica + .doc + .get_change_by_hash(&hash) + .map(|change| change.raw_bytes().to_vec()) + }) + .await + .ok() + .flatten() + .map(Arc::new)? + } + BlobId::Content(hash) => state.remake.content(&hash)?, + }; + let chunk_size = state.blobs.chunk_size; + let prepared = tokio::task::spawn_blocking(move || { + let hashes = chunk_hashes(&bytes, chunk_size); + Prepared { bytes, hashes } + }) + .await + .ok()?; + let prepared = Arc::new(prepared); + state.blobs.remember(id, Arc::clone(&prepared)); + Some(prepared) +} + +async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) { + let message = ParsedMessage::Blob(BlobMessage { + epoch: state.local_epoch(), + id, + body, + }); + if let Err(error) = state.transport.send_message(peer, &message).await { + debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob answer failed"); + } +} + +async fn serve(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) { + let Some(prepared) = prepare(state, id).await else { + return reply(state, peer, id, BlobBody::Missing).await; + }; + let chunk_size = state.blobs.chunk_size; + if chunks.is_empty() { + let body = BlobBody::Manifest { + size: prepared.bytes.len() as u64, + chunk_size, + hashes: prepared.hashes.clone(), + }; + return reply(state, peer, id, body).await; + } + for index in chunks.into_iter().take(WINDOW * 2) { + let start = index as usize * chunk_size as usize; + if start >= prepared.bytes.len() { + continue; + } + let end = (start + chunk_size as usize).min(prepared.bytes.len()); + let Ok(_permit) = state.blobs.serving.acquire().await else { + return; + }; + let data = bytes::Bytes::copy_from_slice(&prepared.bytes[start..end]); + reply(state, peer, id, BlobBody::Chunk { index, data }).await; + } +} + +// --- fetching ---------------------------------------------------------------------- + +/// Removes the download's answer channel when it ends, however it ends. +struct Registration<'a> { + state: &'a AppState, + id: BlobId, +} + +impl Drop for Registration<'_> { + fn drop(&mut self) { + self.state.blobs.fetches().remove(&self.id); + } +} + +async fn request(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) { + let message = ParsedMessage::Blob(BlobMessage { + epoch: state.local_epoch(), + id, + body: BlobBody::Request { chunks }, + }); + if let Err(error) = state.transport.send_message(peer, &message).await { + debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob request failed"); + } +} + +struct Manifest { + size: u64, + chunk_size: u32, + hashes: Vec<[u8; 32]>, +} + +/// One peer during a download. +struct Source { + peer: String, + in_flight: HashMap<u32, Instant>, + window: usize, + bad: u32, + timeout: Duration, +} + +/// Downloads blob `id` from `peers` (all thought to hold it). +pub(crate) async fn fetch(state: &AppState, id: BlobId, peers: Vec<String>) -> Result<Vec<u8>> { + let (sender, mut answers) = mpsc::unbounded_channel(); + if state.blobs.fetches().insert(id, sender).is_some() { + bail!("{id} is being downloaded already"); + } + let _registration = Registration { state, id }; + + // Closest first; peers without a round-trip time last. + let now = Instant::now(); + let mut peers: Vec<(String, Option<Duration>)> = { + let rtt = state.lock_rtt(); + peers + .into_iter() + .map(|peer| { + let estimate = rtt.effective(&peer, now, state.timing.rtt_fresh_for()); + (peer, estimate) + }) + .collect() + }; + peers.sort_by_key(|(_, rtt)| rtt.unwrap_or(Duration::MAX)); + let rtts: HashMap<String, Option<Duration>> = peers.iter().cloned().collect(); + let mut candidates: VecDeque<String> = peers.into_iter().map(|(peer, _)| peer).collect(); + if candidates.is_empty() { + bail!("no peer to download {id} from"); + } + + // The manifest, from the first peer that has the blob. + let mut manifest = None; + let mut missing: HashSet<String> = HashSet::new(); + for peer in &candidates { + request(state, peer, id, Vec::new()).await; + let deadline = Instant::now() + MANIFEST_TIMEOUT; + loop { + let answer = tokio::time::timeout_at(deadline, answers.recv()).await; + match answer { + Ok(Some(( + from, + BlobBody::Manifest { + size, + chunk_size, + hashes, + }, + ))) => { + if size > MAX_SIZE + || chunk_size == 0 + || chunk_size > MAX_CHUNK + || hashes.len() as u64 != chunk_count(size, chunk_size) + { + warn!(peer = %from, blob = %id, size, chunk_size, "Unusable blob manifest; asking another peer"); + missing.insert(from); + break; + } + manifest = Some(Manifest { + size, + chunk_size, + hashes, + }); + break; + } + Ok(Some((from, BlobBody::Missing))) => { + missing.insert(from.clone()); + if &from == peer { + break; + } + } + Ok(Some(_)) => {} // chunks of an earlier attempt + Ok(None) | Err(_) => break, + } + } + if manifest.is_some() { + break; + } + } + let Some(manifest) = manifest else { + bail!("no peer sent a manifest for {id}"); + }; + candidates.retain(|peer| !missing.contains(peer)); + + let total = manifest.hashes.len(); + info!(blob = %id, bytes = manifest.size, chunks = total, peers = candidates.len(), "Downloading blob"); + let started = Instant::now(); + let mut chunks: Vec<Option<bytes::Bytes>> = vec![None; total]; + let mut done = 0usize; + let mut sources: Vec<Source> = Vec::new(); + let mut last_progress = Instant::now(); + + while done < total { + // Fill up the sources, then their windows. + while sources.len() < MAX_SOURCES { + let Some(peer) = candidates.pop_front() else { + break; + }; + let timeout = + CHUNK_TIMEOUT + rtts.get(&peer).copied().flatten().unwrap_or_default() * 4; + sources.push(Source { + peer, + in_flight: HashMap::new(), + window: WINDOW, + bad: 0, + timeout, + }); + } + if sources.is_empty() { + bail!("no peer left to download {id} from ({done} of {total} chunks)"); + } + let now = Instant::now(); + let assigned: HashSet<u32> = sources + .iter() + .flat_map(|source| source.in_flight.keys().copied()) + .collect(); + let mut wanted = (0..total as u32) + .filter(|index| chunks[*index as usize].is_none() && !assigned.contains(index)); + for source in &mut sources { + let free = source.window.saturating_sub(source.in_flight.len()); + let batch: Vec<u32> = wanted.by_ref().take(free).collect(); + if batch.is_empty() { + continue; + } + for index in &batch { + source.in_flight.insert(*index, now); + } + request(state, &source.peer, id, batch).await; + } + + match tokio::time::timeout(Duration::from_millis(500), answers.recv()).await { + Ok(Some((from, BlobBody::Chunk { index, data }))) => { + let Some(source) = sources.iter_mut().find(|source| source.peer == from) else { + continue; + }; + let sent_at = source.in_flight.remove(&index); + let slot = index as usize; + if slot >= total || chunks[slot].is_some() { + continue; + } + let expected = expected_len(&manifest, slot); + if data.len() != expected + || *blake3::hash(&data).as_bytes() != manifest.hashes[slot] + { + source.bad += 1; + warn!(peer = %from, blob = %id, index, "Bad blob chunk"); + continue; + } + if let Some(sent_at) = sent_at { + debug!(peer = %from, blob = %id, index, ms = sent_at.elapsed().as_millis() as u64, "Blob chunk"); + } + chunks[slot] = Some(data); + done += 1; + last_progress = Instant::now(); + // A peer that delivers gets its full window back. + source.window = WINDOW; + } + Ok(Some((from, BlobBody::Missing))) => { + sources.retain(|source| source.peer != from); + } + Ok(Some(_)) => {} + Ok(None) => bail!("download of {id} cancelled"), + Err(_) => {} + } + + // Chunks that took too long go to another peer; the slow peer gets a smaller + // window. Peers that sent bad chunks are dropped. + let now = Instant::now(); + for source in &mut sources { + let timeout = source.timeout; + let before = source.in_flight.len(); + source + .in_flight + .retain(|_, sent| now.saturating_duration_since(*sent) < timeout); + if source.in_flight.len() < before { + source.window = 1; + debug!(peer = %source.peer, blob = %id, "Blob chunks timed out; asking others"); + } + } + sources.retain(|source| { + let keep = source.bad < MAX_BAD; + if !keep { + warn!(peer = %source.peer, blob = %id, "Dropping a peer that sent bad blob chunks"); + } + keep + }); + if last_progress.elapsed() > STALL_TIMEOUT { + bail!( + "download of {id} stalled ({done} of {total} chunks, nothing for {}s)", + STALL_TIMEOUT.as_secs() + ); + } + } + + let mut bytes = Vec::with_capacity(manifest.size as usize); + for chunk in chunks.into_iter().flatten() { + bytes.extend_from_slice(&chunk); + } + info!( + blob = %id, + bytes = bytes.len(), + secs = started.elapsed().as_secs_f32(), + "Downloaded blob" + ); + Ok(bytes) +} + +fn expected_len(manifest: &Manifest, slot: usize) -> usize { + let chunk_size = manifest.chunk_size as u64; + let start = slot as u64 * chunk_size; + (manifest.size - start).min(chunk_size) as usize +} + +/// Downloads change `hash` from `peers` and checks it. +/// Downloads the content with BLAKE3 hash `hash` from `peers` and checks it. +pub(crate) async fn fetch_content( + state: &AppState, + hash: [u8; 32], + peers: Vec<String>, +) -> Result<Vec<u8>> { + let id = BlobId::Content(hash); + let bytes = fetch(state, id, peers).await?; + if *blake3::hash(&bytes).as_bytes() != hash { + bail!("downloaded {id} doesn't match its hash"); + } + Ok(bytes) +} + +pub(crate) async fn fetch_change( + state: &AppState, + hash: ChangeHash, + peers: Vec<String>, +) -> Result<automerge::Change> { + let bytes = fetch(state, BlobId::Change(hash), peers).await?; + let change = automerge::Change::from_bytes(bytes) + .map_err(|error| anyhow!("downloaded change {hash} doesn't load: {error}"))?; + if change.hash() != hash { + bail!("downloaded change has hash {}, not {hash}", change.hash()); + } + Ok(change) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn chunking() { + let bytes: Vec<u8> = (0..1000u32).map(|n| n as u8).collect(); + let hashes = chunk_hashes(&bytes, 300); + assert_eq!(hashes.len(), 4); + assert_eq!(chunk_count(1000, 300), 4); + assert_eq!(chunk_count(900, 300), 3); + let manifest = Manifest { + size: 1000, + chunk_size: 300, + hashes, + }; + assert_eq!(expected_len(&manifest, 0), 300); + assert_eq!(expected_len(&manifest, 3), 100); + assert_eq!(manifest.hashes[3], *blake3::hash(&bytes[900..]).as_bytes()); + } +} diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 6f194dc..adb5d1c 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -2,6 +2,7 @@ //! Automerge synchronization with every member (`sync`). pub(crate) mod auth; +pub(crate) mod blob; pub(crate) mod bootstrap; pub(crate) mod members; pub(crate) mod pair; diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 3b85539..3d7da32 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -108,12 +108,12 @@ pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> { /// Our keepalive for `peer`. pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive { - // AutoCommit::get_heads takes &mut self (it closes the pending transaction). - // Heads and epoch come from the same locked replica, so they always match. + // Heads and epoch come from the same locked replica, so they always match. The heads + // are cached (`Replica`), so the read lock is enough. let (heads, epoch) = { - let mut replica = state.doc.write().await; + let replica = state.doc.read().await; ( - transport::HeadsSummary::of(replica.doc.get_heads()), + transport::HeadsSummary::of(replica.heads().to_vec()), replica.epoch, ) }; @@ -824,6 +824,9 @@ pub(crate) async fn handle_message( debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed"); } } + transport::ParsedMessage::Blob(blob) => { + crate::swarm::blob::handle(state, &peer, blob).await; + } transport::ParsedMessage::Pong(remote) => { state.lock_rtt().pong_received(&peer, Instant::now()); if reactivated || peer_swarm(state, &peer).await.is_none() { diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 454a46a..703bc3f 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -9,8 +9,16 @@ //! receives a change about once; the rest is announcements. Regular keepalives work the //! same way, so any gap is noticed within a keepalive interval. //! +//! Announcements are coalesced: the first applied change schedules one announcement +//! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A +//! round costs one PING and one PONG per neighbour (plus the pulls), whatever the +//! write rate; the delay is the price per hop. Without coalescing, every applied sync +//! message would announce to every neighbour, and announcements would outnumber the +//! data they announce (about 15 to 1 at 250 nodes). +//! //! In a full mesh the writer reaches everyone itself, so nothing is announced. +use std::collections::HashSet; use std::sync::atomic::Ordering; use std::time::Duration; @@ -26,18 +34,64 @@ use crate::transport::{self, HeadsSummary}; /// A pull that brought nothing for this long no longer blocks pulling from another peer. const PULL_TIMEOUT: Duration = Duration::from_secs(15); +/// How long an announcement waits for further changes to ride along. Added once per +/// hop to propagation; well below a round trip over Mycelium. +const ANNOUNCE_DELAY: Duration = Duration::from_millis(200); + /// After changes from `from` were applied and our heads changed: tell our other -/// neighbours. +/// neighbours, in the next announcement round. pub(crate) async fn announce(state: &AppState, from: &str) { if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) { return; } + let schedule = { + let mut pending = state + .announce_pending + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + match pending.as_mut() { + Some(senders) => { + senders.insert(from.to_owned()); + false + } + None => { + *pending = Some(HashSet::from([from.to_owned()])); + true + } + } + }; + if !schedule { + return; + } + let state = state.clone(); + tokio::spawn(async move { + tokio::time::sleep(ANNOUNCE_DELAY).await; + let senders = state + .announce_pending + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take() + .unwrap_or_default(); + announce_now(&state, &senders).await; + }); +} + +/// One announcement round. The peer whose changes we announce has them already, so a +/// round with a single sender skips it; with several, each lacks the others' changes, +/// so every neighbour is told. +async fn announce_now(state: &AppState, senders: &HashSet<String>) { + let skip = match senders.len() { + 1 => senders.iter().next().cloned(), + _ => None, + }; let neighbours: Vec<String> = state .peer_states .read() .await .keys() - .filter(|peer| peer.as_str() != from && topology::is_neighbour(state, peer)) + .filter(|peer| { + skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer) + }) .cloned() .collect(); for peer in neighbours { @@ -52,6 +106,10 @@ pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transp if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) { return; } + // The common case, the same heads, needs only the cached heads (read lock). + if HeadsSummary::of(state.doc.read().await.heads().to_vec()) == remote.heads { + return; + } let missing = match &remote.heads { HeadsSummary::Heads(heads) => { // AutoCommit's lookups take &mut self. diff --git a/src/swarm/source.rs b/src/swarm/source.rs index 4025f0e..378c40a 100644 --- a/src/swarm/source.rs +++ b/src/swarm/source.rs @@ -199,7 +199,7 @@ pub(crate) fn finished(state: &AppState, from: &str) { } /// After giving up a source: let every peer be considered again. -fn resync_everyone(state: &AppState) { +pub(crate) fn resync_everyone(state: &AppState) { let state = state.clone(); tokio::spawn(async move { let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect(); diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 3284015..80d6d56 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -32,6 +32,12 @@ pub(crate) struct ChangeSet { } async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { + if state.remake.fetching_snapshot() { + // Rejoining: the snapshot comes as a download, not in a sync message. Sync + // starts once it is here (`remake::catchup`). + trace!("Waiting for the snapshot download; not syncing yet"); + return Ok(()); + } let Some(peer_state_mutex) = state.get_peer_state(peer).await else { trace!("Peer was forgotten; skipping synchronization"); return Ok(()); @@ -337,7 +343,7 @@ pub(crate) async fn handle_sync( let mut peer_state = peer_state_mutex.blocking_lock(); peer_state.reports_ready = Some(from_ready); - let heads_before = replica.doc.get_heads(); + let heads_before = replica.heads().to_vec(); replica .doc @@ -354,7 +360,8 @@ pub(crate) async fn handle_sync( peer_state.sync.reply_settled(); resend_if_needed(&peer, &mut peer_state.sync.state); - let heads_after = replica.doc.get_heads(); + replica.refresh_heads(); + let heads_after = replica.heads().to_vec(); stats.heads_seen(heads_after.len()); let changed = heads_after != heads_before; if changed && measure_propagation { @@ -372,8 +379,11 @@ pub(crate) async fn handle_sync( } Ok(Some(( - // Only from a ready peer: a peer still catching up has empty heads, which - // every document "contains". + // Only from a ready peer, and only if it has heads at all: empty heads are + // "contained" in every document, so they prove nothing. A peer still + // catching up has them, and so does a ready peer whose document is empty + // (a misconfigured second swarm, or a genesis node in its first seconds, + // before it registered itself). check_caught_up && from_ready && peer_state @@ -382,9 +392,10 @@ pub(crate) async fn handle_sync( .their_heads .as_ref() .is_some_and(|heads| { - heads - .iter() - .all(|h| replica.doc.get_change_by_hash(h).is_some()) + !heads.is_empty() + && heads + .iter() + .all(|h| replica.doc.get_change_by_hash(h).is_some()) }), changed, ))) diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index c5a02f1..57856b6 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -204,6 +204,13 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { const RING_LABEL: &[u8] = b"continuum/topology-ring/1"; const PAIR_LABEL: &[u8] = b"continuum/topology-pair/1"; +/// How many members may prefer the same stable member, as a multiple of `k`. Without a +/// bound, 20 stable members among 2000 get about 500 neighbours each: every +/// announcement in the swarm lands on them, and the backbone saturates first. Members +/// that find a stable member full prefer the next one, then go by pair score, so the +/// graph stays connected and symmetric. +const STABLE_FANIN_FACTOR: usize = 4; + /// The neighbours of `local` among `members` (which may or may not contain `local`), /// skipping members for which `unreachable` holds. pub(crate) fn neighbours( @@ -222,7 +229,7 @@ pub(crate) fn neighbours( /// Every reachable member's neighbours. Costs about N² hashes (a few milliseconds for /// hundreds of members). `stable` members (`swarm::stability`) are preferred for half of -/// each member's scored slots. +/// each member's scored slots, each by at most `STABLE_FANIN_FACTOR * k` members. /// /// Each reachable member walks the ring (by key hash) both ways to the first reachable /// member, which gives a ring over the reachable members, and adds the `k - 2` @@ -231,6 +238,8 @@ pub(crate) fn neighbours( /// /// Each node judges reachability by itself, so nodes may briefly disagree; that only /// adds links (whoever chose a link initiates, the other answers), it never cuts one. +/// The preference budget is spent in key order, which every node shares, so nodes that +/// agree on reachability compute the same links. pub(crate) fn graph( members: &[String], k: usize, @@ -275,6 +284,11 @@ pub(crate) fn graph( .expect("more than k + 1 live members") }; + // How many members have preferred each stable member so far (`live` iterates in key + // order, the same on every node). + let budget = STABLE_FANIN_FACTOR * k; + let mut preferred_by: BTreeMap<&str, usize> = BTreeMap::new(); + for member in &live { let at = position[member]; let next = first_live(at, 1); @@ -289,15 +303,18 @@ pub(crate) fn graph( .collect(); candidates.sort_unstable(); // Half the slots (rounded up) prefer stable members, so long-running nodes form - // a backbone; the rest go by pair score alone, keeping paths short and the load - // on stable members bounded. + // a backbone; the rest go by pair score alone, keeping paths short. A stable + // member enough others prefer already is skipped, so no hub grows with the swarm. let stable_slots = (k - 2).div_ceil(2); let preferred: Vec<&str> = candidates .iter() .map(|(_, other)| *other) - .filter(|other| stable(other)) + .filter(|other| stable(other) && preferred_by.get(other).copied().unwrap_or(0) < budget) .take(stable_slots) .collect(); + for other in preferred.iter().copied() { + *preferred_by.entry(other).or_default() += 1; + } let rest = candidates .iter() .map(|(_, other)| *other) @@ -476,4 +493,29 @@ mod tests { "a stable member has {stable_max} neighbours" ); } + + #[test] + fn hub_degree_is_bounded_however_few_stable_members() { + // Few stable members among many: without the fan-in budget each would be chosen + // by most of the swarm (about 130 neighbours with 10 of 250). With it, a hub has + // at most the budget plus the links others choose by pair score and the ring. + for (n, stable_count) in [(250, 10), (1000, 10), (1000, 20)] { + let k = 12; + let ms = members(n); + let stable: BTreeSet<String> = ms.iter().step_by(n / stable_count).cloned().collect(); + let is_stable = |m: &str| stable.contains(m); + let g = super::graph(&ms, k, &all_reachable, &is_stable); + assert!(connected(&g), "n={n} stable={stable_count}"); + let hub_max = g + .iter() + .filter(|(m, _)| stable.contains(*m)) + .map(|(_, ns)| ns.len()) + .max() + .unwrap(); + assert!( + hub_max <= 8 * k, + "n={n} stable={stable_count}: a stable member has {hub_max} neighbours" + ); + } + } } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index b2f5c9d..c5769ca 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -100,6 +100,86 @@ pub enum ParsedMessage { Sync(SyncEnvelope), Ping(Keepalive), Pong(Keepalive), + Blob(BlobMessage), +} + +/// Content-addressed bytes that any node holding them can serve, in chunks +/// (`swarm::blob`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum BlobId { + /// An Automerge change, by its hash: e.g. an epoch's snapshot change. + Change(ChangeHash), + /// Bytes by their BLAKE3 hash: e.g. the content of a remake. + Content([u8; 32]), +} + +impl std::fmt::Display for BlobId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BlobId::Change(hash) => write!(f, "change:{hash}"), + BlobId::Content(hash) => { + write!(f, "content:")?; + hash.iter().try_for_each(|byte| write!(f, "{byte:02x}")) + } + } + } +} + +/// A blob request or one of its answers. Each is a message of its own, so no message +/// is larger than one chunk. +#[derive(Debug, Clone)] +pub struct BlobMessage { + /// The sender's epoch (every message carries it). + pub epoch: Epoch, + pub id: BlobId, + pub body: BlobBody, +} + +#[derive(Clone)] +pub enum BlobBody { + /// Asks for these chunks; for the manifest if empty. + Request { + chunks: Vec<u32>, + }, + /// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter. + Manifest { + size: u64, + chunk_size: u32, + hashes: Vec<[u8; 32]>, + }, + Chunk { + index: u32, + data: bytes::Bytes, + }, + /// The sender doesn't have the blob. + Missing, +} + +/// Sizes instead of contents: logged messages stay short. +impl std::fmt::Debug for BlobBody { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BlobBody::Request { chunks } => { + f.debug_struct("Request").field("chunks", chunks).finish() + } + BlobBody::Manifest { + size, + chunk_size, + hashes, + } => f + .debug_struct("Manifest") + .field("size", size) + .field("chunk_size", chunk_size) + .field("chunks", &hashes.len()) + .finish(), + BlobBody::Chunk { index, data } => f + .debug_struct("Chunk") + .field("index", index) + .field("bytes", &data.len()) + .finish(), + BlobBody::Missing => f.write_str("Missing"), + } + } } /// What a message is, without its contents. @@ -110,6 +190,7 @@ pub enum MessageKind { Sync, Ping, Pong, + Blob, } impl ParsedMessage { @@ -120,6 +201,7 @@ impl ParsedMessage { ParsedMessage::Sync(_) => MessageKind::Sync, ParsedMessage::Ping(_) => MessageKind::Ping, ParsedMessage::Pong(_) => MessageKind::Pong, + ParsedMessage::Blob(_) => MessageKind::Blob, } } @@ -130,6 +212,7 @@ impl ParsedMessage { ParsedMessage::HelloReply(reply) => reply.epoch, ParsedMessage::Sync(envelope) => envelope.epoch, ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch, + ParsedMessage::Blob(blob) => blob.epoch, } } } diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index cc7f25f..82e6016 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -19,6 +19,7 @@ use tokio::process::{Child, Command}; use url::Url; use crate::transport::EncodedMessage; +use crate::transport::{BlobBody, BlobId, BlobMessage}; use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope}; use crate::transport::{HeadsSummary, Keepalive, MAX_LISTED_HEADS}; use crate::transport::{MessageHandler, ParsedMessage, Transport}; @@ -31,6 +32,7 @@ const TOPIC_HELLO_REPLY: &str = "continuum.v1.hello-reply"; const TOPIC_SYNC: &str = "continuum.v1.sync"; const TOPIC_PING: &str = "continuum.v1.ping"; const TOPIC_PONG: &str = "continuum.v1.pong"; +const TOPIC_BLOB: &str = "continuum.v1.blob"; const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25); @@ -747,6 +749,7 @@ fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, T } TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)), TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)), + TOPIC_BLOB => Ok(ParsedMessage::Blob(bytes_to_blob(&bytes)?)), t => Err(TransportError::UnknownTopic(t.to_string())), } } @@ -774,6 +777,7 @@ fn format_message( ParsedMessage::Sync(env) => (TOPIC_SYNC, sync_envelope_to_bytes(env)?), ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?), ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?), + ParsedMessage::Blob(blob) => (TOPIC_BLOB, blob_to_bytes(blob)?), }; let outbound = OutboundMessage { dst: Destination { @@ -1021,6 +1025,110 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr }) } +// --- BlobMessage --- +#[derive(Serialize, Deserialize)] +enum BlobIdShadow { + Change(ByteBuf), + Content(ByteBuf), +} + +#[derive(Serialize, Deserialize)] +enum BlobBodyShadow { + Request { + chunks: Vec<u32>, + }, + Manifest { + size: u64, + chunk_size: u32, + hashes: Vec<ByteBuf>, + }, + Chunk { + index: u32, + data: ByteBuf, + }, + Missing, +} + +#[derive(Serialize, Deserialize)] +struct BlobShadow { + epoch: EpochShadow, + id: BlobIdShadow, + body: BlobBodyShadow, +} + +fn blob_to_bytes(msg: &BlobMessage) -> Result<Vec<u8>, rmp_serde::encode::Error> { + let id = match msg.id { + BlobId::Change(hash) => BlobIdShadow::Change(ByteBuf::from(hash.0.to_vec())), + BlobId::Content(hash) => BlobIdShadow::Content(ByteBuf::from(hash.to_vec())), + }; + let body = match &msg.body { + BlobBody::Request { chunks } => BlobBodyShadow::Request { + chunks: chunks.clone(), + }, + BlobBody::Manifest { + size, + chunk_size, + hashes, + } => BlobBodyShadow::Manifest { + size: *size, + chunk_size: *chunk_size, + hashes: hashes + .iter() + .map(|hash| ByteBuf::from(hash.to_vec())) + .collect(), + }, + BlobBody::Chunk { index, data } => BlobBodyShadow::Chunk { + index: *index, + data: ByteBuf::from(data.to_vec()), + }, + BlobBody::Missing => BlobBodyShadow::Missing, + }; + rmp_serde::to_vec(&BlobShadow { + epoch: epoch_to_shadow(&msg.epoch), + id, + body, + }) +} + +fn hash32(bytes: &[u8], what: &str) -> Result<[u8; 32], TransportError> { + bytes + .try_into() + .map_err(|_| TransportError::InvalidPayload(format!("{what} must be 32 bytes"))) +} + +fn bytes_to_blob(bytes: &[u8]) -> Result<BlobMessage, TransportError> { + let shadow: BlobShadow = rmp_serde::from_slice(bytes)?; + let id = match shadow.id { + BlobIdShadow::Change(hash) => BlobId::Change(ChangeHash(hash32(&hash, "a change hash")?)), + BlobIdShadow::Content(hash) => BlobId::Content(hash32(&hash, "a content hash")?), + }; + let body = match shadow.body { + BlobBodyShadow::Request { chunks } => BlobBody::Request { chunks }, + BlobBodyShadow::Manifest { + size, + chunk_size, + hashes, + } => BlobBody::Manifest { + size, + chunk_size, + hashes: hashes + .iter() + .map(|hash| hash32(hash, "a chunk hash")) + .collect::<Result<_, _>>()?, + }, + BlobBodyShadow::Chunk { index, data } => BlobBody::Chunk { + index, + data: data.into_vec().into(), + }, + BlobBodyShadow::Missing => BlobBody::Missing, + }; + Ok(BlobMessage { + epoch: epoch_from_shadow(shadow.epoch)?, + id, + body, + }) +} + #[derive(Error, Debug)] pub enum TransportError { #[error("Mycelium delivered a message without a sender public key")] @@ -1087,6 +1195,47 @@ mod tests { } #[test] + fn blob_messages_round_trip() { + let epoch = Epoch { + generation: 3, + id: [7; 32], + }; + let id = BlobId::Change(ChangeHash([9; 32])); + let chunk = BlobMessage { + epoch, + id, + body: BlobBody::Chunk { + index: 5, + data: vec![1u8; 64 * 1024].into(), + }, + }; + let bytes = blob_to_bytes(&chunk).unwrap(); + // A 64 KiB chunk costs only a few bytes of framing. + assert!(bytes.len() < 64 * 1024 + 128, "{} bytes", bytes.len()); + let decoded = bytes_to_blob(&bytes).unwrap(); + assert_eq!(decoded.id, id); + assert_eq!(decoded.epoch, epoch); + assert!( + matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024) + ); + + let manifest = BlobMessage { + epoch, + id, + body: BlobBody::Manifest { + size: 100, + chunk_size: 64, + hashes: vec![[1; 32], [2; 32]], + }, + }; + let decoded = bytes_to_blob(&blob_to_bytes(&manifest).unwrap()).unwrap(); + assert!(matches!( + decoded.body, + BlobBody::Manifest { size: 100, chunk_size: 64, ref hashes } if hashes.len() == 2 + )); + } + + #[test] fn listed_heads_equal_their_digest() { let heads: Vec<_> = (0..3u8).map(|n| ChangeHash([n; 32])).collect(); let listed = HeadsSummary::of(heads.clone()); diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index bb93d48..9dcd91a 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -39,6 +39,7 @@ pub struct ByKind { pub sync: Counter, pub ping: Counter, pub pong: Counter, + pub blob: Counter, } impl ByKind { @@ -49,6 +50,7 @@ impl ByKind { MessageKind::Sync => &mut self.sync, MessageKind::Ping => &mut self.ping, MessageKind::Pong => &mut self.pong, + MessageKind::Blob => &mut self.blob, } } } |