diff --git a/crates/commitlog/src/payload/txdata.rs b/crates/commitlog/src/payload/txdata.rs index 36a09a8848b..ab9dc570d27 100644 --- a/crates/commitlog/src/payload/txdata.rs +++ b/crates/commitlog/src/payload/txdata.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::{io, sync::Arc}; use bitflags::bitflags; use spacetimedb_sats::buffer::{BufReader, BufWriter, DecodeError}; @@ -528,6 +528,8 @@ pub enum DecoderError { Visitor(V), #[error(transparent)] Traverse(#[from] error::Traversal), + #[error(transparent)] + Io(#[from] io::Error), } /// A free standing implementation of [`crate::Decoder::skip_record`] diff --git a/crates/datastore/src/locking_tx_datastore/replay.rs b/crates/datastore/src/locking_tx_datastore/replay.rs index d803360551d..eeb60a82da8 100644 --- a/crates/datastore/src/locking_tx_datastore/replay.rs +++ b/crates/datastore/src/locking_tx_datastore/replay.rs @@ -29,6 +29,7 @@ use spacetimedb_schema::table_name::TableName; use spacetimedb_table::indexes::RowPointer; use spacetimedb_table::table::{InsertError, RowRef}; use std::cell::RefCell; +use std::io; use std::sync::Arc; use thiserror::Error; @@ -104,6 +105,8 @@ pub enum ReplayError { Db(#[from] DatastoreError), #[error(transparent)] Any(#[from] anyhow::Error), + #[error(transparent)] + Io(#[from] io::Error), } /// A [`spacetimedb_commitlog::Decoder`] suitable for replaying a transaction diff --git a/crates/durability/src/imp/local.rs b/crates/durability/src/imp/local.rs index e3eca56e5d9..4411523da6a 100644 --- a/crates/durability/src/imp/local.rs +++ b/crates/durability/src/imp/local.rs @@ -1,5 +1,6 @@ use std::{ io, + marker::PhantomData, num::NonZeroUsize, sync::{ atomic::{AtomicU64, Ordering::Relaxed}, @@ -12,13 +13,13 @@ use itertools::Itertools as _; use log::{info, trace, warn}; use scopeguard::ScopeGuard; use spacetimedb_commitlog::{ - error, + commitlog, error, payload::Txdata, - repo::{Fs, Repo, RepoWithoutLockFile}, - Commit, Commitlog, CompressionStats, Decoder, Encode, Transaction, + repo::{self, Fs, Repo, RepoWithoutLockFile}, + Commit, Commitlog, CompressionStats, Decoder, Encode, Transaction, DEFAULT_LOG_FORMAT_VERSION, }; use spacetimedb_fs_utils::lockfile::advisory::{LockError, LockedFile}; -use spacetimedb_paths::server::ReplicaDir; +use spacetimedb_paths::server::{CommitLogDir, ReplicaDir}; use spacetimedb_runtime::{Handle, JoinHandle}; use thiserror::Error; use tokio::sync::watch; @@ -92,6 +93,10 @@ where { /// The [`Commitlog`] this [`Durability`] and [`History`] impl wraps. clog: Arc, R>>, + /// Copy of the underlying [`Repo`]. + /// + /// Used to obtain a [LocalHistory] via [Self::as_history]. + repo: R, /// The durable transaction offset, as reported by the background /// [`FlushAndSyncTask`]. durable_offset: watch::Receiver>, @@ -133,12 +138,9 @@ impl Local { // yet for backwards-compatibility, we keep using the `db.lock` file. let lock = LockedFile::lock(replica_dir.0.join("db.lock"))?; - let clog = Arc::new(Commitlog::open( - replica_dir.commit_log(), - opts.commitlog, - on_new_segment, - )?); - Self::open_inner(clog, rt, opts, Some(lock)) + let repo = repo::Fs::new(replica_dir.commit_log(), on_new_segment)?; + let clog = Commitlog::open_with_repo(repo.clone(), opts.commitlog).map(Arc::new)?; + Self::open_inner(clog, repo, rt, opts, Some(lock)) } } @@ -150,8 +152,8 @@ where /// Create a [`Local`] instance backed by the provided commitlog repo. pub fn open_with_repo(repo: R, rt: Handle, opts: Options) -> Result { info!("open local durability"); - let clog = Arc::new(Commitlog::open_with_repo(repo, opts.commitlog)?); - Self::open_inner(clog, rt, opts, None) + let clog = Arc::new(Commitlog::open_with_repo(repo.clone(), opts.commitlog)?); + Self::open_inner(clog, repo, rt, opts, None) } } @@ -162,6 +164,7 @@ where { fn open_inner( clog: Arc, R>>, + repo: R, rt: Handle, opts: Options, lock: Option, @@ -184,6 +187,7 @@ where Ok(Self { clog, + repo, durable_offset: durable_rx, queue, queue_depth, @@ -193,7 +197,16 @@ where /// Obtain a read-only copy of the durable state that implements [History]. pub fn as_history(&self) -> impl History> + use { - self.clog.clone() + let tx_range = { + let min = self.clog.min_committed_offset().unwrap_or_default(); + let max = self.clog.max_committed_offset(); + (min, max) + }; + LocalHistory { + repo: self.repo.clone(), + tx_range, + _payload: PhantomData, + } } } @@ -383,39 +396,65 @@ where } } -impl History for Commitlog, R> -where - T: Encode + 'static, - R: Repo + Send + Sync + 'static, -{ +/// [History] impl that operates on a local [Repo] `R`. +pub struct LocalHistory { + repo: R, + tx_range: (TxOffset, Option), + _payload: PhantomData, +} + +impl LocalHistory { + /// Open the commitlog at `dir` as a read-only [History]. + pub fn open(dir: CommitLogDir) -> io::Result { + let repo = repo::Fs::new(dir, None)?; + let meta = commitlog::committed_meta(&repo)?; + let tx_range = match meta { + // Log is empty + None => (0, None), + Some(committed) => { + let max = committed.metadata().tx_range.end.saturating_sub(1); + let min = repo.existing_offsets()?.first().copied().unwrap_or_default(); + + (min, Some(max)) + } + }; + + Ok(Self { + repo, + tx_range, + _payload: PhantomData, + }) + } +} + +impl History for LocalHistory { type TxData = Txdata; fn fold_transactions_from(&self, offset: TxOffset, decoder: D) -> Result<(), D::Error> where D: Decoder, - D::Error: From, + D::Error: From + From, { - self.fold_transactions_from(offset, decoder) + commitlog::fold_transactions_from(&self.repo, DEFAULT_LOG_FORMAT_VERSION, offset, decoder) } fn transactions_from<'a, D>( - &self, + &'a self, offset: TxOffset, decoder: &'a D, - ) -> impl Iterator, D::Error>> + ) -> impl Iterator, D::Error>> + 'a where D: Decoder, - D::Error: From, + D::Error: From + From, Self::TxData: 'a, { - self.transactions_from(offset, decoder) + commitlog::transactions_from(&self.repo, DEFAULT_LOG_FORMAT_VERSION, offset, decoder) + .into_iter() + .flatten() } fn tx_range_hint(&self) -> (TxOffset, Option) { - let min = self.min_committed_offset().unwrap_or_default(); - let max = self.max_committed_offset(); - - (min, max) + self.tx_range } } diff --git a/crates/durability/src/lib.rs b/crates/durability/src/lib.rs index 8bdcbb3b922..3ef4a28de1c 100644 --- a/crates/durability/src/lib.rs +++ b/crates/durability/src/lib.rs @@ -1,4 +1,4 @@ -use std::{iter, marker::PhantomData, sync::Arc}; +use std::{io, iter, marker::PhantomData, sync::Arc}; use futures::future::BoxFuture; use thiserror::Error; @@ -190,17 +190,17 @@ pub trait History { fn fold_transactions_from(&self, offset: TxOffset, decoder: D) -> Result<(), D::Error> where D: Decoder, - D::Error: From; + D::Error: From + From; /// Obtain an iterator over the history of transactions, starting from `offset`. fn transactions_from<'a, D>( - &self, + &'a self, offset: TxOffset, decoder: &'a D, - ) -> impl Iterator, D::Error>> + ) -> impl Iterator, D::Error>> + 'a where D: Decoder, - D::Error: From, + D::Error: From + From, Self::TxData: 'a; /// Get the maximum transaction offset contained in this history. @@ -224,19 +224,19 @@ impl History for Arc { fn fold_transactions_from(&self, offset: TxOffset, decoder: D) -> Result<(), D::Error> where D: Decoder, - D::Error: From, + D::Error: From + From, { (**self).fold_transactions_from(offset, decoder) } fn transactions_from<'a, D>( - &self, + &'a self, offset: TxOffset, decoder: &'a D, - ) -> impl Iterator, D::Error>> + ) -> impl Iterator, D::Error>> + 'a where D: Decoder, - D::Error: From, + D::Error: From + From, Self::TxData: 'a, { (**self).transactions_from(offset, decoder) @@ -273,7 +273,7 @@ impl History for EmptyHistory { &self, _offset: TxOffset, _decoder: &'a D, - ) -> impl Iterator, D::Error>> + ) -> impl Iterator, D::Error>> + 'a where D: Decoder, D::Error: From, diff --git a/crates/engine/src/relational_db.rs b/crates/engine/src/relational_db.rs index 5ba71da15cb..9428a6d5caf 100644 --- a/crates/engine/src/relational_db.rs +++ b/crates/engine/src/relational_db.rs @@ -8,7 +8,7 @@ use crate::MetricsRecorderQueue; use anyhow::{anyhow, Context}; use enum_map::EnumMap; use spacetimedb_commitlog::repo::OnNewSegmentFn; -use spacetimedb_commitlog::{self as commitlog, Commitlog, SizeOnDisk}; +use spacetimedb_commitlog::{self as commitlog, SizeOnDisk}; use spacetimedb_data_structures::map::HashSet; use spacetimedb_datastore::db_metrics::DB_METRICS; use spacetimedb_datastore::error::{DatastoreError, TableError, ViewError}; @@ -35,6 +35,7 @@ use spacetimedb_datastore::{ }, traits::TxData, }; +use spacetimedb_durability::local::LocalHistory; use spacetimedb_durability::{self as durability, History}; use spacetimedb_lib::bsatn::ToBsatn; use spacetimedb_lib::db::auth::StAccess; @@ -1767,7 +1768,7 @@ pub async fn local_history( runtime: &Handle, ) -> io::Result + use<>> { let commitlog_dir = replica_dir.commit_log(); - asyncify(runtime, move || Commitlog::open(commitlog_dir, <_>::default(), None)).await + asyncify(runtime, move || LocalHistory::open(commitlog_dir)).await } /// Watches snapshot creation events and compresses all commitlog segments older diff --git a/crates/smoketests/src/lib.rs b/crates/smoketests/src/lib.rs index 1fc9521dade..67d4f8d2dcb 100644 --- a/crates/smoketests/src/lib.rs +++ b/crates/smoketests/src/lib.rs @@ -1111,6 +1111,10 @@ impl Smoketest { pub fn spacetime_cmd(&self, args: &[&str]) -> Output { let start = Instant::now(); let cli_path = self.cli_path(); + + let cmd_name = args.first().unwrap_or(&"unknown"); + eprintln!("[TIMING] spacetime {cmd_name}: starting at {start:?}"); + let output = Command::new(&cli_path) .arg("--config-path") .arg(&self.config_path) @@ -1119,8 +1123,8 @@ impl Smoketest { .output() .expect("Failed to execute spacetime command"); - let cmd_name = args.first().unwrap_or(&"unknown"); - eprintln!("[TIMING] spacetime {}: {:?}", cmd_name, start.elapsed()); + eprintln!("[TIMING] spacetime {}: completed in {:?}", cmd_name, start.elapsed()); + output }