Standalone: delete snapshotted history by default in new data directories

crates/commitlog/src/commitlog.rs

@@ -242,6 +242,38 @@ impl<R: Repo, T> Generic<R, T> {
         Self::open(self.repo.clone(), self.opts)
     }
 
+    /// Remove the segments at the given offsets from the log.
+    ///
+    /// `offsets` must contain the exact segment offsets, no rounding to the
+    /// nearest offset is performed.
+    ///
+    /// The segments' contents are discarded permanently, including their
+    /// offset indexes. The current writable segment cannot be removed: if
+    /// `offsets` contains an offset at or beyond it, an error is returned
+    /// and no segments are removed.
+    ///
+    /// Segments are removed oldest-first, so that a failure partway through
+    /// does not leave a gap in the log. If removing a segment fails, the
+    /// error is returned and the remaining, newer segments are retained.
+    pub fn remove_segments(&mut self, offsets: &[u64]) -> io::Result<()> {
+        let head_offset = self.head.min_tx_offset();
+        if let Some(offset) = offsets.iter().find(|&&offset| offset >= head_offset) {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                format!("refusing to remove mutable segment {offset}"),
+            ));
+        }
+        let mut offsets = offsets.to_vec();
+        offsets.sort_unstable();
+        for offset in offsets {
+            debug!("removing segment {offset}");
+            self.repo.remove_segment(offset)?;
+            self.tail.retain(|&tail_offset| tail_offset != offset);
+        }
+
+        Ok(())
+    }
+
     /// Start a new segment, preserving the current head's `Commit`.
     ///
     /// The caller must ensure that the current head is synced to disk as
@@ -1091,6 +1123,32 @@ mod tests {
         assert_eq!(&offsets, &[0, 3]);
     }
 
+    #[test]
+    fn remove_segments() {
+        let mut log = mem_log::<[u8; 32]>(32);
+        fill_log(&mut log, 10, repeat(1));
+
+        let offsets = log.repo.existing_offsets().unwrap();
+        assert!(offsets.len() > 3, "expected more than 3 segments, got {offsets:?}");
+
+        // Refuses to remove the writable segment, leaving the log untouched.
+        let err = log.remove_segments(&offsets).unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
+        assert_eq!(log.repo.existing_offsets().unwrap(), offsets);
+
+        // Removes historical segments and keeps the in-memory state consistent.
+        let (remove, retain) = offsets.split_at(2);
+        log.remove_segments(remove).unwrap();
+        assert_eq!(&log.repo.existing_offsets().unwrap(), retain);
+        assert_eq!(&log.tail, &retain[..retain.len() - 1]);
+
+        // The log is still traversable from the first retained segment.
+        let first_retained = retain[0];
+        for (offset, commit) in (first_retained..10).zip(log.commits_from(first_retained)) {
+            assert_eq!(offset, commit.unwrap().min_tx_offset);
+        }
+    }
+
     #[test]
     fn huge_commit() {
         let mut log = mem_log::<[u8; 32]>(32);

crates/commitlog/src/lib.rs

@@ -398,6 +398,26 @@ where
         Ok(stats)
     }
 
+    /// Remove the segments at the given offsets from the log.
+    ///
+    /// `offsets` must contain the exact segment offsets, no rounding to the
+    /// nearest offset is performed.
+    ///
+    /// The segments' contents are discarded permanently, including their
+    /// offset indexes. The latest, writable segment cannot be removed: if
+    /// `offsets` contains its offset, an error is returned and no segments
+    /// are removed.
+    ///
+    /// Segments are removed oldest-first, so that a failure partway through
+    /// does not leave a gap in the log. If removing a segment fails, the
+    /// error is returned and the remaining, newer segments are retained.
+    ///
+    /// This method acquires a write lock on this `Commitlog` instance for the
+    /// duration of the removal.
+    pub fn remove_segments(&self, offsets: &[u64]) -> io::Result<()> {
+        self.inner.write().unwrap().remove_segments(offsets)
+    }
+
     /// Remove all data from the log and reopen it.
     ///
     /// Log segments are deleted starting from the newest. As multiple segments
@@ -611,6 +631,25 @@ pub fn committed_meta(root: CommitLogDir) -> io::Result<Option<CommittedMeta>> {
     commitlog::committed_meta(repo::Fs::new(root, None)?)
 }
 
+/// Given a sorted (ascending) list of segment offsets, return the prefix of
+/// segments whose entire contents are strictly before `tx_offset`.
+///
+/// Segments are named for the earliest transaction they contain, so the
+/// segment which contains `tx_offset` itself is excluded.
+///
+/// Useful to determine which segments can be compressed, archived or removed
+/// once a snapshot at `tx_offset` exists: the returned segments will never be
+/// needed again to restore the database from that snapshot.
+pub fn segments_older_than(offsets: &[u64], tx_offset: u64) -> &[u64] {
+    let end_idx = offsets
+        .binary_search(&tx_offset)
+        // If `tx_offset` is in the middle of a segment, exclude the segment
+        // that contains it by rounding down.
+        .unwrap_or_else(|i| i.saturating_sub(1));
+    &offsets[..end_idx]
+}
+
+
 /// Obtain an iterator which traverses the commitlog located at the `root`
 /// directory from the start, yielding [`StoredCommit`]s.
 ///
