put remake content into blobs

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 },