diff --git a/Cargo.lock b/Cargo.lock index 6f9455400ec..74e64fb602a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6285,6 +6285,15 @@ dependencies = [ "rand_core 0.9.3", ] +[[package]] +name = "rand_xoshiro" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f703f4665700daf5512dcca5f43afa6af89f09db47fb56be587f80636bda2d41" +dependencies = [ + "rand_core 0.9.3", +] + [[package]] name = "rayon" version = "1.11.0" @@ -8223,6 +8232,8 @@ dependencies = [ "pretty_assertions", "prometheus", "proptest", + "rand 0.9.2", + "rand_xoshiro", "smallvec", "spacetimedb-commitlog", "spacetimedb-data-structures", diff --git a/Cargo.toml b/Cargo.toml index 9f7852863c8..2205e920a45 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -281,6 +281,7 @@ quick-junit = { version = "0.3.2" } quote = "1.0.8" rand08 = { package = "rand", version = "0.8" } rand = "0.9" +rand_xoshiro = "0.7" rayon = "1.8" rayon-core = "1.11.0" regex = "1" diff --git a/crates/datastore/Cargo.toml b/crates/datastore/Cargo.toml index bb900ad8926..c8e14e0ac8d 100644 --- a/crates/datastore/Cargo.toml +++ b/crates/datastore/Cargo.toml @@ -34,6 +34,11 @@ strum.workspace = true thiserror.workspace = true thin-vec.workspace = true +# For simulating sequence value skips. +# See `fn should_simulate_sequence_reallocation` in src/locking_tx_datastore/mut_tx.rs +rand.workspace = true +rand_xoshiro.workspace = true + [features] # Print a warning when doing an unindexed `iter_by_col_range` on a large table. unindexed_iter_by_col_range_warn = [] diff --git a/crates/datastore/src/error.rs b/crates/datastore/src/error.rs index 912abca4fce..c2451a6efdd 100644 --- a/crates/datastore/src/error.rs +++ b/crates/datastore/src/error.rs @@ -1,6 +1,6 @@ use super::system_tables::SystemTable; use spacetimedb_lib::db::raw_def::{v9::RawSql, RawIndexDefV8}; -use spacetimedb_primitives::{ColId, ColList, IndexId, SequenceId, TableId, ViewId}; +use spacetimedb_primitives::{ColId, IndexId, SequenceId, TableId, ViewId}; use spacetimedb_sats::buffer::DecodeError; use spacetimedb_sats::product_value::InvalidFieldError; use spacetimedb_sats::raw_identifier::RawNamespacedIdentifier; @@ -116,22 +116,10 @@ pub enum SequenceError { Exist(String), #[error("Sequence `{0}`: The increment is 0, and this means the sequence can't advance.")] IncrementIsZero(String), - #[error("Sequence `{0}`: The min_value {1} must < max_value {2}.")] - MinMax(String, i128, i128), - #[error("Sequence `{0}`: The start value {1} must be >= min_value {2}.")] - MinStart(String, i128, i128), - #[error("Sequence `{0}`: The start value {1} must be <= min_value {2}.")] - MaxStart(String, i128, i128), - #[error("Sequence `{0}` failed to decode value from Sled (not a u128).")] - SequenceValue(String), #[error("Sequence ID `{0}` not found.")] NotFound(SequenceId), - #[error("Sequence applied to a non-integer field. Column `{col}` is of type {{found.to_sats()}}.")] - NotInteger { col: String, found: AlgebraicType }, - #[error("Sequence ID `{0}` still had no values left after allocation.")] - UnableToAllocate(SequenceId), - #[error("Autoinc constraint on table {0:?} spans more than one column: {1:?}")] - MultiColumnAutoInc(TableId, ColList), + #[error("Incrementing sequence with previous value {0} would result in integer overflow")] + Overflow(i128), } impl From for DatastoreError { diff --git a/crates/datastore/src/locking_tx_datastore/committed_state.rs b/crates/datastore/src/locking_tx_datastore/committed_state.rs index 0ce629524c4..6142d431a27 100644 --- a/crates/datastore/src/locking_tx_datastore/committed_state.rs +++ b/crates/datastore/src/locking_tx_datastore/committed_state.rs @@ -1,7 +1,6 @@ use super::{ datastore::Result, delete_table::DeleteTable, - sequence::{Sequence, SequencesState}, state_view::StateView, tx_state::{IndexIdMap, PendingSchemaChange, TxState}, IterByColEqTx, @@ -36,13 +35,15 @@ use crate::{ }; use anyhow::anyhow; use core::{convert::Infallible, ops::RangeBounds}; +use rand::SeedableRng; +use rand_xoshiro::Xoshiro128PlusPlus; use spacetimedb_data_structures::map::{HashMap, HashSet, IntMap, IntSet}; use spacetimedb_durability::TxOffset; use spacetimedb_lib::{db::auth::StTableType, Identity}; use spacetimedb_primitives::{ColList, IndexId, TableId}; use spacetimedb_sats::memory_usage::MemoryUsage; use spacetimedb_sats::{AlgebraicValue, ProductValue}; -use spacetimedb_schema::schema::TableSchema; +use spacetimedb_schema::schema::{SequenceSchema, TableSchema}; use spacetimedb_table::{ blob_store::{BlobStore, HashMapBlobStore}, indexes::{RowPointer, SquashedOffset}, @@ -95,6 +96,17 @@ pub struct CommittedState { /// - system tables: `st_view_sub`, `st_view_arg` /// - Tables which back views. pub(super) ephemeral_tables: EphemeralTables, + + /// RNG source for deciding when advancing a sequence should simulate a reallocation. + /// + /// We don't want users to depend on sequence values being strictly sequential, + /// as we have in the past and may in the future used optimizations that would cause values to be skipped. + /// To prevent this, [`get_next_sequence_value`](super::mut_tx::get_next_sequence_value) occasionally simulates a skip ahead. + /// + /// We use an explicit PRNG here rather than reading from the thread RNG because we'd like our tests to be deterministic. + /// + /// We chose Xoshiro128++ because it is fast and small, and we do not need a cryptographically secure PRNG for this purpose. + pub(super) sequence_advance_simulate_reallocation_rng: Xoshiro128PlusPlus, } impl CommittedState { @@ -130,6 +142,9 @@ impl MemoryUsage for CommittedState { read_sets, view_instances, ephemeral_tables, + // Don't include the PRNG; it doesn't live on or use the heap, + // and it's easier to just ignore it here than to write a trait impl that returns zero. + sequence_advance_simulate_reallocation_rng: _, } = self; // NOTE(centril): We do not want to include the heap usage of `page_pool` as it's a shared resource. next_tx_offset.heap_usage() @@ -211,6 +226,16 @@ impl CommittedState { page_pool, datastore_page_bytes: 0, ephemeral_tables: <_>::default(), + sequence_advance_simulate_reallocation_rng: { + #[cfg(test)] + { + Xoshiro128PlusPlus::seed_from_u64(0) + } + #[cfg(not(test))] + { + Xoshiro128PlusPlus::from_rng(&mut rand::rng()) + } + }, } } @@ -367,9 +392,9 @@ impl CommittedState { sequence_name: seq.sequence_name.clone(), table_id: seq.table_id, col_pos: seq.col_pos, - increment: seq.increment, - min_value: seq.min_value, - max_value: seq.max_value, + increment: SequenceSchema::INCREMENT, + min_value: SequenceSchema::MIN_VALUE, + max_value: SequenceSchema::MAX_VALUE, start: seq.start, // In practice, this means we will actually start at start - 1, since `allocated` // overrides start, but we keep these fields set this way to match databases @@ -420,31 +445,6 @@ impl CommittedState { Ok(()) } - /// Builds the in-memory state of sequences from `st_sequence` system table. - /// The tables store the lasted allocated value, which tells us where to start generating. - pub(super) fn build_sequence_state(&mut self) -> Result { - let mut sequence_state = SequencesState::default(); - let st_sequences = self.tables.get(&ST_SEQUENCE_ID).unwrap(); - for row_ref in st_sequences.scan_rows(&self.blob_store) { - let sequence = StSequenceRow::try_from(row_ref)?; - let seq = Sequence::new(sequence.clone().into(), Some(sequence.allocated)); - - // Clobber any existing in-memory `Sequence`. - // Such a value may exist because, when replaying without a snapshot, - // `build_sequence_state` is called twice: - // once when bootstrapping the empty datastore, - // and then again after replaying the commitlog. - // At this latter time, `sequence_state.get(seq.id())` for the system table sequences - // will return a sequence with incorrect `allocated`, - // as it will reflect the state after initializing the system tables, - // but before creating any user tables. - // The `sequence` we read out of `row_ref` above, and used to construct `seq`, - // will correctly reflect the state after creating user tables. - sequence_state.insert(seq); - } - Ok(sequence_state) - } - /// Returns an iterator doing a full table scan on `table_id`. pub(super) fn table_scan<'a>(&'a self, table_id: TableId) -> Option> { Some(self.get_table(table_id)?.scan_rows(&self.blob_store)) @@ -769,20 +769,16 @@ impl CommittedState { } /// Rolls back the changes immediately made to the committed state during a transaction. - pub(super) fn rollback(&mut self, seq_state: &mut SequencesState, tx_state: TxState) -> TxOffset { + pub(super) fn rollback(&mut self, tx_state: TxState) -> TxOffset { // Roll back the changes in the reverse order in which they were made // so that e.g., the last change is undone first. for change in tx_state.pending_schema_changes.into_iter().rev() { - self.rollback_pending_schema_change(seq_state, change); + self.rollback_pending_schema_change(change); } self.next_tx_offset.saturating_sub(1) } - fn rollback_pending_schema_change( - &mut self, - seq_state: &mut SequencesState, - change: PendingSchemaChange, - ) -> Option<()> { + fn rollback_pending_schema_change(&mut self, change: PendingSchemaChange) -> Option<()> { use PendingSchemaChange::*; match change { // An index was removed. Add it back. @@ -923,16 +919,14 @@ impl CommittedState { } } // A sequence was removed. Add it back. - SequenceRemoved(table_id, seq, schema) => { + SequenceRemoved(table_id, schema) => { let table = self.tables.get_mut(&table_id)?; table.with_mut_schema(|s| s.update_sequence(schema)); - seq_state.insert(seq); } // A sequence was added. Remove it. SequenceAdded(table_id, sequence_id) => { let table = self.tables.get_mut(&table_id)?; table.with_mut_schema(|s| s.remove_sequence(sequence_id)); - seq_state.remove(sequence_id); } } diff --git a/crates/datastore/src/locking_tx_datastore/datastore.rs b/crates/datastore/src/locking_tx_datastore/datastore.rs index 7293a81fba0..457f3fae4f3 100644 --- a/crates/datastore/src/locking_tx_datastore/datastore.rs +++ b/crates/datastore/src/locking_tx_datastore/datastore.rs @@ -1,9 +1,6 @@ -use super::{ - committed_state::CommittedState, mut_tx::MutTxId, sequence::SequencesState, state_view::StateView, tx::TxId, - tx_state::TxState, -}; +use super::{committed_state::CommittedState, mut_tx::MutTxId, state_view::StateView, tx::TxId, tx_state::TxState}; use crate::execution_context::{Workload, WorkloadType}; -use crate::locking_tx_datastore::replay::{build_sequence_state, ErrorBehavior, Replay}; +use crate::locking_tx_datastore::replay::{ErrorBehavior, Replay}; use crate::{ db_metrics::DB_METRICS, error::{DatastoreError, TableError}, @@ -25,7 +22,7 @@ use crate::{ }; use anyhow::anyhow; use core::ops::RangeBounds; -use parking_lot::{Mutex, RwLock}; +use parking_lot::RwLock; use spacetimedb_data_structures::map::{HashCollectionExt, HashMap}; use spacetimedb_durability::TxOffset; use spacetimedb_lib::{db::auth::StAccess, metrics::ExecutionMetrics}; @@ -57,7 +54,6 @@ pub type Result = std::result::Result; /// Lock Acquisition Order: /// 1. `memory` /// 2. `committed_state` -/// 3. `sequence_state` /// /// All locking mechanisms are encapsulated within the struct through local methods. #[derive(Clone)] @@ -66,8 +62,6 @@ pub struct Locking { // TODO(cloutiertyler): This was made `pub` for the datastore split. This should be // made private again. pub committed_state: Arc>, - /// The state of sequence generation in this database. - pub(super) sequence_state: Arc>, /// The identity of this database. pub(crate) database_identity: Identity, } @@ -76,14 +70,9 @@ impl MemoryUsage for Locking { fn heap_usage(&self) -> usize { let Self { committed_state, - sequence_state, database_identity, } = self; - std::mem::size_of_val(&**committed_state) - + committed_state.read().heap_usage() - + std::mem::size_of_val(&**sequence_state) - + sequence_state.lock().heap_usage() - + database_identity.heap_usage() + std::mem::size_of_val(&**committed_state) + committed_state.read().heap_usage() + database_identity.heap_usage() } } @@ -91,7 +80,6 @@ impl Locking { pub fn new(database_identity: Identity, page_pool: PagePool) -> Self { Self { committed_state: Arc::new(RwLock::new(CommittedState::new(page_pool))), - sequence_state: <_>::default(), database_identity, } } @@ -114,9 +102,6 @@ impl Locking { // Create the system tables and insert information about themselves into commit_state.bootstrap_system_tables(database_identity)?; - // The database tables are now initialized with the correct data. - // Now we have to build our in memory structures. - build_sequence_state(&datastore, &mut commit_state)?; // We don't want to build indexes here; we'll build those later, // in `rebuild_state_after_replay`. @@ -203,11 +188,6 @@ impl Locking { // Double check that our in-memory system table ids match the on-disk schemas. // committed_state.assert_system_table_schemas_match()?; - // Set the sequence state. In practice we will end up doing this again after replaying - // the commit log, but we do it here too just to avoid having an incorrectly restored - // snapshot. - build_sequence_state(&datastore, &mut committed_state)?; - // The next TX offset after restoring from a snapshot is one greater than the snapshotted offset. committed_state.next_tx_offset = tx_offset + 1; @@ -979,12 +959,10 @@ impl MutTx for Locking { let timer = Instant::now(); let committed_state_write_lock = self.committed_state.write_arc(); - let sequence_state_lock = self.sequence_state.lock_arc(); let lock_wait_time = timer.elapsed(); MutTxId { committed_state_write_lock, - sequence_state_lock, tx_state: TxState::default(), lock_wait_time, read_sets: <_>::default(), @@ -1015,12 +993,10 @@ impl Locking { let timer = Instant::now(); let committed_state_write_lock = self.committed_state.try_write_arc()?; - let sequence_state_lock = self.sequence_state.try_lock_arc()?; let lock_wait_time = timer.elapsed(); Some(MutTxId { committed_state_write_lock, - sequence_state_lock, tx_state: TxState::default(), lock_wait_time, read_sets: <_>::default(), @@ -1324,10 +1300,7 @@ pub(crate) mod tests { sequence_name: RawNamespacedIdentifier::new(value.name), table_id: value.table.into(), col_pos: value.col_pos.into(), - increment: 1, start: value.start, - min_value: 1, - max_value: i128::MAX, } } } @@ -1435,9 +1408,6 @@ pub(crate) mod tests { col_pos: 0.into(), sequence_name: "Foo_id_seq".into(), start: 1, - increment: 1, - min_value: 1, - max_value: i128::MAX, }; user_public_table( map_array(basic_table_schema_cols()), @@ -2127,11 +2097,42 @@ pub(crate) mod tests { let _ = datastore.rollback_mut_tx(tx); let mut tx = begin_mut_tx(&datastore); insert(&datastore, &mut tx, table_id, &row)?; + // The rolled-back insert did not consume the first auto-inc value. #[rustfmt::skip] - assert_eq!(all_rows(&datastore, &tx, table_id), vec![u32_str_u32(2, "Foo", 18)]); + assert_eq!(all_rows(&datastore, &tx, table_id), vec![u32_str_u32(1, "Foo", 18)]); Ok(()) } + #[test] + fn sequence_occasionally_skips_values_to_simulate_reallocation() -> ResultTest<()> { + let datastore = get_datastore()?; + let mut tx = begin_mut_tx(&datastore); + let mut schema = basic_table_schema_with_indices(basic_indices(), basic_constraints()); + schema.primary_key = Some(0.into()); + let table_id = datastore.create_table_mut_tx(&mut tx, schema)?; + commit(&datastore, tx)?; + + let mut tx = begin_mut_tx(&datastore); + let mut previous_value = 0; + + // Determined experimentally; + // the fixed seed we use for tests first hits a simulated reallocation point + // somewhere between rows 8192 and 16384. + const MAX_ROWS: u32 = 16384; + + for row_number in 0..MAX_ROWS { + let row = product![0_u32, format!("row_{row_number}"), 0_u32]; + let (_, row_ref) = insert(&datastore, &mut tx, table_id, &row)?; + let value = row_ref.read_col::(0).unwrap(); + if value != previous_value + 1 { + return Ok(()); + } + previous_value = value; + } + + panic!("did not simulate a sequence reallocation after inserting {MAX_ROWS} rows"); + } + fn assert_st_indices(tx: &MutTxId, include_age: bool) -> ResultTest<()> { let seq_start = FIRST_NON_SYSTEM_ID; #[rustfmt::skip] @@ -2325,9 +2326,6 @@ pub(crate) mod tests { col_pos: 0.into(), sequence_name: "seq".into(), start: 1, - increment: 1, - min_value: 1, - max_value: i128::MAX, }; let seq_id = datastore.create_sequence_mut_tx(&mut tx, sequence.clone())?; assert_matches!( @@ -2386,7 +2384,8 @@ pub(crate) mod tests { let _ = datastore.rollback_mut_tx(tx); let mut tx = begin_mut_tx(&datastore); assert_eq!(tx.pending_schema_changes(), []); - insert_assert_and_remove(&mut tx, &zero, &product![2])?; + // The auto-inc value generated before the rollback remains available. + insert_assert_and_remove(&mut tx, &zero, &one)?; // Drop the seq and commit this time around. In the next tx, we witness that there's no seq. datastore.drop_sequence_mut_tx(&mut tx, seq_id)?; @@ -3500,10 +3499,12 @@ pub(crate) mod tests { "Unexpected delete entries after altering the table" ); + // The rolled-back migration did not consume 7, so the committed migration uses it + // after the two initial committed rows with IDs 5 and 6. let inserted_rows = [ product![5u64, AlgebraicValue::sum(0, 1u16.into()), 42u8], product![6u64, AlgebraicValue::sum(0, 1u16.into()), 42u8], - product![8u64, AlgebraicValue::sum(0, 1u16.into()), 42u8], + product![7u64, AlgebraicValue::sum(0, 1u16.into()), 42u8], ]; let new_entry = tx_data diff --git a/crates/datastore/src/locking_tx_datastore/mod.rs b/crates/datastore/src/locking_tx_datastore/mod.rs index 2eebaf4e619..cb787fececc 100644 --- a/crates/datastore/src/locking_tx_datastore/mod.rs +++ b/crates/datastore/src/locking_tx_datastore/mod.rs @@ -4,7 +4,6 @@ pub mod committed_state; pub mod datastore; mod mut_tx; pub use mut_tx::{FuncCallType, IndexScanPointOrRange, MutTxId, ViewCallInfo, ViewInstanceArgs}; -mod sequence; pub mod state_view; pub use state_view::{IterByColEqTx, IterByColRangeTx}; pub mod delete_table; @@ -17,11 +16,10 @@ mod tx_state; pub use tx_state::PendingSchemaChange; use parking_lot::{ - lock_api::{ArcMutexGuard, ArcRwLockReadGuard, ArcRwLockWriteGuard}, - RawMutex, RawRwLock, + lock_api::{ArcRwLockReadGuard, ArcRwLockWriteGuard}, + RawRwLock, }; // Type aliases for lock guards type SharedWriteGuard = ArcRwLockWriteGuard; -type SharedMutexGuard = ArcMutexGuard; type SharedReadGuard = ArcRwLockReadGuard; diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index 042fa6cfc5f..807f3e56606 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -2,11 +2,10 @@ use super::{ committed_state::{CommitTableForInsertion, CommittedState}, datastore::{Result, TxMetrics}, delete_table::DeleteTable, - sequence::{Sequence, SequencesState}, state_view::{IterByColEqMutTx, IterByColRangeMutTx, IterMutTx, StateView}, tx::TxId, tx_state::{IndexIdMap, PendingSchemaChange, TxState, TxTableForInsertion}, - SharedMutexGuard, SharedWriteGuard, + SharedWriteGuard, }; use crate::{ error::ViewError, @@ -36,15 +35,14 @@ use crate::{ }; use core::{cell::RefCell, iter, ops::RangeBounds}; use itertools::Either; +use rand::Rng; +use rand_xoshiro::Xoshiro128PlusPlus; use smallvec::SmallVec; use spacetimedb_data_structures::map::{HashMap, HashSet, IntMap}; use spacetimedb_durability::TxOffset; use spacetimedb_execution::{dml::MutDatastore, Datastore, DeltaStore, Row}; use spacetimedb_lib::{ - db::raw_def::v9::RawSql, - db::{auth::StAccess, raw_def::SEQUENCE_ALLOCATION_STEP}, - empty_view_arg_hash_value, - metrics::ExecutionMetrics, + db::auth::StAccess, db::raw_def::v9::RawSql, empty_view_arg_hash_value, metrics::ExecutionMetrics, sender_view_arg_hash_value, ConnectionId, Identity, Timestamp, }; use spacetimedb_primitives::{ @@ -444,7 +442,6 @@ pub enum FuncCallType { pub struct MutTxId { pub(super) tx_state: TxState, pub(super) committed_state_write_lock: SharedWriteGuard, - pub(super) sequence_state_lock: SharedMutexGuard, pub(super) lock_wait_time: Duration, pub(super) read_sets: ViewReadSets, pub(super) view_instances: ViewInstanceTxState, @@ -1674,19 +1671,14 @@ impl MutTxId { // Store sequence values to restore them later with new table. // Using a map from name to value as the new sequence ids will be different. // and I am not sure if we should rely on the order of sequences in the table schema. - let mut seq_values: HashMap<_, (i128, i128)> = HashMap::default(); + let mut seq_values: HashMap<_, i128> = HashMap::default(); for seq in &original_table_schema.sequences { - let value = self - .sequence_state_lock - .get_sequence_mut(seq.sequence_id) - .expect("sequence exists in original schema and should in sequence state.") - .get_value(); let allocated = self .iter_by_col_eq(ST_SEQUENCE_ID, StSequenceFields::SequenceId, &seq.sequence_id.into())? .last() .ok_or(SequenceError::NotFound(seq.sequence_id))? .read_col(StSequenceFields::Allocated)?; - seq_values.insert(seq.sequence_name.clone(), (value, allocated)); + seq_values.insert(seq.sequence_name.clone(), allocated); } // Drop existing table first due to unique constraints on table name in `st_table` @@ -1721,32 +1713,22 @@ impl MutTxId { /// schema change (for example `add_columns_to_table`). /// /// `create_table(...)` generates fresh table/sequence IDs and inserts fresh - /// rows into `st_sequence`. We then restore preserved `(value, allocated)` + /// rows into `st_sequence`. We then restore preserved `allocated` /// by sequence name: - /// - update in-memory sequence state (`SequencesState`) so this process keeps - /// allocating from the same point; /// - patch the newly created `st_sequence` row so reopen/replay restores the /// same allocation cursor instead of sequence start. fn create_table_and_update_seq( &mut self, table_schema: TableSchema, - seq_values: HashMap, + seq_values: HashMap, ) -> Result { let table_id = self.create_table(table_schema)?; let table_schema = self.schema_for_table(table_id)?; for seq in table_schema.sequences.iter() { - let (value, allocated) = *seq_values + let allocated = *seq_values .get(&seq.sequence_name) - .ok_or_else(|| SequenceError::NotFound(seq.sequence_id))?; - { - let new_seq = self - .sequence_state_lock - .get_sequence_mut(seq.sequence_id) - .expect("sequence just created"); - new_seq.update_value(value); - new_seq.update_allocation(allocated); - } + .ok_or(SequenceError::NotFound(seq.sequence_id))?; // This updates the new `st_sequence` row created by `create_table(...)` // above (old table rows are already dropped). @@ -2051,12 +2033,7 @@ impl MutTxId { } pub fn get_next_sequence_value(&mut self, seq_id: SequenceId) -> Result { - get_next_sequence_value( - &mut self.tx_state, - &self.committed_state_write_lock, - &mut self.sequence_state_lock, - seq_id, - ) + get_next_sequence_value(&mut self.tx_state, &mut self.committed_state_write_lock, seq_id) } } @@ -2070,32 +2047,30 @@ pub enum IndexScanPointOrRange<'de, 'a> { Range(IndexScanRanged<'a>), } -fn get_sequence_mut(seq_state: &mut SequencesState, seq_id: SequenceId) -> Result<&mut Sequence> { - seq_state - .get_sequence_mut(seq_id) - .ok_or_else(|| SequenceError::NotFound(seq_id).into()) +const SEQUENCE_SIMULATED_ALLOCATION_CHUNK: u16 = 4096; + +/// Should the next sequence allocation skip values? +/// +/// We don't want users to depend on sequence values being strictly sequential, +/// as we have in the past and may in the future used optimizations that would cause values to be skipped. +/// To prevent this, [`get_next_sequence_value`] occasionally simulates a skip ahead. +/// +/// The supplied `rng` should be the one in the committed state for this purpose, +/// so that tests which rely on auto-inc sequences will be deterministic. +/// See [`CommittedState::sequence_advance_simulate_reallocation_rng`]. +fn should_simulate_sequence_reallocation(rng: &mut Xoshiro128PlusPlus) -> bool { + // Skip an average of once every 4096 values, the same as the chunk size. + // Chosen completely arbitrarily. + rng.random::().is_multiple_of(SEQUENCE_SIMULATED_ALLOCATION_CHUNK) } fn get_next_sequence_value( tx_state: &mut TxState, - committed_state: &CommittedState, - seq_state: &mut SequencesState, + committed_state: &mut CommittedState, seq_id: SequenceId, ) -> Result { - { - let sequence = get_sequence_mut(seq_state, seq_id)?; - - // If there are allocated sequence values, return the new value. - // `gen_next_value` internally checks that the new allocation is acceptable, - // i.e. is less than or equal to the allocation amount. - // Note that on restart we start one after the allocation amount. - if let Some(value) = sequence.gen_next_value() { - return Ok(value); - } - } - - // Allocate new sequence values - // If we're out of allocations, then update the sequence row in st_sequences to allocate a fresh batch of sequences. + // Allocate a new sequence value + // Read and update the `st_sequence` row to record the new value. let old_seq_row_ref = iter_by_col_eq( tx_state, committed_state, @@ -2106,13 +2081,35 @@ fn get_next_sequence_value( .last() .unwrap(); let old_seq_row_ptr = old_seq_row_ref.pointer(); - let seq_row = { + let (seq_row, value) = { let mut seq_row = StSequenceRow::try_from(old_seq_row_ref)?; - let sequence = get_sequence_mut(seq_state, seq_id)?; - let new_allocated = sequence.allocate_steps(SEQUENCE_ALLOCATION_STEP as usize); - seq_row.allocated = new_allocated; - seq_row + let value = seq_row.allocated; + + // We don't want users to depend on sequence values being strictly sequential, + // as we have in the past and may in the future used optimizations that would cause values to be skipped. + // To prevent this, skip sequence values at a low rate. + if should_simulate_sequence_reallocation(&mut committed_state.sequence_advance_simulate_reallocation_rng) { + // Simulate an event where you skip to the next block of 4096 values. + // Do this by masking off the low 11 bits of the counter, + // then incrementing the 12th bit, + // resulting in a value that is strictly larger than the previous value, + // and which is divisible by 4096. + let mask = !(SEQUENCE_SIMULATED_ALLOCATION_CHUNK as i128 - 1); + let lower = seq_row.allocated & mask; + let higher = lower + .checked_add(SEQUENCE_SIMULATED_ALLOCATION_CHUNK as i128) + .ok_or(SequenceError::Overflow(value))?; + assert!(higher > value); + + seq_row.allocated = higher; + } else { + seq_row.allocated = value + .checked_add(SequenceSchema::INCREMENT) + .ok_or(SequenceError::Overflow(value))?; + }; + + (seq_row, value) }; delete(tx_state, committed_state, ST_SEQUENCE_ID, old_seq_row_ptr)?; @@ -2124,12 +2121,10 @@ fn get_next_sequence_value( // has ID 0, and would otherwise trigger autoinc. with_sys_table_buf(|buf| { to_writer(buf, &seq_row).unwrap(); - insert::(tx_state, committed_state, seq_state, ST_SEQUENCE_ID, buf) + insert::(tx_state, committed_state, ST_SEQUENCE_ID, buf) })?; - get_sequence_mut(seq_state, seq_id)? - .gen_next_value() - .ok_or_else(|| SequenceError::UnableToAllocate(seq_id).into()) + Ok(value) } impl MutTxId { @@ -2172,10 +2167,10 @@ impl MutTxId { table_id, col_pos: seq.col_pos, allocated: seq.start, - increment: seq.increment, + increment: SequenceSchema::INCREMENT, start: seq.start, - min_value: seq.min_value, - max_value: seq.max_value, + min_value: SequenceSchema::MIN_VALUE, + max_value: SequenceSchema::MAX_VALUE, }; let row = self.insert_via_serialize_bsatn(ST_SEQUENCE_ID, &sequence_row)?; let seq_id = row.1.collapse().read_col(StSequenceFields::SequenceId)?; @@ -2185,7 +2180,6 @@ impl MutTxId { let ((tx_table, ..), (commit_table, ..)) = self.get_or_create_insert_table_mut(table_id)?; // This won't clone-write when creating a table but likely to otherwise. tx_table.with_mut_schema_and_clone(commit_table, |s| s.update_sequence(schema.clone())); - self.sequence_state_lock.insert(Sequence::new(schema, None)); self.push_schema_change(PendingSchemaChange::SequenceAdded(table_id, seq_id)); log::trace!("SEQUENCE CREATED: id = {seq_id}"); @@ -2204,18 +2198,13 @@ impl MutTxId { // Delete from system tables. self.delete(ST_SEQUENCE_ID, st_sequence_ref.pointer())?; - // Drop the sequence from in-memory tables. - let sequence = self - .sequence_state_lock - .remove(sequence_id) - .expect("there should be a sequence in the committed state if we reach here"); let ((tx_table, ..), (commit_table, ..)) = self.get_or_create_insert_table_mut(table_id)?; // This likely will do a clone-write as over time? // The schema might have found other referents. let schema = commit_table .with_mut_schema_and_clone(tx_table, |s| s.remove_sequence(sequence_id)) .expect("there should be a schema in the committed state if we reach here"); - self.push_schema_change(PendingSchemaChange::SequenceRemoved(table_id, sequence, schema)); + self.push_schema_change(PendingSchemaChange::SequenceRemoved(table_id, schema)); Ok(()) } @@ -2787,9 +2776,7 @@ impl MutTxId { /// - [`TxMetrics`], various measurements of the work performed by this transaction. /// - `ReducerName`, the name of the reducer which ran during this transaction. pub fn rollback(mut self) -> (TxOffset, TxMetrics, Option) { - let offset = self - .committed_state_write_lock - .rollback(&mut self.sequence_state_lock, self.tx_state); + let offset = self.committed_state_write_lock.rollback(self.tx_state); // Compute and keep enough info that we can // record metrics after the transaction has ended @@ -2816,8 +2803,7 @@ impl MutTxId { /// - [`TxMetrics`], various measurements of the work performed by this transaction. /// - [`TxId`], a read-only transaction with a shared lock on the committed state. pub fn rollback_downgrade(mut self, workload: Workload) -> (TxMetrics, TxId) { - self.committed_state_write_lock - .rollback(&mut self.sequence_state_lock, self.tx_state); + self.committed_state_write_lock.rollback(self.tx_state); // Compute and keep enough info that we can // record metrics after the transaction has ended @@ -3326,13 +3312,7 @@ impl MutTxId { table_id: TableId, row: &[u8], ) -> Result<(ColList, RowRefInsertion<'_>, InsertFlags)> { - insert::( - &mut self.tx_state, - &self.committed_state_write_lock, - &mut self.sequence_state_lock, - table_id, - row, - ) + insert::(&mut self.tx_state, &mut self.committed_state_write_lock, table_id, row) } } @@ -3355,8 +3335,7 @@ impl MutTxId { /// - The "commit table for insertion" for further processing. fn insert_physically_maybe_generate<'a, const GENERATE: bool>( tx_state: &'a mut TxState, - committed_state: &'a CommittedState, - seq_state: &mut SequencesState, + committed_state: &'a mut CommittedState, table_id: TableId, row: &[u8], ) -> Result<( @@ -3366,12 +3345,11 @@ fn insert_physically_maybe_generate<'a, const GENERATE: bool>( TxTableForInsertion<'a>, CommitTableForInsertion<'a>, )> { - // Get commit table and friends. - let commit_parts = committed_state.get_table_and_blob_store(table_id)?; - let (commit_table, ..) = commit_parts; - // Get the insert table, so we can write the row into it. - let (tx_table, tx_blob_store, _) = tx_state.get_table_and_blob_store_or_create_from(table_id, commit_table); + let (tx_table, tx_blob_store, _) = { + let (commit_table, ..) = committed_state.get_table_and_blob_store(table_id)?; + tx_state.get_table_and_blob_store_or_create_from(table_id, commit_table) + }; // 1. Insert the physical row. let page_pool = &committed_state.page_pool; @@ -3386,12 +3364,7 @@ fn insert_physically_maybe_generate<'a, const GENERATE: bool>( // Generate a value for every column in the row that needs it. let mut seq_vals: SmallVec<[i128; 1]> = <_>::default(); for sequence_id in seqs_to_use { - seq_vals.push(get_next_sequence_value( - tx_state, - committed_state, - seq_state, - sequence_id, - )?); + seq_vals.push(get_next_sequence_value(tx_state, committed_state, sequence_id)?); } // Write the generated values to the physical row at `tx_row_ptr`. @@ -3414,6 +3387,7 @@ fn insert_physically_maybe_generate<'a, const GENERATE: bool>( (tx_parts, ColList::empty()) }; + let commit_parts = committed_state.get_table_and_blob_store(table_id)?; Ok((tx_row_ptr, gen_cols, blob_bytes, tx_parts, commit_parts)) } @@ -3434,8 +3408,7 @@ fn insert_physically_maybe_generate<'a, const GENERATE: bool>( /// - any insert flags. pub(super) fn insert<'a, const GENERATE: bool>( tx_state: &'a mut TxState, - committed_state: &'a CommittedState, - seq_state: &mut SequencesState, + committed_state: &'a mut CommittedState, table_id: TableId, row: &[u8], ) -> Result<(ColList, RowRefInsertion<'a>, InsertFlags)> { @@ -3445,7 +3418,7 @@ pub(super) fn insert<'a, const GENERATE: bool>( blob_bytes, (tx_table, tx_blob_store, delete_table), (commit_table, commit_blob_store, _), - ) = insert_physically_maybe_generate::(tx_state, committed_state, seq_state, table_id, row)?; + ) = insert_physically_maybe_generate::(tx_state, committed_state, table_id, row)?; let insert_flags = InsertFlags { is_scheduler_table: tx_table.is_scheduler(), @@ -3587,8 +3560,7 @@ impl MutTxId { (commit_table, commit_blob_store, _), ) = insert_physically_maybe_generate::( &mut self.tx_state, - &self.committed_state_write_lock, - &mut self.sequence_state_lock, + &mut self.committed_state_write_lock, table_id, row, )?; diff --git a/crates/datastore/src/locking_tx_datastore/replay.rs b/crates/datastore/src/locking_tx_datastore/replay.rs index c42185e2bb2..3e0d3e41573 100644 --- a/crates/datastore/src/locking_tx_datastore/replay.rs +++ b/crates/datastore/src/locking_tx_datastore/replay.rs @@ -84,7 +84,7 @@ pub fn apply_history( .set((end_tx_offset - start_tx_offset) as _); log::info!("[{database_identity}] DATABASE: applied transaction history"); - replay.committed_state().rebuild_state_after_replay(datastore)?; + replay.committed_state().rebuild_state_after_replay()?; log::info!("[{database_identity}] DATABASE: rebuilt state after replay"); Ok(()) @@ -489,7 +489,7 @@ impl<'cs> ReplayCommittedState<'cs> { /// This is necessary because, for example, inserting a row into `st_table` /// is not equivalent to calling `create_table`. /// There may eventually be better way to do this, but this will have to do for now. - pub fn rebuild_state_after_replay(&mut self, datastore: &Locking) -> Result<()> { + pub fn rebuild_state_after_replay(&mut self) -> Result<()> { // Prior versions of `RelationalDb::migrate_system_tables` (defined in the `core` crate) // initialized newly-created system sequences to `allocation: 4097`, // while `committed_state::bootstrap_system_tables` sets `allocation: 4096`. @@ -515,9 +515,6 @@ impl<'cs> ReplayCommittedState<'cs> { self.collect_ephemeral_tables()?; self.rebuild_datastore_page_bytes(); - // Figure out where to pick up for each sequence. - build_sequence_state(datastore, self)?; - Ok(()) } @@ -1075,13 +1072,6 @@ impl<'cs> ReplayCommittedState<'cs> { } } -pub(super) fn build_sequence_state(datastore: &Locking, cs: &mut CommittedState) -> Result<()> { - let sequence_state = cs.build_sequence_state()?; - // Reset our sequence state so that they start in the right places. - *datastore.sequence_state.lock() = sequence_state; - Ok(()) -} - impl StateView for ReplayCommittedState<'_> { /// Find the `st_table` row for `table_id`, /// first inspecting [`Self::replay_table_updated`], diff --git a/crates/datastore/src/locking_tx_datastore/sequence.rs b/crates/datastore/src/locking_tx_datastore/sequence.rs deleted file mode 100644 index 1d29dda29df..00000000000 --- a/crates/datastore/src/locking_tx_datastore/sequence.rs +++ /dev/null @@ -1,487 +0,0 @@ -use spacetimedb_data_structures::map::IntMap; -use spacetimedb_primitives::SequenceId; -use spacetimedb_sats::memory_usage::MemoryUsage; -use spacetimedb_schema::schema::SequenceSchema; - -#[derive(Debug, PartialEq)] -// TODO(cloutiertyler): The below was made `pub` for the datastore split. We should -// investigate if this should be private again. -pub struct Sequence { - schema: SequenceSchema, - // The next value to be returned by this sequence. - value: i128, - // The number we have persisted as a lower bound for the next restart. - // This is the first value to be returned after a restart, so when we - // reach this value, the user needs to call allocate_steps and update - // the corresponding system table row. - allocated: i128, -} - -impl MemoryUsage for Sequence { - fn heap_usage(&self) -> usize { - // MEMUSE: intentionally ignoring schema - self.value.heap_usage() - } -} - -impl Sequence { - pub(super) fn new(schema: SequenceSchema, previous_allocation: Option) -> Self { - if schema.start < schema.min_value || schema.start > schema.max_value { - panic!( - "Invalid sequence: start value {} is out of bounds for sequence with min_value {} and max_value {}", - schema.start, schema.min_value, schema.max_value - ); - } - if schema.max_value <= schema.min_value { - panic!("Invalid sequence: max_value must be greater than min_value"); - } - if schema.increment == 0 { - panic!("Invalid sequence: increment must be non-zero"); - } - if schema.increment.unsigned_abs() >= (schema.max_value - schema.min_value) as u128 { - panic!( - "Invalid sequence: increment must be less than or equal to the range between min_value and max_value" - ); - } - let start = if let Some(prev) = previous_allocation { - if prev < schema.min_value || prev > schema.max_value { - // Previous versions set allocated to 0 as a default, - // so we have this special case. - if prev == 0 { - schema.start - } else { - panic!( - "Invalid sequence: previous allocation value {prev} is out of bounds for sequence with min_value {} and max_value {}", - schema.min_value, schema.max_value - ); - } - } else { - prev - } - } else { - schema.start - }; - // We will always need to allocate before generating any values. - Self { - value: start, - allocated: start, - schema, - } - } - - /// Update the current value of the sequence. - /// This is used on very specific occasions, - /// such as cloning a sequence - pub(super) fn update_value(&mut self, new_value: i128) { - if new_value < self.schema.min_value || new_value > self.schema.max_value { - panic!( - "Invalid sequence update: new value {} is out of bounds for sequence with min_value {} and max_value {}", - new_value, self.schema.min_value, self.schema.max_value - ); - } - self.value = new_value; - } - - /// Update the persisted allocation cursor for the sequence. - pub(super) fn update_allocation(&mut self, new_allocated: i128) { - if !(self.schema.min_value..=self.schema.max_value).contains(&new_allocated) { - panic!( - "Invalid sequence allocation update: new allocated {} is out of bounds for sequence with min_value {} and max_value {}", - new_allocated, self.schema.min_value, self.schema.max_value - ); - } - self.allocated = new_allocated; - } - - pub(super) fn get_value(&self) -> i128 { - self.value - } - - pub(super) fn id(&self) -> SequenceId { - self.schema.sequence_id - } - - /// Returns the next value in the sequence given the params. - /// - /// Examples: - /// (min: 1, max: 10, increment: 1, value: 9) -> 1 - /// (min: 1, max: 10, increment: 20, value: 5) -> 5 - /// (min: 1, max: 10, increment: 3, value: 5) -> 8 - /// (min: 1, max: 10, increment: 3, value: 9) -> 2 - /// (min: 1, max: 10, increment: -3, value: 4) -> 1 - /// (min: 1, max: 10, increment: -3, value: 1) -> 8 - fn next_in_sequence(min: i128, max: i128, increment: i128, value: i128) -> i128 { - // calculate the next value - let mut next = value + increment; - // handle wrapping around the sequence - if increment > 0 { - if next > max { - next = min + (next - max - 1) % (max - min + 1); - } - } else if next < min { - next = max - (min - next - 1) % (max - min + 1); - } - next - } - - /// Returns the next value iff no allocation is needed. - pub(super) fn gen_next_value(&mut self) -> Option { - if self.needs_allocation() { - return None; - } - let value = self.value; - self.value = self.next_value(); - Some(value) - } - - fn next_value(&self) -> i128 { - self.nth_value(1) - } - - fn nth_value(&self, n: usize) -> i128 { - let mut value = self.value; - for _ in 0..n { - value = Self::next_in_sequence( - self.schema.min_value, - self.schema.max_value, - self.schema.increment, - value, - ); - } - value - } - - /// The allocated value represents the place where the sequence would - /// start from if the system memory was lost. Therefore we cannot generate - /// the next value in the sequence without the risk of using the same - /// value twice in two separate transactions. - /// e.g. - /// 1. incr = 1, allocated = 10, value = 10 - /// 2. next_value() -> 11 - /// 3. commit transaction - /// 4. crash - /// 5. restart - /// 6. incr = 1 allocated = 10, value = 10 - /// 7. next_value() -> 11 - fn needs_allocation(&self) -> bool { - // On restart we are allowed to begin at the allocation amount, so we stop before we - // reach it. It is important that the allocated value is one that would be returned - // by the sequence, so we can use equality here. Otherwise, to handle wrapping sequences - // correctly, we would need to check if the next value is on the same side of the allocated - // value as the current value, which seems more complex. - self.value == self.allocated - } - - /// Allocate up to `steps` new values in the sequence. This returns the new allocated value, - /// which should be written to the corresponding system table row, so that we start generating - /// at that value on the next restart. - /// This may allocate fewer steps if it is possible to fully loop around the sequence in that - /// many steps. - pub(super) fn allocate_steps(&mut self, steps: usize) -> i128 { - if !self.needs_allocation() { - // No allocation needed, return the current allocation. - return self.allocated; - } - let original_allocation = self.allocated; - for _ in 0..steps { - let next = Self::next_in_sequence( - self.schema.min_value, - self.schema.max_value, - self.schema.increment, - self.allocated, - ); - if next == original_allocation { - // We have looped all the way around, stop here. - break; - } - self.allocated = next; - } - if self.needs_allocation() { - // This should only be possible if |max - min| == |increment|. - // This should be unreachable, since `new` will panic if this would happen. - panic!("Unable to allocate new sequence value. Sequence parameters are invalid.") - } - self.allocated - } -} - -/// A map of [`SequenceId`] -> [`Sequence`]. -#[derive(Default, Debug)] -pub(super) struct SequencesState { - sequences: IntMap, -} - -impl MemoryUsage for SequencesState { - fn heap_usage(&self) -> usize { - let Self { sequences } = self; - sequences.heap_usage() - } -} - -impl SequencesState { - pub(super) fn get_sequence_mut(&mut self, seq_id: SequenceId) -> Option<&mut Sequence> { - self.sequences.get_mut(&seq_id) - } - - pub(super) fn insert(&mut self, seq: Sequence) { - self.sequences.insert(seq.id(), seq); - } - - pub(super) fn remove(&mut self, seq_id: SequenceId) -> Option { - self.sequences.remove(&seq_id) - } -} - -#[cfg(test)] -mod tests { - - use crate::locking_tx_datastore::sequence::Sequence; - use spacetimedb_primitives::{ColId, SequenceId, TableId}; - use spacetimedb_schema::schema::SequenceSchema; - - #[derive(Clone, Copy)] - struct SequenceParams { - min: i128, - max: i128, - increment: i128, - start: i128, - previous_allocation: Option, - } - fn make_test_sequence_schema(params: SequenceParams) -> Sequence { - let schema = SequenceSchema { - sequence_id: SequenceId(1), - min_value: params.min, - max_value: params.max, - increment: params.increment, - start: params.start, - col_pos: ColId(1), - table_id: TableId(1), - sequence_name: "test_sequence".into(), - }; - Sequence::new(schema, params.previous_allocation) - } - #[test] - fn test_double_allocation_noops() { - // A simple sequence that increments by 1 from 1 to 10. - let seq_params = SequenceParams { - min: 1, - max: 10, - increment: 1, - start: 1, - previous_allocation: None, - }; - let mut seq = make_test_sequence_schema(seq_params); - assert_eq!(seq.gen_next_value(), None); - let new_alloc = seq.allocate_steps(1); - assert_eq!(new_alloc, 2); - // Check that trying to allocate again will do nothing if we haven't exhausted - // the existing allocation. - let new_alloc = seq.allocate_steps(2); - assert_eq!(new_alloc, 2); - assert_eq!(seq.gen_next_value(), Some(1)); - assert_eq!(seq.gen_next_value(), None); - } - - #[test] - fn test_simple_loop() { - // A simple sequence that increments by 1 from 1 to 10. - let seq_params = SequenceParams { - min: 1, - max: 10, - increment: 1, - start: 1, - previous_allocation: None, - }; - let mut seq = make_test_sequence_schema(seq_params); - assert_sequence_works(&mut seq, seq_params, seq_params.start, 100); - } - - #[test] - fn test_loop_with_odd_increment() { - // A simple sequence that increments by 1 from 1 to 10. - let seq_params = SequenceParams { - min: 1, - max: 100, - increment: 3, - start: 1, - previous_allocation: None, - }; - let mut seq = make_test_sequence_schema(seq_params); - assert_sequence_works(&mut seq, seq_params, seq_params.start, 100); - } - - #[test] - fn test_loop_with_odd_increment_and_even_start() { - // A simple sequence that increments by 1 from 1 to 10. - let seq_params = SequenceParams { - min: 1, - max: 100, - increment: 3, - start: 10, - previous_allocation: None, - }; - let mut seq = make_test_sequence_schema(seq_params); - assert_sequence_works(&mut seq, seq_params, seq_params.start, 100); - } - - #[test] - fn test_loop_with_fully_negative_range() { - // A simple sequence that increments by 1 from 1 to 10. - let seq_params = SequenceParams { - min: -100, - max: -1, - increment: 3, - start: -50, - previous_allocation: None, - }; - let mut seq = make_test_sequence_schema(seq_params); - assert_sequence_works(&mut seq, seq_params, seq_params.start, 100); - } - - #[test] - fn test_simple_negative_loop() { - // A simple sequence that increments by 1 from 1 to 10. - let seq_params = SequenceParams { - min: 1, - max: 10, - increment: -1, - start: 1, - previous_allocation: None, - }; - let mut seq = make_test_sequence_schema(seq_params); - assert_sequence_works(&mut seq, seq_params, seq_params.start, 100); - } - - // This function tests that a sequence works correctly by generating `steps` values. - // This uses a different way of calculating the next value is that will be more likely to hit - // overflow issues, so it won't work if steps * increment overflows i128. - fn assert_sequence_works(seq: &mut Sequence, seq_params: SequenceParams, initial_value: i128, steps: i128) { - for i in 0..steps { - if seq.needs_allocation() { - seq.allocate_steps(10); - } - let val = seq.gen_next_value().unwrap(); - assert!( - val >= seq_params.min && val <= seq_params.max, - "Generated value {val} out of bounds [{}, {}]", - seq_params.min, - seq_params.max - ); - - let range = seq_params.max - seq_params.min + 1; - let raw_next = initial_value + i * seq_params.increment; - // This is an alternate way to handling wrapping. Since the mod operator can return - // negative values in rust, we do the `(n % max + max) % max` trick. - let wrapped_next = ((raw_next - seq_params.min) % range + range) % range + seq_params.min; - assert_eq!(val, wrapped_next, "Failed at iteration {i} (0 indexed)"); - } - } - - #[test] - fn test_restarting_after_allocation() { - // A simple sequence that increments by 1 from 1 to 100. - let seq_params = SequenceParams { - min: 1, - max: 100, - increment: 1, - start: 1, - previous_allocation: None, - }; - let mut seq = make_test_sequence_schema(seq_params); - assert!(seq.needs_allocation()); - // We are picking a number lower than the max to avoid wrapping. - let new_allocation = seq.allocate_steps(40); - let mut previous_value = 0; - // Keep going until we exhaust the allocation. - while !seq.needs_allocation() { - previous_value = seq.gen_next_value().unwrap(); - // Since this won't wrap, we should get values strictly less than the allocation. - assert!(previous_value <= new_allocation); - } - assert_eq!(previous_value, new_allocation - 1); - let restarted_params = SequenceParams { - previous_allocation: Some(new_allocation), - ..seq_params - }; - let mut restarted_seq = make_test_sequence_schema(restarted_params); - assert!(restarted_seq.needs_allocation()); - restarted_seq.allocate_steps(1); - let next_value = restarted_seq.gen_next_value().unwrap(); - assert_eq!(next_value, new_allocation); - } - - #[test] - fn test_first_value_is_prev_allocation() { - // A simple sequence that increments by 1 from 1 to 100. - // The start is set to 1, but this is overridden by the previous allocation of 7. - let seq_params = SequenceParams { - min: 1, - max: 100, - increment: 1, - start: 1, - previous_allocation: Some(7), - }; - let mut seq = make_test_sequence_schema(seq_params); - assert!(seq.needs_allocation()); - // We are picking a number lower than the max to avoid wrapping. - let _ = seq.allocate_steps(1); - assert_eq!(7, seq.gen_next_value().unwrap()); - } - #[test] - #[should_panic(expected = "Invalid sequence:")] - fn test_increment_range() { - // This is a sequence that would only ever be able to generate one value. - let seq_params = SequenceParams { - min: 1, - max: 10, - increment: 10, - start: 1, - previous_allocation: None, - }; - make_test_sequence_schema(seq_params); - } - - #[test] - #[should_panic(expected = "Invalid sequence:")] - fn test_previous_out_of_range() { - // This is a sequence that would only ever be able to generate one value. - let seq_params = SequenceParams { - min: 1, - max: 10, - increment: 1, - start: 1, - previous_allocation: Some(100), - }; - make_test_sequence_schema(seq_params); - } - - #[test] - fn test_previous_out_of_range_but_zero() { - // This is a sequence that would only ever be able to generate one value. - let seq_params = SequenceParams { - min: 1, - max: 10, - increment: 1, - start: 1, - previous_allocation: Some(0), - }; - let mut seq = make_test_sequence_schema(seq_params); - assert!(seq.needs_allocation()); - seq.allocate_steps(1); - assert_eq!(1, seq.gen_next_value().unwrap()); - } - - #[test] - #[should_panic(expected = "Invalid sequence:")] - fn test_start_out_of_range() { - // This is a sequence that would only ever be able to generate one value. - let seq_params = SequenceParams { - min: 1, - max: 10, - increment: 1, - start: 100, - previous_allocation: None, - }; - make_test_sequence_schema(seq_params); - } -} diff --git a/crates/datastore/src/locking_tx_datastore/tx_state.rs b/crates/datastore/src/locking_tx_datastore/tx_state.rs index 1d7680ae05a..acfee9cdc4b 100644 --- a/crates/datastore/src/locking_tx_datastore/tx_state.rs +++ b/crates/datastore/src/locking_tx_datastore/tx_state.rs @@ -1,4 +1,4 @@ -use super::{delete_table::DeleteTable, sequence::Sequence}; +use super::delete_table::DeleteTable; use spacetimedb_data_structures::map::IntMap; use spacetimedb_lib::db::auth::StAccess; use spacetimedb_primitives::{ColId, ColList, ConstraintId, IndexId, SequenceId, TableId}; @@ -149,8 +149,8 @@ pub enum PendingSchemaChange { /// The constraint with [`ConstraintId`] was added to the table with [`TableId`]. /// If indices were made unique, their [`IndexId`]s and the taken [`PointerMap`] are stored. ConstraintAdded(TableId, ConstraintId, Vec, Option), - /// The [`Sequence`] with [`SequenceSchema`] was added to the table with [`TableId`]. - SequenceRemoved(TableId, Sequence, SequenceSchema), + /// The sequence with [`SequenceSchema`] was added to the table with [`TableId`]. + SequenceRemoved(TableId, SequenceSchema), /// The sequence with [`SequenceId`] was added to the table with [`TableId`]. SequenceAdded(TableId, SequenceId), } @@ -178,9 +178,7 @@ impl MemoryUsage for PendingSchemaChange { Self::ConstraintAdded(table_id, constraint_id, index_ids, pointer_map) => { table_id.heap_usage() + constraint_id.heap_usage() + index_ids.heap_usage() + pointer_map.heap_usage() } - Self::SequenceRemoved(table_id, sequence, sequence_schema) => { - table_id.heap_usage() + sequence.heap_usage() + sequence_schema.heap_usage() - } + Self::SequenceRemoved(table_id, sequence_schema) => table_id.heap_usage() + sequence_schema.heap_usage(), Self::SequenceAdded(table_id, sequence_id) => table_id.heap_usage() + sequence_id.heap_usage(), Self::TableAlterAccessorName(table_id, alias) => { table_id.heap_usage() + alias.as_ref().map(|a| a.as_ref().len()).unwrap_or(0) diff --git a/crates/datastore/src/system_tables.rs b/crates/datastore/src/system_tables.rs index 2635093cc2d..351e9fdec92 100644 --- a/crates/datastore/src/system_tables.rs +++ b/crates/datastore/src/system_tables.rs @@ -1335,9 +1335,6 @@ impl From for SequenceSchema { table_id: sequence.table_id, col_pos: sequence.col_pos, start: sequence.start, - increment: sequence.increment, - min_value: sequence.min_value, - max_value: sequence.max_value, } } } diff --git a/crates/engine/src/relational_db.rs b/crates/engine/src/relational_db.rs index 19cda5b6fbe..88381e90cad 100644 --- a/crates/engine/src/relational_db.rs +++ b/crates/engine/src/relational_db.rs @@ -3166,16 +3166,17 @@ mod tests { let mut tx = begin_mut_tx(&stdb); insert(&stdb, &mut tx, table_id, &product![0i64]).unwrap(); - // Check the second row start after `SEQUENCE_PREALLOCATION_AMOUNT` - assert_eq!(collect_from_sorted(&stdb, &tx, table_id, 0i64)?, vec![1, 4097]); + // Check the second row is directly after the first, as we perform no preallocation. + // The property we actually care about is that the next value is greater than the previous. + assert_eq!(collect_from_sorted(&stdb, &tx, table_id, 0i64)?, vec![1, 2]); stdb.commit_tx(tx)?; let stdb = stdb.reopen()?; let mut tx = begin_mut_tx(&stdb); insert(&stdb, &mut tx, table_id, &product![0i64]).unwrap(); - // The next value will have a gap because we preallocate, but asserting the specific number - // seems brittle. - assert!(collect_from_sorted(&stdb, &tx, table_id, 0i64)?[2] > 4097); + // The next value will not have a gap because we do not preallocate. + // The property we actually care about is that it is larger than the previous two values. + assert_eq!(collect_from_sorted(&stdb, &tx, table_id, 0i64)?[2], 3); Ok(()) } diff --git a/crates/engine/src/update.rs b/crates/engine/src/update.rs index 5b9e6e0fde2..a125be0196c 100644 --- a/crates/engine/src/update.rs +++ b/crates/engine/src/update.rs @@ -233,18 +233,14 @@ fn auto_migrate_database( .ty .clone(); - // Convert `SequenceDef` min/max to `AlgebraicValue`s of the correct type. + // Convert `SequenceSchema` min/max to `AlgebraicValue`s of the correct type. let min = ty - .saturating_value_from_i128(sequence_def.min_value.unwrap_or(1)) + .saturating_value_from_i128(SequenceSchema::MIN_VALUE) .ok_or_else(|| { anyhow::anyhow!("Precheck failed: added sequence {sequence_name} has invalid min value") })?; - let max = match sequence_def.max_value { - Some(max) => ty.saturating_value_from_i128(max), - None => ty.saturating_value_from_i128(i128::MAX), - } - .ok_or_else(|| { + let max = ty.saturating_value_from_i128(i128::MAX).ok_or_else(|| { anyhow::anyhow!("Precheck failed: added sequence {sequence_name} has invalid max value") })?; @@ -1640,11 +1636,13 @@ mod test { let stdb = stdb.reopen()?; // After replay, the allocation cursor should be preserved. + // We only care that the next value is strictly higher than all the previous ones, + // but we happen to get the value 4 here because we do not perform any sequence allocation batching. { let ids = insert_and_collect_ids(&stdb, product![0i64, 99u64].into())?; assert!( - ids.iter().last().unwrap() == &4097, - "expected id 4097 after reopen, got {ids:?}" + ids.iter().last().unwrap() == &4, + "expected id 4 after reopen, got {ids:?}" ); } diff --git a/crates/lib/src/db/raw_def/v10.rs b/crates/lib/src/db/raw_def/v10.rs index b21d8470b84..b2664f50513 100644 --- a/crates/lib/src/db/raw_def/v10.rs +++ b/crates/lib/src/db/raw_def/v10.rs @@ -417,20 +417,16 @@ pub struct RawSequenceDefV10 { /// This must be the unique `RawSequenceDef` for this column. pub column: ColId, - /// The value to start assigning to this column. - /// Will be incremented by 1 for each new row. - /// If not present, an arbitrary start point may be selected. + /// Deprecated; should be `None`. pub start: Option, - /// The minimum allowed value in this column. - /// If not present, no minimum. + /// Deprecated; should be `None`. pub min_value: Option, - /// The maximum allowed value in this column. - /// If not present, no maximum. + /// Deprecated; should be `None`. pub max_value: Option, - /// The increment used when updating the SequenceDef. + /// Deprecated; should be `1i128`. pub increment: i128, } diff --git a/crates/lib/src/db/raw_def/v8.rs b/crates/lib/src/db/raw_def/v8.rs index d0266ba5ea7..7166de8449c 100644 --- a/crates/lib/src/db/raw_def/v8.rs +++ b/crates/lib/src/db/raw_def/v8.rs @@ -14,12 +14,6 @@ use spacetimedb_sats::raw_identifier::RawIdentifier; pub use crate::ModuleDefBuilder as RawModuleDefV8Builder; pub use crate::RawModuleDefV8; -/// The amount sequences allocate each time they over-run their allocation. -/// -/// Note that we do not perform an initial allocation during `create_sequence` or at startup. -/// Newly-created sequences will allocate the first time they are advanced. -pub const SEQUENCE_ALLOCATION_STEP: i128 = 4096; - /// Represents a sequence definition for a database table column. #[derive(Debug, Clone, Eq, PartialEq, PartialOrd, Ord, SpacetimeType)] #[sats(crate = crate)] @@ -28,13 +22,13 @@ pub struct RawSequenceDefV8 { pub sequence_name: RawIdentifier, /// The position of the column associated with this sequence. pub col_pos: ColId, - /// The increment value for the sequence. + /// Deprecated; should be `1i128`. pub increment: i128, - /// The starting value for the sequence. + /// Deprecated; should be `None`. pub start: Option, - /// The minimum value for the sequence. + /// Deprecated; should be `None`. pub min_value: Option, - /// The maximum value for the sequence. + /// Deprecated; should be `None`. pub max_value: Option, /// The number of values to preallocate for the sequence. /// Deprecated, in the future this concept will no longer exist. @@ -69,8 +63,9 @@ impl RawSequenceDefV8 { start: None, min_value: None, max_value: None, - // Start with no values allocated. The first time we advance the sequence, - // we will allocate [`SEQUENCE_ALLOCATION_STEP`] values. + // Start with no values allocated. + // The first time we advance the sequence, + // we will allocate a block of multiple values. allocated: 0, } } diff --git a/crates/lib/src/db/raw_def/v9.rs b/crates/lib/src/db/raw_def/v9.rs index ba5361c6cdb..e0a92e3e6a2 100644 --- a/crates/lib/src/db/raw_def/v9.rs +++ b/crates/lib/src/db/raw_def/v9.rs @@ -258,20 +258,16 @@ pub struct RawSequenceDefV9 { /// This must be the unique `RawSequenceDef` for this column. pub column: ColId, - /// The value to start assigning to this column. - /// Will be incremented by 1 for each new row. - /// If not present, an arbitrary start point may be selected. + /// Deprecated; should be `None`. pub start: Option, - /// The minimum allowed value in this column. - /// If not present, no minimum. + /// Deprecated; should be `None`. pub min_value: Option, - /// The maximum allowed value in this column. - /// If not present, no maximum. + /// Deprecated; should be `None`. pub max_value: Option, - /// The increment used when updating the SequenceDef. + /// Deprecated; should be `1i128`. pub increment: i128, } diff --git a/crates/schema/src/def.rs b/crates/schema/src/def.rs index 521bcb35aa9..877d182435a 100644 --- a/crates/schema/src/def.rs +++ b/crates/schema/src/def.rs @@ -1428,6 +1428,15 @@ impl From for TableDef { } /// A sequence definition for a database table column. +/// +/// Previous versions of this definition exposed options `start`, `min_value`, `max_value` and `increment`. +/// SpacetimeDB never exercised these options in any useful way, +/// and supporting them caused considerable implementation burden, +/// so we chose to remove them. +/// All sequences start at some arbitrary nonnegative value near zero, +/// have the range of the non-negative `i128`s, +/// and increment by 1. +/// Raw defs still have these values, but we reject any def that uses values other than the defaults. #[derive(Debug, Clone, Eq, PartialEq)] pub struct SequenceDef { /// The name of the sequence. Must be unique within the containing `ModuleDef`. @@ -1444,22 +1453,11 @@ pub struct SequenceDef { /// The column must have integral type. /// This must be the unique `RawSequenceDef` for this column. pub column: ColId, +} - /// The value to start assigning to this column. - /// Will be incremented by 1 for each new row. - /// If not present, an arbitrary start point may be selected. - pub start: Option, - - /// The minimum allowed value in this column. - /// If not present, no minimum. - pub min_value: Option, - - /// The maximum allowed value in this column. - /// If not present, no maximum. - pub max_value: Option, - - /// The increment to use when updating the sequence. - pub increment: i128, +impl SequenceDef { + /// All sequences increment by 1. + pub const INCREMENT: i128 = 1; } impl From for RawSequenceDefV9 { @@ -1467,10 +1465,10 @@ impl From for RawSequenceDefV9 { RawSequenceDefV9 { name: Some(val.name), column: val.column, - start: val.start, - min_value: val.min_value, - max_value: val.max_value, - increment: val.increment, + start: None, + min_value: None, + max_value: None, + increment: SequenceDef::INCREMENT, } } } @@ -1480,10 +1478,10 @@ impl From for RawSequenceDefV10 { RawSequenceDefV10 { source_name: Some(val.name), column: val.column, - start: val.start, - min_value: val.min_value, - max_value: val.max_value, - increment: val.increment, + start: None, + min_value: None, + max_value: None, + increment: SequenceDef::INCREMENT, } } } diff --git a/crates/schema/src/def/validate/v9.rs b/crates/schema/src/def/validate/v9.rs index 8de4927afa2..743213077f2 100644 --- a/crates/schema/src/def/validate/v9.rs +++ b/crates/schema/src/def/validate/v9.rs @@ -1154,41 +1154,53 @@ impl<'a, 'b> TableValidator<'a, 'b> { } }); - /// Compare two `Option` values, returning `true` if `lo <= hi`, - /// or if either is `None`. - pub(crate) fn le(lo: Option, hi: Option) -> bool { - match (lo, hi) { - (Some(lo), Some(hi)) => lo <= hi, - _ => true, - } - } - let valid = le(min_value, start) && le(start, max_value) && le(min_value, max_value); - - let min_start_max = if valid { - Ok((min_value, start, max_value)) + let increment_is_one: Result<()> = if increment == 1 { + Ok(()) } else { - Err(ValidationError::InvalidSequenceRange { + Err(ValidationError::InvalidSequenceDefOption { sequence: name.clone(), - min_value, - start, - max_value, + option: "increment", + supplied_value: increment, + expected_value: "1", } .into()) }; + fn validate_option_typed_option_is_none( + name: &RawIdentifier, + option_name: &'static str, + supplied_value: Option, + ) -> Result<()> { + if let Some(supplied_value) = supplied_value { + Err(ValidationError::InvalidSequenceDefOption { + sequence: name.clone(), + option: option_name, + supplied_value, + expected_value: "None", + } + .into()) + } else { + Ok(()) + } + } + + let no_supplied_removed_options = ( + validate_option_typed_option_is_none(&name, "start", start), + validate_option_typed_option_is_none(&name, "min_value", min_value), + validate_option_typed_option_is_none(&name, "max_value", max_value), + increment_is_one, + ) + .combine_errors(); + let name = self.add_to_global_namespace(name); - let (name, column, (min_value, start, max_value)) = (name, column, min_start_max).combine_errors()?; + let (name, column, ((), (), (), ())) = (name, column, no_supplied_removed_options).combine_errors()?; Ok(SequenceDef { // Set by `ModuleDef::apply_namespace` once the module tree is assembled. namespace: NamespacePath::root(), name, column, - min_value, - start, - max_value, - increment, }) } diff --git a/crates/schema/src/error.rs b/crates/schema/src/error.rs index e9408a35482..39ceee1ea29 100644 --- a/crates/schema/src/error.rs +++ b/crates/schema/src/error.rs @@ -74,12 +74,12 @@ pub enum ValidationError { column: RawColumnName, column_type: PrettyAlgebraicType, }, - #[error("invalid sequence range information: expected {min_value:?} <= {start:?} <= {max_value:?} in sequence `{sequence}`")] - InvalidSequenceRange { + #[error("sequence definition `{sequence}` specifies unsupported option `{option}` with value {supplied_value}, should be {expected_value}")] + InvalidSequenceDefOption { sequence: RawIdentifier, - min_value: Option, - start: Option, - max_value: Option, + option: &'static str, + supplied_value: i128, + expected_value: &'static str, }, #[error("View {view} has invalid return type {ty}")] InvalidViewReturnType { diff --git a/crates/schema/src/schema.rs b/crates/schema/src/schema.rs index db854fab15e..ab900552190 100644 --- a/crates/schema/src/schema.rs +++ b/crates/schema/src/schema.rs @@ -1245,6 +1245,15 @@ impl From> for ProductTypeElement { } /// Represents a schema definition for a database sequence. +/// +/// Previous versions of this definition exposed options `start`, `min_value`, `max_value` and `increment`. +/// SpacetimeDB never exercised these options in any useful way, +/// and supporting them caused considerable implementation burden, +/// so we chose to remove them. +/// All sequences start at some arbitrary nonnegative value near zero, +/// have the range of the non-negative `i128`s, +/// and increment by 1. +/// Raw defs still have these values, but we reject any def that uses values other than the defaults. #[derive(Debug, Clone, PartialEq, Eq)] pub struct SequenceSchema { /// The unique identifier for the sequence within a database. @@ -1256,14 +1265,12 @@ pub struct SequenceSchema { pub table_id: TableId, /// The position of the column associated with this sequence. pub col_pos: ColId, - /// The increment value for the sequence. - pub increment: i128, - /// The initial value to be returned by this sequence. + /// The starting point for this schema, i.e. the first value from it. + /// + /// For user-defined sequences, this will be [`Self::START`]. + /// For system-defiend sequences, it will be a higher value, + /// as we reserve a range of values in system-defined sequences for IDs of future system-defined rows. pub start: i128, - /// The minimum value for the sequence. - pub min_value: i128, - /// The maximum value for the sequence. - pub max_value: i128, } impl spacetimedb_memory_usage::MemoryUsage for SequenceSchema { @@ -1273,22 +1280,35 @@ impl spacetimedb_memory_usage::MemoryUsage for SequenceSchema { sequence_name, table_id, col_pos, - increment, start, - min_value, - max_value, } = self; sequence_id.heap_usage() + sequence_name.heap_usage() + table_id.heap_usage() + col_pos.heap_usage() - + increment.heap_usage() + start.heap_usage() - + min_value.heap_usage() - + max_value.heap_usage() } } +impl SequenceSchema { + /// Value to fill into the `increment` field of `StSequenceRow`. + /// + /// All sequences increment by 1. + pub const INCREMENT: i128 = SequenceDef::INCREMENT; + /// Value to fill into the `start` field of `StSequenceRow`. + /// + /// User-defined sequences start at 1. + const START: i128 = 1; + /// Value to fill into the `min_value` field of `StSequenceRow`. + /// + /// All sequences have a minimum value the same as their start, which is 1. + pub const MIN_VALUE: i128 = Self::START; + /// Value to fill into the `max_value` field of `StSequenceRow`. + /// + /// All sequences have a max value of the largest representable value. + pub const MAX_VALUE: i128 = i128::MAX; +} + impl Schema for SequenceSchema { type Def = SequenceDef; type Id = SequenceId; @@ -1302,11 +1322,7 @@ impl Schema for SequenceSchema { sequence_name: def.name.clone().into(), table_id: parent_id, col_pos: def.column, - increment: def.increment, - start: def.start.unwrap_or(1), - min_value: def.min_value.unwrap_or(1), - max_value: def.max_value.unwrap_or(i128::MAX), - // allocated: 0, // TODO: information not available in the `Def`s anymore, which is correct, but this may need to be overridden later. + start: Self::START, } } @@ -1318,16 +1334,6 @@ impl Schema for SequenceSchema { "Sequence name mismatch" ); ensure_eq!(self.col_pos, def.column, "Sequence column mismatch"); - ensure_eq!(self.increment, def.increment, "Sequence increment mismatch"); - if let Some(start) = &def.start { - ensure_eq!(self.start, *start, "Sequence start mismatch"); - } - if let Some(min_value) = &def.min_value { - ensure_eq!(self.min_value, *min_value, "Sequence min_value mismatch"); - } - if let Some(max_value) = &def.max_value { - ensure_eq!(self.max_value, *max_value, "Sequence max_value mismatch"); - } Ok(()) } } diff --git a/crates/smoketests/modules/autoinc-unique/src/lib.rs b/crates/smoketests/modules/autoinc-unique/src/lib.rs index a20e6ad2a0c..13d9ed07c16 100644 --- a/crates/smoketests/modules/autoinc-unique/src/lib.rs +++ b/crates/smoketests/modules/autoinc-unique/src/lib.rs @@ -22,6 +22,18 @@ macro_rules! autoinc_unique { Ok(()) } + #[spacetimedb::reducer] + pub fn [](ctx: &ReducerContext) -> Result<(), Box> { + ctx.db.[]().try_insert([] { + key_col: 0, + name: "rolled_back".into(), + })?; + Err(Box::new(std::io::Error::new( + std::io::ErrorKind::Other, + "intentional failure after auto_inc insert", + ))) + } + #[spacetimedb::reducer] pub fn [](ctx: &ReducerContext, name: String, new_id: $ty) { ctx.db.[]().name().delete(&name); diff --git a/crates/smoketests/tests/cluster/auto_inc.rs b/crates/smoketests/tests/cluster/auto_inc.rs index 96d25385d18..f35d66fc14b 100644 --- a/crates/smoketests/tests/cluster/auto_inc.rs +++ b/crates/smoketests/tests/cluster/auto_inc.rs @@ -120,3 +120,47 @@ fn test_autoinc_unique() { ); } } + +/// A rolled-back auto-inc insert must leave durable sequence metadata consistent with later +/// committed auto-inc values. +/// +/// This is a regression test for a bug which we fixed in [PR 5880](https://github.com/clockworklabs/SpacetimeDB/pull/5880). +/// Prior to that PR, sequences kept a non-transactional and non-persistent in-memory side table +/// as an optimization rather than updating `st_sequence` rows on each sequence read. +/// A bug in the implementation of that optimization caused the in-memory state to remain updated +/// even when the persistent `st_sequence` change rolled back. +#[test] +fn autoinc_sequence_allocation_remains_consistent_after_rollback() { + let test = Smoketest::builder().precompiled_module("autoinc-unique").build(); + + assert!( + test.call("add_and_fail_u_64", &[]).is_err(), + "Reducer that intentionally fails after an auto-inc insert should fail" + ); + test.call("add_new_u_64", &[r#""committed""#]).unwrap(); + + let inserted_value = parse_single_u64( + &test + .sql("SELECT key_col FROM person_u_64 WHERE name = 'committed'") + .unwrap(), + ); + let persisted_allocation = parse_single_u64( + &test + .sql("SELECT allocated FROM st_sequence WHERE sequence_name = 'person_u_64_key_col_seq'") + .unwrap(), + ); + + assert!( + inserted_value <= persisted_allocation, + "st_sequence allocation is inconsistent after a rolled-back auto-inc insert: \ + committed value {inserted_value}, persisted allocation {persisted_allocation}" + ); +} + +fn parse_single_u64(output: &str) -> u64 { + output + .lines() + .map(str::trim) + .find_map(|line| line.parse().ok()) + .unwrap_or_else(|| panic!("SQL query should return one unsigned integer, got:\n{output}")) +}