schema generation

crates/dst/Cargo.toml

@@ -7,7 +7,15 @@ rust-version.workspace = true
 [dependencies]
 anyhow.workspace = true
 clap.workspace = true
-spacetimedb-runtime = { path = "../runtime/",default-features = false,  features = ["simulation"]}
+spacetimedb-datastore = { path = "../datastore", default-features = false, features = ["simulation"] }
+spacetimedb-durability = { path = "../durability", default-features = false, features = ["simulation"] }
+spacetimedb-engine = { path = "../engine", default-features = false, features = ["simulation"] }
+spacetimedb-lib.workspace = true
+spacetimedb-primitives.workspace = true
+spacetimedb-runtime = { path = "../runtime/", default-features = false, features = ["simulation"] }
+spacetimedb-sats.workspace = true
+spacetimedb-schema.workspace = true
+spacetimedb-table = { path = "../table", default-features = false }
 tracing.workspace = true
 
 [lints]

crates/dst/README.md

@@ -0,0 +1,21 @@
+# SpacetimeDB DST
+
+Deterministic Simulation Testing framework for SpacetimeDB.
+
+## Test
+
+```sh
+cargo test -p spacetimedb-dst
+```
+
+## Run
+
+```sh
+cargo run -p spacetimedb-dst -- run --seed 42 --tables 5
+```
+
+Options:
+
+- `--seed <u64>` — RNG seed (defaults to wall-clock nanos)
+- `--tables <usize>` — number of tables to generate (default 3)
+- `--max-interactions <usize>` — interaction budget

crates/dst/src/core.rs

@@ -1,5 +1,17 @@
 use anyhow::Error;
 
