summaryrefslogtreecommitdiffstats
path: root/src/compaction/catchup.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/compaction/catchup.rs')
-rw-r--r--src/compaction/catchup.rs283
1 files changed, 212 insertions, 71 deletions
diff --git a/src/compaction/catchup.rs b/src/compaction/catchup.rs
index f166f56..4694fbc 100644
--- a/src/compaction/catchup.rs
+++ b/src/compaction/catchup.rs
@@ -2,9 +2,9 @@
//!
//! Peers on another epoch don't sync, so a node that missed a switch (new, late, or
//! one whose own build failed) can't learn the new epoch through ordinary sync. Instead,
-//! once an established peer advertises a newer generation, the node asks it for its
-//! whole document (`SnapshotRequest`), checks that the document starts with the
-//! advertised snapshot change, and switches to it:
+//! once an active peer advertises a newer generation, the node downloads that peer's
+//! document (`SnapshotRequest`/`SnapshotReply`), checks that it starts with the advertised
+//! snapshot change, and switches to it:
//!
//! * a node that wrote nothing the snapshot lacks (e.g. a brand-new node) adopts the
//! document as is, whatever the gap;
@@ -13,17 +13,24 @@
//! * a node further behind with such changes stays where it is and reports it; that is
//! the returning-node case (replay window or quarantine, step 7).
//!
-//! Only one fetch runs at a time, and the worker retries after `RETRY`. The reply is the
-//! peer's full document in one message; chunking comes later.
+//! The document travels in chunks of `CHUNK_BYTES`, one request per chunk, so a large
+//! document neither needs one huge transport message nor one long timeout. The serving
+//! node saves its document once per download and serves every chunk from that copy; a
+//! SHA-256 digest of the whole copy is in every reply and checked at the end.
+//!
+//! Peers refuse explicitly (non-members, busy, other epoch), so the requester moves on
+//! to the next peer right away instead of timing out.
+use std::sync::Arc;
use std::time::Duration;
use automerge::{ActorId, AutoCommit};
+use sha2::{Digest, Sha256};
use tokio::time::Instant;
use tracing::{debug, error, info, warn};
use super::delta::{self, ReplayStats};
-use super::local::{CatchUp, LocalStatus};
+use super::local::{CatchUp, LocalStatus, Served};
use super::records::unix_millis;
use super::snapshot;
use super::switch::swap_in;
@@ -31,21 +38,31 @@ use super::Host;
use crate::replica::Epoch;
use crate::transport::{ParsedMessage, SnapshotReply, SnapshotRequest};
+/// Pause before the next attempt (next peer) after a failed download.
const RETRY: Duration = Duration::from_secs(10);
-const FETCH_TIMEOUT: Duration = Duration::from_secs(60);
+/// Payload per reply.
+pub(crate) const CHUNK_BYTES: usize = 1024 * 1024;
+/// How long to wait for one chunk, and how often to ask for it.
+const CHUNK_TIMEOUT: Duration = Duration::from_secs(60);
+const CHUNK_ATTEMPTS: u32 = 3;
+/// A served copy older than this is replaced when a new download starts.
+const SERVED_FRESH_FOR: Duration = Duration::from_secs(60);
/// Called by the switch worker when there is nothing to switch to locally.
pub(super) async fn check<H: Host>(host: &H) {
let local = host.current_epoch();
- let ahead: Vec<(String, Epoch)> = host
+ // Any active peer will do: it proved liveness recently, and whether it accepts us
+ // as a member is its decision (it refuses explicitly otherwise).
+ let mut ahead: Vec<(String, Epoch)> = host
.peer_summaries()
.await
.into_iter()
- .filter(|peer| peer.active && peer.established)
+ .filter(|peer| peer.active)
.filter_map(|peer| peer.epoch.map(|epoch| (peer.id, epoch)))
.filter(|(_, epoch)| epoch.generation > local.generation)
.collect();
+ ahead.sort_by(|a, b| a.0.cmp(&b.0));
let Some(generation) = ahead.iter().map(|(_, epoch)| epoch.generation).max() else {
if host.compaction().catch_up().is_some() {
@@ -81,7 +98,7 @@ pub(super) async fn check<H: Host>(host: &H) {
}
let attempts = previous.map_or(0, |catch_up| catch_up.attempts) + 1;
- // Rotate through the peers on the target epoch.
+ // Rotate through the peers on the target epoch (sorted, so the rotation is stable).
let sources: Vec<&String> = ahead
.iter()
.filter(|(_, epoch)| *epoch == target)
@@ -98,7 +115,7 @@ pub(super) async fn check<H: Host>(host: &H) {
info!(local = %local, target = %target, peer = %source, attempt = attempts, "Fetching newer epoch from peer");
- let result = match fetch(host, &source, local, target).await {
+ let result = match download(host, &source, local, target).await {
Ok(document) => adopt(host, local, target, document).await,
Err(error) => Err(error),
};
@@ -117,38 +134,95 @@ pub(super) async fn check<H: Host>(host: &H) {
}
}
-async fn fetch<H: Host>(
+/// Downloads `peer`'s document of `target`, chunk by chunk.
+async fn download<H: Host>(
host: &H,
peer: &str,
local: Epoch,
target: Epoch,
) -> Result<Vec<u8>, String> {
- let reply = host.compaction().await_reply(peer, target);
- let request = ParsedMessage::SnapshotRequest(SnapshotRequest {
- epoch: local,
- wanted: target,
- });
- if let Err(error) = host.send_message(peer, request).await {
- host.compaction().stop_awaiting();
- return Err(format!("sending the request failed: {error:#}"));
+ let mut document: Vec<u8> = Vec::new();
+ let mut expected: Option<([u8; 32], u64)> = None;
+ let mut last_progress = 0;
+
+ loop {
+ let offset = document.len() as u64;
+ let reply = request_chunk(host, peer, local, target, offset).await?;
+
+ if !reply.refused.is_empty() {
+ return Err(format!("peer refused: {}", reply.refused));
+ }
+ if reply.epoch != target {
+ return Err(format!("peer is on {} instead of {target}", reply.epoch));
+ }
+ match expected {
+ None => {
+ info!(peer = %peer, document_bytes = reply.total, "Downloading document");
+ expected = Some((reply.digest, reply.total));
+ }
+ Some((digest, total)) if digest != reply.digest || total != reply.total => {
+ return Err("the peer's document changed during the download".to_owned());
+ }
+ Some(_) => {}
+ }
+ let total = reply.total;
+ if reply.data.is_empty() && offset < total {
+ return Err(format!("empty chunk at offset {offset} of {total}"));
+ }
+ document.extend_from_slice(&reply.data);
+
+ let percent = (document.len() as u64 * 100)
+ .checked_div(total)
+ .unwrap_or(100);
+ if percent >= last_progress + 10 {
+ last_progress = percent;
+ debug!(peer = %peer, percent, "Download progress");
+ }
+
+ if document.len() as u64 >= total {
+ break;
+ }
+ }
+
+ let (digest, total) = expected.expect("set by the first reply");
+ if document.len() as u64 != total || sha256(&document) != digest {
+ return Err("downloaded document does not match its digest".to_owned());
}
+ Ok(document)
+}
- let reply = match tokio::time::timeout(FETCH_TIMEOUT, reply).await {
- Ok(Ok(reply)) => reply,
- Ok(Err(_)) => return Err("request superseded".to_owned()),
- Err(_) => {
+async fn request_chunk<H: Host>(
+ host: &H,
+ peer: &str,
+ local: Epoch,
+ target: Epoch,
+ offset: u64,
+) -> Result<SnapshotReply, String> {
+ for attempt in 1..=CHUNK_ATTEMPTS {
+ let reply = host.compaction().await_reply(peer, target, offset);
+ let request = ParsedMessage::SnapshotRequest(SnapshotRequest {
+ epoch: local,
+ wanted: target,
+ offset,
+ });
+ if let Err(error) = host.send_message(peer, request).await {
host.compaction().stop_awaiting();
- return Err(format!("no reply within {}s", FETCH_TIMEOUT.as_secs()));
+ return Err(format!("sending the request failed: {error:#}"));
}
- };
- if reply.epoch != target || reply.document.is_empty() {
- return Err(format!(
- "peer is on {} and did not send {target}",
- reply.epoch
- ));
+ match tokio::time::timeout(CHUNK_TIMEOUT, reply).await {
+ Ok(Ok(reply)) => return Ok(reply),
+ Ok(Err(_)) => return Err("request superseded".to_owned()),
+ Err(_) => {
+ host.compaction().stop_awaiting();
+ debug!(peer = %peer, offset, attempt, "No reply for chunk; asking again");
+ }
+ }
}
- Ok(reply.document)
+ Err(format!(
+ "no reply for the chunk at offset {offset} after {CHUNK_ATTEMPTS} attempts of {}s",
+ CHUNK_TIMEOUT.as_secs()
+ ))
}
/// Verifies the fetched document, carries over this node's own missing changes, and
@@ -250,58 +324,125 @@ struct Prepared {
metadata: snapshot::EpochMetadata,
}
-/// Answers a peer's snapshot request with this node's document, if it is on the wanted
-/// epoch. Only for members of our cluster; one at a time.
+/// Answers a peer's chunk request from a saved copy of this node's document, if this
+/// node is on the wanted epoch and the peer is a member of our cluster. Refuses
+/// explicitly otherwise.
pub(crate) async fn serve_snapshot<H: Host>(host: H, peer: String, request: SnapshotRequest) {
- if !host.is_cluster_member(&peer).await {
- debug!(peer = %peer, "Ignoring snapshot request from a non-member");
- return;
- }
- if !host.compaction().try_start_serving() {
- debug!(peer = %peer, "Already serving a snapshot; the peer will retry");
- return;
- }
-
let current = host.current_epoch();
let wanted = request.wanted;
- let document = if current == wanted {
- // Clone under the read lock, save without one. The history check guards against
- // a switch between reading the epoch and cloning.
- let mut doc = host.read_doc(|doc| doc.clone()).await;
- tokio::task::spawn_blocking(move || {
- if doc
- .get_change_by_hash(&automerge::ChangeHash(wanted.id))
- .is_some()
- {
- doc.save()
- } else {
- Vec::new()
+ let reply = match served_copy(&host, &peer, current, wanted, request.offset).await {
+ Ok(served) => {
+ let start = (request.offset as usize).min(served.bytes.len());
+ let end = (start + CHUNK_BYTES).min(served.bytes.len());
+ if request.offset == 0 {
+ info!(peer = %peer, wanted = %wanted, document_bytes = served.bytes.len(), "Serving document");
}
- })
+ SnapshotReply {
+ epoch: served.epoch,
+ wanted,
+ digest: served.digest,
+ total: served.bytes.len() as u64,
+ offset: request.offset,
+ data: served.bytes[start..end].to_vec(),
+ refused: String::new(),
+ }
+ }
+ Err(reason) => {
+ info!(peer = %peer, wanted = %wanted, reason = %reason, "Refusing document request");
+ SnapshotReply {
+ epoch: current,
+ wanted,
+ digest: [0; 32],
+ total: 0,
+ offset: request.offset,
+ data: Vec::new(),
+ refused: reason,
+ }
+ }
+ };
+
+ if let Err(error) = host
+ .send_message(&peer, ParsedMessage::SnapshotReply(reply))
.await
- .unwrap_or_else(|error| {
+ {
+ warn!(peer = %peer, error = %format!("{error:#}"), "Sending document chunk failed");
+ }
+}
+
+/// The saved document to serve chunks from: reused while a download is in progress,
+/// saved anew when a download starts and the copy is stale.
+async fn served_copy<H: Host>(
+ host: &H,
+ peer: &str,
+ current: Epoch,
+ wanted: Epoch,
+ offset: u64,
+) -> Result<Arc<Served>, String> {
+ if !host.is_cluster_member(peer).await {
+ return Err("not a member of this cluster (no completed handshake)".to_owned());
+ }
+ if current != wanted {
+ return Err(format!("this node is on {current}"));
+ }
+
+ if let Some(served) = host.compaction().served(wanted) {
+ if offset > 0 || served.saved_at.elapsed() < SERVED_FRESH_FOR {
+ return Ok(served);
+ }
+ } else if offset > 0 {
+ return Err("download expired; start again".to_owned());
+ }
+
+ if !host.compaction().try_start_serving() {
+ return Err("busy saving the document; try again".to_owned());
+ }
+
+ // Clone under the read lock, save without one. The history check guards against
+ // a switch between reading the epoch and cloning.
+ let mut doc = host.read_doc(|doc| doc.clone()).await;
+ let saved = tokio::task::spawn_blocking(move || {
+ if doc
+ .get_change_by_hash(&automerge::ChangeHash(wanted.id))
+ .is_some()
+ {
+ Some(doc.save())
+ } else {
+ None
+ }
+ })
+ .await;
+ host.compaction().done_serving();
+
+ let bytes = match saved {
+ Ok(Some(bytes)) => bytes,
+ Ok(None) => return Err("this node switched epochs".to_owned()),
+ Err(error) => {
error!(error = %error, "Saving the document for a peer failed");
- Vec::new()
- })
- } else {
- Vec::new()
+ return Err("saving the document failed".to_owned());
+ }
};
- host.compaction().done_serving();
- info!(peer = %peer, wanted = %wanted, document_bytes = document.len(), "Serving snapshot");
- let reply = ParsedMessage::SnapshotReply(SnapshotReply {
- epoch: if document.is_empty() { current } else { wanted },
- document,
+ let served = Arc::new(Served {
+ epoch: wanted,
+ digest: sha256(&bytes),
+ bytes,
+ saved_at: Instant::now(),
});
- if let Err(error) = host.send_message(&peer, reply).await {
- warn!(peer = %peer, error = %format!("{error:#}"), "Sending snapshot failed");
- }
+ host.compaction().set_served(Arc::clone(&served));
+ Ok(served)
+}
+
+fn sha256(bytes: &[u8]) -> [u8; 32] {
+ let mut digest = [0; 32];
+ digest.copy_from_slice(&Sha256::digest(bytes));
+ digest
}
-/// Hands a peer's snapshot reply to the waiting fetch.
+/// Hands a peer's chunk to the waiting download.
pub(crate) fn receive_snapshot_reply<H: Host>(host: &H, peer: &str, reply: SnapshotReply) {
+ let offset = reply.offset;
if !host.compaction().deliver_reply(peer, reply) {
- debug!(peer = %peer, "Unexpected snapshot reply; dropped");
+ debug!(peer = %peer, offset, "Unexpected document chunk; dropped");
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-22 16:48:09 +0000