summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/api/stats/prometheus.rs1
-rw-r--r--src/config/mod.rs15
-rw-r--r--src/main.rs1
-rw-r--r--src/remake/catchup.rs49
-rw-r--r--src/remake/local.rs19
-rw-r--r--src/remake/mod.rs10
-rw-r--r--src/state.rs24
-rw-r--r--src/swarm/blob.rs491
-rw-r--r--src/swarm/mod.rs1
-rw-r--r--src/swarm/peer.rs3
-rw-r--r--src/swarm/source.rs2
-rw-r--r--src/swarm/sync.rs6
-rw-r--r--src/transport/mod.rs77
-rw-r--r--src/transport/mycelium.rs146
-rw-r--r--src/transport/traffic.rs2
15 files changed, 844 insertions, 3 deletions
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/config/mod.rs b/src/config/mod.rs
index 4f6c5e0..4914fb6 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -131,6 +131,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",
@@ -205,6 +212,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 +236,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 +286,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/main.rs b/src/main.rs
index 5337f90..228fa54 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -139,6 +139,7 @@ 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)),
config: Arc::new(config.report()),
neighbours: Default::default(),
pulling: Default::default(),
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index db25251..42b6526 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -7,8 +7,10 @@
//! 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):
@@ -50,6 +52,10 @@ const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60);
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 {
@@ -147,6 +153,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.
diff --git a/src/remake/local.rs b/src/remake/local.rs
index d2ea2c2..787bd5e 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -34,6 +34,8 @@ 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>,
}
@@ -211,6 +213,23 @@ impl RemakeState {
self.lock().rejoin.take()
}
+ /// 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..1c2be36 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -68,6 +68,16 @@ 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;
+
+ /// 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.
diff --git a/src/state.rs b/src/state.rs
index df053a6..0f30732 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -597,6 +597,8 @@ 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>,
/// The effective configuration with sources, imported values left out
/// (`config::Loaded::report`), for `GET /admin/config`.
pub(crate) config: Arc<serde_json::Value>,
@@ -1029,6 +1031,28 @@ 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 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();
+ crate::swarm::blob::fetch_change(&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 }
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
new file mode 100644
index 0000000..6ff1618
--- /dev/null
+++ b/src/swarm/blob.rs
@@ -0,0 +1,491 @@
+//! 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. 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.
+//!
+//! * 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: 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()?
+ }
+ };
+ 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.
+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..dd70bbb 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -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/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..58a08db 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(());
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index b2f5c9d..c96b053 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -100,6 +100,80 @@ 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),
+}
+
+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}"),
+ }
+ }
+}
+
+/// 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 +184,7 @@ pub enum MessageKind {
Sync,
Ping,
Pong,
+ Blob,
}
impl ParsedMessage {
@@ -120,6 +195,7 @@ impl ParsedMessage {
ParsedMessage::Sync(_) => MessageKind::Sync,
ParsedMessage::Ping(_) => MessageKind::Ping,
ParsedMessage::Pong(_) => MessageKind::Pong,
+ ParsedMessage::Blob(_) => MessageKind::Blob,
}
}
@@ -130,6 +206,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..a8ab104 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,107 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr
})
}
+// --- BlobMessage ---
+#[derive(Serialize, Deserialize)]
+enum BlobIdShadow {
+ Change(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())),
+ };
+ 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")?)),
+ };
+ 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 +1192,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,
}
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 20:58:57 +0000