+pub trait Target {
+    const NAME: &'static str;
+
+    fn prepare(config: &crate::RunConfig) -> Result<(), Error>
+    where
+        Self: Sized;
+
+    fn run_streaming(config: crate::RunConfig) -> Result<String, Error>
+    where
+        Self: Sized;
+}
+
 pub trait TargetDriver<I> {
     type Observation;
     type Outcome;

crates/dst/src/main.rs

@@ -1,10 +1,15 @@
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use clap::{Args, Parser, Subcommand};
-use spacetimedb_runtime::sim::Runtime;
+use spacetimedb_runtime::sim::Rng;
 
-mod core;
-mod target;
+pub mod core;
+pub mod source;
+pub mod target;
+
+use source::schema_gen::{SchemaGenerator, SchemaProfile};
+use source::table_ops::{Interaction, InteractionGen, Model};
+use target::engine::EngineTarget;
 
 #[derive(Parser, Debug)]
 #[command(name = "spacetimedb-dst")]
@@ -21,11 +26,9 @@ enum Command {
 
 #[derive(Args, Debug)]
 struct RunArgs {
-    #[arg(long, default_value = Engine::NAME, help = "Target to run.")]
-    target: String,
     #[arg(long, help = "Seed for generated choices. Defaults to wall-clock time.")]
     seed: Option<u64>,
-    #[arg(long, help = "Deterministic interaction budget. Preferred for replayable failures.")]
+    #[arg(long, help = "Deterministic interaction budget.")]
     max_interactions: Option<usize>,
 }
 
@@ -39,27 +42,77 @@ fn main() -> anyhow::Result<()> {
 fn init_tracing() {}
 
 fn run_command(args: RunArgs) -> anyhow::Result<()> {
-    //resolve_target(&args.target)?;
     let seed = resolve_seed(args.seed);
     let config = RunConfig {
         max_interactions: args.max_interactions,
         seed,
     };
 
-    run_prepared_target::<Engine>(config)
-}
+    eprintln!("seed: {}", config.seed);
 
-fn run_prepared_target<T>(config: RunConfig) -> anyhow::Result<()>
-where
-    T: Target + 'static,
-{
-    T::prepare(&config)?;
-    std::thread::spawn(move || {
-        let mut runtime = Runtime::new(config.seed);
-        runtime.block_on(run_target::<T>(config))
-    })
-    .join()
-    .unwrap_or_else(|payload| std::panic::resume_unwind(payload))
+    // Generate schema from seed.
+    let rng = Rng::new(config.seed);
+    let schema = SchemaGenerator::new(&rng, SchemaProfile::default()).gen_schema();
+
+    eprintln!("generated {} tables:", schema.tables.len());
+    for table in &schema.tables {
+        eprintln!("  {} ({} columns)", table.name, table.columns.len());
+    }
+
+    // Open engine and create tables.
+    let engine = EngineTarget::prepare(&config, schema.clone())?;
+    eprintln!("engine ready");
+
+    // Generate and execute interactions.
+    let budget = config.max_interactions.unwrap_or(100);
+    let source = InteractionGen::new(&rng, &schema);
+    let mut model = Model::new(&schema);
+
+    let mut inserts = 0u64;
+    let mut deletes = 0u64;
+    let mut counts = 0u64;
+
+    for _ in 0..budget {
+        let ix = source.next_interaction(&model);
+        let expected = model.apply(&ix);
+        let got = engine.execute(&ix).unwrap();
+        assert_eq!(expected, got, "model mismatch");
+        match &ix {
+            Interaction::Insert { .. } => inserts += 1,
+            Interaction::Delete { .. } => deletes += 1,
+            Interaction::Count { .. } => counts += 1,
+        }
+    }
+
+    eprintln!("done: {inserts} inserts, {deletes} deletes, {counts} counts, {budget} total");
+
+    // Final verification: model row counts match engine.
+    for (i, table) in schema.tables.iter().enumerate() {
+        let table_id = engine.db().with_auto_commit(
+            spacetimedb_datastore::execution_context::Workload::Internal,
+            |tx| {
+                engine
+                    .db()
+                    .table_id_from_name_mut(tx, &table.name)
+                    .map(|t| t.unwrap())
+            },
+        )?;
+        let actual = engine.db().with_auto_commit(
+            spacetimedb_datastore::execution_context::Workload::Internal,
+            |tx| engine.db().iter_mut(tx, table_id).map(|it| it.count() as u64),
+        )?;
+        assert_eq!(
+            model.row_count(i),
+            actual,
+            "table '{}': model={} engine={}",
+            table.name,
+            model.row_count(i),
+            actual,
+        );
+    }
+    eprintln!("model consistency verified");
+
+    Ok(())
 }
 
 fn resolve_seed(seed: Option<u64>) -> u64 {
@@ -71,46 +124,8 @@ fn resolve_seed(seed: Option<u64>) -> u64 {
     })
 }
 
-//fn resolve_target(target: &str) -> anyhow::Result<()> {
-//    if target == RelationalDbCommitlogDescriptor::NAME {
-//        Ok(())
-//    } else {
-//        anyhow::bail!(
-//            "unsupported target: {target}; expected: {}",
-//            RelationalDbCommitlogDescriptor::NAME
-//        )
-//    }
-//}
-//
-//
-async fn run_target<T: Target>(config: RunConfig) -> anyhow::Result<()> {
-    let line = T::run_streaming(config).await?;
-    println!("{line}");
-    Ok(())
-}
-
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct RunConfig {
-    /// Hard cap on generated interactions. `None` means no interaction budget.
-    ///
-    /// This is the preferred budget for exact seed replay: the same target,
-    /// scenario, seed, max-interactions value, and fault profile should produce
-    /// the same generated interaction stream.
     pub max_interactions: Option<usize>,
-
     pub seed: u64,
 }
-
-struct Engine;
-
-impl Target for Engine {
-    const NAME: &'static str = "engine";
-
-    fn prepare(config: &RunConfig) {
-        todo!()
-    }
-
-    fn run_streaming(config: RunConfig) {
-        todo!()
-    }
-}

crates/dst/src/source/mod.rs

@@ -0,0 +1,3 @@
+pub mod schema;
+pub mod schema_gen;
+pub mod table_ops;

crates/dst/src/source/schema.rs

@@ -0,0 +1,323 @@
+//! Custom schema types for DST table/index definitions.
+//!
+//! These types are the canonical source of truth for generated schemas.
+//! They lower into [`RawModuleDefV10`] via [`lower_schema`].
+
+use spacetimedb_lib::db::raw_def::v10::*;
+use spacetimedb_lib::db::raw_def::v9::{RawIndexAlgorithm, TableAccess, TableType};
+use spacetimedb_primitives::{ColId, ColList};
+use spacetimedb_sats::{AlgebraicType, AlgebraicValue, ArrayType, ArrayValue, ProductType, ProductTypeElement};
+
+// ---------------------------------------------------------------------------
+// Column types — closed set, expand deliberately.
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum Type {
+    Bool,
+    I64,
+    U64,
+    String,
+    Bytes,
+}
+
+impl Type {
+    pub const ALL: &'static [Type] = &[
+        Type::Bool,
+        Type::I64,
+        Type::U64,
+        Type::String,
+        Type::Bytes,
+    ];
+
+    pub fn to_algebraic(self) -> AlgebraicType {
+        match self {
+            Type::Bool => AlgebraicType::Bool,
+            Type::I64 => AlgebraicType::I64,
+            Type::U64 => AlgebraicType::U64,
+            Type::String => AlgebraicType::String,
+            Type::Bytes => AlgebraicType::Array(ArrayType {
+                elem_ty: Box::new(AlgebraicType::U8),
+            }),
+        }
+    }
+
+    pub fn is_integral(self) -> bool {
+        matches!(self, Type::I64 | Type::U64)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum Value {
+    Bool(bool),
+    I64(i64),
+    U64(u64),
+    String(String),
+    Bytes(Vec<u8>),
+}
+
+impl Value {
+    pub fn type_of(&self) -> Type {
+        match self {
+            Value::Bool(_) => Type::Bool,
+            Value::I64(_) => Type::I64,
+            Value::U64(_) => Type::U64,
+            Value::String(_) => Type::String,
+            Value::Bytes(_) => Type::Bytes,
+        }
+    }
+
+    fn to_algebraic(&self) -> AlgebraicValue {
+        match self {
+            Value::Bool(b) => AlgebraicValue::Bool(*b),
+            Value::I64(v) => AlgebraicValue::I64(*v),
+            Value::U64(v) => AlgebraicValue::U64(*v),
+            Value::String(s) => AlgebraicValue::String(s.clone().into()),
+            Value::Bytes(b) => AlgebraicValue::Array(ArrayValue::U8(b.clone().into())),
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Schema plan — the canonical source of truth.
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone)]
+pub struct SchemaPlan {
+    pub tables: Vec<TablePlan>,
+}
+
+#[derive(Debug, Clone)]
+pub struct TablePlan {
+    pub name: String,
+    pub columns: Vec<ColumnPlan>,
+    pub primary_key: Option<usize>,
+    pub indexes: Vec<IndexPlan>,
+    pub unique_constraints: Vec<UniqueConstraintPlan>,
+    pub sequences: Vec<SequencePlan>,
+    pub default_values: Vec<DefaultPlan>,
+    pub is_event: bool,
+    pub is_public: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct ColumnPlan {
+    pub name: String,
+    pub ty: Type,
+}
+
+#[derive(Debug, Clone)]
+pub struct IndexPlan {
+    /// Indices into `TablePlan.columns`.
+    pub columns: Vec<usize>,
+    pub algorithm: IndexAlgorithm,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum IndexAlgorithm {
+    BTree,
+    Hash,
+}
+
+#[derive(Debug, Clone)]
+pub struct UniqueConstraintPlan {
+    /// Indices into `TablePlan.columns`. Non-empty.
+    pub columns: Vec<usize>,
+}
+
+/// A sequence on a specific column. The column's type is carried inline
+/// so callers cannot create a sequence on a non-integral column —
+/// the constructor requires `ty.is_integral()`.
+#[derive(Debug, Clone)]
+pub struct SequencePlan {
+    /// Index into `TablePlan.columns`.
+    pub column: usize,
+    /// The type of that column. Must be integral (I64 or U64).
+    pub ty: Type,
+    pub start: Option<i128>,
+    pub min_value: Option<i128>,
+    pub max_value: Option<i128>,
+    pub increment: i128,
+}
+
+impl SequencePlan {
+    /// Create a sequence plan. Returns `None` if the type is not integral.
+    pub fn new(column: usize, ty: Type) -> Option<Self> {
+        if !ty.is_integral() {
+            return None;
+        }
+        Some(Self {
+            column,
+            ty,
+            start: None,
+            min_value: None,
+            max_value: None,
+            increment: 1,
+        })
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct DefaultPlan {
+    /// Index into `TablePlan.columns`.
+    pub column: usize,
+    pub value: Value,
+}
+
+// ---------------------------------------------------------------------------
+// Lowering into RawModuleDefV10.
+// ---------------------------------------------------------------------------
+
+pub fn lower_schema(schema: &SchemaPlan) -> RawModuleDefV10 {
+    let mut builder = RawModuleDefV10Builder::new();
+    builder.set_case_conversion_policy(CaseConversionPolicy::None);
+
+    for table in &schema.tables {
+        lower_table(&mut builder, table);
+    }
+
+    builder.finish()
+}
+
+fn lower_table(builder: &mut RawModuleDefV10Builder, table: &TablePlan) {
+    let product_type = ProductType {
+        elements: table
+            .columns
+            .iter()
+            .map(|col| ProductTypeElement {
+                name: Some(col.name.clone().into()),
+                algebraic_type: col.ty.to_algebraic(),
+            })
+            .collect(),
+    };
+
+    let mut tbl = builder.build_table_with_new_type(table.name.clone(), product_type, false);
+
+    tbl = tbl.with_type(TableType::User);
+    tbl = tbl.with_access(if table.is_public {
+        TableAccess::Public
+    } else {
+        TableAccess::Private
+    });
+    tbl = tbl.with_event(table.is_event);
+
+    // Primary key.
+    if let Some(pk) = table.primary_key {
+        tbl = tbl.with_primary_key(ColId(pk as u16));
+    }
+
+    // Unique constraints — all of them, including PK-matching.
+    for constraint in &table.unique_constraints {
+        let col_list: ColList = constraint
+            .columns
+            .iter()
+            .map(|&c| ColId(c as u16))
+            .collect();
+        tbl = tbl.with_unique_constraint(col_list);
+    }
+
+    // Indexes.
+    for index in &table.indexes {
+        let col_list: ColList = index
+            .columns
+            .iter()
+            .map(|&c| ColId(c as u16))
+            .collect();
+
+        let algorithm = match index.algorithm {
+            IndexAlgorithm::BTree => RawIndexAlgorithm::BTree { columns: col_list },
+            IndexAlgorithm::Hash => RawIndexAlgorithm::Hash { columns: col_list },
+        };
+
+        let source_name = format!(
+            "{}_{}_idx",
+            table.name,
+            index
+                .columns
+                .iter()
+                .map(|&c| table.columns[c].name.as_str())
+                .collect::<Vec<_>>()
+                .join("_")
+        );
+
+        tbl = tbl.with_index_no_accessor_name(algorithm, source_name);
+    }
+
+    // Sequences — all of them.
+    for seq in &table.sequences {
+        tbl = tbl.with_column_sequence(ColId(seq.column as u16));
+    }
+
+    // Default values.
+    for default in &table.default_values {
+        let algebraic_val = default.value.to_algebraic();
+        tbl = tbl.with_default_column_value(ColId(default.column as u16), algebraic_val);
+    }
+
+    tbl.finish();
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn lower_single_table() {
+        let schema = SchemaPlan {
+            tables: vec![TablePlan {
+                name: "users".into(),
+                columns: vec![
+                    ColumnPlan { name: "id".into(), ty: Type::U64 },
+                    ColumnPlan { name: "name".into(), ty: Type::String },
+                    ColumnPlan { name: "score".into(), ty: Type::I64 },
+                ],
+                primary_key: Some(0),
+                indexes: vec![IndexPlan {
+                    columns: vec![2],
+                    algorithm: IndexAlgorithm::BTree,
+                }],
+                unique_constraints: vec![UniqueConstraintPlan {
+                    columns: vec![0],
+                }],
+                sequences: vec![SequencePlan::new(0, Type::U64).unwrap()],
+                default_values: vec![],
+                is_event: false,
+                is_public: true,
+            }],
+        };
+
+        let raw = lower_schema(&schema);
+
+        // Should have Typespace, Types, and Tables sections.
+        assert!(raw.typespace().is_some());
+        assert!(raw.types().is_some());
+        let tables = raw.tables().unwrap();
+        assert_eq!(tables.len(), 1);
+
+        let t = &tables[0];
+        assert_eq!(t.source_name.as_ref(), "users");
+        assert_eq!(t.table_type, TableType::User);
+        assert_eq!(t.table_access, TableAccess::Public);
+        assert!(!t.is_event);
+        assert_eq!(t.primary_key.len(), 1);
+        assert_eq!(t.indexes.len(), 1);
+        assert_eq!(t.sequences.len(), 1);
+    }
+
+    #[test]
+    fn sequence_rejects_non_integral() {
+        assert!(SequencePlan::new(0, Type::Bool).is_none());
+        assert!(SequencePlan::new(0, Type::String).is_none());
+        assert!(SequencePlan::new(0, Type::Bytes).is_none());
+        assert!(SequencePlan::new(0, Type::I64).is_some());
+        assert!(SequencePlan::new(0, Type::U64).is_some());
+    }
+
+    #[test]
+    fn type_roundtrip() {
+        for ty in Type::ALL {
+            // Every DST type should roundtrip through AlgebraicType.
+            let _ = ty.to_algebraic();
+        }
+    }
+}

crates/dst/src/source/schema_gen.rs

@@ -0,0 +1,235 @@
+//! Seed-based random schema generation.
+
+use spacetimedb_runtime::sim::Rng;
+
+use super::schema::*;
+
+/// Controls the shape of generated schemas.
+#[derive(Debug, Clone)]
+pub struct SchemaProfile {
+    pub table_count: (usize, usize),
+    pub columns: (usize, usize),
+    pub pk_prob: f64,
+    pub auto_inc_prob: f64,
+    pub indexes: (usize, usize),
+    pub unique_constraints: (usize, usize),
+    pub btree_prob: f64,
+    pub event_prob: f64,
+    pub private_prob: f64,
+}
+
+impl Default for SchemaProfile {
+    fn default() -> Self {
+        Self {
+            table_count: (2, 5),
+            columns: (1, 8),
+            pk_prob: 0.7,
+            auto_inc_prob: 0.8,
+            indexes: (0, 3),
+            unique_constraints: (0, 2),
+            btree_prob: 0.7,
+            event_prob: 0.1,
+            private_prob: 0.1,
+        }
+    }
+}
+
+pub struct SchemaGenerator<'a> {
+    rng: &'a Rng,
+    profile: SchemaProfile,
+}
+
+impl<'a> SchemaGenerator<'a> {
+    pub fn new(rng: &'a Rng, profile: SchemaProfile) -> Self {
+        Self { rng, profile }
+    }
+
+    fn range(&self, (lo, hi): (usize, usize)) -> usize {
+        if lo >= hi {
+            return lo;
+        }
+        lo + (self.rng.next_u64() as usize % (hi - lo + 1))
+    }
+
+    fn gen_type(&self) -> Type {
+        Type::ALL[self.rng.index(Type::ALL.len())]
+    }
+
+    fn gen_columns(&self) -> Vec<ColumnPlan> {
+        let n = self.range(self.profile.columns);
+        (0..n)
+            .map(|i| ColumnPlan {
+                name: format!("col_{i}"),
+                ty: self.gen_type(),
+            })
+            .collect()
+    }
+
+    fn gen_unique_constraints(
+        &self,
+        columns: &[ColumnPlan],
+        pk: &Option<usize>,
+    ) -> Vec<UniqueConstraintPlan> {
+        let n = self.range(self.profile.unique_constraints);
+        let mut seen: Vec<Vec<usize>> = Vec::new();
+        let mut result = Vec::new();
+        for _ in 0..n {
+            let num_cols = 1 + self.rng.index(columns.len().min(3));
+            let mut cols: Vec<usize> = (0..num_cols)
+                .map(|_| self.rng.index(columns.len()))
+                .collect();
+            cols.sort();
+            cols.dedup();
+            if !cols.is_empty() && !seen.contains(&cols) {
+                seen.push(cols.clone());
+                result.push(UniqueConstraintPlan { columns: cols });
+            }
+        }
+        // Ensure PK has a matching unique constraint.
+        if let Some(pk) = pk {
+            if !seen.contains(&vec![*pk]) {
+                result.push(UniqueConstraintPlan {
+                    columns: vec![*pk],
+                });
+            }
+        }
+        result
+    }
+
+    fn gen_indexes(
+        &self,
+        columns: &[ColumnPlan],
+        unique_constraints: &[UniqueConstraintPlan],
+        pk: &Option<usize>,
+    ) -> Vec<IndexPlan> {
+        // Every unique constraint and PK needs a matching index.
+        let mut seen_cols: Vec<Vec<usize>> = Vec::new();
+        let mut indexes: Vec<IndexPlan> = Vec::new();
+
+        // Index for PK.
+        if let Some(pk) = pk {
+            seen_cols.push(vec![*pk]);
+            indexes.push(IndexPlan {
+                columns: vec![*pk],
+                algorithm: IndexAlgorithm::BTree,
+            });
+        }
+
+        // Indexes for unique constraints.
+        for constraint in unique_constraints {
+            if seen_cols.contains(&constraint.columns) {
+                continue;
+            }
+            seen_cols.push(constraint.columns.clone());
+            indexes.push(IndexPlan {
+                columns: constraint.columns.clone(),
+                algorithm: IndexAlgorithm::BTree,
+            });
+        }
+
+        // Additional random indexes.
+        let n = self.range(self.profile.indexes);
+        for _ in 0..n {
+            let num_cols = 1 + self.rng.index(columns.len().min(3));
+            let mut cols: Vec<usize> = (0..num_cols)
+                .map(|_| self.rng.index(columns.len()))
+                .collect();
+            cols.sort();
+            cols.dedup();
+            if cols.is_empty() || seen_cols.contains(&cols) {
+                continue;
+            }
+            seen_cols.push(cols.clone());
+            let algorithm = if self.rng.sample_probability(self.profile.btree_prob) {
+                IndexAlgorithm::BTree
+            } else {
+                IndexAlgorithm::Hash
+            };
+            indexes.push(IndexPlan {
+                columns: cols,
+                algorithm,
+            });
+        }
+
+        indexes
+    }
+
+    fn gen_table(&self, _table_index: usize) -> TablePlan {
+        let columns = self.gen_columns();
+
+        let primary_key = if self.rng.sample_probability(self.profile.pk_prob) && !columns.is_empty()
+        {
+            Some(self.rng.index(columns.len()))
+        } else {
+            None
+        };
+
+        let unique_constraints = self.gen_unique_constraints(&columns, &primary_key);
+
+        let sequences = if let Some(pk) = primary_key {
+            if columns[pk].ty.is_integral()
+                && self.rng.sample_probability(self.profile.auto_inc_prob)
+            {
+                SequencePlan::new(pk, columns[pk].ty).into_iter().collect()
+            } else {
+                vec![]
+            }
+        } else {
+            vec![]
+        };
+
+        let indexes = self.gen_indexes(&columns, &unique_constraints, &primary_key);
+
+        // Generate a name from the RNG so different seeds produce different names.
+        let name = format!("tbl_{}", self.rng.next_u64());
+
+        TablePlan {
+            name,
+            columns,
+            primary_key,
+            indexes,
+            unique_constraints,
+            sequences,
+            default_values: vec![],
+            is_event: self.rng.sample_probability(self.profile.event_prob),
+            is_public: !self.rng.sample_probability(self.profile.private_prob),
+        }
+    }
+
+    pub fn gen_schema(&self) -> SchemaPlan {
+        let table_count = self.range(self.profile.table_count);
+        let tables = (0..table_count)
+            .map(|i| self.gen_table(i))
+            .collect();
+        SchemaPlan { tables }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use spacetimedb_runtime::sim::Rng;
+
+    #[test]
+    fn deterministic_from_seed() {
+        let rng1 = Rng::new(42);
+        let rng2 = Rng::new(42);
+        let s1 = SchemaGenerator::new(&rng1, SchemaProfile::default()).gen_schema();
+        let s2 = SchemaGenerator::new(&rng2, SchemaProfile::default()).gen_schema();
+        assert_eq!(s1.tables.len(), s2.tables.len());
+        for (a, b) in s1.tables.iter().zip(s2.tables.iter()) {
+            assert_eq!(a.name, b.name);
+            assert_eq!(a.columns.len(), b.columns.len());
+        }
+    }
+
+    #[test]
+    fn different_seeds_differ() {
+        let rng1 = Rng::new(1);
+        let rng2 = Rng::new(2);
+        let s1 = SchemaGenerator::new(&rng1, SchemaProfile::default()).gen_schema();
+        let s2 = SchemaGenerator::new(&rng2, SchemaProfile::default()).gen_schema();
+        // At least one table name should differ.
+        assert_ne!(s1.tables[0].name, s2.tables[0].name);
+    }
+}

crates/dst/src/source/table_ops.rs

@@ -1,7 +1,191 @@
-type Row = AlgebraicValue;
+use spacetimedb_lib::bsatn::to_vec;
+use spacetimedb_lib::{AlgebraicValue, ProductValue};
+use spacetimedb_runtime::sim::Rng;
 
-enum RowOps {
-    Insert(TableId, Row),
-    Delete(TableId, Row),
-    Update(TableId, Row, Row),
+use super::schema::{SchemaPlan, TablePlan, Type};
+
+/// A row is a product value aligned to a table's columns.
+pub type Row = ProductValue;
+
+/// A single interaction against the database.
+#[derive(Debug, Clone)]
+pub enum Interaction {
+    Insert { table: usize, row: Row },
+    Delete { table: usize, row_index: usize },
+    Count { table: usize },
+}
+
+/// Observation returned by executing an interaction.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Observation {
+    Inserted { count_after: u64 },
+    Deleted { count_after: u64 },
+    Counted { count: u64 },
+}
+
+/// Model stores all rows per table — the ground truth.
+#[derive(Debug)]
+pub struct Model {
+    rows: Vec<Vec<Row>>,
+}
+
+impl Model {
+    pub fn new(schema: &SchemaPlan) -> Self {
+        Self {
+            rows: schema.tables.iter().map(|_| Vec::new()).collect(),
+        }
+    }
+
+    pub fn apply(&mut self, interaction: &Interaction) -> Observation {
+        match interaction {
+            Interaction::Insert { table, row } => {
+                self.rows[*table].push(row.clone());
+                Observation::Inserted {
+                    count_after: self.rows[*table].len() as u64,
+                }
+            }
+            Interaction::Delete { table, row_index } => {
+                if *row_index < self.rows[*table].len() {
+                    self.rows[*table].remove(*row_index);
+                }
+                Observation::Deleted {
+                    count_after: self.rows[*table].len() as u64,
+                }
+            }
+            Interaction::Count { table } => Observation::Counted {
+                count: self.rows[*table].len() as u64,
+            },
+        }
+    }
+
+    pub fn row_count(&self, table: usize) -> u64 {
+        self.rows[table].len() as u64
+    }
+
+    pub fn rows(&self, table: usize) -> &[Row] {
+        &self.rows[table]
+    }
+}
+
+/// Generates random interactions from a schema plan.
+pub struct InteractionGen<'a> {
+    rng: &'a Rng,
+    schema: &'a SchemaPlan,
+}
+
+impl<'a> InteractionGen<'a> {
+    pub fn new(rng: &'a Rng, schema: &'a SchemaPlan) -> Self {
+        Self { rng, schema }
+    }
+
+    fn gen_value(&self, ty: Type) -> AlgebraicValue {
+        match ty {
+            Type::Bool => AlgebraicValue::Bool(self.rng.next_u64() % 2 == 0),
+            Type::I64 => AlgebraicValue::I64(self.rng.next_u64() as i64),
+            Type::U64 => AlgebraicValue::U64(self.rng.next_u64()),
+            Type::String => {
+                AlgebraicValue::String(format!("v_{}", self.rng.next_u64()).into())
+            }
+            Type::Bytes => {
+                let len = (self.rng.next_u64() % 16) as usize;
+                let bytes: Vec<u8> = (0..len).map(|_| self.rng.next_u64() as u8).collect();
+                AlgebraicValue::Array(ArrayValue::U8(bytes.into()))
+            }
+        }
+    }
+
+    fn gen_row(&self, table: &TablePlan) -> Row {
+        table
+            .columns
+            .iter()
+            .map(|c| self.gen_value(c.ty))
+            .collect::<ProductValue>()
+    }
+
+    /// Generate an interaction given the current model state.
+    pub fn next_interaction(&self, model: &Model) -> Interaction {
+        let table_idx = self.rng.index(self.schema.tables.len());
+
+        // ~60% insert, ~20% delete (if rows exist), ~20% count
+        let coin = self.rng.next_u64() % 10;
+        if coin < 6 {
+            Interaction::Insert {
+                table: table_idx,
+                row: self.gen_row(&self.schema.tables[table_idx]),
+            }
+        } else if coin < 8 && !model.rows(table_idx).is_empty() {
+            let row_index = self.rng.index(model.rows(table_idx).len());
+            Interaction::Delete {
+                table: table_idx,
+                row_index,
+            }
+        } else {
+            Interaction::Count { table: table_idx }
+        }
+    }
+}
+
+use spacetimedb_sats::ArrayValue;
+
+/// Serialize a row to BSATN bytes for the engine insert API.
+pub fn row_to_bytes(row: &Row) -> Vec<u8> {
+    to_vec(row).expect("row serialization must not fail")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn model_tracks_rows() {
+        let schema = SchemaPlan {
+            tables: vec![TablePlan {
+                name: "t".into(),
+                columns: vec![],
+                primary_key: None,
+                indexes: vec![],
+                unique_constraints: vec![],
+                sequences: vec![],
+                default_values: vec![],
+                is_event: false,
+                is_public: true,
+            }],
+        };
+        let mut model = Model::new(&schema);
+
+        let obs = model.apply(&Interaction::Insert {
+            table: 0,
+            row: ProductValue::default(),
+        });
+        assert_eq!(obs, Observation::Inserted { count_after: 1 });
+        assert_eq!(model.row_count(0), 1);
+
+        let obs = model.apply(&Interaction::Delete {
+            table: 0,
+            row_index: 0,
+        });
+        assert_eq!(obs, Observation::Deleted { count_after: 0 });
+        assert_eq!(model.row_count(0), 0);
+    }
+
+    #[test]
+    fn gen_produces_valid_interactions() {
+        use spacetimedb_runtime::sim::Rng;
+        let rng = Rng::new(42);
+        let schema = super::super::schema_gen::SchemaGenerator::new(
+            &rng,
+            super::super::schema_gen::SchemaProfile::default(),
+        )
+        .gen_schema();
+        let model = Model::new(&schema);
+        let source = InteractionGen::new(&rng, &schema);
+        for _ in 0..100 {
+            let ix = source.next_interaction(&model);
+            match ix {
+                Interaction::Insert { table, .. } => assert!(table < schema.tables.len()),
+                Interaction::Delete { table, .. } => assert!(table < schema.tables.len()),
+                Interaction::Count { table } => assert!(table < schema.tables.len()),
+            }
+        }
+    }
 }