#![cfg(not(target_arch = "wasm32"))]
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;
use regolith::{IsolationLevel, OptimisticTransactionDb, Options, TransactionError};
use tempfile::TempDir;
fn write_skew_pairs(isolation: IsolationLevel, rounds: usize) -> usize {
let dir = TempDir::new().expect("tempdir");
let db = Arc::new(
OptimisticTransactionDb::open(dir.path(), Options::default())
.expect("open")
.with_isolation(isolation),
);
let both = Arc::new(AtomicUsize::new(0));
for round in 0..rounds {
let x = format!("x{round:04}").into_bytes();
let y = format!("y{round:04}").into_bytes();
db.db().put(&x, b"0").expect("seed x");
db.db().put(&y, b"0").expect("seed y");
let barrier = Arc::new(std::sync::Barrier::new(2));
let mut handles = Vec::new();
for (mine, theirs) in [(x.clone(), y.clone()), (y.clone(), x.clone())] {
let db = Arc::clone(&db);
let barrier = Arc::clone(&barrier);
handles.push(thread::spawn(move || {
let tx = db.begin_transaction();
let seen = tx.get(&theirs).expect("read");
barrier.wait();
if seen.as_deref() == Some(b"0".as_ref()) {
tx.put(&mine, b"1").expect("write");
}
matches!(tx.commit(), Ok(()))
}));
}
let committed = handles
.into_iter()
.filter(|_| true)
.map(|h| h.join().expect("join"))
.filter(|ok| *ok)
.count();
if committed == 2 {
let vx = db.db().get(&x).expect("get x");
let vy = db.db().get(&y).expect("get y");
if vx.as_deref() == Some(b"1".as_ref()) && vy.as_deref() == Some(b"1".as_ref()) {
both.fetch_add(1, Ordering::Relaxed);
}
}
}
both.load(Ordering::Relaxed)
}
#[test]
fn serializable_excludes_write_skew() {
let skews = write_skew_pairs(IsolationLevel::Serializable, 40);
assert_eq!(
skews, 0,
"{skews} write-skew pair(s) committed under Serializable; validating the whole \
read set should have aborted the second of each pair"
);
}
#[test]
fn snapshot_isolation_admits_the_write_skew_serializable_excludes() {
let skews = write_skew_pairs(IsolationLevel::SnapshotIsolation, 40);
assert!(
skews > 0,
"the schedule produced no write skew even under snapshot isolation, so it is not \
exercising the anomaly and the serializable test above proves nothing"
);
}
#[test]
fn serializable_commits_transactions_that_do_not_conflict() {
let dir = TempDir::new().expect("tempdir");
let db = OptimisticTransactionDb::open(dir.path(), Options::default())
.expect("open")
.with_isolation(IsolationLevel::Serializable);
for i in 0..200u64 {
let tx = db.begin_transaction();
let k = format!("k{i:04}");
tx.get(k.as_bytes()).expect("read");
tx.put(k.as_bytes(), b"v").expect("write");
tx.commit()
.expect("a transaction touching only its own key must commit");
}
for i in 0..200u64 {
let k = format!("k{i:04}");
assert_eq!(db.db().get(k.as_bytes()).expect("get"), Some(b"v".to_vec()));
}
}
#[test]
fn serializable_read_only_transactions_commit_under_concurrent_writes() {
let dir = TempDir::new().expect("tempdir");
let db = OptimisticTransactionDb::open(dir.path(), Options::default())
.expect("open")
.with_isolation(IsolationLevel::Serializable);
for i in 0..50u64 {
db.db()
.put(format!("r{i:04}").as_bytes(), b"v")
.expect("seed");
}
let tx = db.begin_transaction();
for i in 0..50u64 {
tx.get(format!("r{i:04}").as_bytes()).expect("read");
}
for i in 0..50u64 {
db.db()
.put(format!("other{i:04}").as_bytes(), b"v")
.expect("write");
}
tx.commit()
.expect("a read-only transaction must not conflict with writes it never read");
}
#[test]
fn the_level_can_be_chosen_per_transaction() {
let dir = TempDir::new().expect("tempdir");
let db = OptimisticTransactionDb::open(dir.path(), Options::default()).expect("open");
assert_eq!(db.isolation(), IsolationLevel::SnapshotIsolation);
db.db().put(b"a", b"0").expect("seed");
let tx = db.begin_transaction_with(IsolationLevel::Serializable);
tx.get(b"a").expect("read");
db.db().put(b"a", b"1").expect("concurrent write");
match tx.commit() {
Err(TransactionError::Conflict { .. }) => {}
other => panic!(
"a serializable transaction must abort when a key it read was written; got {other:?}"
),
}
}