diff --git a/crates/commitlog/src/commitlog.rs b/crates/commitlog/src/commitlog.rs index 9082000dcfd..a2855f953f1 100644 --- a/crates/commitlog/src/commitlog.rs +++ b/crates/commitlog/src/commitlog.rs @@ -242,6 +242,38 @@ impl Generic { 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); diff --git a/crates/commitlog/src/lib.rs b/crates/commitlog/src/lib.rs index 599c71c0136..b9167adc527 100644 --- a/crates/commitlog/src/lib.rs +++ b/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> { 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); + } +} diff --git a/crates/durability/src/imp/local.rs b/crates/durability/src/imp/local.rs index 4411523da6a..8ecd2c4c5eb 100644 --- a/crates/durability/src/imp/local.rs +++ b/crates/durability/src/imp/local.rs @@ -235,6 +235,14 @@ where pub fn compress_segments(&self, offsets: &[TxOffset]) -> io::Result { 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 Local { diff --git a/crates/engine/src/persistence.rs b/crates/engine/src/persistence.rs index de152b0c6f7..ff547c71db6 100644 --- a/crates/engine/src/persistence.rs +++ b/crates/engine/src/persistence.rs @@ -41,6 +41,99 @@ pub struct CommitlogConfig { pub write_buffer_size: Option, } +/// 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,13 +290,20 @@ 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, durability: DurabilityConfig, + retention: RetentionConfig, } impl LocalPersistenceProvider { @@ -211,6 +311,7 @@ impl LocalPersistenceProvider { Self { data_dir: data_dir.into(), durability: DurabilityConfig::default(), + retention: RetentionConfig::default(), } } @@ -218,6 +319,11 @@ impl LocalPersistenceProvider { 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, diff --git a/crates/engine/src/relational_db.rs b/crates/engine/src/relational_db.rs index 7edb9fe20fc..44b735aa405 100644 --- a/crates/engine/src/relational_db.rs +++ b/crates/engine/src/relational_db.rs @@ -64,6 +64,7 @@ use spacetimedb_table::table::{RowRef, TableScanIter}; use spacetimedb_table::table_index::IndexKey; use std::borrow::Cow; use std::io; +use std::num::NonZeroUsize; use std::ops::RangeBounds; use std::sync::Arc; @@ -1825,13 +1826,9 @@ pub async fn snapshot_watching_commitlog_compressor( // if the snapshot is in the middle of a segment, we want to round down. // [0, 2].binary_search(1) will return Err(1), so we subtract 1. .unwrap_or_else(|i| i.saturating_sub(1)); - let segment_offsets = &segment_offsets[start_idx..]; - let end_idx = segment_offsets - .binary_search(&snapshot_offset) - .unwrap_or_else(|i| i.saturating_sub(1)); - // in this case, segment_offsets[end_idx] is the segment that contains the snapshot, - // which we don't want to compress, so an exclusive range is correct. - let segment_offsets = &segment_offsets[..end_idx]; + // Exclude the segment which contains the snapshot, + // as it may still be written to. + let segment_offsets = commitlog::segments_older_than(&segment_offsets[start_idx..], snapshot_offset); durability.compress_segments(segment_offsets)?; let n = segment_offsets.len(); let last_compressed_segment = if n > 0 { Some(segment_offsets[n - 1]) } else { None }; @@ -1856,6 +1853,101 @@ pub async fn snapshot_watching_commitlog_compressor( } } +/// Watches snapshot creation events and deletes historical data that is no +/// longer needed to restart the database. +/// +/// Whenever a new snapshot is created, [`prune_history`] is invoked with the +/// given `retain_snapshots`, deleting all but the most recent snapshots along +/// with the prefix of the commitlog that precedes them. +/// +/// Suitable **only** for non-replicated databases. +pub async fn snapshot_watching_history_pruner( + mut snapshot_rx: watch::Receiver, + retain_snapshots: NonZeroUsize, + snapshot_repo: Arc, + durability: LocalDurability, + runtime: Handle, +) { + // Prune once at startup, then whenever a snapshot is created. The initial + // run deletes history that accumulated while the database was offline, or + // was running with a different retention policy, without waiting for the + // next snapshot. + loop { + let snapshot_repo = snapshot_repo.clone(); + let durability = durability.clone(); + let res = asyncify(&runtime, move || { + prune_history(&snapshot_repo, &durability, retain_snapshots) + }) + .await; + if let Err(err) = res { + tracing::warn!("failed to prune history: {err:#}"); + } + + if snapshot_rx.changed().await.is_err() { + break; + } + snapshot_rx.borrow_and_update(); + } +} + +/// Delete snapshots and commitlog segments that are no longer needed to +/// restart the database. +/// +/// The `retain_snapshots` most recent snapshots are retained, all older +/// snapshots are deleted. Commitlog segments whose entire contents precede +/// the oldest retained snapshot are also deleted, as the database will never +/// replay them again. +/// +/// Snapshots are captured from the committed, not necessarily durable, state +/// of the database, and a snapshot can only be restored from if its offset is +/// durable (see `RelationalDB::maybe_do_snapshot`). Snapshots above the +/// durable [`TxOffset`] are therefore ignored: they neither count towards +/// `retain_snapshots`, nor are they ever deleted. +/// +/// Nothing is deleted unless more than `retain_snapshots` durable snapshots +/// exist, or the commitlog extends before the oldest retained snapshot, +/// respectively. +pub fn prune_history( + snapshot_repo: &SnapshotRepository, + durability: &durability::Local, + retain_snapshots: NonZeroUsize, +) -> anyhow::Result<()> { + use durability::Durability as _; + + let database_identity = snapshot_repo.database_identity(); + + let Some(durable_offset) = durability.durable_tx_offset().get().context("durability has exited")? else { + // Nothing is durable yet, so no snapshot can be restored from. + return Ok(()); + }; + let Some(oldest_retained) = snapshot_repo.nth_latest_snapshot(retain_snapshots, durable_offset)? else { + // Fewer durable snapshots than the retention policy asks for, + // nothing to do. + return Ok(()); + }; + + // Delete all snapshots older than the oldest retained one. + snapshot_repo + .prune_snapshots_before(oldest_retained) + .with_context(|| format!("failed to delete snapshots of database {database_identity}"))?; + + // Delete all commitlog segments whose entire contents precede the oldest + // retained snapshot. + let segment_offsets = durability.existing_segment_offsets()?; + let segment_offsets = commitlog::segments_older_than(&segment_offsets, oldest_retained); + if !segment_offsets.is_empty() { + durability.remove_segments(segment_offsets).with_context(|| { + format!("failed to delete commitlog segments before {oldest_retained} of database {database_identity}") + })?; + tracing::info!( + "deleted {} commitlog segment(s) before {oldest_retained} of database {database_identity}", + segment_offsets.len() + ); + } + + Ok(()) +} + /// Open a [`SnapshotRepository`] at `db_path/snapshots`, /// configured to store snapshots of the database `database_identity`/`replica_id`. pub fn open_snapshot_repo( @@ -3809,6 +3901,115 @@ mod tests { Ok(()) } + /// Tests that [`prune_history`] deletes exactly the snapshots and + /// commitlog segments that are no longer needed to restart the database. + #[tokio::test(flavor = "multi_thread")] + async fn prune_history_deletes_stale_snapshots_and_segments() -> ResultTest<()> { + use durability::Durability as _; + use tests_utils::TempReplicaDir; + + /// Fabricate an entry on disk which [`SnapshotRepository::all_snapshots`] + /// recognizes as a complete snapshot. + fn fake_snapshot(repo: &SnapshotRepository, offset: TxOffset) { + let dir = repo.snapshot_dir_path(offset); + std::fs::create_dir_all(&dir.0).unwrap(); + std::fs::write(dir.snapshot_file(offset).0, []).unwrap(); + } + + let retain_snapshots = NonZeroUsize::new(2).unwrap(); + let dir = TempReplicaDir::new()?; + let durability = durability::Local::open( + (*dir).clone(), + Handle::tokio_current(), + durability::local::Options { + commitlog: commitlog::Options { + max_segment_size: 128, + ..<_>::default() + }, + ..<_>::default() + }, + None, + )?; + + // Commit transactions one at a time, waiting for each to become + // durable, so that the log rotates into multiple small segments. + for offset in 0..16u64 { + durability.append_tx(Box::new(durability::Transaction::from(( + offset, + Txdata { + inputs: None, + outputs: None, + mutations: Some(txdata::Mutations { + inserts: Box::new([txdata::Ops { + table_id: 8000.into(), + rowdata: Arc::new([product![offset]]), + }]), + deletes: Box::new([]), + truncates: Box::new([]), + }), + }, + )))); + durability.durable_tx_offset().wait_for(offset).await.unwrap(); + } + + let segments = durability.existing_segment_offsets()?; + assert!(segments.len() > 3, "expected more than 3 segments, got {segments:?}"); + + let snapshots_path = dir.snapshots(); + snapshots_path.create()?; + let repo = SnapshotRepository::open(snapshots_path, Identity::ZERO, 0)?; + + // With fewer snapshots than the retention policy asks for, + // nothing is deleted. + let oldest_snapshot = segments[1]; + fake_snapshot(&repo, oldest_snapshot); + prune_history(&repo, &durability, retain_snapshots).unwrap(); + assert_eq!(repo.all_snapshots()?.collect::>(), vec![oldest_snapshot]); + assert_eq!(durability.existing_segment_offsets()?, segments); + + // A leftover of an interrupted prune run. + fake_snapshot(&repo, 0); + repo.snapshot_dir_path(0).rename_as_archived()?; + + // A snapshot above the durable offset neither counts towards the + // retention limit, nor is it ever deleted. + let undurable_snapshot = 100; + fake_snapshot(&repo, undurable_snapshot); + + // Two more snapshots. The segment containing the older one, and all + // segments after it, must be retained. + let retained_snapshot = 14; + let latest_snapshot = 15; + let first_retained_segment = segments.iter().rposition(|&s| s <= retained_snapshot).unwrap(); + assert!( + (2..segments.len() - 1).contains(&first_retained_segment), + "test wants a snapshot in the middle of the log, \ + but segment offsets are {segments:?}" + ); + fake_snapshot(&repo, retained_snapshot); + fake_snapshot(&repo, latest_snapshot); + prune_history(&repo, &durability, retain_snapshots).unwrap(); + + let mut snapshots = repo.all_snapshots()?.collect::>(); + snapshots.sort_unstable(); + assert_eq!(snapshots, vec![retained_snapshot, latest_snapshot, undurable_snapshot]); + assert_eq!(repo.all_archived_snapshots()?.count(), 0); + assert_eq!( + durability.existing_segment_offsets()?, + segments[first_retained_segment..].to_vec() + ); + + // The remaining log covers the retained snapshot up to the latest + // transaction, i.e. the database can restart from it. + durability.close().await; + let commits = + commitlog::commits_from(dir.commit_log(), retained_snapshot + 1)?.collect::, _>>()?; + let last_commit = commits.last().expect("no commits after the retained snapshot"); + assert_eq!(last_commit.min_tx_offset + last_commit.n as u64 - 1, latest_snapshot); + + Ok(()) + } + // For test compression into an existing database. // Must supply the path to the database and the identity of the replica using the `ENV`: // - `SNAPSHOT` the path to the database, like `/tmp/db/replicas/.../8/database` diff --git a/crates/snapshot/src/lib.rs b/crates/snapshot/src/lib.rs index b04ee20216f..be34be4a710 100644 --- a/crates/snapshot/src/lib.rs +++ b/crates/snapshot/src/lib.rs @@ -54,6 +54,7 @@ use std::{ ffi::OsStr, fmt, io::{BufWriter, Read, Write}, + num::NonZeroUsize, ops::{Add, AddAssign}, path::PathBuf, }; @@ -1227,6 +1228,56 @@ impl SnapshotRepository { self.latest_snapshot_older_than(TxOffset::MAX) } + /// Return the `TxOffset` of the `n`-th most recent complete snapshot + /// at or below `upper_bound`, if one exists. + /// + /// When deleting or archiving historical snapshots while retaining the + /// `n` most recent ones, the returned offset is the cutoff: all snapshots + /// strictly before it can be removed from the repository, e.g. via + /// [`Self::prune_snapshots_before`]. + pub fn nth_latest_snapshot( + &self, + n: NonZeroUsize, + upper_bound: TxOffset, + ) -> Result, SnapshotError> { + let mut snapshots = self + .all_snapshots()? + .filter(|tx_offset| *tx_offset <= upper_bound) + .collect::>(); + snapshots.sort_unstable_by(|a, b| b.cmp(a)); + Ok(snapshots.get(n.get() - 1).copied()) + } + + /// Delete all complete snapshots strictly before `cutoff` from the + /// repository. + /// + /// Snapshot directories are renamed with [`ARCHIVED_SNAPSHOT_EXT`] before + /// deletion, so that a partially deleted snapshot is never mistaken for a + /// valid one. Leftover renamed directories of an earlier, interrupted run + /// are also deleted, regardless of their offset. + /// + /// Does not delete invalidated or locked snapshots. + pub fn prune_snapshots_before(&self, cutoff: TxOffset) -> Result<(), SnapshotError> { + for path in self.all_archived_snapshots()? { + match Self::remove_archived_snapshot(&path) { + Ok(()) => log::info!("deleted archived snapshot directory {}", path.display()), + Err(e) => warn!("failed to delete archived snapshot directory {}: {e}", path.display()), + } + } + + let stale = self + .all_snapshots()? + .filter(|tx_offset| *tx_offset < cutoff) + .collect::>(); + for tx_offset in stale { + let archived = self.snapshot_dir_path(tx_offset).rename_as_archived()?; + Self::remove_archived_snapshot(&archived)?; + log::info!("deleted snapshot {tx_offset} of database {}", self.database_identity()); + } + + Ok(()) + } + /// Rename any snapshot newer than `upper_bound` with [`INVALID_SNAPSHOT_DIR_EXT`]. /// /// When rebuilding a database, we will call this method @@ -1635,6 +1686,61 @@ mod tests { Ok(()) } + /// Create the on-disk representation of a complete snapshot at `tx_offset`. + fn complete_snapshot(repo: &SnapshotRepository, tx_offset: TxOffset) -> anyhow::Result<()> { + let dir = repo.snapshot_dir_path(tx_offset); + dir.create()?; + dir.snapshot_file(tx_offset) + .open_file(OpenOptions::new().write(true).create_new(true)) + .map(drop)?; + Ok(()) + } + + #[test] + fn nth_latest_snapshot_respects_count_and_upper_bound() -> anyhow::Result<()> { + let tmp = tempdir()?; + let root = SnapshotsPath::from_path_unchecked(tmp.path()); + let repo = SnapshotRepository::open(root, Identity::ZERO, 42)?; + + let one = NonZeroUsize::new(1).unwrap(); + let two = NonZeroUsize::new(2).unwrap(); + + assert_eq!(repo.nth_latest_snapshot(two, TxOffset::MAX)?, None); + + for tx_offset in [10, 20, 30] { + complete_snapshot(&repo, tx_offset)?; + } + + assert_eq!(repo.nth_latest_snapshot(two, TxOffset::MAX)?, Some(20)); + assert_eq!(repo.nth_latest_snapshot(two, 25)?, Some(10)); + assert_eq!(repo.nth_latest_snapshot(two, 15)?, None); + assert_eq!(repo.nth_latest_snapshot(one, 25)?, Some(20)); + + Ok(()) + } + + #[test] + fn prune_snapshots_before_deletes_stale_snapshots_and_leftovers() -> anyhow::Result<()> { + let tmp = tempdir()?; + let root = SnapshotsPath::from_path_unchecked(tmp.path()); + let repo = SnapshotRepository::open(root, Identity::ZERO, 42)?; + + for tx_offset in [10, 20, 30] { + complete_snapshot(&repo, tx_offset)?; + } + // A leftover of an earlier prune run that was interrupted after + // renaming a snapshot, but before deleting it from disk. + complete_snapshot(&repo, 5)?; + repo.snapshot_dir_path(5).rename_as_archived()?; + + repo.prune_snapshots_before(30)?; + + assert_eq!(vec![30], repo.all_snapshots()?.collect::>()); + assert_eq!(0, repo.all_archived_snapshots()?.count()); + + Ok(()) + } + #[test] fn listing_ignores_if_lockfile_exists() -> anyhow::Result<()> { let tmp = tempdir()?; diff --git a/crates/standalone/config.toml b/crates/standalone/config.toml index 9eeef5d3535..e1ce4966f2d 100644 --- a/crates/standalone/config.toml +++ b/crates/standalone/config.toml @@ -64,4 +64,23 @@ directives = [ # Size in bytes of the memory buffer holding commit data before flushing to storage. # write-buffer-size = 131072 +[retention] +# What to do with historical data, i.e. data that is no longer needed to +# restart the database: commitlog segments whose entire contents precede the +# most recent snapshots, and older snapshots. +# +# "delete": delete historical data from disk whenever a new snapshot is +# taken. This bounds disk usage. +# "keep": keep historical data on disk indefinitely. Historical commitlog +# segments are compressed, but disk usage grows without bound. +# +# If this key is omitted, historical data is kept, preserving the behavior of +# SpacetimeDB versions that predate this option. +policy = "delete" + +# Number of most recent snapshots to retain when policy = "delete". +# The commitlog is retained from the oldest retained snapshot onwards, so +# that the database can be restored from any retained snapshot. +# retain-snapshots = 2 + # vim: set nowritebackup: << otherwise triggers cargo-watch diff --git a/crates/standalone/src/lib.rs b/crates/standalone/src/lib.rs index f806f18e995..8fdaeffb671 100644 --- a/crates/standalone/src/lib.rs +++ b/crates/standalone/src/lib.rs @@ -12,7 +12,7 @@ use http::StatusCode; use spacetimedb::client::ClientActorIndex; use spacetimedb::config::{CertificateAuthority, MetadataFile, V8Config, WasmConfig}; use spacetimedb::db; -use spacetimedb::db::persistence::{DurabilityConfig, LocalPersistenceProvider}; +use spacetimedb::db::persistence::{DurabilityConfig, LocalPersistenceProvider, RetentionConfig}; use spacetimedb::energy::{EnergyBalance, EnergyQuanta, NullEnergyMonitor}; use spacetimedb::host::{DiskStorage, HostController, HostRuntimeConfig, MigratePlanResult, UpdateDatabaseResult}; use spacetimedb::identity::{AuthCtx, Identity}; @@ -43,6 +43,7 @@ pub use spacetimedb_client_api::routes::subscribe::{BIN_PROTOCOL, TEXT_PROTOCOL} pub struct StandaloneOptions { pub db_config: db::Config, pub durability: DurabilityConfig, + pub retention: RetentionConfig, pub websocket: WebSocketOptions, pub wasm: WasmConfig, pub v8: V8Config, @@ -78,8 +79,11 @@ impl StandaloneEnv { let energy_monitor = Arc::new(NullEnergyMonitor); let program_store = Arc::new(DiskStorage::new(data_dir.program_bytes().0).await?); - let persistence_provider = - Arc::new(LocalPersistenceProvider::new(data_dir.clone()).with_durability_config(config.durability)); + let persistence_provider = Arc::new( + LocalPersistenceProvider::new(data_dir.clone()) + .with_durability_config(config.durability) + .with_retention_config(config.retention), + ); let host_controller = HostController::new( data_dir, config.db_config, @@ -673,6 +677,7 @@ mod tests { page_pool_max_size: None, }, durability: DurabilityConfig::default(), + retention: RetentionConfig::default(), websocket: WebSocketOptions::default(), wasm: WasmConfig::default(), v8: V8Config::default(), diff --git a/crates/standalone/src/subcommands/start.rs b/crates/standalone/src/subcommands/start.rs index 01078f7af8d..9ccc87f6451 100644 --- a/crates/standalone/src/subcommands/start.rs +++ b/crates/standalone/src/subcommands/start.rs @@ -11,7 +11,7 @@ use axum::extract::DefaultBodyLimit; use clap::ArgAction::SetTrue; use clap::{Arg, ArgMatches}; use spacetimedb::config::{parse_config, CertificateAuthority}; -use spacetimedb::db::persistence::{CommitlogConfig, DurabilityConfig}; +use spacetimedb::db::persistence::{CommitlogConfig, DurabilityConfig, RetentionConfig}; use spacetimedb::db::{self, Storage}; use spacetimedb::startup::{self, TracingOptions}; use spacetimedb::util::jobs::JobCores; @@ -102,6 +102,8 @@ struct ConfigFile { #[serde(default)] commitlog: CommitlogConfig, #[serde(default)] + retention: RetentionConfig, + #[serde(default)] websocket: WebSocketOptions, } @@ -187,6 +189,7 @@ pub async fn exec(args: &ArgMatches, db_cores: JobCores) -> anyhow::Result<()> { durability: DurabilityConfig { commitlog: config.commitlog, }, + retention: config.retention, websocket: config.websocket, wasm: config.common.wasm, v8: config.common.v8, @@ -505,6 +508,7 @@ fn banner() { #[cfg(test)] mod tests { use super::*; + use spacetimedb::db::persistence::RetentionPolicy; use std::time::Duration; #[test] @@ -539,6 +543,10 @@ mod tests { offset-index-require-segment-fsync = false preallocate-segments = true write-buffer-size = 131072 + + [retention] + policy = "keep" + retain-snapshots = 5 "#; let config: ConfigFile = toml::from_str(toml).unwrap(); @@ -581,6 +589,24 @@ mod tests { ..<_>::default() } ); + + assert_eq!(config.retention.policy, RetentionPolicy::Keep); + assert_eq!(config.retention.retain_snapshots.get(), 5); + } + + /// The shipped `config.toml` template, which is written into every newly + /// created data directory, deletes historical data after it has been + /// snapshotted. A config without a `[retention]` section, as found in + /// data directories that predate the option, keeps historical data. + #[test] + fn retention_defaults() { + let template: ConfigFile = toml::from_str(include_str!("../../config.toml")).unwrap(); + assert_eq!(template.retention.policy, RetentionPolicy::Delete); + assert_eq!(template.retention.retain_snapshots.get(), 2); + + let unconfigured: ConfigFile = toml::from_str("").unwrap(); + assert_eq!(unconfigured.retention.policy, RetentionPolicy::Keep); + assert_eq!(unconfigured.retention.retain_snapshots.get(), 2); } #[test] diff --git a/crates/testing/src/modules.rs b/crates/testing/src/modules.rs index b03b87df9b5..b8f5fb97072 100644 --- a/crates/testing/src/modules.rs +++ b/crates/testing/src/modules.rs @@ -274,6 +274,7 @@ impl CompiledModule { spacetimedb_standalone::StandaloneOptions { db_config: config, durability: Default::default(), + retention: Default::default(), websocket: WebSocketOptions::default(), wasm: Default::default(), v8: Default::default(), diff --git a/docs/docs/00300-resources/00200-reference/00100-cli-reference/00200-standalone-config.md b/docs/docs/00300-resources/00200-reference/00100-cli-reference/00200-standalone-config.md index 6863d642a98..ec81cc8df33 100644 --- a/docs/docs/00300-resources/00200-reference/00100-cli-reference/00200-standalone-config.md +++ b/docs/docs/00300-resources/00200-reference/00100-cli-reference/00200-standalone-config.md @@ -20,6 +20,8 @@ On Linux and macOS, this directory is by default `~/.local/share/spacetime/data` - [`commitlog`](#commitlog) +- [`retention`](#retention) + - [`websocket`](#websocket) ### `certificate-authority` @@ -101,6 +103,32 @@ If `true`, preallocate disk space for commitlog segments up to `commitlog.max-se Size in bytes of the memory buffer holding commit data before flushing to storage. +### `retention` + +```toml +[retention] +policy = "delete" +retain-snapshots = 2 +``` + +The `retention` table configures what happens to historical data, i.e. data that is no longer needed to restart the database. This comprises commitlog segments whose entire contents precede the most recent snapshots, and all but the most recent snapshots. + +#### `retention.policy` + +Can be one of: + +- `"delete"`: Delete historical data from disk whenever a new snapshot is taken. This bounds disk usage. The `config.toml` written into newly created data directories sets this policy. + +- `"keep"`: Keep historical data on disk indefinitely. Historical commitlog segments are compressed, but disk usage grows without bound. Use this if you want to preserve the full transaction history of the database. + +If the key is omitted, historical data is kept. Data directories created by SpacetimeDB versions that predate this option therefore keep their history when the server is upgraded; add `policy = "delete"` to their `config.toml` to reclaim disk space. + +#### `retention.retain-snapshots` + +Number of most recent snapshots to retain when `retention.policy` is `"delete"`. Must be at least 1, the default is 2. + +The commitlog is retained from the oldest retained snapshot onwards, so that the database can be restored from any retained snapshot even if a newer one turns out to be unreadable. + ### `websocket` ```toml