separate mode;

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",
 ]
 

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<Observation> {
         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 } => {

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<TableState>,
+    pending_tables: Option<Vec<TableState>>,
+}
+
+#[derive(Debug, Clone)]
+struct TableState {
+    rows: Vec<Row>,
+}
+
+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<TableSummary> {
+        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<Row> {
+    rows.iter().filter(|row| !other.contains(row)).cloned().collect()
+}

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<Box<dyn EngineProperty>>,
 }
 
 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<Interaction, Observation> 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<TableSummary> {
-        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(())

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<TableSummary> },
+    Committed { delta: CommitDelta },
     Counted { count: u64 },
     Replayed { summaries: Vec<TableSummary> },
 }
@@ -32,130 +33,17 @@ pub struct TableSummary {
     pub hash: u64,
 }
 
-#[derive(Debug)]
-pub struct Model {
-    schema: SchemaPlan,
-    committed_tables: Vec<TableState>,
-    pending_tables: Option<Vec<TableState>>,
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CommitDelta {
+    pub tables: Vec<TableDelta>,
 }
 
-#[derive(Debug, Clone)]
-struct TableState {
-    rows: Vec<Row>,
-}
-
-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<TableSummary> {
-        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 {