summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/remake/api.rs28
-rw-r--r--src/remake/catchup.rs2
-rw-r--r--src/remake/content.rs95
-rw-r--r--src/remake/coordinator.rs9
-rw-r--r--src/remake/kind.rs52
-rw-r--r--src/remake/local.rs48
-rw-r--r--src/remake/mod.rs7
-rw-r--r--src/remake/records.rs37
-rw-r--r--src/remake/snapshot.rs10
-rw-r--r--src/remake/switch.rs15
-rw-r--r--src/remake/vote.rs33
-rw-r--r--src/state.rs28
-rw-r--r--src/swarm/blob.rs25
-rw-r--r--src/transport/mod.rs6
-rw-r--r--src/transport/mycelium.rs3
15 files changed, 319 insertions, 79 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs
index de8166c..3345086 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -29,8 +29,9 @@
//! `{ "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
@@ -61,7 +62,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, EpochRecord, Phase, Policy, Proposal, Record, Retention, Vote,
+ unix_millis,
};
use super::{Host, kind, vote};
use crate::document::store::StoreError;
@@ -160,10 +162,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>,
@@ -195,12 +193,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,
}
@@ -394,7 +390,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 +410,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
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 42b6526..39b9550 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -510,7 +510,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..370d5fc 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -20,6 +20,7 @@ use std::time::Duration;
use tracing::{info, warn};
use super::Host;
+use super::content;
use super::records::{self, Decision, Record};
use super::snapshot;
use super::vote;
@@ -118,7 +119,7 @@ fn abort_reason<H: Host>(
}
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 +127,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;
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 787bd5e..ec2f70a 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;
@@ -37,8 +38,16 @@ struct Inner {
/// 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]>,
}
+/// 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 {
@@ -213,6 +222,37 @@ 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()
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 1c2be36..dacd1db 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;
@@ -75,6 +77,11 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
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);
diff --git a/src/remake/records.rs b/src/remake/records.rs
index ed018ca..e3a0a7c 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -137,14 +137,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()
}
@@ -229,8 +252,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 +262,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(),
})
}
@@ -390,7 +413,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,
}
}
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index efe7e37..d308dcd 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -51,8 +51,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 +92,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 +105,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,
}
}
@@ -367,7 +367,7 @@ mod tests {
policy,
proposed_at: 0,
kind: "compact",
- content: &NULL,
+ content: None,
translation: &NULL,
}
}
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index e9f49ee..c18905c 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -26,6 +26,7 @@ use tracing::{error, info, warn};
use super::Host;
use super::catchup;
+use super::content;
use super::delta::{self, ReplayStats};
use super::kind::Translator;
use super::local::LocalStatus;
@@ -82,6 +83,12 @@ 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);
let committed: Vec<(String, Record)> = all
@@ -157,10 +164,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..4416293 100644
--- a/src/remake/vote.rs
+++ b/src/remake/vote.rs
@@ -10,7 +10,9 @@
//! may vote again.
//!
//! 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.
+//! 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.
//!
//! 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
@@ -21,6 +23,7 @@ use std::collections::HashSet;
use tracing::{info, warn};
use super::Host;
+use super::content;
use super::records::{self, EpochRecord, Phase, Record, Vote, unix_millis};
use crate::document::store::StoreError;
@@ -147,18 +150,23 @@ pub(super) async fn cast<H: Host>(host: &H) {
let (voter, wanted) = (me.clone(), epoch.clone());
let due = host
.read_doc(move |doc| {
- choice(
- &voter,
- &wanted,
- &records::load_all(doc),
- unix_millis(),
- grace,
- )
- .is_some()
+ let all = records::load_all(doc);
+ let id = choice(&voter, &wanted, &all, unix_millis(), grace)?;
+ 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,7 +175,8 @@ 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 Some(id) = choice(&voter, &epoch, &all, now, grace).filter(|id| *id == chosen)
+ else {
return Ok(None);
};
records::write_vote(doc, &id, &voter, &Vote { at: now })?;
@@ -206,7 +215,7 @@ 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,
diff --git a/src/state.rs b/src/state.rs
index 0f30732..1ea5aae 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -640,6 +640,17 @@ impl AppState {
self.transport.local_id()
}
+ /// Peers that are active and last said they are on our 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.active && peer.epoch == Some(epoch))
+ .map(|peer| peer.id)
+ .collect()
+ }
+
pub(crate) fn lock_rtt(&self) -> std::sync::MutexGuard<'_, crate::swarm::rtt::RttTable> {
self.rtt
.lock()
@@ -1037,18 +1048,19 @@ impl Host for AppState {
) -> impl Future<Output = Result<automerge::Change>> + Send {
let state = self.clone();
async move {
- let epoch = state.local_epoch();
- let peers: Vec<String> = state
- .peer_summaries()
- .await
- .into_iter()
- .filter(|peer| peer.active && peer.epoch == Some(epoch))
- .map(|peer| peer.id)
- .collect();
+ 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);
}
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 6ff1618..8c891ae 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -2,8 +2,9 @@
//!
//! 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. Today the only kind of 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.
+//! 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
@@ -62,7 +63,7 @@ pub(crate) struct Blobs {
}
struct Prepared {
- bytes: Vec<u8>,
+ bytes: Arc<Vec<u8>>,
hashes: Vec<[u8; 32]>,
}
@@ -152,8 +153,10 @@ async fn prepare(state: &AppState, id: BlobId) -> Option<Arc<Prepared>> {
})
.await
.ok()
- .flatten()?
+ .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 || {
@@ -454,6 +457,20 @@ fn expected_len(manifest: &Manifest, slot: usize) -> 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,
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index c96b053..c5769ca 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -109,12 +109,18 @@ pub enum ParsedMessage {
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}"))
+ }
}
}
}
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index a8ab104..82e6016 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -1029,6 +1029,7 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr
#[derive(Serialize, Deserialize)]
enum BlobIdShadow {
Change(ByteBuf),
+ Content(ByteBuf),
}
#[derive(Serialize, Deserialize)]
@@ -1058,6 +1059,7 @@ struct BlobShadow {
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 {
@@ -1098,6 +1100,7 @@ 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 },
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-25 09:33:00 +0000