rename compaction to remake

1 files changed, 248 insertions, 0 deletions

diff --git a/src/remake/switch.rs b/src/remake/switch.rs
new file mode 100644
index 0000000..b4ab829
--- /dev/null
+++ b/src/remake/switch.rs
@@ -0,0 +1,248 @@
+//! Switching this node to a committed epoch.
+//!
+//! 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:
+//!
+//! 1. clones the live document under the read lock;
+//! 2. without any lock: takes the coordinator's prebuilt snapshot or builds it, checks
+//!    that its hash is the committed epoch ID, and replays the clone's changes since
+//!    the base into it;
+//! 3. under the write lock: replays whatever arrived since the clone, swaps document and
+//!    epoch, and hands back the old document;
+//! 4. archives the old document according to the policy.
+//!
+//! The write lock is held only for step 3, which is proportional to the changes made
+//! during steps 1-2.
+//!
+//! A node whose build doesn't match the committed epoch, or that never saw the commit,
+//! rejoins the new epoch from peers instead (`catchup`).
+
+use std::time::Duration;
+
+use automerge::{ActorId, AutoCommit, ChangeHash};
+use tokio::sync::broadcast;
+use tracing::{error, info, warn};
+
+use super::catchup;
+use super::delta::{self, ReplayStats};
+use super::local::LocalStatus;
+use super::records::{self, unix_millis, EpochRecord, Record};
+use super::snapshot::{self, META_ROOT, REMAKES};
+use super::Host;
+use crate::document::replica::Epoch;
+use crate::document::store;
+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 remake worker");
+
+    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);
+    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+
+    loop {
+        // React promptly to changes of the records; the timer covers the switch delay.
+        let should_check = tokio::select! {
+            _ = tick.tick() => true,
+            received = changes.recv(), if !changes_closed => match received {
+                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; remake worker continues on its timer only");
+                    changes_closed = true;
+                    false
+                }
+            }
+        };
+
+        if should_check {
+            check(&host).await;
+        }
+    }
+}
+
+async fn check<H: Host>(host: &H) {
+    let local = host.current_epoch();
+    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.remake().is_rejoining() {
+        return catchup::check(host).await;
+    }
+
+    let all = host.read_doc(records::load_all).await;
+    let local_record = EpochRecord::from(local);
+
+    let committed: Vec<(String, Record)> = all
+        .into_iter()
+        .filter(|(_, record)| {
+            record.proposal.from_epoch == local_record && record.committed_epoch().is_some()
+        })
+        .collect();
+
+    let (id, record) = match committed.as_slice() {
+        // Nothing committed here; maybe peers moved on without us.
+        [] => return catchup::check(host).await,
+        [(id, record)] => (id.clone(), record.clone()),
+        several => {
+            // 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!(remakes = ?ids, "Several remakes committed for this epoch; not switching on my own");
+            return catchup::check(host).await;
+        }
+    };
+
+    if host
+        .remake()
+        .status(&id)
+        .is_some_and(|status| status.error.is_some())
+    {
+        // 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.remake().first_seen(&id).elapsed() < switch_delay {
+        return;
+    }
+
+    match switch(host, &id, &record).await {
+        Ok(stats) => {
+            host.remake().set_status(
+                &id,
+                LocalStatus {
+                    switched_at: Some(unix_millis()),
+                    replayed_paths: Some(stats.replayed),
+                    failed_paths: Some(stats.failed),
+                    quarantine: None,
+                    error: None,
+                },
+            );
+        }
+        Err(reason) => {
+            error!(remake = %id, reason = %reason, "Epoch switch failed");
+            host.remake().set_status(
+                &id,
+                LocalStatus {
+                    error: Some(reason),
+                    ..LocalStatus::default()
+                },
+            );
+        }
+    }
+}
+
+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 remake record".to_owned());
+    };
+    let previous = request.previous;
+    let base = request.base_heads.clone();
+    let retention = record.proposal.policy.archive_retention.clone();
+    // Only the coordinator carries the record into the new epoch: one writer, so the
+    // nodes don't all write the same path concurrently right after switching (a merge
+    // conflict, plus one change per node to sync). The epoch metadata in the snapshot
+    // keeps the essentials on every node anyway.
+    let carry_record = record.proposal.coordinator == host.node_id();
+    let record_path = records::record_path(id, "");
+
+    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
+        .remake()
+        .take_prebuilt(id)
+        .filter(|snapshot| snapshot.epoch == expected);
+
+    // 2. Build (or reuse), verify and replay without holding any lock.
+    let (new_doc, copy_heads, stats) = tokio::task::spawn_blocking(
+        move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> {
+            let mut copy = copy;
+            let snapshot = match prebuilt {
+                Some(snapshot) => snapshot,
+                None => snapshot::build(&mut copy, &request.params())
+                    .map_err(|error| format!("snapshot build failed: {error}"))?,
+            };
+            if snapshot.epoch != expected {
+                return Err(format!(
+                    "built snapshot {} does not match the committed epoch {expected}",
+                    snapshot.epoch
+                ));
+            }
+
+            // A fresh actor: this node's writes in the new epoch.
+            let mut doc = snapshot.doc.with_actor(ActorId::random());
+
+            // 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 remake record into the new epoch");
+                }
+            }
+
+            let heads = copy.get_heads();
+            let stats = delta::replay(&mut copy, &base, &heads, &mut doc);
+            Ok((doc, heads, stats))
+        },
+    )
+    .await
+    .map_err(|error| format!("switch task failed: {error}"))??;
+
+    // 3. Swap, replaying what arrived since the copy.
+    let old = swap_in(host, expected, new_doc, copy_heads).await?;
+
+    // 4. Keep the old document as configured.
+    host.remake()
+        .archive(previous, old, retention, expected.generation);
+
+    info!(
+        remake = %id,
+        epoch = %expected,
+        replayed = stats.replayed,
+        failed = stats.failed,
+        "Switched to the new epoch"
+    );
+    Ok(stats)
+}
+
+/// Under the write lock: replays into `new_doc` whatever the live document received
+/// since `copy_heads` (the heads of the copy `new_doc` was prepared from), then swaps it
+/// in as `target`. Returns the old document.
+pub(super) async fn swap_in<H: Host>(
+    host: &H,
+    target: Epoch,
+    new_doc: AutoCommit,
+    copy_heads: Vec<ChangeHash>,
+) -> Result<AutoCommit, String> {
+    host.switch_epoch(target, move |live| {
+        // Only one worker switches, but be sure the copy still describes the live document.
+        if !copy_heads
+            .iter()
+            .all(|head| live.get_change_by_hash(head).is_some())
+        {
+            return Err("the document was replaced while preparing the switch".to_owned());
+        }
+        let mut new_doc = new_doc;
+        let live_heads = live.get_heads();
+        let late = delta::replay(live, &copy_heads, &live_heads, &mut new_doc);
+        if late.replayed + late.failed > 0 {
+            info!(
+                replayed = late.replayed,
+                failed = late.failed,
+                "Replayed late changes"
+            );
+        }
+        Ok(new_doc)
+    })
+    .await
+}