summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/api/mod.rs2
-rw-r--r--src/document/replica.rs10
-rw-r--r--src/main.rs8
-rw-r--r--src/remake/api.rs (renamed from src/compaction/api.rs)65
-rw-r--r--src/remake/catchup.rs (renamed from src/compaction/catchup.rs)36
-rw-r--r--src/remake/coordinator.rs (renamed from src/compaction/coordinator.rs)14
-rw-r--r--src/remake/delta.rs (renamed from src/compaction/delta.rs)14
-rw-r--r--src/remake/local.rs (renamed from src/compaction/local.rs)30
-rw-r--r--src/remake/mod.rs (renamed from src/compaction/mod.rs)28
-rw-r--r--src/remake/quarantine.rs (renamed from src/compaction/quarantine.rs)12
-rw-r--r--src/remake/records.rs (renamed from src/compaction/records.rs)16
-rw-r--r--src/remake/snapshot.rs (renamed from src/compaction/snapshot.rs)61
-rw-r--r--src/remake/switch.rs (renamed from src/compaction/switch.rs)42
-rw-r--r--src/state.rs12
-rw-r--r--src/swarm/bootstrap.rs2
-rw-r--r--src/swarm/peer.rs2
16 files changed, 179 insertions, 175 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs
index ac6f2d0..ea3ad41 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -37,7 +37,7 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) ->
.route("/stream/{*key}", get(store::stream_store_entry))
.route("/peers", get(peers::list_peers))
.route("/peers/{peer}", delete(peers::delete_peer))
- .merge(crate::compaction::router::<AppState>());
+ .merge(crate::remake::router::<AppState>());
if debug_api {
warn!("Debug API enabled");
diff --git a/src/document/replica.rs b/src/document/replica.rs
index 4579029..ff3af4c 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -1,7 +1,7 @@
//! The local replica: the Automerge document together with the epoch it belongs to.
//!
-//! Compaction (later steps) replaces the document with a snapshot whose history is
-//! unrelated to the old one. Syncing two unrelated histories would make Automerge merge
+//! A remake replaces the document with a new one (for compaction: a snapshot) whose
+//! history is unrelated to the old one. Syncing two unrelated histories would make Automerge merge
//! them, turning every key into a conflict with its own copy, so nodes on different
//! epochs must never sync. Keeping the epoch and the document under one lock makes it
//! impossible to label a sync message from one document with the other's epoch.
@@ -13,10 +13,10 @@ use automerge::AutoCommit;
/// One generation of the cluster document.
///
-/// * `generation` orders epochs: every compaction increments it, so a node can tell
+/// * `generation` orders epochs: every remake increments it, so a node can tell
/// whether a peer is behind or ahead.
/// * `id` is the hash of the epoch's snapshot change. It tells two epochs of the same
-/// generation apart, which only happens if two compactions committed concurrently.
+/// generation apart, which only happens if two remakes committed concurrently.
#[derive(Clone, Copy, PartialEq, Eq, Hash)]
pub struct Epoch {
pub generation: u64,
@@ -31,7 +31,7 @@ pub enum EpochRelation {
PeerBehind,
/// The peer is on a newer generation; we have to migrate.
PeerAhead,
- /// Same generation, different snapshot: competing compactions. Never sync.
+ /// Same generation, different snapshot: competing remakes. Never sync.
Diverged,
}
diff --git a/src/main.rs b/src/main.rs
index 1793523..ab91490 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -19,8 +19,8 @@ use tracing::{debug, error, info, trace};
use url::Url;
mod api;
-mod compaction;
mod document;
+mod remake;
mod state;
mod swarm;
mod transport;
@@ -155,7 +155,7 @@ async fn main() -> Result<()> {
initial_sync_done: Arc::new(AtomicBool::new(false)),
change_events,
timing: Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)),
- compaction: Arc::new(compaction::Compaction::default()),
+ remake: Arc::new(remake::RemakeState::default()),
stale: Default::default(),
};
@@ -189,8 +189,8 @@ async fn main() -> Result<()> {
// Switches to committed epochs and catches up with newer ones. Starts before
// bootstrapping: a node joining a compacted cluster can only finish its initial sync
// after fetching the current epoch from a peer.
- tokio::spawn(compaction::run_worker(state.clone()));
- trace!("Compaction worker task spawned");
+ tokio::spawn(remake::run_worker(state.clone()));
+ trace!("Remake worker task spawned");
bootstrap::bootstrap(&state, &seeds)
.await
diff --git a/src/compaction/api.rs b/src/remake/api.rs
index 9fae7c9..b9c0625 100644
--- a/src/compaction/api.rs
+++ b/src/remake/api.rs
@@ -1,17 +1,18 @@
-//! `/admin/compaction(s)` handlers.
+//! `/admin/remakes` handlers.
//!
-//! * `POST /admin/compaction` proposes a compaction of the current epoch. The receiving
+//! * `POST /admin/remakes` proposes a remake of the current epoch (`POST
+//! /admin/compaction` is a shortcut, kept from before). The receiving
//! node becomes its coordinator. `202` with the report, or with `?wait=true` the final
//! report once decided. A dropped request does not abort anything. `?force=true`
//! accepts a `settleSecs` below the dormancy threshold (tests, single-site clusters).
-//! * `GET /admin/compactions` lists all compactions.
-//! * `GET /admin/compactions/{id}` reports one; `?wait=true` long-polls until decided
+//! * `GET /admin/remakes` lists all remakes.
+//! * `GET /admin/remakes/{id}` reports one; `?wait=true` long-polls until decided
//! (bounded by the proposal's own expiry).
//! * `GET /admin/epochs` shows the local epoch, the archived previous ones and whether
//! the node is catching up with a newer epoch from a peer.
-//! * `DELETE /admin/compactions/{id}` asks the coordinator to abort. Only possible while
+//! * `DELETE /admin/remakes/{id}` asks the coordinator to abort. Only possible while
//! proposed; `409` once committed (a switch can't be undone) or already over.
-//! `DELETE /admin/compaction` does the same for the compaction of the local epoch that
+//! `DELETE /admin/remakes` does the same for the remake of the local epoch that
//! is in progress, without having to know its ID.
//!
//! Request body for POST (all optional):
@@ -72,14 +73,14 @@ struct Report {
phase: Phase,
#[serde(flatten)]
record: Record,
- /// What happened to this compaction on the node answering the request.
+ /// What happened to this remake on the node answering the request.
#[serde(skip_serializing_if = "Option::is_none")]
local: Option<LocalStatus>,
}
#[derive(Debug, thiserror::Error)]
enum StartError {
- #[error("compaction {0} is already open for this epoch")]
+ #[error("remake {0} is already open for this epoch")]
Open(String),
#[error(transparent)]
Store(#[from] StoreError),
@@ -104,7 +105,7 @@ fn grace<H: Host>(host: &H) -> u64 {
fn report<H: Host>(host: &H, id: String, record: Record) -> Report {
Report {
phase: record.phase(unix_millis(), grace(host)),
- local: host.compaction().status(&id),
+ local: host.remake().status(&id),
id,
record,
}
@@ -118,7 +119,7 @@ async fn load_report<H: Host>(host: &H, id: &str) -> Option<Report> {
Some(report(host, id.to_owned(), record))
}
-/// Waits until the compaction leaves `proposed`. Bounded by the proposal's own expiry,
+/// Waits until the remake leaves `proposed`. Bounded by the proposal's own expiry,
/// at which point it is reported as `expired`.
async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> {
// A record that was readable may be briefly unreadable or absent while concurrent
@@ -131,7 +132,7 @@ async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> {
loop {
let Some(report) = load_report(host, id).await else {
misses += 1;
- warn!(compaction = %id, seen, misses, "Compaction record not readable while waiting");
+ warn!(remake = %id, seen, misses, "Remake record not readable while waiting");
if !seen || misses >= MAX_MISSES {
return None;
}
@@ -209,7 +210,7 @@ pub(crate) async fn start<H: Host>(
if !dormant.is_empty() {
return error_response(
StatusCode::CONFLICT,
- "dormant peers block compaction; forget them first (DELETE /peers/{pk})",
+ "dormant peers block remake; forget them first (DELETE /peers/{pk})",
Some(json!({ "dormantPeers": dormant })),
);
}
@@ -261,12 +262,12 @@ pub(crate) async fn start<H: Host>(
Err(StartError::Open(open)) => {
return error_response(
StatusCode::CONFLICT,
- "another compaction of this epoch is still open",
- Some(json!({ "compaction": open })),
+ "another remake of this epoch is still open",
+ Some(json!({ "remake": open })),
);
}
Err(StartError::Store(error)) => {
- error!(error = %format!("{error:#}"), "Failed to write compaction proposal");
+ error!(error = %format!("{error:#}"), "Failed to write remake proposal");
return error_response(
StatusCode::INTERNAL_SERVER_ERROR,
"failed to write the proposal",
@@ -276,19 +277,19 @@ pub(crate) async fn start<H: Host>(
};
info!(
- compaction = %id,
+ remake = %id,
settle_secs = settle.as_secs(),
- "Compaction proposed"
+ "Remake proposed"
);
if query.wait {
return match wait_until_final(&host, &id).await {
Some(report) => Json(report).into_response(),
- None => error_response(StatusCode::NOT_FOUND, "compaction record vanished", None),
+ None => error_response(StatusCode::NOT_FOUND, "remake record vanished", None),
};
}
- let location = format!("/admin/compactions/{id}");
+ let location = format!("/admin/remakes/{id}");
(
StatusCode::ACCEPTED,
[(LOCATION, location)],
@@ -356,11 +357,11 @@ pub(crate) async fn get<H: Host>(
};
match found {
Some(report) => Json(report).into_response(),
- None => error_response(StatusCode::NOT_FOUND, "no such compaction", None),
+ None => error_response(StatusCode::NOT_FOUND, "no such remake", None),
}
}
-/// `DELETE /admin/compactions/{id}`.
+/// `DELETE /admin/remakes/{id}`.
pub(crate) async fn abort<H: Host>(
State(host): State<H>,
Path(id): Path<String>,
@@ -369,7 +370,7 @@ pub(crate) async fn abort<H: Host>(
abort_by_id(host, id, query.wait).await
}
-/// `DELETE /admin/compaction`: aborts whichever compaction of the local epoch is still
+/// `DELETE /admin/remakes`: aborts whichever remake of the local epoch is still
/// proposed. There is at most one, unless two coordinators raced (then both abort anyway
/// and this picks the oldest).
pub(crate) async fn abort_current<H: Host>(
@@ -396,7 +397,7 @@ pub(crate) async fn abort_current<H: Host>(
Some(id) => abort_by_id(host, id, query.wait).await,
None => error_response(
StatusCode::NOT_FOUND,
- "no compaction of the local epoch is in progress",
+ "no remake of the local epoch is in progress",
None,
),
}
@@ -404,7 +405,7 @@ pub(crate) async fn abort_current<H: Host>(
async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
let Some(current) = load_report(&host, &id).await else {
- return error_response(StatusCode::NOT_FOUND, "no such compaction", None);
+ return error_response(StatusCode::NOT_FOUND, "no such remake", None);
};
match current.phase {
@@ -413,14 +414,14 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
return error_response(
StatusCode::CONFLICT,
"already committed; the switch cannot be undone",
- Some(json!({ "compaction": current })),
+ Some(json!({ "remake": current })),
)
}
Phase::Aborted | Phase::Expired => {
return error_response(
StatusCode::CONFLICT,
"already over",
- Some(json!({ "compaction": current })),
+ Some(json!({ "remake": current })),
)
}
}
@@ -437,14 +438,14 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
.commit_doc(move |doc| records::write_abort_request(doc, &record_id, &request))
.await
{
- error!(compaction = %id, error = %format!("{error:#}"), "Failed to write abort request");
+ error!(remake = %id, error = %format!("{error:#}"), "Failed to write abort request");
return error_response(
StatusCode::INTERNAL_SERVER_ERROR,
"failed to record the abort request",
None,
);
}
- info!(compaction = %id, "Abort requested");
+ info!(remake = %id, "Abort requested");
}
let report = if wait {
@@ -455,7 +456,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
match report {
Some(report) if report.phase.is_final() => Json(report).into_response(),
Some(report) => (StatusCode::ACCEPTED, Json(report)).into_response(),
- None => error_response(StatusCode::NOT_FOUND, "no such compaction", None),
+ None => error_response(StatusCode::NOT_FOUND, "no such remake", None),
}
}
@@ -465,9 +466,9 @@ pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response {
let current = EpochRecord::from(host.current_epoch());
Json(json!({
"current": current,
- "archived": host.compaction().archived_epochs(),
- "catchUp": host.compaction().catch_up_report(),
- "quarantine": host.compaction().list_quarantine(|entry| entry.id.clone()),
+ "archived": host.remake().archived_epochs(),
+ "catchUp": host.remake().catch_up_report(),
+ "quarantine": host.remake().list_quarantine(|entry| entry.id.clone()),
}))
.into_response()
}
diff --git a/src/compaction/catchup.rs b/src/remake/catchup.rs
index 5085669..22aa0f8 100644
--- a/src/compaction/catchup.rs
+++ b/src/remake/catchup.rs
@@ -47,12 +47,12 @@ const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60);
/// Called by the switch worker when there is nothing to switch to locally, and on every
/// tick while a rejoin is in progress.
pub(super) async fn check<H: Host>(host: &H) {
- if let Some((target, started)) = host.compaction().rejoin_progress() {
+ if let Some((target, started)) = host.remake().rejoin_progress() {
let timeout = host.dormant_after().max(MIN_REJOIN_TIMEOUT);
if host.initial_sync_done() {
finish(host).await;
} else if started.elapsed() > timeout {
- if let Some(rejoin) = host.compaction().take_rejoin() {
+ if let Some(rejoin) = host.remake().take_rejoin() {
let reason = format!(
"no peer on {target} completed a sync within {}s",
timeout.as_secs()
@@ -75,8 +75,8 @@ pub(super) async fn check<H: Host>(host: &H) {
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() {
- host.compaction().set_catch_up(None);
+ if host.remake().catch_up().is_some() {
+ host.remake().set_catch_up(None);
}
return;
};
@@ -97,7 +97,7 @@ pub(super) async fn check<H: Host>(host: &H) {
// Back off between attempts for the same target.
let previous = host
- .compaction()
+ .remake()
.catch_up()
.filter(|catch_up| catch_up.target == target);
if previous
@@ -112,7 +112,7 @@ pub(super) async fn check<H: Host>(host: &H) {
}
async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
- host.compaction().set_catch_up(Some(CatchUp {
+ host.remake().set_catch_up(Some(CatchUp {
target,
attempts,
last_attempt: Instant::now(),
@@ -139,7 +139,7 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
host.reset_initial_sync();
info!(from = %local, to = %target, attempt = attempts, "Rejoining newer epoch");
- host.compaction().start_rejoin(Rejoin {
+ host.remake().start_rejoin(Rejoin {
previous: local,
target,
old,
@@ -172,7 +172,7 @@ enum Prepared {
}
async fn finish<H: Host>(host: &H) {
- let Some(rejoin) = host.compaction().take_rejoin() else {
+ let Some(rejoin) = host.remake().take_rejoin() else {
return;
};
let Rejoin {
@@ -197,7 +197,7 @@ async fn finish<H: Host>(host: &H) {
}) => (old, metadata, plan, own_missing, had_history),
Ok(Prepared::NotYet(old)) => {
host.reset_initial_sync();
- host.compaction().start_rejoin(Rejoin {
+ host.remake().start_rejoin(Rejoin {
previous,
target,
old,
@@ -249,7 +249,7 @@ async fn finish<H: Host>(host: &H) {
Plan::Quarantine { paths, reason } => {
let id = Uuid::new_v4().to_string();
warn!(quarantine = %id, epoch = %previous, paths = paths.len(), reason = %reason, "Quarantined local changes");
- host.compaction().add_quarantine(Quarantined {
+ host.remake().add_quarantine(Quarantined {
id: id.clone(),
epoch: previous,
target,
@@ -264,7 +264,7 @@ async fn finish<H: Host>(host: &H) {
}
if let (true, true, Some(document)) = (had_history, keep_archive, old) {
- host.compaction().archive(
+ host.remake().archive(
previous,
document,
metadata.policy.archive_retention.clone(),
@@ -272,8 +272,8 @@ async fn finish<H: Host>(host: &H) {
);
}
- host.compaction().set_status(
- &metadata.compaction_id,
+ host.remake().set_status(
+ &metadata.remake_id,
LocalStatus {
switched_at: Some(unix_millis()),
replayed_paths: Some(stats.replayed),
@@ -282,7 +282,7 @@ async fn finish<H: Host>(host: &H) {
error: None,
},
);
- host.compaction().set_catch_up(None);
+ host.remake().set_catch_up(None);
if was_ready {
host.resume().await;
}
@@ -405,11 +405,11 @@ async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) {
host.set_stale_view(None);
let attempts = host
- .compaction()
+ .remake()
.catch_up()
.filter(|catch_up| catch_up.target == target)
.map_or(1, |catch_up| catch_up.attempts);
- host.compaction().set_catch_up(Some(CatchUp {
+ host.remake().set_catch_up(Some(CatchUp {
target,
attempts,
last_attempt: Instant::now(),
@@ -420,7 +420,7 @@ async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) {
#[cfg(test)]
mod tests {
use super::*;
- use crate::compaction::snapshot::{build, SnapshotParams};
+ use crate::remake::snapshot::{build, SnapshotParams};
use automerge::{transaction::Transactable, ActorId, ROOT};
use serde_json::{json, Value as JsonValue};
@@ -435,7 +435,7 @@ mod tests {
let snapshot = build(
&mut old,
&SnapshotParams {
- compaction_id: "c-1",
+ remake_id: "c-1",
previous: Epoch::GENESIS,
base_heads: &base,
policy,
diff --git a/src/compaction/coordinator.rs b/src/remake/coordinator.rs
index 8ce3928..b84b0a9 100644
--- a/src/compaction/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -28,7 +28,7 @@ use crate::document::store::StoreError;
const TICK: Duration = Duration::from_secs(1);
pub(crate) async fn run<H: Host>(host: H, id: String) {
- info!(compaction = %id, "Coordinating compaction");
+ info!(remake = %id, "Coordinating remake");
loop {
tokio::time::sleep(TICK).await;
@@ -37,7 +37,7 @@ pub(crate) async fn run<H: Host>(host: H, id: String) {
Ok(true) => return,
Ok(false) => {}
Err(error) => {
- warn!(compaction = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying");
+ warn!(remake = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying");
}
}
}
@@ -51,7 +51,7 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
.await;
let Some(record) = record else {
- warn!(compaction = %id, "Compaction record vanished; stopping coordinator");
+ warn!(remake = %id, "Remake record vanished; stopping coordinator");
return Ok(true);
};
if record.decision.is_some() {
@@ -97,7 +97,7 @@ async fn abort_reason<H: Host>(
&& candidate.proposal.from_epoch == proposal.from_epoch
&& candidate.phase(now, grace).is_open()
}) {
- return Some(format!("competing compaction {other} for the same epoch"));
+ return Some(format!("competing remake {other} for the same epoch"));
}
// Peers forgotten in the meantime are ignored: an operator decided they are gone.
@@ -157,14 +157,14 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor
write_decision(host, id, decision).await?;
// The local switch reuses it instead of building again.
- host.compaction().store_prebuilt(id, snapshot);
+ host.remake().store_prebuilt(id, snapshot);
- info!(compaction = %id, to_epoch = %to_epoch, snapshot_bytes, "Compaction committed");
+ info!(remake = %id, to_epoch = %to_epoch, snapshot_bytes, "Remake committed");
Ok(())
}
async fn decide_abort<H: Host>(host: &H, id: &str, reason: String) -> Result<(), StoreError> {
- warn!(compaction = %id, reason = %reason, "Compaction aborted");
+ warn!(remake = %id, reason = %reason, "Remake aborted");
write_decision(
host,
id,
diff --git a/src/compaction/delta.rs b/src/remake/delta.rs
index cf556b2..b07c862 100644
--- a/src/compaction/delta.rs
+++ b/src/remake/delta.rs
@@ -10,11 +10,11 @@
//! * Map entries are replayed per key, so changes to sibling keys made on different
//! nodes don't overwrite each other.
//! * Lists and text are replayed whole: concurrent edits of the same list or text on two
-//! nodes around a compaction resolve as last-writer-wins.
+//! nodes around a remake resolve as last-writer-wins.
//! * Edits inside a map that lost a merge conflict produce no patch in Automerge and are
//! not replayed. Such maps only exist briefly between concurrent creations.
//!
-//! The protocol's own metadata (`_continuum/compactions`, `_continuum/epochs`) is never
+//! The protocol's own metadata (`_continuum/remakes`, `_continuum/epochs`) is never
//! replayed: the snapshot owns it.
use std::collections::{BTreeMap, HashSet};
@@ -22,7 +22,7 @@ use std::collections::{BTreeMap, HashSet};
use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop};
use tracing::warn;
-use super::snapshot::{COMPACTIONS, EPOCHS, META_ROOT};
+use super::snapshot::{EPOCHS, META_ROOT, REMAKES};
use crate::document::store::{self, StoreError};
#[derive(Debug, Default, Clone, Copy)]
@@ -96,7 +96,7 @@ fn log_failure(path: &[String], error: &StoreError) {
}
fn is_protocol_path(path: &[String]) -> bool {
- matches!(path, [root, part, ..] if root == META_ROOT && (part == COMPACTIONS || part == EPOCHS))
+ matches!(path, [root, part, ..] if root == META_ROOT && (part == REMAKES || part == EPOCHS))
}
/// True for the root and for `_continuum` itself.
@@ -362,7 +362,7 @@ mod tests {
.unwrap();
new.commit();
- // The old epoch replaces `_continuum` wholesale and writes a compaction record.
+ // The old epoch replaces `_continuum` wholesale and writes a remake record.
store::write_path(
&mut old,
&[META_ROOT.into()],
@@ -371,7 +371,7 @@ mod tests {
.unwrap();
store::write_path(
&mut old,
- &[META_ROOT.into(), COMPACTIONS.into(), "c".into()],
+ &[META_ROOT.into(), REMAKES.into(), "c".into()],
&json!({ "x": 1 }),
)
.unwrap();
@@ -382,6 +382,6 @@ mod tests {
let meta = &view(&new)[META_ROOT];
assert_eq!(meta[EPOCHS]["1"]["generation"], json!(1));
assert_eq!(meta["forgotten"]["pk"], json!(1));
- assert!(meta.get(COMPACTIONS).is_none());
+ assert!(meta.get(REMAKES).is_none());
}
}
diff --git a/src/compaction/local.rs b/src/remake/local.rs
index ec40b06..35a4ff1 100644
--- a/src/compaction/local.rs
+++ b/src/remake/local.rs
@@ -1,4 +1,4 @@
-//! Per-node, in-memory compaction state: snapshots built ahead of the switch, switch
+//! 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.
//!
@@ -17,7 +17,7 @@ use super::snapshot::Snapshot;
use crate::document::replica::Epoch;
#[derive(Default)]
-pub(crate) struct Compaction {
+pub(crate) struct RemakeState {
// A std mutex: every access is short and never spans an await.
inner: Mutex<Inner>,
}
@@ -26,7 +26,7 @@ pub(crate) struct Compaction {
struct Inner {
/// Built by the coordinator at commit time; consumed by the local switch.
prebuilt: HashMap<String, Snapshot>,
- /// When this node first saw a compaction committed.
+ /// When this node first saw a remake committed.
first_seen: HashMap<String, Instant>,
status: HashMap<String, LocalStatus>,
archive: Vec<Archived>,
@@ -82,7 +82,7 @@ pub(crate) struct CatchUpReport {
pub(crate) error: Option<String>,
}
-/// What happened to a compaction on this node.
+/// What happened to a remake on this node.
#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct LocalStatus {
@@ -118,7 +118,7 @@ pub(crate) struct ArchivedEpoch {
pub(crate) archived_at: u64,
}
-impl Compaction {
+impl RemakeState {
fn lock(&self) -> MutexGuard<'_, Inner> {
self.inner
.lock()
@@ -133,7 +133,7 @@ impl Compaction {
self.lock().prebuilt.remove(id)
}
- /// When this node first saw the compaction committed (now, on the first call).
+ /// When this node first saw the remake committed (now, on the first call).
pub(crate) fn first_seen(&self, id: &str) -> Instant {
*self
.lock()
@@ -273,14 +273,14 @@ mod tests {
#[test]
fn archive_retention() {
- let compaction = Compaction::default();
+ let remake = RemakeState::default();
let epoch = |generation| Epoch {
generation,
id: [generation as u8; 32],
};
// Kept by epoch count only.
- compaction.archive(
+ remake.archive(
epoch(0),
AutoCommit::new(),
Retention {
@@ -289,12 +289,12 @@ mod tests {
},
1,
);
- assert_eq!(compaction.archived_epochs().len(), 1);
- compaction.prune_archive(2);
- assert!(compaction.archived_epochs().is_empty());
+ assert_eq!(remake.archived_epochs().len(), 1);
+ remake.prune_archive(2);
+ assert!(remake.archived_epochs().is_empty());
// Kept by time while the epoch count has passed.
- compaction.archive(
+ remake.archive(
epoch(1),
AutoCommit::new(),
Retention {
@@ -303,10 +303,10 @@ mod tests {
},
5,
);
- assert_eq!(compaction.archived_epochs().len(), 1);
+ assert_eq!(remake.archived_epochs().len(), 1);
// No limits at all: not kept.
- compaction.archive(
+ remake.archive(
epoch(2),
AutoCommit::new(),
Retention {
@@ -315,6 +315,6 @@ mod tests {
},
3,
);
- assert_eq!(compaction.archived_epochs().len(), 1);
+ assert_eq!(remake.archived_epochs().len(), 1);
}
}
diff --git a/src/compaction/mod.rs b/src/remake/mod.rs
index 41f011f..8cca57c 100644
--- a/src/compaction/mod.rs
+++ b/src/remake/mod.rs
@@ -1,7 +1,8 @@
-//! Compaction: replacing the document with a snapshot in a new epoch.
+//! Remakes: coordinated moves of the whole cluster to a new document, a new epoch.
+//! Compaction is one: the new document has the old one's content without its history.
//!
-//! * `snapshot`: deterministic snapshot build (step 3).
-//! * `records`: the replicated compaction records under `_continuum/compactions/<id>`.
+//! * `snapshot`: deterministic build of the next epoch's initial document.
+//! * `records`: the replicated remake records under `_continuum/remakes/<id>`.
//! * `coordinator`: proposal, settle period and decision on the node that was asked.
//! * `switch`: switching this node to a committed epoch (worker).
//! * `catchup`: rejoining a newer epoch from peers (late, new or failed nodes), then
@@ -9,7 +10,7 @@
//! * `quarantine`: inspecting, replaying or discarding quarantined changes.
//! * `delta`: replaying this node's changes since the base into the new epoch.
//! * `local`: per-node in-memory state (prebuilt snapshots, switch status, archive).
-//! * `api`: `/admin/compaction(s)` handlers.
+//! * `api`: `/admin/remakes` handlers.
//!
//! The rest of the node is reached exclusively through [`Host`], so this module never
//! depends on `AppState`, peers or sync internals.
@@ -24,7 +25,7 @@ mod records;
pub(crate) mod snapshot;
mod switch;
-pub(crate) use local::Compaction;
+pub(crate) use local::RemakeState;
pub(crate) use switch::run_worker;
use std::future::Future;
@@ -41,12 +42,12 @@ use tokio::sync::broadcast;
use crate::document::replica::Epoch;
use crate::swarm::sync::ChangeSet;
-/// What compaction needs from the node. Implemented by `AppState`.
+/// What remake needs from the node. Implemented by `AppState`.
///
/// Used generically (`H: Host`), never as `dyn Host`, so the generic methods are fine.
pub(crate) trait Host: Clone + Send + Sync + 'static {
- /// This node's in-memory compaction state.
- fn compaction(&self) -> &Compaction;
+ /// This node's in-memory remake state.
+ fn remake(&self) -> &RemakeState;
/// This node's peer ID.
fn node_id(&self) -> String;
@@ -129,12 +130,15 @@ pub(crate) struct PeerSummary {
pub(crate) fn router<H: Host>() -> Router<H> {
Router::new()
.route(
- "/admin/compaction",
- post(api::start::<H>).delete(api::abort_current::<H>),
+ "/admin/remakes",
+ get(api::list::<H>)
+ .post(api::start::<H>)
+ .delete(api::abort_current::<H>),
)
- .route("/admin/compactions", get(api::list::<H>))
+ // The original endpoint, kept as a shortcut for a compacting remake.
+ .route("/admin/compaction", post(api::start::<H>))
.route(
- "/admin/compactions/{id}",
+ "/admin/remakes/{id}",
get(api::get::<H>).delete(api::abort::<H>),
)
.route("/admin/epochs", get(api::epochs::<H>))
diff --git a/src/compaction/quarantine.rs b/src/remake/quarantine.rs
index ad3c616..5643549 100644
--- a/src/compaction/quarantine.rs
+++ b/src/remake/quarantine.rs
@@ -59,11 +59,11 @@ fn not_found() -> Response {
}
pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response {
- Json(host.compaction().list_quarantine(summary)).into_response()
+ Json(host.remake().list_quarantine(summary)).into_response()
}
pub(crate) async fn get<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
- let Some((summary, document, paths)) = host.compaction().with_quarantined(&id, |entry| {
+ let Some((summary, document, paths)) = host.remake().with_quarantined(&id, |entry| {
(summary(entry), entry.document.clone(), entry.paths.clone())
}) else {
return not_found();
@@ -93,7 +93,7 @@ pub(crate) async fn get<H: Host>(State(host): State<H>, Path(id): Path<String>)
pub(crate) async fn document<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
let Some(mut document) = host
- .compaction()
+ .remake()
.with_quarantined(&id, |entry| entry.document.clone())
else {
return not_found();
@@ -123,7 +123,7 @@ pub(crate) async fn replay<H: Host>(State(host): State<H>, Path(id): Path<String
if !host.is_node_ready().await {
return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
}
- let Some(entry) = host.compaction().take_quarantined(&id) else {
+ let Some(entry) = host.remake().take_quarantined(&id) else {
return not_found();
};
@@ -137,7 +137,7 @@ pub(crate) async fn replay<H: Host>(State(host): State<H>, Path(id): Path<String
if stats.failed > 0 && stats.replayed == 0 {
// Nothing went in; keep the entry for another look.
- host.compaction().restore_quarantined(entry);
+ host.remake().restore_quarantined(entry);
return error_response(
StatusCode::CONFLICT,
"none of the writes could be replayed",
@@ -150,7 +150,7 @@ pub(crate) async fn replay<H: Host>(State(host): State<H>, Path(id): Path<String
}
pub(crate) async fn discard<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
- match host.compaction().take_quarantined(&id) {
+ match host.remake().take_quarantined(&id) {
Some(_) => {
info!(quarantine = %id, "Discarded quarantined changes");
StatusCode::NO_CONTENT.into_response()
diff --git a/src/compaction/records.rs b/src/remake/records.rs
index f533ee2..219fc9b 100644
--- a/src/compaction/records.rs
+++ b/src/remake/records.rs
@@ -1,4 +1,4 @@
-//! Compaction records: `_continuum/compactions/<id>` in the replicated document.
+//! Remake records: `_continuum/remakes/<id>` in the replicated document.
//!
//! A record has up to three parts, each with a single kind of writer, so concurrent
//! writes never fight over the same key:
@@ -19,7 +19,7 @@ use automerge::{AutoCommit, ChangeHash};
use serde::{Deserialize, Serialize};
use tracing::warn;
-use super::snapshot::{SnapshotRequest, COMPACTIONS, META_ROOT};
+use super::snapshot::{SnapshotRequest, META_ROOT, REMAKES};
use crate::document::replica::Epoch;
use crate::document::store::{self, StoreError};
@@ -187,7 +187,7 @@ impl Phase {
}
impl Record {
- /// The epoch this compaction committed to, if committed and readable.
+ /// The epoch this remake committed to, if committed and readable.
pub(crate) fn committed_epoch(&self) -> Option<Epoch> {
match &self.decision {
Some(Decision::Committed { to_epoch, .. }) => to_epoch.to_epoch(),
@@ -195,11 +195,11 @@ impl Record {
}
}
- /// Everything needed to (re)build this compaction's snapshot; None if the proposal
+ /// Everything needed to (re)build this remake's snapshot; None if the proposal
/// is unreadable.
pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> {
Some(SnapshotRequest {
- compaction_id: id.to_owned(),
+ remake_id: id.to_owned(),
previous: self.proposal.from_epoch.to_epoch()?,
base_heads: self.proposal.base_change_hashes()?,
policy: serde_json::to_value(&self.proposal.policy).ok()?,
@@ -220,7 +220,7 @@ impl Record {
// --- document access ---------------------------------------------------------
fn records_dir() -> Vec<String> {
- vec![META_ROOT.to_owned(), COMPACTIONS.to_owned()]
+ vec![META_ROOT.to_owned(), REMAKES.to_owned()]
}
pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> {
@@ -238,14 +238,14 @@ pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> {
Ok(Some(value)) => value,
Ok(None) => return None,
Err(error) => {
- warn!(compaction = %id, error = %error, "Failed to read compaction record");
+ warn!(remake = %id, error = %error, "Failed to read remake record");
return None;
}
};
match serde_json::from_value(value) {
Ok(record) => Some(record),
Err(error) => {
- warn!(compaction = %id, error = %error, "Skipping unreadable compaction record");
+ warn!(remake = %id, error = %error, "Skipping unreadable remake record");
None
}
}
diff --git a/src/compaction/snapshot.rs b/src/remake/snapshot.rs
index 0d49d48..da23fdf 100644
--- a/src/compaction/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -1,6 +1,6 @@
//! Deterministic snapshot of the document at agreed base heads.
//!
-//! Every node builds the snapshot of a committed compaction on its own. For that to
+//! Every node builds the snapshot of a committed remake on its own. For that to
//! work, the snapshot must be byte-identical everywhere: same operations, same order,
//! same actor, same timestamp. Its change hash then identifies the new epoch, and a
//! node whose build produces a different hash knows it must fetch the snapshot instead.
@@ -8,7 +8,7 @@
//! Determinism rests on:
//! * the content: the merged view at the base heads (`store::materialize_into`), which
//! depends only on the base, not on how a node's history got there;
-//! * a fixed actor derived from the compaction ID and generation;
+//! * a fixed actor derived from the remake ID and generation;
//! * commit time 0 and no message;
//! * canonical (sorted) JSON for the metadata;
//! * the same Automerge version on every node (encodings may differ across versions).
@@ -33,20 +33,20 @@ use crate::document::replica::Epoch;
use crate::document::store::{self, StoreError};
pub(crate) const META_ROOT: &str = "_continuum";
-pub(crate) const COMPACTIONS: &str = "compactions";
+pub(crate) const REMAKES: &str = "remakes";
pub(crate) const EPOCHS: &str = "epochs";
/// Everything that determines a snapshot besides the base content.
pub(crate) struct SnapshotParams<'a> {
- /// ID of the committed compaction; seeds the snapshot actor.
- pub(crate) compaction_id: &'a str,
+ /// ID of the committed remake; seeds the snapshot actor.
+ pub(crate) remake_id: &'a str,
/// The epoch being compacted. The snapshot starts generation `previous + 1`.
pub(crate) previous: Epoch,
/// Heads of the base. Every node must hold all of these changes.
pub(crate) base_heads: &'a [ChangeHash],
- /// Retention policy from the compaction request; stored verbatim (canonicalized).
+ /// Retention policy from the remake request; stored verbatim (canonicalized).
pub(crate) policy: &'a JsonValue,
- /// When the compaction was proposed (unix ms); recorded for replay windows.
+ /// When the remake was proposed (unix ms); recorded for replay windows.
pub(crate) proposed_at: u64,
}
@@ -77,7 +77,7 @@ pub(crate) enum SnapshotError {
/// Owned version of [`SnapshotParams`], for moving into a blocking task.
#[derive(Clone, Debug)]
pub(crate) struct SnapshotRequest {
- pub(crate) compaction_id: String,
+ pub(crate) remake_id: String,
pub(crate) previous: Epoch,
pub(crate) base_heads: Vec<ChangeHash>,
pub(crate) policy: JsonValue,
@@ -87,7 +87,7 @@ pub(crate) struct SnapshotRequest {
impl SnapshotRequest {
pub(crate) fn params(&self) -> SnapshotParams<'_> {
SnapshotParams {
- compaction_id: &self.compaction_id,
+ remake_id: &self.remake_id,
previous: self.previous,
base_heads: &self.base_heads,
policy: &self.policy,
@@ -114,12 +114,12 @@ pub(crate) async fn build_detached(
/// The protocol's view of an epoch, as recorded in its snapshot.
#[derive(Clone, Debug)]
pub(crate) struct EpochMetadata {
- pub(crate) compaction_id: String,
+ pub(crate) remake_id: String,
pub(crate) previous: Epoch,
/// Highest sequence number per actor of `previous` contained in the snapshot.
pub(crate) clock: BTreeMap<String, u64>,
pub(crate) policy: Policy,
- /// When the compaction was proposed (unix ms): the point the cluster moved on.
+ /// When the remake was proposed (unix ms): the point the cluster moved on.
pub(crate) proposed_at: u64,
}
@@ -127,7 +127,7 @@ pub(crate) struct EpochMetadata {
#[serde(rename_all = "camelCase")]
struct MetadataRecord {
generation: u64,
- compaction_id: String,
+ remake_id: String,
previous: EpochRecord,
clock: BTreeMap<String, u64>,
policy: Policy,
@@ -153,7 +153,7 @@ fn parse_metadata(doc: &AutoCommit, generation: u64) -> Result<EpochMetadata, St
.ok_or_else(|| "invalid previous epoch in metadata".to_owned())?;
Ok(EpochMetadata {
- compaction_id: record.compaction_id,
+ remake_id: record.remake_id,
previous,
clock: record.clock,
policy: record.policy,
@@ -183,7 +183,7 @@ pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<Epoc
}
/// Metadata of an earlier generation, as the current document holds it. Snapshots
-/// carry the metadata of all earlier epochs forward (only compaction records stay
+/// carry the metadata of all earlier epochs forward (only remake records stay
/// behind), so a node many epochs behind can still find the epoch that followed its own.
pub(crate) fn read_epoch_metadata(
doc: &AutoCommit,
@@ -217,18 +217,17 @@ pub(crate) fn build(
let clock = clock_of(&mut base);
let generation = params.previous.generation + 1;
- let mut snapshot =
- AutoCommit::new().with_actor(snapshot_actor(params.compaction_id, generation));
+ let mut snapshot = AutoCommit::new().with_actor(snapshot_actor(params.remake_id, generation));
- // Compaction records describe the old epoch's protocol and stay behind.
- store::materialize_into(&base, &mut snapshot, &is_compaction_record)?;
+ // Remake records describe the old epoch's protocol and stay behind.
+ store::materialize_into(&base, &mut snapshot, &is_remake_record)?;
let mut base_heads: Vec<String> = params.base_heads.iter().map(|h| h.to_string()).collect();
base_heads.sort();
let metadata = canonical(json!({
"generation": generation,
- "compactionId": params.compaction_id,
+ "remakeId": params.remake_id,
"previous": {
"generation": params.previous.generation,
"id": params.previous.id_hex(),
@@ -267,19 +266,19 @@ fn clock_of(doc: &mut AutoCommit) -> BTreeMap<String, u64> {
clock
}
-/// The same 16-byte actor on every node, unique per compaction and generation.
-fn snapshot_actor(compaction_id: &str, generation: u64) -> ActorId {
+/// The same 16-byte actor on every node, unique per remake and generation.
+fn snapshot_actor(remake_id: &str, generation: u64) -> ActorId {
let digest = Sha256::new()
.chain_update(b"continuum.snapshot\0")
- .chain_update(compaction_id.as_bytes())
+ .chain_update(remake_id.as_bytes())
.chain_update(b"\0")
.chain_update(generation.to_be_bytes())
.finalize();
ActorId::from(&digest[..16])
}
-fn is_compaction_record(path: &[String]) -> bool {
- matches!(path, [root, records] if root == META_ROOT && records == COMPACTIONS)
+fn is_remake_record(path: &[String]) -> bool {
+ matches!(path, [root, records] if root == META_ROOT && records == REMAKES)
}
/// Rebuilds every object with sorted keys, so the result is the same whichever key
@@ -309,7 +308,7 @@ mod tests {
fn params<'a>(heads: &'a [ChangeHash], policy: &'a JsonValue) -> SnapshotParams<'a> {
SnapshotParams {
- compaction_id: "c-1",
+ remake_id: "c-1",
previous: Epoch::GENESIS,
base_heads: heads,
policy,
@@ -459,11 +458,11 @@ mod tests {
}
#[test]
- fn metadata_written_and_compaction_records_dropped() {
+ fn metadata_written_and_remake_records_dropped() {
let mut a = doc(b"actor-a");
store::write_path(
&mut a,
- &[META_ROOT.into(), COMPACTIONS.into(), "c-1".into()],
+ &[META_ROOT.into(), REMAKES.into(), "c-1".into()],
&json!({ "state": "committed" }),
)
.unwrap();
@@ -482,12 +481,12 @@ mod tests {
let meta = store::read_path(&snap.doc, &[META_ROOT.into()])
.unwrap()
.unwrap();
- assert!(meta.get(COMPACTIONS).is_none());
+ assert!(meta.get(REMAKES).is_none());
assert_eq!(meta["forgotten"]["pk"]["at"], json!(1));
let epoch = &meta[EPOCHS]["1"];
assert_eq!(epoch["generation"], json!(1));
- assert_eq!(epoch["compactionId"], json!("c-1"));
+ assert_eq!(epoch["remakeId"], json!("c-1"));
assert_eq!(epoch["previous"]["generation"], json!(0));
assert_eq!(epoch["policy"], policy);
assert_eq!(epoch["baseHeads"].as_array().unwrap().len(), heads.len());
@@ -585,7 +584,7 @@ mod tests {
#[test]
fn fetched_document_is_verified_and_late_changes_replayed() {
- use crate::compaction::delta;
+ use crate::remake::delta;
// Old epoch: the base, then a late write by this node.
let mut late = doc(b"late-node");
@@ -610,7 +609,7 @@ mod tests {
let mut fetched = AutoCommit::load(&bytes).unwrap();
let metadata = verify_document(&mut fetched, snap.epoch).unwrap();
assert_eq!(metadata.previous, Epoch::GENESIS);
- assert_eq!(metadata.compaction_id, "c-1");
+ assert_eq!(metadata.remake_id, "c-1");
assert_eq!(
metadata.policy.replay_window,
Policy::default_replay_window()
diff --git a/src/compaction/switch.rs b/src/remake/switch.rs
index 67ff471..b4ab829 100644
--- a/src/compaction/switch.rs
+++ b/src/remake/switch.rs
@@ -1,6 +1,6 @@
//! Switching this node to a committed epoch.
//!
-//! The worker watches the compaction records. When a compaction of the local epoch is
+//! The worker watches the remake records. When a remake of the local epoch is
//! committed, it waits one keepalive interval (so the commit record can reach peers
//! through ordinary sync before this node stops syncing the old epoch), then:
//!
@@ -28,7 +28,7 @@ use super::catchup;
use super::delta::{self, ReplayStats};
use super::local::LocalStatus;
use super::records::{self, unix_millis, EpochRecord, Record};
-use super::snapshot::{self, COMPACTIONS, META_ROOT};
+use super::snapshot::{self, META_ROOT, REMAKES};
use super::Host;
use crate::document::replica::Epoch;
use crate::document::store;
@@ -37,9 +37,9 @@ use crate::swarm::sync::paths_related;
const TICK: Duration = Duration::from_secs(1);
pub(crate) async fn run_worker<H: Host>(host: H) {
- info!("Starting compaction worker");
+ info!("Starting remake worker");
- let records_dir = vec![META_ROOT.to_owned(), COMPACTIONS.to_owned()];
+ let records_dir = vec![META_ROOT.to_owned(), REMAKES.to_owned()];
let mut changes = host.subscribe_changes();
let mut changes_closed = false;
let mut tick = tokio::time::interval(TICK);
@@ -53,7 +53,7 @@ pub(crate) async fn run_worker<H: Host>(host: H) {
Ok(set) => set.paths.iter().any(|path| paths_related(path, &records_dir)),
Err(broadcast::error::RecvError::Lagged(_)) => true,
Err(broadcast::error::RecvError::Closed) => {
- warn!("Change events closed; compaction worker continues on its timer only");
+ warn!("Change events closed; remake worker continues on its timer only");
changes_closed = true;
false
}
@@ -68,11 +68,11 @@ pub(crate) async fn run_worker<H: Host>(host: H) {
async fn check<H: Host>(host: &H) {
let local = host.current_epoch();
- host.compaction().prune_archive(local.generation);
+ host.remake().prune_archive(local.generation);
// While rejoining, the document is being filled from peers: its records are not
// complete enough to act on.
- if host.compaction().is_rejoining() {
+ if host.remake().is_rejoining() {
return catchup::check(host).await;
}
@@ -94,29 +94,29 @@ async fn check<H: Host>(host: &H) {
// The settle protocol prevents this; never guess which one is right. Peers
// that did switch can still be followed.
let ids: Vec<&str> = several.iter().map(|(id, _)| id.as_str()).collect();
- error!(compactions = ?ids, "Several compactions committed for this epoch; not switching on my own");
+ error!(remakes = ?ids, "Several remakes committed for this epoch; not switching on my own");
return catchup::check(host).await;
}
};
if host
- .compaction()
+ .remake()
.status(&id)
.is_some_and(|status| status.error.is_some())
{
- // Our own switch failed (reported in the compaction's status): fetch the epoch
+ // Our own switch failed (reported in the remake's status): fetch the epoch
// from a peer that made it.
return catchup::check(host).await;
}
let switch_delay = host.dormant_after() / 4; // one keepalive interval
- if host.compaction().first_seen(&id).elapsed() < switch_delay {
+ if host.remake().first_seen(&id).elapsed() < switch_delay {
return;
}
match switch(host, &id, &record).await {
Ok(stats) => {
- host.compaction().set_status(
+ host.remake().set_status(
&id,
LocalStatus {
switched_at: Some(unix_millis()),
@@ -128,8 +128,8 @@ async fn check<H: Host>(host: &H) {
);
}
Err(reason) => {
- error!(compaction = %id, reason = %reason, "Epoch switch failed");
- host.compaction().set_status(
+ error!(remake = %id, reason = %reason, "Epoch switch failed");
+ host.remake().set_status(
&id,
LocalStatus {
error: Some(reason),
@@ -143,7 +143,7 @@ async fn check<H: Host>(host: &H) {
async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplayStats, String> {
let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id))
else {
- return Err("unreadable compaction record".to_owned());
+ return Err("unreadable remake record".to_owned());
};
let previous = request.previous;
let base = request.base_heads.clone();
@@ -155,12 +155,12 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
let carry_record = record.proposal.coordinator == host.node_id();
let record_path = records::record_path(id, "");
- info!(compaction = %id, from = %previous, to = %expected, "Switching epoch");
+ info!(remake = %id, from = %previous, to = %expected, "Switching epoch");
// 1. Consistent copy, under the read lock only.
let copy = host.read_doc(|doc| doc.clone()).await;
let prebuilt = host
- .compaction()
+ .remake()
.take_prebuilt(id)
.filter(|snapshot| snapshot.epoch == expected);
@@ -183,10 +183,10 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
// A fresh actor: this node's writes in the new epoch.
let mut doc = snapshot.doc.with_actor(ActorId::random());
- // Keep this compaction's record visible in the new epoch, for its history.
+ // Keep this remake's record visible in the new epoch, for its history.
if carry_record {
if let Err(error) = store::copy_path(&copy, &mut doc, &record_path) {
- warn!(error = %error, "Could not carry the compaction record into the new epoch");
+ warn!(error = %error, "Could not carry the remake record into the new epoch");
}
}
@@ -202,11 +202,11 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
let old = swap_in(host, expected, new_doc, copy_heads).await?;
// 4. Keep the old document as configured.
- host.compaction()
+ host.remake()
.archive(previous, old, retention, expected.generation);
info!(
- compaction = %id,
+ remake = %id,
epoch = %expected,
replayed = stats.replayed,
failed = stats.failed,
diff --git a/src/state.rs b/src/state.rs
index 18bface..51ca58a 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -12,8 +12,8 @@ use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock};
use tokio::time::Instant;
use tracing::{debug, error, warn};
-use crate::compaction::{Compaction, Host, PeerSummary};
use crate::document::replica::{Epoch, Replica};
+use crate::remake::{Host, PeerSummary, RemakeState};
use crate::swarm::sync::{ChangeSet, SyncEvent};
use crate::transport;
use crate::transport::firewall::Firewall;
@@ -298,8 +298,8 @@ pub(crate) struct AppState {
pub(crate) timing: Timing,
- /// Per-node compaction state (prebuilt snapshots, switch status, archive).
- pub(crate) compaction: Arc<Compaction>,
+ /// Per-node remake state (prebuilt snapshots, switch status, archive).
+ pub(crate) remake: Arc<RemakeState>,
/// While rejoining a newer epoch, the previous document, served read-only (and
/// marked stale) instead of answering 503.
@@ -621,10 +621,10 @@ pub(crate) fn unix_millis() -> u64 {
.unwrap_or(0)
}
-/// The node as seen by the compaction module.
+/// The node as seen by the remake module.
impl Host for AppState {
- fn compaction(&self) -> &Compaction {
- &self.compaction
+ fn remake(&self) -> &RemakeState {
+ &self.remake
}
fn node_id(&self) -> String {
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 2907e93..47fadfa 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -54,7 +54,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()>
if let Some(cluster_id) = state.cluster_id().await {
// A node joining a compacted cluster rejoins the newer epoch first; it is
// ready only once that rejoin verified the epoch.
- if state.initial_sync_done.load(Ordering::Acquire) && !state.compaction.is_rejoining() {
+ if state.initial_sync_done.load(Ordering::Acquire) && !state.remake.is_rejoining() {
state.mark_ready(cluster_id.clone()).await?;
info!(cluster_id = %cluster_id, "Node is ready");
return Ok(());
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index eb97a7e..b092b03 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -534,7 +534,7 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool
.send(SyncEvent::PeerNeedsSync(peer.to_owned()));
}
}
- // Expected after a compaction: either side switches or catches up.
+ // Expected after a remake: either side switches or catches up.
EpochRelation::PeerAhead | EpochRelation::PeerBehind => {
info!(
peer = %peer,
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-23 09:52:51 +0000