@@ -700,3 +739,28 @@ where
 {
     commitlog::fold_transaction_range(repo::Fs::new(root, None)?, DEFAULT_LOG_FORMAT_VERSION, range, de)
 }
+#[cfg(test)]
+mod segments_older_than_tests {
+    use super::segments_older_than;
+
+    const SEGMENTS: &[u64] = &[0, 10, 20, 30];
+    const EMPTY: &[u64] = &[];
+
+    #[test]
+    fn excludes_the_segment_containing_the_offset() {
+        assert_eq!(segments_older_than(SEGMENTS, 15), &[0]);
+        assert_eq!(segments_older_than(SEGMENTS, 7), EMPTY);
+        assert_eq!(segments_older_than(SEGMENTS, 35), &[0, 10, 20]);
+    }
+
+    #[test]
+    fn excludes_the_segment_starting_at_the_offset() {
+        assert_eq!(segments_older_than(SEGMENTS, 20), &[0, 10]);
+        assert_eq!(segments_older_than(SEGMENTS, 0), EMPTY);
+    }
+
+    #[test]
+    fn handles_empty_input() {
+        assert_eq!(segments_older_than(EMPTY, 15), EMPTY);
+    }
+}

crates/durability/src/imp/local.rs

@@ -235,6 +235,14 @@ where
     pub fn compress_segments(&self, offsets: &[TxOffset]) -> io::Result<CompressionStats> {
         self.clog.compress_segments(offsets)
     }
+
+    /// Remove the segments at the offsets provided from the underlying
+    /// [`Commitlog`], discarding their contents permanently.
+    ///
+    /// See [`Commitlog::remove_segments`].
+    pub fn remove_segments(&self, offsets: &[TxOffset]) -> io::Result<()> {
+        self.clog.remove_segments(offsets)
+    }
 }
 
 impl<T: Send + Sync + 'static> Local<T, Fs> {

crates/engine/src/persistence.rs

@@ -41,6 +41,99 @@ pub struct CommitlogConfig {
     pub write_buffer_size: Option<NonZeroUsize>,
 }
 
