From 69fcebff12956265fbffc573dc01ab14d3fd470e Mon Sep 17 00:00:00 2001 From: MasterPtato Date: Thu, 6 Aug 2026 16:25:42 -0700 Subject: [PATCH] [slopfix] fix(universaldb): match fdb conflict semantics in rocksdb + postgres drivers --- Cargo.lock | 2 +- .../universaldb/src/conflict_tracker.rs | 62 ++- .../universaldb/src/driver/postgres/commit.rs | 19 +- .../src/driver/rocksdb/transaction_task.rs | 175 +++----- engine/packages/universaldb/src/tx_ops.rs | 16 + .../universaldb/tests/conflict_parity.rs | 391 ++++++++++++++++++ 6 files changed, 520 insertions(+), 145 deletions(-) create mode 100644 engine/packages/universaldb/tests/conflict_parity.rs diff --git a/Cargo.lock b/Cargo.lock index 868daf76fb..60e1e08b32 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5949,7 +5949,7 @@ dependencies = [ [[package]] name = "rivet-universaldb-commit" -version = "2.3.2" +version = "2.3.7" dependencies = [ "anyhow", "rivet-vbare-compiler", diff --git a/engine/packages/universaldb/src/conflict_tracker.rs b/engine/packages/universaldb/src/conflict_tracker.rs index a257f14ecc..4d200c7755 100644 --- a/engine/packages/universaldb/src/conflict_tracker.rs +++ b/engine/packages/universaldb/src/conflict_tracker.rs @@ -20,12 +20,16 @@ const TXN_CONFLICT_TTL: Duration = Duration::from_secs(10); struct PreviousTransaction { insert_instant: Instant, start_version: u64, - conflict_ranges: Vec<(Vec, Vec, ConflictRangeType)>, + /// Only the write ranges are retained. Conflicts are directional: a transaction aborts when + /// something it read was written under it, so a retained transaction's reads can never abort + /// anyone. + write_ranges: Vec<(Vec, Vec)>, } /// In-process FoundationDB-style resolver. Holds the last `TXN_CONFLICT_TTL` of committed /// transactions and rejects a committing transaction if any retained transaction has both an -/// overlapping version window and an overlapping conflict range of a differing type. +/// overlapping version window and a write range overlapping one of the committing transaction's +/// read ranges. /// /// Used by the rocksdb driver (single process) and by the postgres leader-resolver. The two /// differ only in where the commit version comes from: rocksdb generates it from the in-process @@ -60,6 +64,10 @@ impl TransactionConflictTracker { /// Returns `true` on conflicts. The caller /// supplies `commit_version` (e.g. `nextval('udb_version_seq')` on the postgres leader, or /// `next_global_version()` on rocksdb) so version assignment stays the caller's responsibility. + /// + /// Conflicts are directional, matching FoundationDB: only this transaction's reads are checked, + /// and only against writes retained from transactions that committed inside its version window. + /// Blind write-vs-write does not conflict because neither transaction read what the other wrote. pub async fn check_and_insert( &self, txn1_start_version: u64, @@ -86,16 +94,21 @@ impl TransactionConflictTracker { // Check txn versions overlap (intersection or encapsulation) if txn2.start_version < txn1_commit_version { for (cr1_start, cr1_end, cr1_type) in &txn1_conflict_ranges { - for (cr2_start, cr2_end, cr2_type) in &txn2.conflict_ranges { + // Reads are never the aggressor, so this transaction's own writes are checked + // against nothing. + match cr1_type { + ConflictRangeType::Read => {} + ConflictRangeType::Write => continue, + } + + for (cr2_start, cr2_end) in &txn2.write_ranges { // Check conflict ranges overlap - if cr1_start < cr2_end && cr2_start < cr1_end && cr1_type != cr2_type { + if cr1_start < cr2_end && cr2_start < cr1_end { tracing::debug!( - cr1_start=%hex::encode(cr1_start), - cr1_end=%hex::encode(cr1_end), - ?cr1_type, - cr2_start=%hex::encode(cr2_start), - cr2_end=%hex::encode(cr2_end), - ?cr2_type, + read_start=%hex::encode(cr1_start), + read_end=%hex::encode(cr1_end), + write_start=%hex::encode(cr2_start), + write_end=%hex::encode(cr2_end), txn1_start_version, txn1_commit_version, txn2_start_version = txn2.start_version, @@ -109,15 +122,26 @@ impl TransactionConflictTracker { } } - // If no conflicts were detected, save txn data - txns.insert( - txn1_commit_version, - PreviousTransaction { - insert_instant: Instant::now(), - start_version: txn1_start_version, - conflict_ranges: txn1_conflict_ranges, - }, - ); + // Only writes can abort a later transaction, so a transaction that wrote nothing leaves no + // trace here. + let write_ranges = txn1_conflict_ranges + .into_iter() + .filter_map(|(begin, end, conflict_type)| match conflict_type { + ConflictRangeType::Write => Some((begin, end)), + ConflictRangeType::Read => None, + }) + .collect::>(); + + if !write_ranges.is_empty() { + txns.insert( + txn1_commit_version, + PreviousTransaction { + insert_instant: Instant::now(), + start_version: txn1_start_version, + write_ranges, + }, + ); + } false } diff --git a/engine/packages/universaldb/src/driver/postgres/commit.rs b/engine/packages/universaldb/src/driver/postgres/commit.rs index 21d2335913..8e4bbcf179 100644 --- a/engine/packages/universaldb/src/driver/postgres/commit.rs +++ b/engine/packages/universaldb/src/driver/postgres/commit.rs @@ -6,7 +6,11 @@ use std::{ use anyhow::{Context, Result}; use tokio::sync::oneshot; -use crate::{error::DatabaseError, options::ConflictRangeType, tx_ops::Operation}; +use crate::{ + error::DatabaseError, + options::ConflictRangeType, + tx_ops::{self, Operation}, +}; use super::{ codec, @@ -29,21 +33,16 @@ const RESEND_BACKOFF: Duration = Duration::from_millis(100); /// Submit a follower transaction's commit to the leader and await the result. /// -/// `read_version` is the watermark captured when this transaction opened its read snapshot. A pure -/// snapshot read-only transaction (no operations and no read conflict ranges) submits nothing. +/// `read_version` is the watermark captured when this transaction opened its read snapshot. A +/// read-only transaction submits nothing: its reads already came from one pinned snapshot, so there +/// is nothing to order or validate and nothing for a later transaction to conflict against. pub async fn submit( shared: &Arc, read_version: i64, operations: Vec, conflict_ranges: Vec<(Vec, Vec, ConflictRangeType)>, ) -> Result<()> { - // A transaction with no writes and no serializable read ranges has nothing to order or validate; - // it never needs the leader. - if operations.is_empty() - && conflict_ranges - .iter() - .all(|(_, _, kind)| matches!(kind, ConflictRangeType::Write)) - { + if tx_ops::is_read_only(&operations, &conflict_ranges) { return Ok(()); } diff --git a/engine/packages/universaldb/src/driver/rocksdb/transaction_task.rs b/engine/packages/universaldb/src/driver/rocksdb/transaction_task.rs index 662a1a1f39..5e4642c5ad 100644 --- a/engine/packages/universaldb/src/driver/rocksdb/transaction_task.rs +++ b/engine/packages/universaldb/src/driver/rocksdb/transaction_task.rs @@ -2,7 +2,8 @@ use std::sync::Arc; use anyhow::{Context, Result, bail}; use rocksdb::{ - OptimisticTransactionDB, ReadOptions, Transaction as RocksDbTransaction, WriteOptions, + OptimisticTransactionDB, ReadOptions, SnapshotWithThreadMode, + Transaction as RocksDbTransaction, WriteOptions, }; use tokio::sync::{mpsc, oneshot}; @@ -10,9 +11,8 @@ use crate::{ atomic::apply_atomic_op, conflict_tracker::TransactionConflictTracker, error::DatabaseError, - key_selector::KeySelector, options::{ConflictRangeType, MutationType}, - tx_ops::Operation, + tx_ops::{self, Operation}, value::{KeyValue, Slice, Values}, versionstamp::{generate_versionstamp, substitute_raw_versionstamp}, }; @@ -35,6 +35,9 @@ fn iter_bytes_to_vec(bytes: &[u8]) -> Vec { } } +/// The point-in-time view a single UDB transaction reads from. +type Snapshot<'a> = SnapshotWithThreadMode<'a, OptimisticTransactionDB>; + pub enum TransactionCommand { Get { key: Vec, @@ -92,10 +95,17 @@ impl TransactionTask { } pub async fn run(mut self) { + // One pinned snapshot per UDB transaction, taken at the first read. FDB pins a read version at + // the first read so every read in a transaction observes the same point in time; without this + // a commit landing mid-transaction would be partially visible. It is released on commit so a + // reused task takes a fresh snapshot for the next transaction. + let mut snapshot: Option> = None; + while let Some(command) = self.receiver.recv().await { match command { TransactionCommand::Get { key, response } => { - let result = self.handle_get(&key).await; + let snapshot = snapshot.get_or_insert_with(|| self.db.snapshot()); + let result = Self::handle_get(snapshot, &key); let _ = response.send(result); } TransactionCommand::GetKey { @@ -104,7 +114,8 @@ impl TransactionTask { offset, response, } => { - let result = self.handle_get_key(&key, or_equal, offset).await; + let snapshot = snapshot.get_or_insert_with(|| self.db.snapshot()); + let result = Self::handle_get_key(snapshot, &key, or_equal, offset); let _ = response.send(result); } TransactionCommand::GetRange { @@ -118,18 +129,18 @@ impl TransactionTask { reverse, response, } => { - let result = self - .handle_get_range( - begin, - begin_or_equal, - begin_offset, - end, - end_or_equal, - end_offset, - limit, - reverse, - ) - .await; + let snapshot = snapshot.get_or_insert_with(|| self.db.snapshot()); + let result = Self::handle_get_range( + snapshot, + begin, + begin_or_equal, + begin_offset, + end, + end_or_equal, + end_offset, + limit, + reverse, + ); let _ = response.send(result); } TransactionCommand::Commit { @@ -138,6 +149,10 @@ impl TransactionTask { conflict_ranges, response, } => { + // The commit reads the latest committed state, not this transaction's snapshot, so + // release the snapshot before applying. + snapshot = None; + let result = self .handle_commit(start_version, operations, conflict_ranges) .await; @@ -148,7 +163,7 @@ impl TransactionTask { end, response, } => { - let result = self.handle_get_estimated_range_size(&begin, &end).await; + let result = self.handle_get_estimated_range_size(&begin, &end); let _ = response.send(result); } } @@ -162,27 +177,19 @@ impl TransactionTask { self.db.transaction_opt(&write_opts, &txn_opts) } - async fn handle_get(&mut self, key: &[u8]) -> Result> { - let txn = self.create_transaction(); - - let read_opts = ReadOptions::default(); - - Ok(txn - .get_opt(key, &read_opts) + fn handle_get(snapshot: &Snapshot<'_>, key: &[u8]) -> Result> { + Ok(snapshot + .get(key) .context("failed to read key from rocksdb")? .map(|v| v.into())) } - async fn handle_get_key( - &mut self, + fn handle_get_key( + snapshot: &Snapshot<'_>, key: &[u8], or_equal: bool, offset: i32, ) -> Result> { - let txn = self.create_transaction(); - - let read_opts = ReadOptions::default(); - // Based on PostgreSQL's interpretation: // (false, 1) => first_greater_or_equal // (true, 1) => first_greater_than @@ -192,7 +199,7 @@ impl TransactionTask { match (or_equal, offset) { (false, 1) => { // first_greater_or_equal: find first key >= search_key - let mut iter = txn.raw_iterator_opt(read_opts); + let mut iter = snapshot.raw_iterator(); iter.seek(key); let result = iter.key().map(iter_bytes_to_vec); iter.status() @@ -201,7 +208,7 @@ impl TransactionTask { } (true, 1) => { // first_greater_than: find first key > search_key - let mut iter = txn.raw_iterator_opt(read_opts); + let mut iter = snapshot.raw_iterator(); iter.seek(key); while iter.valid() { let k = iter.key().expect("iterator should be valid"); @@ -219,7 +226,7 @@ impl TransactionTask { (false, 0) => { // last_less_than: find last key < search_key // Use reverse iterator starting just before the key - let mut iter = txn.raw_iterator_opt(read_opts); + let mut iter = snapshot.raw_iterator(); iter.seek_for_prev(key); while iter.valid() { let k = iter.key().expect("iterator should be valid"); @@ -236,7 +243,7 @@ impl TransactionTask { (true, 0) => { // last_less_or_equal: find last key <= search_key // Use reverse iterator starting from the key - let mut iter = txn.raw_iterator_opt(read_opts); + let mut iter = snapshot.raw_iterator(); iter.seek_for_prev(key); while iter.valid() { let k = iter.key().expect("iterator should be valid"); @@ -257,75 +264,19 @@ impl TransactionTask { } } - #[allow(dead_code)] - fn resolve_key_selector( - &self, - txn: &RocksDbTransaction, - selector: &KeySelector<'_>, - _read_opts: &ReadOptions, - ) -> Result> { - let key = selector.key(); - let offset = selector.offset(); - let or_equal = selector.or_equal(); - - if offset == 0 && or_equal { - // Simple case: exact key - return Ok(key.to_vec()); - } - - // Create an iterator to find the key - let mut iter = txn.raw_iterator_opt(ReadOptions::default()); - iter.seek(key); - - let mut keys: Vec> = Vec::new(); - - while iter.valid() { - let k = iter.key().expect("iterator should be valid"); - keys.push(iter_bytes_to_vec(k)); - if keys.len() > (offset.abs() + 1) as usize { - break; - } - iter.next(); - } - iter.status() - .context("failed to iterate rocksdb for key selector")?; - - // Apply the selector logic - let idx = if or_equal { - // If or_equal is true and the key exists, use it - if !keys.is_empty() && keys[0] == key { - offset.max(0) as usize - } else { - // Otherwise, use the next key - if offset >= 0 { - offset as usize - } else { - return Ok(Vec::new()); - } - } - } else { - // If or_equal is false, skip the exact match - let skip = if !keys.is_empty() && keys[0] == key { - 1 - } else { - 0 - }; - (skip + offset.max(0)) as usize - }; - - if idx < keys.len() { - Ok(keys[idx].clone()) - } else { - Ok(Vec::new()) - } - } - async fn handle_commit( - &mut self, + &self, start_version: u64, operations: Vec, conflict_ranges: Vec<(Vec, Vec, ConflictRangeType)>, ) -> Result<()> { + // A read-only transaction is never committed, matching FDB. Its reads all came from one pinned + // snapshot, so there is nothing to validate, and running the conflict check anyway would abort + // a transaction that cannot have observed anything inconsistent. + if tx_ops::is_read_only(&operations, &conflict_ranges) { + return Ok(()); + } + // Create a new transaction for this commit let txn = self.create_transaction(); let transaction_versionstamp = generate_versionstamp(0); @@ -442,8 +393,8 @@ impl TransactionTask { } } - async fn handle_get_range( - &mut self, + fn handle_get_range( + snapshot: &Snapshot<'_>, begin: Vec, begin_or_equal: bool, begin_offset: i32, @@ -453,16 +404,13 @@ impl TransactionTask { limit: Option, reverse: bool, ) -> Result { - let txn = self.create_transaction(); - let read_opts = ReadOptions::default(); - // Resolve the begin selector let resolved_begin = - self.resolve_key_selector_for_range(&txn, &begin, begin_or_equal, begin_offset)?; + Self::resolve_key_selector_for_range(snapshot, &begin, begin_or_equal, begin_offset)?; // Resolve the end selector let resolved_end = - self.resolve_key_selector_for_range(&txn, &end, end_or_equal, end_offset)?; + Self::resolve_key_selector_for_range(snapshot, &end, end_or_equal, end_offset)?; let mut results = Vec::new(); let limit = limit.unwrap_or(usize::MAX); @@ -472,7 +420,7 @@ impl TransactionTask { // during a forward scan and reversing afterward would instead return the // lowest keys, which is wrong for reverse range reads. if reverse { - let mut iter = txn.raw_iterator_opt(read_opts); + let mut iter = snapshot.raw_iterator(); iter.seek_for_prev(&resolved_end); while iter.valid() { @@ -499,7 +447,7 @@ impl TransactionTask { iter.status() .context("failed to iterate rocksdb for get range")?; } else { - let mut iter = txn.raw_iterator_opt(read_opts); + let mut iter = snapshot.raw_iterator(); iter.seek(&resolved_begin); while iter.valid() { @@ -526,8 +474,7 @@ impl TransactionTask { } fn resolve_key_selector_for_range( - &self, - txn: &RocksDbTransaction, + snapshot: &Snapshot<'_>, key: &[u8], or_equal: bool, offset: i32, @@ -538,12 +485,10 @@ impl TransactionTask { // (false, 0) => last_less_than // (true, 0) => last_less_or_equal - let read_opts = ReadOptions::default(); - match (or_equal, offset) { (false, 1) => { // first_greater_or_equal: find first key >= search_key - let mut iter = txn.raw_iterator_opt(read_opts); + let mut iter = snapshot.raw_iterator(); iter.seek(key); let result = iter.key().map(iter_bytes_to_vec); iter.status().context( @@ -554,7 +499,7 @@ impl TransactionTask { } (true, 1) => { // first_greater_than: find first key > search_key - let mut iter = txn.raw_iterator_opt(read_opts); + let mut iter = snapshot.raw_iterator(); iter.seek(key); while iter.valid() { let k = iter.key().expect("iterator should be valid"); @@ -578,7 +523,7 @@ impl TransactionTask { } } - async fn handle_get_estimated_range_size(&mut self, begin: &[u8], end: &[u8]) -> Result { + fn handle_get_estimated_range_size(&self, begin: &[u8], end: &[u8]) -> Result { let range = rocksdb::Range::new(begin, end); Ok(self diff --git a/engine/packages/universaldb/src/tx_ops.rs b/engine/packages/universaldb/src/tx_ops.rs index 995315e00d..c922e223b8 100644 --- a/engine/packages/universaldb/src/tx_ops.rs +++ b/engine/packages/universaldb/src/tx_ops.rs @@ -34,6 +34,22 @@ pub enum Operation { }, } +/// Whether a transaction has nothing to commit: no mutations and no explicitly added write conflict +/// ranges. FDB never commits a read-only transaction, so it can never conflict and never causes +/// another transaction to conflict. Drivers use this to skip the commit path entirely. +pub fn is_read_only( + operations: &[Operation], + conflict_ranges: &[(Vec, Vec, ConflictRangeType)], +) -> bool { + operations.is_empty() + && !conflict_ranges + .iter() + .any(|(_, _, conflict_type)| match conflict_type { + ConflictRangeType::Write => true, + ConflictRangeType::Read => false, + }) +} + #[derive(Debug, Clone)] pub enum GetOutput { Value(Vec), diff --git a/engine/packages/universaldb/tests/conflict_parity.rs b/engine/packages/universaldb/tests/conflict_parity.rs new file mode 100644 index 0000000000..5ed0658c42 --- /dev/null +++ b/engine/packages/universaldb/tests/conflict_parity.rs @@ -0,0 +1,391 @@ +//! FoundationDB conflict-semantics parity for the non-FDB drivers. +//! +//! FDB resolves conflicts directionally: a transaction aborts when something it *read* was *written* +//! under it. Reads are never the aggressor, blind write-vs-write does not conflict, and a read-only +//! transaction is never committed at all so it can never abort. + +use std::{pin::Pin, sync::Arc}; + +use rivet_test_deps_docker::TestDatabase; +use tokio::sync::Notify; +use universaldb::{Database, utils::IsolationLevel::*}; +use uuid::Uuid; + +const KEY: &[u8] = b"conflict_parity/key"; +const OTHER_KEY: &[u8] = b"conflict_parity/other"; + +#[tokio::test] +async fn rocksdb_conflict_parity() { + let _ = tracing_subscriber::fmt::try_init(); + + run_all_tests(&|| async { + let (db_config, _docker_config) = TestDatabase::FileSystem + .config(Uuid::new_v4(), 1) + .await + .unwrap(); + let rivet_config::config::Database::FileSystem(fs_config) = db_config else { + unreachable!() + }; + + let driver = universaldb::driver::RocksDbDatabaseDriver::new(fs_config.path) + .await + .unwrap(); + + Database::new(Arc::new(driver)) + }) + .await; +} + +#[tokio::test] +async fn postgres_conflict_parity() { + let _ = tracing_subscriber::fmt::try_init(); + + let (db_config, docker_config) = TestDatabase::Postgres + .config(Uuid::new_v4(), 1) + .await + .unwrap(); + let mut docker_config = docker_config.unwrap(); + docker_config.start().await.unwrap(); + + let rivet_config::config::Database::Postgres(postgres_config) = db_config else { + unreachable!(); + }; + let connection_string = postgres_config.url.read().clone(); + + wait_for_postgres(&connection_string).await; + + // Every database in this test shares one Postgres, and the leader lease is exclusive, so the + // drivers are created and shut down one at a time rather than through `run_all_tests`. + let cases: [(&str, fn(Database) -> BoxFut); 5] = [ + ("read_only_txn_does_not_abort_a_writer", |db| { + Box::pin(read_only_txn_does_not_abort_a_writer(db)) + }), + ("read_only_txn_never_conflicts", |db| { + Box::pin(read_only_txn_never_conflicts(db)) + }), + ("reads_are_repeatable_within_a_txn", |db| { + Box::pin(reads_are_repeatable_within_a_txn(db)) + }), + ("blind_write_vs_write_does_not_conflict", |db| { + Box::pin(blind_write_vs_write_does_not_conflict(db)) + }), + ("writer_conflicts_when_a_key_it_read_was_written", |db| { + Box::pin(writer_conflicts_when_a_key_it_read_was_written(db)) + }), + ]; + + for (name, case) in cases { + tracing::info!(name, "running postgres conflict parity case"); + + let driver = universaldb::driver::PostgresDatabaseDriver::new_with_config( + universaldb::driver::postgres::PostgresConfig::new(connection_string.clone()), + ) + .await + .unwrap(); + let db = Database::new(Arc::new(driver)); + + db.txn("clear", |tx| async move { + tx.clear(KEY); + tx.clear(OTHER_KEY); + Ok(()) + }) + .await + .unwrap(); + + case(db.clone()).await; + + db.shutdown().await; + } +} + +type BoxFut = Pin>>; + +/// Block until the freshly started Postgres container accepts connections. The container reports +/// started before the server is listening. +async fn wait_for_postgres(connection_string: &str) { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60); + loop { + match tokio_postgres::connect(connection_string, tokio_postgres::NoTls).await { + Ok((_client, connection)) => { + drop(connection); + return; + } + Err(err) => { + assert!( + std::time::Instant::now() < deadline, + "postgres never became reachable: {err}" + ); + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + } + } + } +} + +/// Every case runs against a freshly created database because they change the retry limit and race +/// two transactions against one key. +async fn run_all_tests(new_db: &F) +where + F: Fn() -> Fut, + Fut: Future, +{ + read_only_txn_does_not_abort_a_writer(new_db().await).await; + read_only_txn_never_conflicts(new_db().await).await; + reads_are_repeatable_within_a_txn(new_db().await).await; + blind_write_vs_write_does_not_conflict(new_db().await).await; + writer_conflicts_when_a_key_it_read_was_written(new_db().await).await; +} + +/// A transaction that only read a key must not abort a writer of that key, even though their version +/// windows overlap. FDB would commit the writer: it checks the writer's reads against writes, and the +/// reader wrote nothing. +async fn read_only_txn_does_not_abort_a_writer(db: Database) { + db.txn_retry_limit(1).unwrap(); + + let writer_started = Arc::new(Notify::new()); + let read_committed = Arc::new(Notify::new()); + + let writer = { + let db = db.clone(); + let writer_started = writer_started.clone(); + let read_committed = read_committed.clone(); + + tokio::spawn(async move { + db.txn("writer", |tx| { + let writer_started = writer_started.clone(); + let read_committed = read_committed.clone(); + + async move { + tx.set(KEY, b"written"); + + // The write is staged, so the reader can now open a transaction whose version + // window overlaps this one. + writer_started.notify_one(); + read_committed.notified().await; + + Ok(()) + } + }) + .await + }) + }; + + // Open the read-only transaction after the writer so both windows overlap. + writer_started.notified().await; + db.txn("reader", |tx| async move { + tx.get(KEY, Serializable).await?; + Ok(()) + }) + .await + .expect("read-only transaction should commit"); + read_committed.notify_one(); + + writer + .await + .unwrap() + .expect("a concurrent read must not abort a writer"); +} + +/// A read-only transaction is never committed, so a write landing under it cannot abort it. +async fn read_only_txn_never_conflicts(db: Database) { + db.txn_retry_limit(1).unwrap(); + + let read_taken = Arc::new(Notify::new()); + let write_committed = Arc::new(Notify::new()); + + let reader = { + let db = db.clone(); + let read_taken = read_taken.clone(); + let write_committed = write_committed.clone(); + + tokio::spawn(async move { + db.txn("reader", |tx| { + let read_taken = read_taken.clone(); + let write_committed = write_committed.clone(); + + async move { + tx.get(KEY, Serializable).await?; + + read_taken.notify_one(); + write_committed.notified().await; + + Ok(()) + } + }) + .await + }) + }; + + read_taken.notified().await; + db.txn("writer", |tx| async move { + tx.set(KEY, b"written"); + Ok(()) + }) + .await + .unwrap(); + write_committed.notify_one(); + + reader + .await + .unwrap() + .expect("a read-only transaction must never conflict"); +} + +/// All reads in a transaction come from one point in time, so a commit landing mid-transaction is +/// never partially visible. +async fn reads_are_repeatable_within_a_txn(db: Database) { + db.txn_retry_limit(1).unwrap(); + + db.txn("seed", |tx| async move { + tx.set(KEY, b"first"); + Ok(()) + }) + .await + .unwrap(); + + let read_taken = Arc::new(Notify::new()); + let write_committed = Arc::new(Notify::new()); + + let reader = { + let db = db.clone(); + let read_taken = read_taken.clone(); + let write_committed = write_committed.clone(); + + tokio::spawn(async move { + db.txn("reader", |tx| { + let read_taken = read_taken.clone(); + let write_committed = write_committed.clone(); + + async move { + let before = tx.get(KEY, Serializable).await?; + + read_taken.notify_one(); + write_committed.notified().await; + + let after = tx.get(KEY, Serializable).await?; + + Ok((before, after)) + } + }) + .await + }) + }; + + read_taken.notified().await; + db.txn("writer", |tx| async move { + tx.set(KEY, b"second"); + Ok(()) + }) + .await + .unwrap(); + write_committed.notify_one(); + + let (before, after) = reader.await.unwrap().unwrap(); + assert_eq!( + before.as_ref().map(|v| v.as_slice()), + Some(b"first".as_slice()), + "the first read should see the seeded value" + ); + assert_eq!( + after, before, + "a commit landing mid-transaction must not be visible to a later read" + ); +} + +/// Neither transaction read what the other wrote, so neither aborts. +async fn blind_write_vs_write_does_not_conflict(db: Database) { + db.txn_retry_limit(1).unwrap(); + + let first_staged = Arc::new(Notify::new()); + let second_committed = Arc::new(Notify::new()); + + let first = { + let db = db.clone(); + let first_staged = first_staged.clone(); + let second_committed = second_committed.clone(); + + tokio::spawn(async move { + db.txn("first", |tx| { + let first_staged = first_staged.clone(); + let second_committed = second_committed.clone(); + + async move { + tx.set(KEY, b"first"); + + first_staged.notify_one(); + second_committed.notified().await; + + Ok(()) + } + }) + .await + }) + }; + + first_staged.notified().await; + db.txn("second", |tx| async move { + tx.set(KEY, b"second"); + Ok(()) + }) + .await + .unwrap(); + second_committed.notify_one(); + + first + .await + .unwrap() + .expect("blind write-vs-write must not conflict"); +} + +/// The one case that must still abort: a transaction that writes and whose read was invalidated by a +/// write committed inside its version window. +async fn writer_conflicts_when_a_key_it_read_was_written(db: Database) { + db.txn_retry_limit(1).unwrap(); + + let read_taken = Arc::new(Notify::new()); + let write_committed = Arc::new(Notify::new()); + + let reader = { + let db = db.clone(); + let read_taken = read_taken.clone(); + let write_committed = write_committed.clone(); + + tokio::spawn(async move { + db.txn("read_then_write", |tx| { + let read_taken = read_taken.clone(); + let write_committed = write_committed.clone(); + + async move { + tx.get(KEY, Serializable).await?; + tx.set(OTHER_KEY, b"derived"); + + read_taken.notify_one(); + write_committed.notified().await; + + Ok(()) + } + }) + .await + }) + }; + + read_taken.notified().await; + db.txn("writer", |tx| async move { + tx.set(KEY, b"invalidated"); + Ok(()) + }) + .await + .unwrap(); + write_committed.notify_one(); + + let err = reader + .await + .unwrap() + .expect_err("a write to a key this transaction read must abort it"); + assert!( + err.chain().any(|x| matches!( + x.downcast_ref::(), + Some(universaldb::error::DatabaseError::MaxRetriesReached) + )), + "expected the conflict to exhaust retries, got {err:?}" + ); +}