add epoch switch

1 files changed, 198 insertions, 0 deletions

diff --git a/src/compaction/local.rs b/src/compaction/local.rs
new file mode 100644
index 0000000..6290a5e
--- /dev/null
+++ b/src/compaction/local.rs
@@ -0,0 +1,198 @@
+//! Per-node, in-memory compaction state: snapshots built ahead of the switch, switch
+//! progress for reports, and archived documents of previous epochs.
+//!
+//! Nothing here is replicated or durable. After a restart the node simply has no
+//! archive and rebuilds snapshots when needed.
+
+use std::collections::HashMap;
+use std::sync::{Mutex, MutexGuard};
+
+use automerge::AutoCommit;
+use serde::Serialize;
+use tokio::time::Instant;
+
+use super::records::{unix_millis, Retention};
+use super::snapshot::Snapshot;
+use crate::replica::Epoch;
+
+#[derive(Default)]
+pub(crate) struct Compaction {
+    // A std mutex: every access is short and never spans an await.
+    inner: Mutex<Inner>,
+}
+
+#[derive(Default)]
+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.
+    first_seen: HashMap<String, Instant>,
+    status: HashMap<String, LocalStatus>,
+    archive: Vec<Archived>,
+}
+
+/// What happened to a compaction on this node.
+#[derive(Clone, Debug, Default, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct LocalStatus {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) switched_at: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) replayed_paths: Option<usize>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) failed_paths: Option<usize>,
+    /// Why the switch did not happen (yet).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) error: Option<String>,
+}
+
+struct Archived {
+    epoch: Epoch,
+    // Kept for later steps (serving or auditing old epochs); not read yet.
+    #[allow(dead_code)]
+    doc: AutoCommit,
+    archived_at: u64,
+    retention: Retention,
+}
+
+/// Summary of an archived epoch.
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct ArchivedEpoch {
+    pub(crate) generation: u64,
+    pub(crate) id: String,
+    pub(crate) archived_at: u64,
+}
+
+impl Compaction {
+    fn lock(&self) -> MutexGuard<'_, Inner> {
+        self.inner
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    pub(crate) fn store_prebuilt(&self, id: &str, snapshot: Snapshot) {
+        self.lock().prebuilt.insert(id.to_owned(), snapshot);
+    }
+
+    pub(crate) fn take_prebuilt(&self, id: &str) -> Option<Snapshot> {
+        self.lock().prebuilt.remove(id)
+    }
+
+    /// When this node first saw the compaction committed (now, on the first call).
+    pub(crate) fn first_seen(&self, id: &str) -> Instant {
+        *self
+            .lock()
+            .first_seen
+            .entry(id.to_owned())
+            .or_insert_with(Instant::now)
+    }
+
+    pub(crate) fn status(&self, id: &str) -> Option<LocalStatus> {
+        self.lock().status.get(id).cloned()
+    }
+
+    pub(crate) fn set_status(&self, id: &str, status: LocalStatus) {
+        self.lock().status.insert(id.to_owned(), status);
+    }
+
+    /// Keeps the previous epoch's document, then drops what has outlived its retention.
+    pub(crate) fn archive(
+        &self,
+        epoch: Epoch,
+        doc: AutoCommit,
+        retention: Retention,
+        current_generation: u64,
+    ) {
+        let mut inner = self.lock();
+        inner.archive.push(Archived {
+            epoch,
+            doc,
+            archived_at: unix_millis(),
+            retention,
+        });
+        prune(&mut inner.archive, current_generation);
+    }
+
+    pub(crate) fn prune_archive(&self, current_generation: u64) {
+        prune(&mut self.lock().archive, current_generation);
+    }
+
+    pub(crate) fn archived_epochs(&self) -> Vec<ArchivedEpoch> {
+        self.lock()
+            .archive
+            .iter()
+            .map(|archived| ArchivedEpoch {
+                generation: archived.epoch.generation,
+                id: archived.epoch.id_hex(),
+                archived_at: archived.archived_at,
+            })
+            .collect()
+    }
+}
+
+/// An archive is kept until every limit of its retention has passed.
+fn prune(archive: &mut Vec<Archived>, current_generation: u64) {
+    let now = unix_millis();
+    archive.retain(|archived| {
+        let within_epochs = archived.retention.epochs.is_some_and(|epochs| {
+            current_generation.saturating_sub(archived.epoch.generation) <= epochs
+        });
+        let within_time = archived.retention.seconds.is_some_and(|seconds| {
+            now.saturating_sub(archived.archived_at) <= seconds.saturating_mul(1000)
+        });
+        within_epochs || within_time
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn archive_retention() {
+        let compaction = Compaction::default();
+        let epoch = |generation| Epoch {
+            generation,
+            id: [generation as u8; 32],
+        };
+
+        // Kept by epoch count only.
+        compaction.archive(
+            epoch(0),
+            AutoCommit::new(),
+            Retention {
+                epochs: Some(1),
+                seconds: None,
+            },
+            1,
+        );
+        assert_eq!(compaction.archived_epochs().len(), 1);
+        compaction.prune_archive(2);
+        assert!(compaction.archived_epochs().is_empty());
+
+        // Kept by time while the epoch count has passed.
+        compaction.archive(
+            epoch(1),
+            AutoCommit::new(),
+            Retention {
+                epochs: Some(0),
+                seconds: Some(3600),
+            },
+            5,
+        );
+        assert_eq!(compaction.archived_epochs().len(), 1);
+
+        // No limits at all: not kept.
+        compaction.archive(
+            epoch(2),
+            AutoCommit::new(),
+            Retention {
+                epochs: None,
+                seconds: None,
+            },
+            3,
+        );
+        assert_eq!(compaction.archived_epochs().len(), 1);
+    }
+}