+/// Retention of historical data exposed through server config.
+///
+/// Historical data is data which is no longer needed to restart the database:
+/// commitlog segments whose entire contents precede the most recent snapshots,
+/// and all but the most recent snapshots.
+//
+// TODO(config): Allow databases to override the retention policy individually.
+//
+// This config is currently server-wide, applied uniformly to every database
+// the server hosts. Retention is really a property of a database, not of the
+// server: how long history is worth keeping depends on the application, and
+// deleting history can even be a compliance requirement of a particular
+// database. When per-database configuration becomes available, the values
+// here should become the server's *defaults*: a database's own setting, when
+// present, always takes precedence, and neither level acts as a floor or
+// ceiling on the other. In particular, a database may retain history on a
+// server whose default is to delete it:
+//
+//   server default | database setting | effective
+//   ---------------+------------------+----------
+//   keep           | unset            | keep
+//   delete         | unset            | delete
+//   any            | keep             | keep
+//   any            | delete           | delete
+//
+// The same resolution applies to `retain_snapshots`.
+//
+// - The resolution belongs in [PersistenceProvider::persistence], which
+//   already receives the [Database] it is asked to provide services for.
+// - Storage-level settings must remain server config, as they describe the
+//   node's infrastructure rather than a database's lifecycle. (Standalone has
+//   none today; in SpacetimeDB-cloud, these are the archive backend URL and
+//   the bandwidth limits.)
+// - A per-database policy naming a capability the server does not have (e.g.
+//   `archive` on a server without an archive backend) cannot be rejected at
+//   server startup. It must be rejected when the per-database setting is
+//   accepted, and fall back to `keep` (never to `delete`) with a warning if
+//   it is encountered at runtime regardless.
+// - A change of the effective policy applies from the next prune run onwards.
+//   Switching to `delete` deletes all history accumulated so far; switching
+//   away from it does not bring deleted history back.
+#[derive(Clone, Copy, Debug, serde::Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "kebab-case")]
+pub struct RetentionConfig {
+    /// What to do with historical data.
+    #[serde(default)]
+    pub policy: RetentionPolicy,
+    /// Number of most recent snapshots to retain when `policy` is
+    /// [`RetentionPolicy::Delete`].
+    ///
+    /// The commitlog is retained from the oldest retained snapshot onwards,
+    /// so that the database can be restored from any retained snapshot.
+    ///
+    /// Default: 2
+    #[serde(default = "default_retain_snapshots")]
+    pub retain_snapshots: NonZeroUsize,
+}
+
+fn default_retain_snapshots() -> NonZeroUsize {
+    NonZeroUsize::new(2).unwrap()
+}
+
+impl Default for RetentionConfig {
+    fn default() -> Self {
+        Self {
+            policy: RetentionPolicy::default(),
+            retain_snapshots: default_retain_snapshots(),
+        }
+    }
+}
+
+/// Policy for handling historical data, see [`RetentionConfig`].
+#[derive(Clone, Copy, Debug, Default, serde::Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "kebab-case")]
+pub enum RetentionPolicy {
+    /// Delete historical data from disk whenever a new snapshot is taken,
+    /// bounding disk usage.
+    ///
+    /// SpacetimeDB Standalone writes a `config.toml` which enables this
+    /// policy into every newly created data directory.
+    Delete,
+    /// Keep historical data on disk indefinitely.
+    ///
+    /// Historical commitlog segments are compressed, but disk usage still
+    /// grows without bound.
+    ///
+    /// This is the default when no retention policy is configured, so that
+    /// data directories which predate the introduction of this option keep
+    /// their history when the server is upgraded.
+    #[default]
+    Keep,
+}
+
 impl DurabilityConfig {
     fn into_options(self) -> spacetimedb_durability::local::Options {
         let mut opts = spacetimedb_durability::local::Options::default();
@@ -197,27 +290,40 @@ pub trait PersistenceProvider: Send + Sync {
 /// [Persistence] services are provided for the local [ServerDataDir].
 ///
 /// Note that its [PersistenceProvider::persistence] impl will spawn a
-/// background task that [compresses] older commitlog segments whenever a
-/// snapshot is taken.
+/// background task that handles historical data whenever a snapshot is
+/// taken, according to the configured [RetentionConfig]:
 ///
+/// - [RetentionPolicy::Delete] [deletes] historical snapshots and commitlog
+///   segments,
+/// - [RetentionPolicy::Keep] (the default) [compresses] historical commitlog
+///   segments and keeps all data on disk.
+///
+/// [deletes]: relational_db::snapshot_watching_history_pruner
 /// [compresses]: relational_db::snapshot_watching_commitlog_compressor
 pub struct LocalPersistenceProvider {
     data_dir: Arc<ServerDataDir>,
     durability: DurabilityConfig,
+    retention: RetentionConfig,
 }
 
 impl LocalPersistenceProvider {
     pub fn new(data_dir: impl Into<Arc<ServerDataDir>>) -> Self {
         Self {
             data_dir: data_dir.into(),
             durability: DurabilityConfig::default(),
+            retention: RetentionConfig::default(),
         }
     }
 
     pub fn with_durability_config(mut self, durability: DurabilityConfig) -> Self {
         self.durability = durability;
         self
     }
+
+    pub fn with_retention_config(mut self, retention: RetentionConfig) -> Self {
+        self.retention = retention;
+        self
+    }
 }
 
 #[async_trait]
@@ -228,11 +334,17 @@ impl PersistenceProvider for LocalPersistenceProvider {
         let snapshot_dir = replica_dir.snapshots();
         let runtime = Handle::tokio_current();
 
-        let snapshot_worker = asyncify(&runtime, move || {
+        let snapshot_repo = asyncify(&runtime, move || {
             relational_db::open_snapshot_repo(snapshot_dir, database_identity, replica_id)
         })
-        .await
-        .map(|repo| SnapshotWorker::new(repo, snapshot::Compression::Enabled, runtime.clone()))?;
+        .await?;
+        // When historical snapshots are deleted anyway, compressing them
+        // beforehand is wasted work.
+        let compression = match self.retention.policy {
+            RetentionPolicy::Delete => snapshot::Compression::Disabled,
+            RetentionPolicy::Keep => snapshot::Compression::Enabled,
+        };
+        let snapshot_worker = SnapshotWorker::new(snapshot_repo.clone(), compression, runtime.clone());
         let (durability, disk_size) = relational_db::local_durability_with_options(
             replica_dir,
             runtime.clone(),
@@ -241,13 +353,26 @@ impl PersistenceProvider for LocalPersistenceProvider {
         )
         .await?;
 
-        runtime.spawn(relational_db::snapshot_watching_commitlog_compressor(
-            snapshot_worker.subscribe(),
-            None,
-            None,
-            durability.clone(),
-            runtime.clone(),
-        ));
+        match self.retention.policy {
+            RetentionPolicy::Delete => {
+                runtime.spawn(relational_db::snapshot_watching_history_pruner(
+                    snapshot_worker.subscribe(),
+                    self.retention.retain_snapshots,
+                    snapshot_repo,
+                    durability.clone(),
+                    runtime.clone(),
+                ));
+            }
+            RetentionPolicy::Keep => {
+                runtime.spawn(relational_db::snapshot_watching_commitlog_compressor(
+                    snapshot_worker.subscribe(),
+                    None,
+                    None,
+                    durability.clone(),
+                    runtime.clone(),
+                ));
+            }
+        }
 
         Ok(Persistence {
             durability,