diff --git a/Cargo.lock b/Cargo.lock index 29d90b4db4..ab8d897008 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8211,7 +8211,16 @@ version = "2.5.0" dependencies = [ "anyhow", "clap 4.5.50", + "spacetimedb-commitlog", + "spacetimedb-datastore", + "spacetimedb-durability", + "spacetimedb-engine", + "spacetimedb-lib", + "spacetimedb-primitives", "spacetimedb-runtime", + "spacetimedb-sats", + "spacetimedb-schema", + "spacetimedb-table", "tracing", ] diff --git a/crates/dst/src/engine.rs b/crates/dst/src/engine.rs index 1ad73a01ba..03c5099dc3 100644 --- a/crates/dst/src/engine.rs +++ b/crates/dst/src/engine.rs @@ -2,7 +2,7 @@ use std::{io, sync::Arc}; use spacetimedb_commitlog::SizeOnDisk; use spacetimedb_datastore::execution_context::Workload; -use spacetimedb_datastore::traits::IsolationLevel; +use spacetimedb_datastore::traits::{IsolationLevel, TxData}; use spacetimedb_engine::error::DBError; use spacetimedb_engine::persistence::{DiskSizeFn, Durability as EngineDurability, Persistence}; use spacetimedb_engine::relational_db::{MutTx, RelationalDB}; @@ -14,13 +14,15 @@ use spacetimedb_schema::def::ModuleDef; use spacetimedb_schema::schema::{Schema, TableSchema}; use spacetimedb_table::page_pool::PagePool; +mod model; mod properties; mod workload; -use self::workload::{row_to_bytes, summarize_rows, Interaction, Observation, TableSummary}; +use self::workload::{row_to_bytes, summarize_rows, CommitDelta, Interaction, Observation, TableDelta, TableSummary}; +use crate::engine::model::Model; use crate::engine::properties::EngineProperties; -use crate::engine::workload::{Model, WorkloadGen}; +use crate::engine::workload::WorkloadGen; use crate::schema::{default_schema, lower_schema, SchemaPlan}; use crate::sim::commitlog::{InMemoryCommitlog, InMemoryCommitlogHandle}; use crate::traits::{TargetDriver, TestSuite}; @@ -154,6 +156,32 @@ impl EngineTarget { Ok(summaries) } + fn commit_delta_from_tx_data(&self, tx_data: &TxData) -> CommitDelta { + let mut tables = Vec::new(); + + for (table_id, entry) in tx_data.iter_table_entries() { + let Some(table) = self.table_ids.iter().position(|id| *id == table_id) else { + continue; + }; + + let inserts = summarize_rows(entry.inserts.as_ref()); + let deletes = summarize_rows(entry.deletes.as_ref()); + if inserts.count == 0 && deletes.count == 0 && !entry.truncated { + continue; + } + + tables.push(TableDelta { + table, + inserts, + deletes, + truncated: entry.truncated, + }); + } + + tables.sort_by_key(|delta| delta.table); + CommitDelta { tables } + } + pub fn execute(&mut self, interaction: &Interaction) -> anyhow::Result { match interaction { Interaction::BeginMutTx => { @@ -209,9 +237,11 @@ impl EngineTarget { .db .as_ref() .ok_or_else(|| anyhow::anyhow!("database is not open"))?; - db.finish_tx(tx, Ok::<(), anyhow::Error>(()))?; + let Some((_tx_offset, tx_data, _tx_metrics, _reducer)) = db.commit_tx(tx)? else { + anyhow::bail!("commit produced no transaction data"); + }; Ok(Observation::Committed { - summaries: self.table_summaries()?, + delta: self.commit_delta_from_tx_data(&tx_data), }) } Interaction::Count { table } => { diff --git a/crates/dst/src/engine/model.rs b/crates/dst/src/engine/model.rs new file mode 100644 index 0000000000..4f2dff86d1 --- /dev/null +++ b/crates/dst/src/engine/model.rs @@ -0,0 +1,155 @@ +use super::workload::{summarize_rows, CommitDelta, Interaction, Observation, Row, TableDelta, TableSummary}; +use crate::schema::SchemaPlan; + +#[derive(Debug)] +pub struct Model { + schema: SchemaPlan, + committed_tables: Vec, + pending_tables: Option>, +} + +#[derive(Debug, Clone)] +struct TableState { + rows: Vec, +} + +impl Model { + pub fn new(schema: SchemaPlan) -> Self { + let committed_tables = schema.tables.iter().map(|_| TableState { rows: vec![] }).collect(); + Self { + schema, + committed_tables, + pending_tables: None, + } + } + + pub fn schema(&self) -> &SchemaPlan { + &self.schema + } + + fn tables(&self) -> &[TableState] { + self.pending_tables.as_deref().unwrap_or(&self.committed_tables) + } + + fn pending_tables_mut(&mut self) -> &mut [TableState] { + self.pending_tables + .as_deref_mut() + .expect("mutable interaction without active transaction") + } + + fn violates_unique_constraint_in(&self, tables: &[TableState], table: usize, row: &Row) -> bool { + let table_plan = &self.schema.tables[table]; + let rows = &tables[table].rows; + for constraint in &table_plan.unique_constraints { + if rows + .iter() + .any(|r| constraint.columns.iter().all(|&c| r.elements[c] == row.elements[c])) + { + return true; + } + } + false + } + + pub fn apply(&mut self, interaction: &Interaction) -> Observation { + match interaction { + Interaction::BeginMutTx => { + debug_assert!(self.pending_tables.is_none()); + self.pending_tables = Some(self.committed_tables.clone()); + Observation::BeganMutTx + } + Interaction::Insert { table, row } => { + debug_assert!(self.pending_tables.is_some()); + let primary_key = self.schema.tables[*table].primary_key; + + if self.violates_unique_constraint_in(self.tables(), *table, row) + || self.tables()[*table].rows.contains(row) + { + return Observation::Inserted { + count_after: self.tables()[*table].rows.len() as u64, + }; + } + + let rows = &mut self.pending_tables_mut()[*table].rows; + if let Some(pk_col) = primary_key { + if let Some(pos) = rows.iter().position(|r| r.elements[pk_col] == row.elements[pk_col]) { + rows[pos] = row.clone(); + return Observation::Inserted { + count_after: rows.len() as u64, + }; + } + } + rows.push(row.clone()); + Observation::Inserted { + count_after: rows.len() as u64, + } + } + Interaction::Delete { table, row } => { + debug_assert!(self.pending_tables.is_some()); + let rows = &mut self.pending_tables_mut()[*table].rows; + rows.retain(|r| r != row); + Observation::Deleted { + count_after: rows.len() as u64, + } + } + Interaction::CommitTx => { + debug_assert!(self.pending_tables.is_some()); + let pending_tables = self.pending_tables.take().expect("active transaction"); + let delta = commit_delta_from_tables(&self.committed_tables, &pending_tables); + self.committed_tables = pending_tables; + Observation::Committed { delta } + } + Interaction::Count { table } => { + debug_assert!(self.pending_tables.is_some()); + Observation::Counted { + count: self.tables()[*table].rows.len() as u64, + } + } + Interaction::Replay => { + self.pending_tables = None; + Observation::Replayed { + summaries: self.summaries(), + } + } + } + } + + pub fn in_mut_tx(&self) -> bool { + self.pending_tables.is_some() + } + + pub fn summaries(&self) -> Vec { + self.tables().iter().map(|table| summarize_rows(&table.rows)).collect() + } + + pub fn rows(&self, table: usize) -> &[Row] { + &self.tables()[table].rows + } +} + +fn commit_delta_from_tables(before: &[TableState], after: &[TableState]) -> CommitDelta { + let mut tables = Vec::new(); + + for (table, (before, after)) in before.iter().zip(after).enumerate() { + let inserts = rows_absent_from(&after.rows, &before.rows); + let deletes = rows_absent_from(&before.rows, &after.rows); + let truncated = !before.rows.is_empty() && after.rows.is_empty() && !deletes.is_empty(); + + if inserts.is_empty() && deletes.is_empty() && !truncated { + continue; + } + + tables.push(TableDelta { + table, + inserts: summarize_rows(&inserts), + deletes: summarize_rows(&deletes), + truncated, + }); + } + + CommitDelta { tables } +} + +fn rows_absent_from(rows: &[Row], other: &[Row]) -> Vec { + rows.iter().filter(|row| !other.contains(row)).cloned().collect() +} diff --git a/crates/dst/src/engine/properties.rs b/crates/dst/src/engine/properties.rs index e5e4cd7886..6548cb4a86 100644 --- a/crates/dst/src/engine/properties.rs +++ b/crates/dst/src/engine/properties.rs @@ -1,35 +1,43 @@ -use super::workload::{Interaction, Model, Observation, TableSummary}; +use super::model::Model; +use super::workload::{Interaction, Observation}; use crate::schema::SchemaPlan; use crate::traits::Properties; pub struct EngineProperties { oracle: EngineOracle, - count_visible: CountVisible, - commit_matches: CommitMatches, - replay_matches: ReplayMatches, + properties: Vec>, } impl EngineProperties { pub fn new(schema: SchemaPlan) -> Self { Self { oracle: EngineOracle::new(schema), - count_visible: CountVisible, - commit_matches: CommitMatches, - replay_matches: ReplayMatches, + properties: vec![Box::new(CountVisible), Box::new(CommitMatches), Box::new(ReplayMatches)], } } } impl Properties for EngineProperties { fn observe(&mut self, interaction: &Interaction, observation: &Observation) -> Result<(), anyhow::Error> { - self.oracle.apply(interaction); - self.count_visible.check(interaction, observation, &self.oracle)?; - self.commit_matches.check(interaction, observation, &self.oracle)?; - self.replay_matches.check(interaction, observation, &self.oracle)?; + let expected = self.oracle.apply(interaction); + + for property in &self.properties { + if property.observes(interaction) { + property.check(interaction, observation, &expected)?; + } + } + Ok(()) } } +trait EngineProperty { + fn observes(&self, interaction: &Interaction) -> bool; + + fn check(&self, interaction: &Interaction, observation: &Observation, expected: &Observation) + -> anyhow::Result<()>; +} + struct EngineOracle { model: Model, } @@ -41,33 +49,34 @@ impl EngineOracle { } } - fn apply(&mut self, interaction: &Interaction) { - self.model.apply(interaction); - } - - fn row_count(&self, table: usize) -> u64 { - self.model.row_count(table) - } - - fn summaries(&self) -> Vec { - self.model.summaries() + fn apply(&mut self, interaction: &Interaction) -> Observation { + self.model.apply(interaction) } } struct CountVisible; -impl CountVisible { - fn check(&self, interaction: &Interaction, observation: &Observation, oracle: &EngineOracle) -> anyhow::Result<()> { - let Interaction::Count { table } = interaction else { - return Ok(()); - }; +impl EngineProperty for CountVisible { + fn observes(&self, interaction: &Interaction) -> bool { + matches!(interaction, Interaction::Count { .. }) + } + + fn check( + &self, + _interaction: &Interaction, + observation: &Observation, + expected: &Observation, + ) -> anyhow::Result<()> { let Observation::Counted { count } = observation else { anyhow::bail!("count_visible: count produced unexpected observation"); }; + let Observation::Counted { count: expected } = expected else { + unreachable!("CountVisible only subscribes to count interactions"); + }; anyhow::ensure!( - *count == oracle.row_count(*table), - "count_visible: count did not reflect visible transaction state for table {table}" + count == expected, + "count_visible: count did not reflect visible transaction state" ); Ok(()) } @@ -75,36 +84,51 @@ impl CountVisible { struct CommitMatches; -impl CommitMatches { - fn check(&self, interaction: &Interaction, observation: &Observation, oracle: &EngineOracle) -> anyhow::Result<()> { - if !matches!(interaction, Interaction::CommitTx) { - return Ok(()); - } - let Observation::Committed { summaries } = observation else { +impl EngineProperty for CommitMatches { + fn observes(&self, interaction: &Interaction) -> bool { + matches!(interaction, Interaction::CommitTx) + } + + fn check( + &self, + _interaction: &Interaction, + observation: &Observation, + expected: &Observation, + ) -> anyhow::Result<()> { + let Observation::Committed { delta } = observation else { anyhow::bail!("commit_matches: commit produced unexpected observation"); }; + let Observation::Committed { delta: expected } = expected else { + unreachable!("CommitMatches only subscribes to commit interactions"); + }; - anyhow::ensure!( - summaries == &oracle.summaries(), - "commit_matches: committed target summary diverged from model" - ); + anyhow::ensure!(delta == expected, "commit_matches: committed delta diverged from model"); Ok(()) } } struct ReplayMatches; -impl ReplayMatches { - fn check(&self, interaction: &Interaction, observation: &Observation, oracle: &EngineOracle) -> anyhow::Result<()> { - if !matches!(interaction, Interaction::Replay) { - return Ok(()); - } +impl EngineProperty for ReplayMatches { + fn observes(&self, interaction: &Interaction) -> bool { + matches!(interaction, Interaction::Replay) + } + + fn check( + &self, + _interaction: &Interaction, + observation: &Observation, + expected: &Observation, + ) -> anyhow::Result<()> { let Observation::Replayed { summaries } = observation else { anyhow::bail!("replay_matches: replay produced unexpected observation"); }; + let Observation::Replayed { summaries: expected } = expected else { + unreachable!("ReplayMatches only subscribes to replay interactions"); + }; anyhow::ensure!( - summaries == &oracle.summaries(), + summaries == expected, "replay_matches: replayed target summary diverged from committed model" ); Ok(()) diff --git a/crates/dst/src/engine/workload.rs b/crates/dst/src/engine/workload.rs index 66c7939207..972c7d1f5f 100644 --- a/crates/dst/src/engine/workload.rs +++ b/crates/dst/src/engine/workload.rs @@ -2,6 +2,7 @@ use spacetimedb_lib::bsatn::to_vec; use spacetimedb_lib::{AlgebraicValue, ProductValue}; use spacetimedb_runtime::sim::Rng; +use super::model::Model; use crate::schema::{SchemaPlan, TablePlan, Type}; pub type Row = ProductValue; @@ -21,7 +22,7 @@ pub enum Observation { BeganMutTx, Inserted { count_after: u64 }, Deleted { count_after: u64 }, - Committed { summaries: Vec }, + Committed { delta: CommitDelta }, Counted { count: u64 }, Replayed { summaries: Vec }, } @@ -32,130 +33,17 @@ pub struct TableSummary { pub hash: u64, } -#[derive(Debug)] -pub struct Model { - schema: SchemaPlan, - committed_tables: Vec, - pending_tables: Option>, +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CommitDelta { + pub tables: Vec, } -#[derive(Debug, Clone)] -struct TableState { - rows: Vec, -} - -impl Model { - pub fn new(schema: SchemaPlan) -> Self { - let committed_tables = schema.tables.iter().map(|_| TableState { rows: vec![] }).collect(); - Self { - schema, - committed_tables, - pending_tables: None, - } - } - - fn tables(&self) -> &[TableState] { - self.pending_tables.as_deref().unwrap_or(&self.committed_tables) - } - - fn pending_tables_mut(&mut self) -> &mut [TableState] { - self.pending_tables - .as_deref_mut() - .expect("mutable interaction without active transaction") - } - - fn violates_unique_constraint_in(&self, tables: &[TableState], table: usize, row: &Row) -> bool { - let table_plan = &self.schema.tables[table]; - let rows = &tables[table].rows; - for constraint in &table_plan.unique_constraints { - if rows - .iter() - .any(|r| constraint.columns.iter().all(|&c| r.elements[c] == row.elements[c])) - { - return true; - } - } - false - } - - pub fn apply(&mut self, interaction: &Interaction) -> Observation { - match interaction { - Interaction::BeginMutTx => { - debug_assert!(self.pending_tables.is_none()); - self.pending_tables = Some(self.committed_tables.clone()); - Observation::BeganMutTx - } - Interaction::Insert { table, row } => { - debug_assert!(self.pending_tables.is_some()); - let primary_key = self.schema.tables[*table].primary_key; - - if self.violates_unique_constraint_in(self.tables(), *table, row) - || self.tables()[*table].rows.contains(row) - { - return Observation::Inserted { - count_after: self.tables()[*table].rows.len() as u64, - }; - } - - let rows = &mut self.pending_tables_mut()[*table].rows; - if let Some(pk_col) = primary_key { - if let Some(pos) = rows.iter().position(|r| r.elements[pk_col] == row.elements[pk_col]) { - rows[pos] = row.clone(); - return Observation::Inserted { - count_after: rows.len() as u64, - }; - } - } - rows.push(row.clone()); - Observation::Inserted { - count_after: rows.len() as u64, - } - } - Interaction::Delete { table, row } => { - debug_assert!(self.pending_tables.is_some()); - let rows = &mut self.pending_tables_mut()[*table].rows; - rows.retain(|r| r != row); - Observation::Deleted { - count_after: rows.len() as u64, - } - } - Interaction::CommitTx => { - debug_assert!(self.pending_tables.is_some()); - self.committed_tables = self.pending_tables.take().expect("active transaction"); - Observation::Committed { - summaries: self.summaries(), - } - } - Interaction::Count { table } => { - debug_assert!(self.pending_tables.is_some()); - Observation::Counted { - count: self.tables()[*table].rows.len() as u64, - } - } - Interaction::Replay => { - self.pending_tables = None; - Observation::Replayed { - summaries: self.summaries(), - } - } - } - } - - pub fn in_mut_tx(&self) -> bool { - self.pending_tables.is_some() - } - - pub fn row_count(&self, table: usize) -> u64 { - self.tables()[table].rows.len() as u64 - } - - pub fn summaries(&self) -> Vec { - self.tables().iter().map(|table| summarize_rows(&table.rows)).collect() - } - - pub fn rows(&self, table: usize) -> &[Row] { - &self.tables()[table].rows - } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TableDelta { + pub table: usize, + pub inserts: TableSummary, + pub deletes: TableSummary, + pub truncated: bool, } pub struct WorkloadGen { @@ -169,7 +57,7 @@ impl WorkloadGen { } fn schema(&self) -> &SchemaPlan { - &self.model.schema + self.model.schema() } fn gen_value(&self, ty: Type) -> AlgebraicValue {