1 files changed, 210 insertions, 0 deletions
diff --git a/src/compaction/switch.rs b/src/compaction/switch.rs
new file mode 100644
index 0000000..5eb1962
--- /dev/null
+++ b/src/compaction/switch.rs
@@ -0,0 +1,210 @@
+//! Switching this node to a committed epoch.
+//!
+//! The worker watches the compaction records. When a compaction 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 stays on the old epoch and
+//! reports the mismatch; fetching the snapshot from peers comes with step 6.
+
+use std::time::Duration;
+
+use automerge::{ActorId, AutoCommit};
+use tokio::sync::broadcast;
+use tracing::{error, info, warn};
+
+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::Host;
+use crate::store;
+use crate::sync::paths_related;
+
+const TICK: Duration = Duration::from_secs(1);
+
+pub(crate) async fn run_worker<H: Host>(host: H) {
+    info!("Starting compaction worker");
+
+    let records_dir = vec![META_ROOT.to_owned(), COMPACTIONS.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; compaction 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.compaction().prune_archive(local.generation);
+
+    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() {
+        [] => return,
+        [(id, record)] => (id.clone(), record.clone()),
+        several => {
+            // The settle protocol prevents this; never guess which one is right.
+            let ids: Vec<&str> = several.iter().map(|(id, _)| id.as_str()).collect();
+            error!(compactions = ?ids, "Several compactions committed for this epoch; not switching");
+            return;
+        }
+    };
+
+    if host
+        .compaction()
+        .status(&id)
+        .is_some_and(|status| status.error.is_some())
+    {
+        return; // already failed; reported in the compaction's status
+    }
+
+    let switch_delay = host.dormant_after() / 4; // one keepalive interval
+    if host.compaction().first_seen(&id).elapsed() < switch_delay {
+        return;
+    }
+
+    match switch(host, &id, &record).await {
+        Ok(stats) => {
+            host.compaction().set_status(
+                &id,
+                LocalStatus {
+                    switched_at: Some(unix_millis()),
+                    replayed_paths: Some(stats.replayed),
+                    failed_paths: Some(stats.failed),
+                    error: None,
+                },
+            );
+        }
+        Err(reason) => {
+            error!(compaction = %id, reason = %reason, "Epoch switch failed");
+            host.compaction().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 compaction record".to_owned());
+    };
+    let previous = request.previous;
+    let base = request.base_heads.clone();
+    let retention = record.proposal.policy.archive_retention.clone();
+    let record_path = records::record_path(id, "");
+
+    info!(compaction = %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()
+        .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<automerge::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 compaction's record visible in the new epoch, for its history.
+            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");
+            }
+
+            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 = host
+        .switch_epoch(expected, move |live| {
+            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::<_, String>(new_doc)
+        })
+        .await?;
+
+    // 4. Keep the old document as configured.
+    host.compaction()
+        .archive(previous, old, retention, expected.generation);
+
+    info!(
+        compaction = %id,
+        epoch = %expected,
+        replayed = stats.replayed,
+        failed = stats.failed,
+        "Switched to the new epoch"
+    );
+    Ok(stats)
+}