Struct Database 

pub struct Database { /* private fields */ }

Owns every table, version chain and index, and hands out Transactions.

Every type must be passed to Database::register before a transaction touches it; operations on an unregistered type fail with Error::TableNotRegistered. Registration is the only setup there is — nothing is read from disk, and nothing is written to it.

Shared across threads behind an &, so an Arc<Database> is the usual way to hand it to several of them. Reads take no locks whatever else is running. Dropping it frees every version it ever held.

Database::compact is the exception to all of that: it takes &mut self, so no transaction may be live while it runs.

Implementations

impl Database

pub fn open(config: Config) -> Result<Self>

Create an empty database.

Nothing is read from disk and nothing will be written to it — see the crate docs. Every type it will store must then be passed to Database::register before use.

use mvcc::{Config, Database};

let db = Database::open(Config::in_memory())?;

Errors

Currently infallible; the Result is here so that configuration that can fail may be added without a breaking change.

Examples found in repository

examples/basic.rs (line 25)

fn main() -> Result<()> {
    let db = Database::open(Config::in_memory())?;
    db.register::<Account>()?;

    // ---- insert -----------------------------------------------------------
    // `transaction` runs the closure, commits it, and retries it if it hits a
    // retriable conflict. Snapshot isolation by default.
    db.transaction(|tx| {
        tx.insert(Account {
            id: 1,
            owner: "ada".into(),
            branch: 10,
            balance: 500,
        })?;
        tx.insert(Account {
            id: 2,
            owner: "bob".into(),
            branch: 10,
            balance: 250,
        })?;
        tx.insert(Account {
            id: 3,
            owner: "cleo".into(),
            branch: 20,
            balance: 900,
        })?;
        Ok(())
    })?;

    // ---- read -------------------------------------------------------------
    let mut tx = db.begin();
    let ada = tx.get::<Account>(&1)?.expect("just inserted");
    println!("ada: branch {}, balance {}", ada.branch, ada.balance);

    // A read-only transaction has nothing to commit; dropping it rolls back,
    // which for a reader means simply releasing its snapshot.
    drop(tx);

    // ---- update -----------------------------------------------------------
    db.transaction(|tx| {
        tx.update::<Account>(&1, |a| a.balance -= 100)?;
        tx.update::<Account>(&2, |a| a.balance += 100)?;
        Ok(())
    })?;

    // ---- scan by primary key ----------------------------------------------
    let mut tx = db.begin();
    println!("\nall accounts:");
    for account in tx.scan::<Account>()? {
        println!(
            "  {:>4}  {:<6} branch {}  {:>5}",
            account.id, account.owner, account.branch, account.balance
        );
    }

    // ---- scan by secondary index ------------------------------------------
    println!("\nbranch 10:");
    for account in tx.scan_index(Account::BRANCH, 10u32..=10)? {
        println!("  {} ({})", account.owner, account.balance);
    }
    drop(tx);

    // ---- constraint violations --------------------------------------------
    let mut tx = db.begin();
    let duplicate = tx.insert(Account {
        id: 99,
        owner: "ada".into(),
        branch: 30,
        balance: 0,
    });
    println!("\nreusing owner 'ada': {}", duplicate.unwrap_err());
    drop(tx);

    // ---- delete -----------------------------------------------------------
    db.transaction(|tx| {
        let existed = tx.delete::<Account>(&3)?;
        println!("deleted cleo: {existed}");
        Ok(())
    })?;

    let mut tx = db.begin();
    println!(
        "cleo now: {:?}",
        tx.get::<Account>(&3)?.map(|r| r.to_owned())
    );

    Ok(())
}

examples/concurrent.rs (line 30)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Account>()?;

    db.transaction(|tx| {
        for id in 0..ACCOUNTS {
            tx.insert(Account {
                id,
                balance: STARTING_BALANCE,
            })?;
        }
        Ok(())
    })?;

    let total_before = total(&db)?;
    println!("starting total: {total_before}");

    let stop = Arc::new(AtomicBool::new(false));
    let reads = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));

    // ---- a reader that must never observe a torn transfer -----------------
    // Money moves between accounts constantly. Under snapshot isolation this
    // thread's scans must always sum to the same total: it either sees both
    // halves of a transfer or neither.
    let auditor = {
        let db = Arc::clone(&db);
        let stop = Arc::clone(&stop);
        let reads = Arc::clone(&reads);
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let sum: i64 = tx.scan::<Account>()?.iter().map(|a| a.balance).sum();
                assert_eq!(
                    sum,
                    ACCOUNTS as i64 * STARTING_BALANCE,
                    "a reader observed a partially applied transfer"
                );
                reads.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    // ---- writers ----------------------------------------------------------
    let start = Instant::now();
    let mut writers = Vec::new();
    for t in 0..TRANSFER_THREADS {
        let db = Arc::clone(&db);
        let retries = Arc::clone(&retries);
        writers.push(thread::spawn(move || -> Result<()> {
            // A cheap per-thread PRNG; no dependency needed for this.
            let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t as u64 + 1);
            let mut next = move || {
                seed ^= seed << 13;
                seed ^= seed >> 7;
                seed ^= seed << 17;
                seed
            };

            for _ in 0..TRANSFERS_PER_THREAD {
                let from = next() % ACCOUNTS;
                let to = next() % ACCOUNTS;
                if from == to {
                    continue;
                }

                let mut attempts = 0u64;
                // Serializable, because the transfer's *write* depends on a
                // balance it *read* — the write-skew shape. Snapshot isolation
                // would let two concurrent transfers both pass the check.
                db.transaction_with::<Serializable, _, _>(|tx| {
                    attempts += 1;
                    let balance = tx.get::<Account>(&from)?.map(|a| a.balance).unwrap_or(0);
                    if balance < 10 {
                        return Ok(());
                    }
                    tx.update::<Account>(&from, |a| a.balance -= 10)?;
                    tx.update::<Account>(&to, |a| a.balance += 10)?;
                    Ok(())
                })?;
                retries.fetch_add(attempts - 1, Ordering::Relaxed);
            }
            Ok(())
        }));
    }

    for w in writers {
        w.join().expect("writer panicked")?;
    }
    let elapsed = start.elapsed();

    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    // ---- results ----------------------------------------------------------
    let total_after = total(&db)?;
    let committed = (TRANSFER_THREADS * TRANSFERS_PER_THREAD) as u64;

    println!("final total:    {total_after}");
    assert_eq!(total_before, total_after, "money was created or destroyed");

    println!("\n{committed} transfers in {elapsed:.2?}");
    println!(
        "  {:.0} transfers/sec",
        committed as f64 / elapsed.as_secs_f64()
    );
    println!(
        "  {} concurrent audit scans completed, none of them blocked",
        reads.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.1} per transfer)",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / committed as f64
    );

    // The GC watermark should be advancing. If `active_transactions` climbs and
    // the watermark stalls, some transaction is being leaked — see engine::gc.
    let stats = db.stats();
    println!(
        "\ngc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );

    println!("\n✓ no torn reads, no lost updates, conservation held");

    // Slow readers are what pin the GC watermark. Left here as the shape of the
    // thing to watch for, not as something this example demonstrates going wrong.
    thread::sleep(Duration::from_millis(1));
    Ok(())
}

examples/isolation.rs (line 33)

fn main() -> Result<()> {
    let db = Database::open(Config::in_memory())?;
    db.register::<Doctor>()?;
    db.register::<Counter>()?;

    db.transaction(|tx| {
        tx.insert(Doctor {
            id: 1,
            name: "ada".into(),
            on_call: true,
        })?;
        tx.insert(Doctor {
            id: 2,
            name: "bob".into(),
            on_call: true,
        })?;
        tx.insert(Counter { id: 1, value: 0 })
    })?;

    // ------------------------------------------------------------------
    banner("1. Snapshot: a reader is unaffected by concurrent commits");
    {
        let mut reader = db.begin_with::<Snapshot>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        // Someone else commits, start to finish, while `reader` is open.
        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 42).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_eq!(before, after, "snapshot isolation must be stable");
        println!("  ✓ the snapshot held: readers never block and never change under you");
    }

    // ------------------------------------------------------------------
    banner("2. ReadCommitted: each statement sees a fresh snapshot");
    {
        let mut reader = db.begin_with::<ReadCommitted>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 99).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_ne!(
            before, after,
            "read committed should observe the new commit"
        );
        println!("  ✓ non-repeatable read — the tradeoff this level makes for cheapness");
    }

    // ------------------------------------------------------------------
    banner("3. Write-write conflict: first committer wins");
    {
        let mut first = db.begin_with::<Snapshot>();
        let mut second = db.begin_with::<Snapshot>();

        first.update::<Counter>(&1, |c| c.value += 1)?;
        first.commit()?;

        // `second` took its snapshot before `first` committed, so writing here
        // would silently drop `first`'s update.
        let result = second.update::<Counter>(&1, |c| c.value += 1);
        match result {
            Err(e) => println!(
                "  second transaction: {e}  (retriable: {})",
                e.is_retriable()
            ),
            Ok(_) => unreachable!("the stale write should have been rejected"),
        }
        println!("  ✓ no lost update");
    }

    // ------------------------------------------------------------------
    banner("4. Write skew: allowed under Snapshot");
    {
        // The rule: at least one doctor must stay on call. Each transaction
        // checks the rule, sees it satisfied, and takes a *different* doctor
        // off call — so there is no write-write conflict to catch them.
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();

        let t1_sees = t1.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t1.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let t2_sees = t2.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t2.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let names: Vec<String> = {
            let mut tx = db.begin();
            tx.scan::<Doctor>()?
                .iter()
                .map(|d| d.name.clone())
                .collect()
        };
        println!("  on the rota: {}", names.join(", "));
        println!("  t1 counts {t1_sees} on call, t2 counts {t2_sees} — both think it is safe");

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;
        t1.commit()?;
        t2.commit()?;

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✗ {remaining} doctors on call — the invariant is broken");
        println!("    Both transactions were individually legal. This is write skew,");
        println!("    and it is why Snapshot is not Serializable.");
    }

    // ------------------------------------------------------------------
    banner("5. Write skew: prevented under Serializable");
    {
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();

        // Both read both rows, exactly as before.
        let _ = t1.get::<Doctor>(&1)?;
        let _ = t1.get::<Doctor>(&2)?;
        let _ = t2.get::<Doctor>(&1)?;
        let _ = t2.get::<Doctor>(&2)?;

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;

        t1.commit()?;
        match t2.commit() {
            Err(e) => println!("  t2: {e}  (retriable: {})", e.is_retriable()),
            Ok(()) => unreachable!("t2 read a row t1 changed; it must not commit"),
        }

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✓ {remaining} doctor still on call — the invariant held");
        println!("    t2 read doctor 1, which t1 changed, so t2 could not be serialized.");
    }

    // ------------------------------------------------------------------
    banner("6. Retries are the normal way to use Serializable");
    {
        // `transaction_with` reruns the closure on a retriable failure, so the
        // abort in scenario 5 becomes invisible to the caller.
        let mut attempts = 0;
        db.transaction_with::<Serializable, _, _>(|tx| {
            attempts += 1;
            let ada = tx.get::<Doctor>(&1)?.unwrap().on_call;
            let bob = tx.get::<Doctor>(&2)?.unwrap().on_call;
            if ada && bob {
                tx.update::<Doctor>(&1, |d| d.on_call = false)?;
            }
            Ok(())
        })?;
        println!("  committed after {attempts} attempt(s)");
        println!("  ✓ write your transaction as if it runs alone; let the engine retry it");
    }

    banner("Summary");
    println!("  ReadCommitted   cheapest; non-repeatable reads and phantoms");
    println!("  RepeatableRead  stable snapshot; write skew possible");
    println!("  Snapshot        stable snapshot; write skew possible   ← default");
    println!("  Serializable    no anomalies; expect retriable aborts");

    Ok(())
}

examples/game.rs (line 142)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Hero>()?;
    db.register::<Item>()?;
    db.register::<PartyMember>()?;

    let (ada, bram, cleo) = (1u64, 2u64, 3u64);

    // -----------------------------------------------------------------------
    act(1, "The world is created");

    db.transaction(|tx| {
        tx.insert(Hero {
            id: ada,
            name: "Ada".into(),
            gold: 120,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: bram,
            name: "Bram".into(),
            gold: 80,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: cleo,
            name: "Cleo".into(),
            gold: 200,
            strength: 60,
        })?;

        let mut item = |id, owner, name: &str, weight, value, kind| {
            tx.insert(Item {
                id,
                owner,
                name: name.into(),
                weight,
                value,
                kind,
                flavour: HashMap::new(),
            })
        };
        item(10, ada, "Rusty Sword", 30, 15, Kind::Weapon)?;
        item(11, ada, "Health Potion", 5, 20, Kind::Potion)?;
        item(12, bram, "Oak Shield", 40, 35, Kind::Weapon)?;
        item(13, cleo, "Lucky Charm", 2, 90, Kind::Trinket)?;
        // Loot lying in the dungeon, owned by nobody.
        item(20, GROUND, "Flaming Greatsword", 55, 500, Kind::Weapon)?;
        item(21, GROUND, "Elven Cloak", 20, 240, Kind::Trinket)?;
        item(22, GROUND, "Iron Helm", 35, 60, Kind::Weapon)?;

        // Arbitrary per-item state, in a database row, with no schema to
        // declare and no serialisation format to agree on.
        tx.update::<Item>(&20, |i| {
            i.flavour.insert("enchantment".into(), "flame".into());
            i.flavour.insert("forged_by".into(), "Durin".into());
        })?;
        Ok(())
    })?;

    // A whole transaction that fails partway leaves nothing behind. The unique
    // index on `name` refuses the second Ada, and the gold change goes with it.
    let before = gold_of(&mut db.begin(), ada)?;
    let doomed = db.transaction(|tx| {
        tx.update::<Hero>(&ada, |h| h.gold += 10_000)?;
        tx.insert(Hero {
            id: 99,
            name: "Ada".into(),
            gold: 0,
            strength: 10,
        })
    });
    println!("  a duplicate hero name: {}", doomed.unwrap_err());
    println!(
        "  Ada's gold is still {} — the whole transaction rolled back, not just the insert",
        gold_of(&mut db.begin(), ada)?
    );
    assert_eq!(gold_of(&mut db.begin(), ada)?, before);

    let mut tx = db.begin();
    for hero in [ada, bram, cleo] {
        let name = name_of(&mut tx, hero)?;
        let carried = carried(&mut tx, hero)?;
        let items = inventory(&mut tx, hero)?;
        let names: Vec<_> = items.iter().map(|(_, n, _)| n.as_str()).collect();
        println!("  {name:<5} carries {carried:>3} — {}", names.join(", "));
    }
    drop(tx);

    // -----------------------------------------------------------------------
    act(2, "Two heroes reach for the same greatsword");

    // Both begin, both see the sword unclaimed. The first to write takes it;
    // the second is told immediately rather than being made to wait.
    let mut ada_grabs = db.begin_with::<Snapshot>();
    let mut bram_grabs = db.begin_with::<Snapshot>();

    ada_grabs.update::<Item>(&20, |i| i.owner = ada)?;

    match bram_grabs.update::<Item>(&20, |i| i.owner = bram) {
        Err(e @ Error::WriteConflict { .. }) => {
            println!("  Bram: {e}");
            println!(
                "  ...retriable: {} — he can try for something else",
                e.is_retriable()
            );
        }
        other => unreachable!("expected a conflict, got {other:?}"),
    }

    // A failed write does not end the transaction. Bram takes the cloak instead.
    bram_grabs.update::<Item>(&21, |i| i.owner = bram)?;
    ada_grabs.commit()?;
    bram_grabs.commit()?;
    println!("  Ada takes the greatsword, Bram takes the cloak. Nobody blocked.");

    // -----------------------------------------------------------------------
    act(3, "Cleo overloads herself — write skew, and the fix");

    // Cleo can carry 60 and is carrying 2. Either 35kg helm fits; both do not.
    // Each transaction reads her inventory, checks the total, and picks up a
    // *different* item — so the two writes never touch the same record, and
    // first-updater-wins has nothing to catch.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.insert(Item {
            id: 23,
            owner: GROUND,
            name: "Steel Helm".into(),
            weight: 35,
            value: 70,
            kind: Kind::Weapon,
            flavour: HashMap::new(),
        })
    })?;

    let pick_up = |item: u64| {
        move |tx: &mut mvcc::Transaction<'_, Serializable>| -> Result<bool> {
            let hero = tx.get::<Hero>(&cleo)?.expect("Cleo exists");
            let strength = hero.strength;
            let load = carried(tx, cleo)?;
            let weight = tx.get::<Item>(&item)?.map(|i| i.weight).unwrap_or(0);
            if load + weight > strength {
                return Ok(false);
            }
            tx.update::<Item>(&item, |i| i.owner = cleo)?;
            Ok(true)
        }
    };

    // Under Snapshot, both checks pass against a stale inventory.
    {
        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();
        let (load1, load2) = (carried(&mut t1, cleo)?, carried(&mut t2, cleo)?);
        println!(
            "  Snapshot:     both transactions see {load1}/{load2} carried, both think a 35kg helm fits"
        );
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        t2.commit()?;
        let over = carried(&mut db.begin(), cleo)?;
        println!(
            "  ✗ Cleo now carries {over} of a possible 60. Two legal transactions, one broken rule."
        );
    }

    // Put it back and try again at Serializable.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.update::<Item>(&23, |i| i.owner = GROUND).map(|_| ())
    })?;

    {
        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();
        // Both read the inventory, as before.
        let _ = carried(&mut t1, cleo)?;
        let _ = carried(&mut t2, cleo)?;
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        match t2.commit() {
            Err(e @ Error::SerializationFailure) => println!("  Serializable: second pickup {e}"),
            other => unreachable!("expected a serialization failure, got {other:?}"),
        }
        println!(
            "  ✓ Cleo carries {} — the second pickup read an inventory the first changed.",
            carried(&mut db.begin(), cleo)?
        );
    }

    // In real code you would not hand-roll that. `transaction_with` retries,
    // and the retry re-reads the inventory and correctly declines.
    let took_it = db.transaction_with::<Serializable, _, _>(pick_up(23))?;
    println!("  ...and on retry the second helm is refused on its merits: picked_up = {took_it}");

    // -----------------------------------------------------------------------
    act(4, "The party fills up — a phantom, not a conflict");

    db.transaction(|tx| {
        tx.insert(PartyMember {
            hero: ada,
            party: PARTY,
            role: "Vanguard".into(),
        })?;
        tx.insert(PartyMember {
            hero: bram,
            party: PARTY,
            role: "Shield".into(),
        })?;
        tx.insert(PartyMember {
            hero: cleo,
            party: PARTY,
            role: "Scout".into(),
        })
    })?;

    // Two newcomers apply at once, with three of four slots taken. Neither
    // writes a row the other wrote — they insert *different* rows — so there is
    // no write conflict to detect. What they collide on is a row that did not
    // exist when either of them counted.
    let (dara, finn) = (4u64, 5u64);
    db.transaction(|tx| {
        tx.insert(Hero {
            id: dara,
            name: "Dara".into(),
            gold: 40,
            strength: 80,
        })?;
        tx.insert(Hero {
            id: finn,
            name: "Finn".into(),
            gold: 40,
            strength: 80,
        })
    })?;

    let mut d = db.begin_with::<Serializable>();
    let mut f = db.begin_with::<Serializable>();
    println!(
        "  Dara counts {} members, Finn counts {} — both see a free slot",
        party_size(&mut d)?,
        party_size(&mut f)?
    );
    d.insert(PartyMember {
        hero: dara,
        party: PARTY,
        role: "Healer".into(),
    })?;
    f.insert(PartyMember {
        hero: finn,
        party: PARTY,
        role: "Healer".into(),
    })?;
    d.commit()?;
    match f.commit() {
        Err(e @ Error::SerializationFailure) => println!("  Finn: {e}"),
        other => unreachable!("expected a serialization failure, got {other:?}"),
    }
    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan_where::<PartyMember, _>(|m| m.party == PARTY)?
        .iter()
        .map(|m| (m.hero, m.role.clone()))
        .collect();
    drop(tx);
    roster.sort();
    let size = roster.len();
    let mut tx = db.begin();
    let listed: Vec<String> = roster
        .iter()
        .map(|(h, role)| Ok(format!("{} the {role}", name_of(&mut tx, *h)?)))
        .collect::<Result<_>>()?;
    drop(tx);
    println!("  party: {}", listed.join(", "));
    println!("  ✓ {size}/{PARTY_LIMIT} filled. Finn's insert became a phantom in Dara's count.");
    println!("    (`scan_where` hands the engine the predicate, so it can re-check it at commit.)");
    assert!(size <= PARTY_LIMIT);

    // -----------------------------------------------------------------------
    act(5, "A trade — two records, one atomic step");

    let before = total_gold(&db)?;

    let trade = |seller: u64, buyer: u64, item: u64, price: i64| {
        let db = Arc::clone(&db);
        move || -> Result<bool> {
            db.transaction_with::<Serializable, _, _>(|tx| {
                let buyer_gold = gold_of(tx, buyer)?;
                if buyer_gold < price {
                    return Ok(false);
                }
                let owned_by_seller = tx
                    .get::<Item>(&item)?
                    .map(|i| i.owner == seller)
                    .unwrap_or(false);
                if !owned_by_seller {
                    return Ok(false);
                }
                tx.update::<Hero>(&buyer, |h| h.gold -= price)?;
                tx.update::<Hero>(&seller, |h| h.gold += price)?;
                tx.update::<Item>(&item, |i| i.owner = buyer)?;
                Ok(true)
            })
        }
    };

    let sold = trade(bram, cleo, 21, 150)()?;
    println!("  Bram sells the Elven Cloak to Cleo for 150g: {sold}");
    let mut tx = db.begin();
    let cloak_owner = tx.get::<Item>(&21)?.expect("cloak").owner;
    println!(
        "  Bram {}g, Cleo {}g, cloak owner is {}",
        gold_of(&mut tx, bram)?,
        gold_of(&mut tx, cleo)?,
        name_of(&mut tx, cloak_owner)?
    );
    drop(tx);
    assert_eq!(total_gold(&db)?, before, "gold must be conserved");
    println!("  ✓ total gold unchanged at {before} — no step of that was separately visible");

    // -----------------------------------------------------------------------
    act(6, "The chronicler reads while the world moves");

    // A long report scanning the whole world while combat rewrites it. It never
    // blocks a writer, and its view never shifts underneath it.
    let mut chronicler = db.begin_with::<Snapshot>();
    let opening = total_gold(&db)?;
    let seen_first = chronicler.scan::<Hero>()?.len();

    for _ in 0..50 {
        db.transaction(|tx| {
            tx.update::<Hero>(&ada, |h| h.gold += 1)?;
            tx.update::<Hero>(&bram, |h| h.gold -= 1)?;
            Ok(())
        })?;
    }
    db.transaction(|tx| {
        tx.insert(Hero {
            id: 6,
            name: "Mira".into(),
            gold: 0,
            strength: 50,
        })
    })?;

    let chron_gold: i64 = chronicler.scan::<Hero>()?.iter().map(|h| h.gold).sum();
    let seen_after = chronicler.scan::<Hero>()?.len();
    let in_world = db.begin().scan::<Hero>()?.len();
    println!("  chronicler saw {seen_first} heroes at the start and still sees {seen_after};");
    println!("  the world now holds {in_world}, after 50 commits and one new arrival");
    println!(
        "  chronicler totals {chron_gold}g; the world now totals {}g",
        total_gold(&db)?
    );
    assert_eq!(chron_gold, opening, "a snapshot must not move");
    chronicler.commit()?;
    println!("  ✓ a reader that ran across 51 commits saw exactly one consistent world");

    // -----------------------------------------------------------------------
    act(7, "The raid — many adventurers at once");

    let stop = Arc::new(AtomicBool::new(false));
    let trades = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));
    let audits = Arc::new(AtomicU64::new(0));
    let opening = total_gold(&db)?;
    let heroes = [ada, bram, cleo, dara, finn];

    // An auditor that must never catch a trade half-applied.
    let auditor = {
        let (db, stop, audits) = (Arc::clone(&db), Arc::clone(&stop), Arc::clone(&audits));
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let total: i64 = tx.scan::<Hero>()?.iter().map(|h| h.gold).sum();
                assert_eq!(total, opening, "auditor saw a half-finished trade");
                audits.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    let start = Instant::now();
    let raiders: Vec<_> = (0..4)
        .map(|t| {
            let (db, trades, retries) =
                (Arc::clone(&db), Arc::clone(&trades), Arc::clone(&retries));
            thread::spawn(move || -> Result<()> {
                let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t + 1);
                for _ in 0..400 {
                    seed ^= seed << 13;
                    seed ^= seed >> 7;
                    seed ^= seed << 17;
                    let from = heroes[(seed % heroes.len() as u64) as usize];
                    let to = heroes[((seed >> 8) % heroes.len() as u64) as usize];
                    if from == to {
                        continue;
                    }

                    let mut attempts = 0u64;
                    // Serializable, because the payment depends on a balance we
                    // read — the same shape as Cleo's carry limit in Act 3.
                    db.transaction_with::<Serializable, _, _>(|tx| {
                        attempts += 1;
                        let purse = gold_of(tx, from)?;
                        if purse < 5 {
                            return Ok(());
                        }
                        tx.update::<Hero>(&from, |h| h.gold -= 5)?;
                        tx.update::<Hero>(&to, |h| h.gold += 5)?;
                        Ok(())
                    })?;
                    trades.fetch_add(1, Ordering::Relaxed);
                    retries.fetch_add(attempts - 1, Ordering::Relaxed);
                }
                Ok(())
            })
        })
        .collect();

    for r in raiders {
        r.join().expect("raider panicked")?;
    }
    let elapsed = start.elapsed();
    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    let done = trades.load(Ordering::Relaxed);
    println!("  {done} trades across 4 threads in {elapsed:.2?}");
    println!(
        "  {} audit sweeps completed alongside them, none blocked, none torn",
        audits.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.2} per trade) — contention on 5 heroes, handled by the engine",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / done.max(1) as f64
    );
    assert_eq!(total_gold(&db)?, opening, "gold was created or destroyed");
    println!("  ✓ total gold still {opening}");

    // -----------------------------------------------------------------------
    act(8, "Closing the ledger");

    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan::<Hero>()?
        .iter()
        .map(|h| (h.id, h.name.clone(), h.gold))
        .collect();
    roster.sort_by_key(|(id, _, _)| *id);
    drop(tx);

    println!(
        "  {:<6} {:>6} {:>5} {:>7}  inventory",
        "hero", "gold", "load", "worth"
    );
    let mut tx = db.begin();
    for (id, name, gold) in &roster {
        let held = tx.scan_index(Item::OWNER, *id..=*id)?;
        let load: i32 = held.iter().map(|i| i.weight).sum();
        let worth: i64 = held.iter().map(|i| i.value).sum();
        let names: Vec<String> = held
            .iter()
            .map(|i| match i.kind {
                Kind::Weapon => format!("[wpn] {}", i.name),
                Kind::Potion => format!("[pot] {}", i.name),
                Kind::Trinket => format!("[trk] {}", i.name),
            })
            .collect();
        let names = if names.is_empty() {
            "—".to_string()
        } else {
            names.join(", ")
        };
        println!("  {name:<6} {gold:>5}g {load:>4}kg {worth:>6}g  {names}");
    }

    // Potions anywhere in the world, found by predicate rather than by key —
    // and `kind` is an ordinary Rust enum the engine knows nothing about.
    let potions = tx.scan_where::<Item, _>(|i| i.kind == Kind::Potion)?.len();
    let sword = tx.get::<Item>(&20)?.expect("greatsword");
    let enchantment = sword
        .flavour
        .get("enchantment")
        .cloned()
        .unwrap_or_default();
    let forged_by = sword.flavour.get("forged_by").cloned().unwrap_or_default();
    drop(tx);
    // The rule Act 3 was about, checked rather than asserted in prose.
    let mut tx = db.begin();
    for (id, name, _) in &roster {
        let strength = tx.get::<Hero>(id)?.map(|h| h.strength).unwrap_or(0);
        let load = carried(&mut tx, *id)?;
        assert!(
            load <= strength,
            "{name} is over their carry limit: {load} > {strength}"
        );
    }
    drop(tx);

    println!("\n  potions in the world: {potions}");
    println!("  the greatsword is {enchantment}-enchanted, forged by {forged_by}");

    // Nothing is reclaimed while a transaction that could still reach it is
    // alive. This is the number to watch in a long-lived process: a forgotten
    // transaction pins it and version chains grow without limit.
    let stats = db.stats();
    println!(
        "\n  gc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );
    println!("  (a watermark that stops moving while writes continue is the leak to look for)");

    Ok(())
}

pub fn stats(&self) -> GcStats

Snapshot of engine statistics. See crate::stats for what to watch.

Examples found in repository

examples/concurrent.rs (line 148)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Account>()?;

    db.transaction(|tx| {
        for id in 0..ACCOUNTS {
            tx.insert(Account {
                id,
                balance: STARTING_BALANCE,
            })?;
        }
        Ok(())
    })?;

    let total_before = total(&db)?;
    println!("starting total: {total_before}");

    let stop = Arc::new(AtomicBool::new(false));
    let reads = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));

    // ---- a reader that must never observe a torn transfer -----------------
    // Money moves between accounts constantly. Under snapshot isolation this
    // thread's scans must always sum to the same total: it either sees both
    // halves of a transfer or neither.
    let auditor = {
        let db = Arc::clone(&db);
        let stop = Arc::clone(&stop);
        let reads = Arc::clone(&reads);
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let sum: i64 = tx.scan::<Account>()?.iter().map(|a| a.balance).sum();
                assert_eq!(
                    sum,
                    ACCOUNTS as i64 * STARTING_BALANCE,
                    "a reader observed a partially applied transfer"
                );
                reads.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    // ---- writers ----------------------------------------------------------
    let start = Instant::now();
    let mut writers = Vec::new();
    for t in 0..TRANSFER_THREADS {
        let db = Arc::clone(&db);
        let retries = Arc::clone(&retries);
        writers.push(thread::spawn(move || -> Result<()> {
            // A cheap per-thread PRNG; no dependency needed for this.
            let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t as u64 + 1);
            let mut next = move || {
                seed ^= seed << 13;
                seed ^= seed >> 7;
                seed ^= seed << 17;
                seed
            };

            for _ in 0..TRANSFERS_PER_THREAD {
                let from = next() % ACCOUNTS;
                let to = next() % ACCOUNTS;
                if from == to {
                    continue;
                }

                let mut attempts = 0u64;
                // Serializable, because the transfer's *write* depends on a
                // balance it *read* — the write-skew shape. Snapshot isolation
                // would let two concurrent transfers both pass the check.
                db.transaction_with::<Serializable, _, _>(|tx| {
                    attempts += 1;
                    let balance = tx.get::<Account>(&from)?.map(|a| a.balance).unwrap_or(0);
                    if balance < 10 {
                        return Ok(());
                    }
                    tx.update::<Account>(&from, |a| a.balance -= 10)?;
                    tx.update::<Account>(&to, |a| a.balance += 10)?;
                    Ok(())
                })?;
                retries.fetch_add(attempts - 1, Ordering::Relaxed);
            }
            Ok(())
        }));
    }

    for w in writers {
        w.join().expect("writer panicked")?;
    }
    let elapsed = start.elapsed();

    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    // ---- results ----------------------------------------------------------
    let total_after = total(&db)?;
    let committed = (TRANSFER_THREADS * TRANSFERS_PER_THREAD) as u64;

    println!("final total:    {total_after}");
    assert_eq!(total_before, total_after, "money was created or destroyed");

    println!("\n{committed} transfers in {elapsed:.2?}");
    println!(
        "  {:.0} transfers/sec",
        committed as f64 / elapsed.as_secs_f64()
    );
    println!(
        "  {} concurrent audit scans completed, none of them blocked",
        reads.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.1} per transfer)",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / committed as f64
    );

    // The GC watermark should be advancing. If `active_transactions` climbs and
    // the watermark stalls, some transaction is being leaked — see engine::gc.
    let stats = db.stats();
    println!(
        "\ngc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );

    println!("\n✓ no torn reads, no lost updates, conservation held");

    // Slow readers are what pin the GC watermark. Left here as the shape of the
    // thing to watch for, not as something this example demonstrates going wrong.
    thread::sleep(Duration::from_millis(1));
    Ok(())
}

examples/game.rs (line 651)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Hero>()?;
    db.register::<Item>()?;
    db.register::<PartyMember>()?;

    let (ada, bram, cleo) = (1u64, 2u64, 3u64);

    // -----------------------------------------------------------------------
    act(1, "The world is created");

    db.transaction(|tx| {
        tx.insert(Hero {
            id: ada,
            name: "Ada".into(),
            gold: 120,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: bram,
            name: "Bram".into(),
            gold: 80,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: cleo,
            name: "Cleo".into(),
            gold: 200,
            strength: 60,
        })?;

        let mut item = |id, owner, name: &str, weight, value, kind| {
            tx.insert(Item {
                id,
                owner,
                name: name.into(),
                weight,
                value,
                kind,
                flavour: HashMap::new(),
            })
        };
        item(10, ada, "Rusty Sword", 30, 15, Kind::Weapon)?;
        item(11, ada, "Health Potion", 5, 20, Kind::Potion)?;
        item(12, bram, "Oak Shield", 40, 35, Kind::Weapon)?;
        item(13, cleo, "Lucky Charm", 2, 90, Kind::Trinket)?;
        // Loot lying in the dungeon, owned by nobody.
        item(20, GROUND, "Flaming Greatsword", 55, 500, Kind::Weapon)?;
        item(21, GROUND, "Elven Cloak", 20, 240, Kind::Trinket)?;
        item(22, GROUND, "Iron Helm", 35, 60, Kind::Weapon)?;

        // Arbitrary per-item state, in a database row, with no schema to
        // declare and no serialisation format to agree on.
        tx.update::<Item>(&20, |i| {
            i.flavour.insert("enchantment".into(), "flame".into());
            i.flavour.insert("forged_by".into(), "Durin".into());
        })?;
        Ok(())
    })?;

    // A whole transaction that fails partway leaves nothing behind. The unique
    // index on `name` refuses the second Ada, and the gold change goes with it.
    let before = gold_of(&mut db.begin(), ada)?;
    let doomed = db.transaction(|tx| {
        tx.update::<Hero>(&ada, |h| h.gold += 10_000)?;
        tx.insert(Hero {
            id: 99,
            name: "Ada".into(),
            gold: 0,
            strength: 10,
        })
    });
    println!("  a duplicate hero name: {}", doomed.unwrap_err());
    println!(
        "  Ada's gold is still {} — the whole transaction rolled back, not just the insert",
        gold_of(&mut db.begin(), ada)?
    );
    assert_eq!(gold_of(&mut db.begin(), ada)?, before);

    let mut tx = db.begin();
    for hero in [ada, bram, cleo] {
        let name = name_of(&mut tx, hero)?;
        let carried = carried(&mut tx, hero)?;
        let items = inventory(&mut tx, hero)?;
        let names: Vec<_> = items.iter().map(|(_, n, _)| n.as_str()).collect();
        println!("  {name:<5} carries {carried:>3} — {}", names.join(", "));
    }
    drop(tx);

    // -----------------------------------------------------------------------
    act(2, "Two heroes reach for the same greatsword");

    // Both begin, both see the sword unclaimed. The first to write takes it;
    // the second is told immediately rather than being made to wait.
    let mut ada_grabs = db.begin_with::<Snapshot>();
    let mut bram_grabs = db.begin_with::<Snapshot>();

    ada_grabs.update::<Item>(&20, |i| i.owner = ada)?;

    match bram_grabs.update::<Item>(&20, |i| i.owner = bram) {
        Err(e @ Error::WriteConflict { .. }) => {
            println!("  Bram: {e}");
            println!(
                "  ...retriable: {} — he can try for something else",
                e.is_retriable()
            );
        }
        other => unreachable!("expected a conflict, got {other:?}"),
    }

    // A failed write does not end the transaction. Bram takes the cloak instead.
    bram_grabs.update::<Item>(&21, |i| i.owner = bram)?;
    ada_grabs.commit()?;
    bram_grabs.commit()?;
    println!("  Ada takes the greatsword, Bram takes the cloak. Nobody blocked.");

    // -----------------------------------------------------------------------
    act(3, "Cleo overloads herself — write skew, and the fix");

    // Cleo can carry 60 and is carrying 2. Either 35kg helm fits; both do not.
    // Each transaction reads her inventory, checks the total, and picks up a
    // *different* item — so the two writes never touch the same record, and
    // first-updater-wins has nothing to catch.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.insert(Item {
            id: 23,
            owner: GROUND,
            name: "Steel Helm".into(),
            weight: 35,
            value: 70,
            kind: Kind::Weapon,
            flavour: HashMap::new(),
        })
    })?;

    let pick_up = |item: u64| {
        move |tx: &mut mvcc::Transaction<'_, Serializable>| -> Result<bool> {
            let hero = tx.get::<Hero>(&cleo)?.expect("Cleo exists");
            let strength = hero.strength;
            let load = carried(tx, cleo)?;
            let weight = tx.get::<Item>(&item)?.map(|i| i.weight).unwrap_or(0);
            if load + weight > strength {
                return Ok(false);
            }
            tx.update::<Item>(&item, |i| i.owner = cleo)?;
            Ok(true)
        }
    };

    // Under Snapshot, both checks pass against a stale inventory.
    {
        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();
        let (load1, load2) = (carried(&mut t1, cleo)?, carried(&mut t2, cleo)?);
        println!(
            "  Snapshot:     both transactions see {load1}/{load2} carried, both think a 35kg helm fits"
        );
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        t2.commit()?;
        let over = carried(&mut db.begin(), cleo)?;
        println!(
            "  ✗ Cleo now carries {over} of a possible 60. Two legal transactions, one broken rule."
        );
    }

    // Put it back and try again at Serializable.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.update::<Item>(&23, |i| i.owner = GROUND).map(|_| ())
    })?;

    {
        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();
        // Both read the inventory, as before.
        let _ = carried(&mut t1, cleo)?;
        let _ = carried(&mut t2, cleo)?;
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        match t2.commit() {
            Err(e @ Error::SerializationFailure) => println!("  Serializable: second pickup {e}"),
            other => unreachable!("expected a serialization failure, got {other:?}"),
        }
        println!(
            "  ✓ Cleo carries {} — the second pickup read an inventory the first changed.",
            carried(&mut db.begin(), cleo)?
        );
    }

    // In real code you would not hand-roll that. `transaction_with` retries,
    // and the retry re-reads the inventory and correctly declines.
    let took_it = db.transaction_with::<Serializable, _, _>(pick_up(23))?;
    println!("  ...and on retry the second helm is refused on its merits: picked_up = {took_it}");

    // -----------------------------------------------------------------------
    act(4, "The party fills up — a phantom, not a conflict");

    db.transaction(|tx| {
        tx.insert(PartyMember {
            hero: ada,
            party: PARTY,
            role: "Vanguard".into(),
        })?;
        tx.insert(PartyMember {
            hero: bram,
            party: PARTY,
            role: "Shield".into(),
        })?;
        tx.insert(PartyMember {
            hero: cleo,
            party: PARTY,
            role: "Scout".into(),
        })
    })?;

    // Two newcomers apply at once, with three of four slots taken. Neither
    // writes a row the other wrote — they insert *different* rows — so there is
    // no write conflict to detect. What they collide on is a row that did not
    // exist when either of them counted.
    let (dara, finn) = (4u64, 5u64);
    db.transaction(|tx| {
        tx.insert(Hero {
            id: dara,
            name: "Dara".into(),
            gold: 40,
            strength: 80,
        })?;
        tx.insert(Hero {
            id: finn,
            name: "Finn".into(),
            gold: 40,
            strength: 80,
        })
    })?;

    let mut d = db.begin_with::<Serializable>();
    let mut f = db.begin_with::<Serializable>();
    println!(
        "  Dara counts {} members, Finn counts {} — both see a free slot",
        party_size(&mut d)?,
        party_size(&mut f)?
    );
    d.insert(PartyMember {
        hero: dara,
        party: PARTY,
        role: "Healer".into(),
    })?;
    f.insert(PartyMember {
        hero: finn,
        party: PARTY,
        role: "Healer".into(),
    })?;
    d.commit()?;
    match f.commit() {
        Err(e @ Error::SerializationFailure) => println!("  Finn: {e}"),
        other => unreachable!("expected a serialization failure, got {other:?}"),
    }
    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan_where::<PartyMember, _>(|m| m.party == PARTY)?
        .iter()
        .map(|m| (m.hero, m.role.clone()))
        .collect();
    drop(tx);
    roster.sort();
    let size = roster.len();
    let mut tx = db.begin();
    let listed: Vec<String> = roster
        .iter()
        .map(|(h, role)| Ok(format!("{} the {role}", name_of(&mut tx, *h)?)))
        .collect::<Result<_>>()?;
    drop(tx);
    println!("  party: {}", listed.join(", "));
    println!("  ✓ {size}/{PARTY_LIMIT} filled. Finn's insert became a phantom in Dara's count.");
    println!("    (`scan_where` hands the engine the predicate, so it can re-check it at commit.)");
    assert!(size <= PARTY_LIMIT);

    // -----------------------------------------------------------------------
    act(5, "A trade — two records, one atomic step");

    let before = total_gold(&db)?;

    let trade = |seller: u64, buyer: u64, item: u64, price: i64| {
        let db = Arc::clone(&db);
        move || -> Result<bool> {
            db.transaction_with::<Serializable, _, _>(|tx| {
                let buyer_gold = gold_of(tx, buyer)?;
                if buyer_gold < price {
                    return Ok(false);
                }
                let owned_by_seller = tx
                    .get::<Item>(&item)?
                    .map(|i| i.owner == seller)
                    .unwrap_or(false);
                if !owned_by_seller {
                    return Ok(false);
                }
                tx.update::<Hero>(&buyer, |h| h.gold -= price)?;
                tx.update::<Hero>(&seller, |h| h.gold += price)?;
                tx.update::<Item>(&item, |i| i.owner = buyer)?;
                Ok(true)
            })
        }
    };

    let sold = trade(bram, cleo, 21, 150)()?;
    println!("  Bram sells the Elven Cloak to Cleo for 150g: {sold}");
    let mut tx = db.begin();
    let cloak_owner = tx.get::<Item>(&21)?.expect("cloak").owner;
    println!(
        "  Bram {}g, Cleo {}g, cloak owner is {}",
        gold_of(&mut tx, bram)?,
        gold_of(&mut tx, cleo)?,
        name_of(&mut tx, cloak_owner)?
    );
    drop(tx);
    assert_eq!(total_gold(&db)?, before, "gold must be conserved");
    println!("  ✓ total gold unchanged at {before} — no step of that was separately visible");

    // -----------------------------------------------------------------------
    act(6, "The chronicler reads while the world moves");

    // A long report scanning the whole world while combat rewrites it. It never
    // blocks a writer, and its view never shifts underneath it.
    let mut chronicler = db.begin_with::<Snapshot>();
    let opening = total_gold(&db)?;
    let seen_first = chronicler.scan::<Hero>()?.len();

    for _ in 0..50 {
        db.transaction(|tx| {
            tx.update::<Hero>(&ada, |h| h.gold += 1)?;
            tx.update::<Hero>(&bram, |h| h.gold -= 1)?;
            Ok(())
        })?;
    }
    db.transaction(|tx| {
        tx.insert(Hero {
            id: 6,
            name: "Mira".into(),
            gold: 0,
            strength: 50,
        })
    })?;

    let chron_gold: i64 = chronicler.scan::<Hero>()?.iter().map(|h| h.gold).sum();
    let seen_after = chronicler.scan::<Hero>()?.len();
    let in_world = db.begin().scan::<Hero>()?.len();
    println!("  chronicler saw {seen_first} heroes at the start and still sees {seen_after};");
    println!("  the world now holds {in_world}, after 50 commits and one new arrival");
    println!(
        "  chronicler totals {chron_gold}g; the world now totals {}g",
        total_gold(&db)?
    );
    assert_eq!(chron_gold, opening, "a snapshot must not move");
    chronicler.commit()?;
    println!("  ✓ a reader that ran across 51 commits saw exactly one consistent world");

    // -----------------------------------------------------------------------
    act(7, "The raid — many adventurers at once");

    let stop = Arc::new(AtomicBool::new(false));
    let trades = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));
    let audits = Arc::new(AtomicU64::new(0));
    let opening = total_gold(&db)?;
    let heroes = [ada, bram, cleo, dara, finn];

    // An auditor that must never catch a trade half-applied.
    let auditor = {
        let (db, stop, audits) = (Arc::clone(&db), Arc::clone(&stop), Arc::clone(&audits));
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let total: i64 = tx.scan::<Hero>()?.iter().map(|h| h.gold).sum();
                assert_eq!(total, opening, "auditor saw a half-finished trade");
                audits.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    let start = Instant::now();
    let raiders: Vec<_> = (0..4)
        .map(|t| {
            let (db, trades, retries) =
                (Arc::clone(&db), Arc::clone(&trades), Arc::clone(&retries));
            thread::spawn(move || -> Result<()> {
                let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t + 1);
                for _ in 0..400 {
                    seed ^= seed << 13;
                    seed ^= seed >> 7;
                    seed ^= seed << 17;
                    let from = heroes[(seed % heroes.len() as u64) as usize];
                    let to = heroes[((seed >> 8) % heroes.len() as u64) as usize];
                    if from == to {
                        continue;
                    }

                    let mut attempts = 0u64;
                    // Serializable, because the payment depends on a balance we
                    // read — the same shape as Cleo's carry limit in Act 3.
                    db.transaction_with::<Serializable, _, _>(|tx| {
                        attempts += 1;
                        let purse = gold_of(tx, from)?;
                        if purse < 5 {
                            return Ok(());
                        }
                        tx.update::<Hero>(&from, |h| h.gold -= 5)?;
                        tx.update::<Hero>(&to, |h| h.gold += 5)?;
                        Ok(())
                    })?;
                    trades.fetch_add(1, Ordering::Relaxed);
                    retries.fetch_add(attempts - 1, Ordering::Relaxed);
                }
                Ok(())
            })
        })
        .collect();

    for r in raiders {
        r.join().expect("raider panicked")?;
    }
    let elapsed = start.elapsed();
    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    let done = trades.load(Ordering::Relaxed);
    println!("  {done} trades across 4 threads in {elapsed:.2?}");
    println!(
        "  {} audit sweeps completed alongside them, none blocked, none torn",
        audits.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.2} per trade) — contention on 5 heroes, handled by the engine",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / done.max(1) as f64
    );
    assert_eq!(total_gold(&db)?, opening, "gold was created or destroyed");
    println!("  ✓ total gold still {opening}");

    // -----------------------------------------------------------------------
    act(8, "Closing the ledger");

    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan::<Hero>()?
        .iter()
        .map(|h| (h.id, h.name.clone(), h.gold))
        .collect();
    roster.sort_by_key(|(id, _, _)| *id);
    drop(tx);

    println!(
        "  {:<6} {:>6} {:>5} {:>7}  inventory",
        "hero", "gold", "load", "worth"
    );
    let mut tx = db.begin();
    for (id, name, gold) in &roster {
        let held = tx.scan_index(Item::OWNER, *id..=*id)?;
        let load: i32 = held.iter().map(|i| i.weight).sum();
        let worth: i64 = held.iter().map(|i| i.value).sum();
        let names: Vec<String> = held
            .iter()
            .map(|i| match i.kind {
                Kind::Weapon => format!("[wpn] {}", i.name),
                Kind::Potion => format!("[pot] {}", i.name),
                Kind::Trinket => format!("[trk] {}", i.name),
            })
            .collect();
        let names = if names.is_empty() {
            "—".to_string()
        } else {
            names.join(", ")
        };
        println!("  {name:<6} {gold:>5}g {load:>4}kg {worth:>6}g  {names}");
    }

    // Potions anywhere in the world, found by predicate rather than by key —
    // and `kind` is an ordinary Rust enum the engine knows nothing about.
    let potions = tx.scan_where::<Item, _>(|i| i.kind == Kind::Potion)?.len();
    let sword = tx.get::<Item>(&20)?.expect("greatsword");
    let enchantment = sword
        .flavour
        .get("enchantment")
        .cloned()
        .unwrap_or_default();
    let forged_by = sword.flavour.get("forged_by").cloned().unwrap_or_default();
    drop(tx);
    // The rule Act 3 was about, checked rather than asserted in prose.
    let mut tx = db.begin();
    for (id, name, _) in &roster {
        let strength = tx.get::<Hero>(id)?.map(|h| h.strength).unwrap_or(0);
        let load = carried(&mut tx, *id)?;
        assert!(
            load <= strength,
            "{name} is over their carry limit: {load} > {strength}"
        );
    }
    drop(tx);

    println!("\n  potions in the world: {potions}");
    println!("  the greatsword is {enchantment}-enchanted, forged by {forged_by}");

    // Nothing is reclaimed while a transaction that could still reach it is
    // alive. This is the number to watch in a long-lived process: a forgotten
    // transaction pins it and version chains grow without limit.
    let stats = db.stats();
    println!(
        "\n  gc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );
    println!("  (a watermark that stops moving while writes continue is the leak to look for)");

    Ok(())
}

pub fn compact(&mut self) -> usize

Reclaim the memory of deleted records, returning how many were freed.

Ordinary reclamation frees a record’s versions but not its slot: about 180 bytes per key stay behind, because the map that resolves a key to a slot is append-only, and that is what lets a lookup take no locks and write nothing. This is the operation that gives those bytes back.

It takes &mut self, and that is the entire safety argument. A Transaction borrows the database, so an exclusive borrow is a compile-time proof that none exist — which is what makes it sound to free a slot some transaction might otherwise have already resolved.

This is deliberately a separate call rather than something the engine does for you, and the reason is speed. Reclaiming slots as it went would mean either revalidating every write against the key map or making transactions announce themselves to a lock — putting synchronisation back onto paths that currently have none. That append-only property is what lets a lookup take no locks and write nothing to shared memory, and it is worth 1.9x on uniform point reads and 9.8x on contended ones at four threads — the shipped before-and-after, not the throwaway lock-deletion experiment crate::engine::slotmap tabulates. Confining reclamation to a moment when nothing else is running keeps all of it, so call it during a quiet one.

let mut db = Database::open(Config::in_memory())?;
db.register::<Session>()?;

db.transaction(|tx| tx.insert(Session { id: 1, token: 42 }))?;
db.transaction(|tx| tx.delete::<Session>(&1))?;

// No transaction may be alive here — the borrow checker enforces it.
let freed = db.compact();

Only keys whose records are gone are reclaimed — deleted, and their tombstone already collected. A record that merely has not been touched in a while is untouched. If a delete is very recent its tombstone may still be live, in which case that key is reclaimed by a later call rather than this one.

It also drops secondary index entries that can no longer resolve. Those were never unsafe — an index holds primary keys, not slot pointers — but they are dead weight in memory and in every scan that walks them.

pub fn register<T: Versioned>(&self) -> Result<()>

Register a type. Assigns its [TableId] and builds its indexes.

Must happen before any transaction touches T. Registering twice is a no-op rather than an error, so a library that registers its own types defensively does not conflict with an application that did the same.

Examples found in repository

examples/basic.rs (line 26)

fn main() -> Result<()> {
    let db = Database::open(Config::in_memory())?;
    db.register::<Account>()?;

    // ---- insert -----------------------------------------------------------
    // `transaction` runs the closure, commits it, and retries it if it hits a
    // retriable conflict. Snapshot isolation by default.
    db.transaction(|tx| {
        tx.insert(Account {
            id: 1,
            owner: "ada".into(),
            branch: 10,
            balance: 500,
        })?;
        tx.insert(Account {
            id: 2,
            owner: "bob".into(),
            branch: 10,
            balance: 250,
        })?;
        tx.insert(Account {
            id: 3,
            owner: "cleo".into(),
            branch: 20,
            balance: 900,
        })?;
        Ok(())
    })?;

    // ---- read -------------------------------------------------------------
    let mut tx = db.begin();
    let ada = tx.get::<Account>(&1)?.expect("just inserted");
    println!("ada: branch {}, balance {}", ada.branch, ada.balance);

    // A read-only transaction has nothing to commit; dropping it rolls back,
    // which for a reader means simply releasing its snapshot.
    drop(tx);

    // ---- update -----------------------------------------------------------
    db.transaction(|tx| {
        tx.update::<Account>(&1, |a| a.balance -= 100)?;
        tx.update::<Account>(&2, |a| a.balance += 100)?;
        Ok(())
    })?;

    // ---- scan by primary key ----------------------------------------------
    let mut tx = db.begin();
    println!("\nall accounts:");
    for account in tx.scan::<Account>()? {
        println!(
            "  {:>4}  {:<6} branch {}  {:>5}",
            account.id, account.owner, account.branch, account.balance
        );
    }

    // ---- scan by secondary index ------------------------------------------
    println!("\nbranch 10:");
    for account in tx.scan_index(Account::BRANCH, 10u32..=10)? {
        println!("  {} ({})", account.owner, account.balance);
    }
    drop(tx);

    // ---- constraint violations --------------------------------------------
    let mut tx = db.begin();
    let duplicate = tx.insert(Account {
        id: 99,
        owner: "ada".into(),
        branch: 30,
        balance: 0,
    });
    println!("\nreusing owner 'ada': {}", duplicate.unwrap_err());
    drop(tx);

    // ---- delete -----------------------------------------------------------
    db.transaction(|tx| {
        let existed = tx.delete::<Account>(&3)?;
        println!("deleted cleo: {existed}");
        Ok(())
    })?;

    let mut tx = db.begin();
    println!(
        "cleo now: {:?}",
        tx.get::<Account>(&3)?.map(|r| r.to_owned())
    );

    Ok(())
}

examples/concurrent.rs (line 31)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Account>()?;

    db.transaction(|tx| {
        for id in 0..ACCOUNTS {
            tx.insert(Account {
                id,
                balance: STARTING_BALANCE,
            })?;
        }
        Ok(())
    })?;

    let total_before = total(&db)?;
    println!("starting total: {total_before}");

    let stop = Arc::new(AtomicBool::new(false));
    let reads = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));

    // ---- a reader that must never observe a torn transfer -----------------
    // Money moves between accounts constantly. Under snapshot isolation this
    // thread's scans must always sum to the same total: it either sees both
    // halves of a transfer or neither.
    let auditor = {
        let db = Arc::clone(&db);
        let stop = Arc::clone(&stop);
        let reads = Arc::clone(&reads);
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let sum: i64 = tx.scan::<Account>()?.iter().map(|a| a.balance).sum();
                assert_eq!(
                    sum,
                    ACCOUNTS as i64 * STARTING_BALANCE,
                    "a reader observed a partially applied transfer"
                );
                reads.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    // ---- writers ----------------------------------------------------------
    let start = Instant::now();
    let mut writers = Vec::new();
    for t in 0..TRANSFER_THREADS {
        let db = Arc::clone(&db);
        let retries = Arc::clone(&retries);
        writers.push(thread::spawn(move || -> Result<()> {
            // A cheap per-thread PRNG; no dependency needed for this.
            let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t as u64 + 1);
            let mut next = move || {
                seed ^= seed << 13;
                seed ^= seed >> 7;
                seed ^= seed << 17;
                seed
            };

            for _ in 0..TRANSFERS_PER_THREAD {
                let from = next() % ACCOUNTS;
                let to = next() % ACCOUNTS;
                if from == to {
                    continue;
                }

                let mut attempts = 0u64;
                // Serializable, because the transfer's *write* depends on a
                // balance it *read* — the write-skew shape. Snapshot isolation
                // would let two concurrent transfers both pass the check.
                db.transaction_with::<Serializable, _, _>(|tx| {
                    attempts += 1;
                    let balance = tx.get::<Account>(&from)?.map(|a| a.balance).unwrap_or(0);
                    if balance < 10 {
                        return Ok(());
                    }
                    tx.update::<Account>(&from, |a| a.balance -= 10)?;
                    tx.update::<Account>(&to, |a| a.balance += 10)?;
                    Ok(())
                })?;
                retries.fetch_add(attempts - 1, Ordering::Relaxed);
            }
            Ok(())
        }));
    }

    for w in writers {
        w.join().expect("writer panicked")?;
    }
    let elapsed = start.elapsed();

    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    // ---- results ----------------------------------------------------------
    let total_after = total(&db)?;
    let committed = (TRANSFER_THREADS * TRANSFERS_PER_THREAD) as u64;

    println!("final total:    {total_after}");
    assert_eq!(total_before, total_after, "money was created or destroyed");

    println!("\n{committed} transfers in {elapsed:.2?}");
    println!(
        "  {:.0} transfers/sec",
        committed as f64 / elapsed.as_secs_f64()
    );
    println!(
        "  {} concurrent audit scans completed, none of them blocked",
        reads.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.1} per transfer)",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / committed as f64
    );

    // The GC watermark should be advancing. If `active_transactions` climbs and
    // the watermark stalls, some transaction is being leaked — see engine::gc.
    let stats = db.stats();
    println!(
        "\ngc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );

    println!("\n✓ no torn reads, no lost updates, conservation held");

    // Slow readers are what pin the GC watermark. Left here as the shape of the
    // thing to watch for, not as something this example demonstrates going wrong.
    thread::sleep(Duration::from_millis(1));
    Ok(())
}

examples/isolation.rs (line 34)

fn main() -> Result<()> {
    let db = Database::open(Config::in_memory())?;
    db.register::<Doctor>()?;
    db.register::<Counter>()?;

    db.transaction(|tx| {
        tx.insert(Doctor {
            id: 1,
            name: "ada".into(),
            on_call: true,
        })?;
        tx.insert(Doctor {
            id: 2,
            name: "bob".into(),
            on_call: true,
        })?;
        tx.insert(Counter { id: 1, value: 0 })
    })?;

    // ------------------------------------------------------------------
    banner("1. Snapshot: a reader is unaffected by concurrent commits");
    {
        let mut reader = db.begin_with::<Snapshot>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        // Someone else commits, start to finish, while `reader` is open.
        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 42).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_eq!(before, after, "snapshot isolation must be stable");
        println!("  ✓ the snapshot held: readers never block and never change under you");
    }

    // ------------------------------------------------------------------
    banner("2. ReadCommitted: each statement sees a fresh snapshot");
    {
        let mut reader = db.begin_with::<ReadCommitted>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 99).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_ne!(
            before, after,
            "read committed should observe the new commit"
        );
        println!("  ✓ non-repeatable read — the tradeoff this level makes for cheapness");
    }

    // ------------------------------------------------------------------
    banner("3. Write-write conflict: first committer wins");
    {
        let mut first = db.begin_with::<Snapshot>();
        let mut second = db.begin_with::<Snapshot>();

        first.update::<Counter>(&1, |c| c.value += 1)?;
        first.commit()?;

        // `second` took its snapshot before `first` committed, so writing here
        // would silently drop `first`'s update.
        let result = second.update::<Counter>(&1, |c| c.value += 1);
        match result {
            Err(e) => println!(
                "  second transaction: {e}  (retriable: {})",
                e.is_retriable()
            ),
            Ok(_) => unreachable!("the stale write should have been rejected"),
        }
        println!("  ✓ no lost update");
    }

    // ------------------------------------------------------------------
    banner("4. Write skew: allowed under Snapshot");
    {
        // The rule: at least one doctor must stay on call. Each transaction
        // checks the rule, sees it satisfied, and takes a *different* doctor
        // off call — so there is no write-write conflict to catch them.
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();

        let t1_sees = t1.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t1.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let t2_sees = t2.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t2.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let names: Vec<String> = {
            let mut tx = db.begin();
            tx.scan::<Doctor>()?
                .iter()
                .map(|d| d.name.clone())
                .collect()
        };
        println!("  on the rota: {}", names.join(", "));
        println!("  t1 counts {t1_sees} on call, t2 counts {t2_sees} — both think it is safe");

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;
        t1.commit()?;
        t2.commit()?;

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✗ {remaining} doctors on call — the invariant is broken");
        println!("    Both transactions were individually legal. This is write skew,");
        println!("    and it is why Snapshot is not Serializable.");
    }

    // ------------------------------------------------------------------
    banner("5. Write skew: prevented under Serializable");
    {
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();

        // Both read both rows, exactly as before.
        let _ = t1.get::<Doctor>(&1)?;
        let _ = t1.get::<Doctor>(&2)?;
        let _ = t2.get::<Doctor>(&1)?;
        let _ = t2.get::<Doctor>(&2)?;

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;

        t1.commit()?;
        match t2.commit() {
            Err(e) => println!("  t2: {e}  (retriable: {})", e.is_retriable()),
            Ok(()) => unreachable!("t2 read a row t1 changed; it must not commit"),
        }

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✓ {remaining} doctor still on call — the invariant held");
        println!("    t2 read doctor 1, which t1 changed, so t2 could not be serialized.");
    }

    // ------------------------------------------------------------------
    banner("6. Retries are the normal way to use Serializable");
    {
        // `transaction_with` reruns the closure on a retriable failure, so the
        // abort in scenario 5 becomes invisible to the caller.
        let mut attempts = 0;
        db.transaction_with::<Serializable, _, _>(|tx| {
            attempts += 1;
            let ada = tx.get::<Doctor>(&1)?.unwrap().on_call;
            let bob = tx.get::<Doctor>(&2)?.unwrap().on_call;
            if ada && bob {
                tx.update::<Doctor>(&1, |d| d.on_call = false)?;
            }
            Ok(())
        })?;
        println!("  committed after {attempts} attempt(s)");
        println!("  ✓ write your transaction as if it runs alone; let the engine retry it");
    }

    banner("Summary");
    println!("  ReadCommitted   cheapest; non-repeatable reads and phantoms");
    println!("  RepeatableRead  stable snapshot; write skew possible");
    println!("  Snapshot        stable snapshot; write skew possible   ← default");
    println!("  Serializable    no anomalies; expect retriable aborts");

    Ok(())
}

examples/game.rs (line 143)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Hero>()?;
    db.register::<Item>()?;
    db.register::<PartyMember>()?;

    let (ada, bram, cleo) = (1u64, 2u64, 3u64);

    // -----------------------------------------------------------------------
    act(1, "The world is created");

    db.transaction(|tx| {
        tx.insert(Hero {
            id: ada,
            name: "Ada".into(),
            gold: 120,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: bram,
            name: "Bram".into(),
            gold: 80,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: cleo,
            name: "Cleo".into(),
            gold: 200,
            strength: 60,
        })?;

        let mut item = |id, owner, name: &str, weight, value, kind| {
            tx.insert(Item {
                id,
                owner,
                name: name.into(),
                weight,
                value,
                kind,
                flavour: HashMap::new(),
            })
        };
        item(10, ada, "Rusty Sword", 30, 15, Kind::Weapon)?;
        item(11, ada, "Health Potion", 5, 20, Kind::Potion)?;
        item(12, bram, "Oak Shield", 40, 35, Kind::Weapon)?;
        item(13, cleo, "Lucky Charm", 2, 90, Kind::Trinket)?;
        // Loot lying in the dungeon, owned by nobody.
        item(20, GROUND, "Flaming Greatsword", 55, 500, Kind::Weapon)?;
        item(21, GROUND, "Elven Cloak", 20, 240, Kind::Trinket)?;
        item(22, GROUND, "Iron Helm", 35, 60, Kind::Weapon)?;

        // Arbitrary per-item state, in a database row, with no schema to
        // declare and no serialisation format to agree on.
        tx.update::<Item>(&20, |i| {
            i.flavour.insert("enchantment".into(), "flame".into());
            i.flavour.insert("forged_by".into(), "Durin".into());
        })?;
        Ok(())
    })?;

    // A whole transaction that fails partway leaves nothing behind. The unique
    // index on `name` refuses the second Ada, and the gold change goes with it.
    let before = gold_of(&mut db.begin(), ada)?;
    let doomed = db.transaction(|tx| {
        tx.update::<Hero>(&ada, |h| h.gold += 10_000)?;
        tx.insert(Hero {
            id: 99,
            name: "Ada".into(),
            gold: 0,
            strength: 10,
        })
    });
    println!("  a duplicate hero name: {}", doomed.unwrap_err());
    println!(
        "  Ada's gold is still {} — the whole transaction rolled back, not just the insert",
        gold_of(&mut db.begin(), ada)?
    );
    assert_eq!(gold_of(&mut db.begin(), ada)?, before);

    let mut tx = db.begin();
    for hero in [ada, bram, cleo] {
        let name = name_of(&mut tx, hero)?;
        let carried = carried(&mut tx, hero)?;
        let items = inventory(&mut tx, hero)?;
        let names: Vec<_> = items.iter().map(|(_, n, _)| n.as_str()).collect();
        println!("  {name:<5} carries {carried:>3} — {}", names.join(", "));
    }
    drop(tx);

    // -----------------------------------------------------------------------
    act(2, "Two heroes reach for the same greatsword");

    // Both begin, both see the sword unclaimed. The first to write takes it;
    // the second is told immediately rather than being made to wait.
    let mut ada_grabs = db.begin_with::<Snapshot>();
    let mut bram_grabs = db.begin_with::<Snapshot>();

    ada_grabs.update::<Item>(&20, |i| i.owner = ada)?;

    match bram_grabs.update::<Item>(&20, |i| i.owner = bram) {
        Err(e @ Error::WriteConflict { .. }) => {
            println!("  Bram: {e}");
            println!(
                "  ...retriable: {} — he can try for something else",
                e.is_retriable()
            );
        }
        other => unreachable!("expected a conflict, got {other:?}"),
    }

    // A failed write does not end the transaction. Bram takes the cloak instead.
    bram_grabs.update::<Item>(&21, |i| i.owner = bram)?;
    ada_grabs.commit()?;
    bram_grabs.commit()?;
    println!("  Ada takes the greatsword, Bram takes the cloak. Nobody blocked.");

    // -----------------------------------------------------------------------
    act(3, "Cleo overloads herself — write skew, and the fix");

    // Cleo can carry 60 and is carrying 2. Either 35kg helm fits; both do not.
    // Each transaction reads her inventory, checks the total, and picks up a
    // *different* item — so the two writes never touch the same record, and
    // first-updater-wins has nothing to catch.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.insert(Item {
            id: 23,
            owner: GROUND,
            name: "Steel Helm".into(),
            weight: 35,
            value: 70,
            kind: Kind::Weapon,
            flavour: HashMap::new(),
        })
    })?;

    let pick_up = |item: u64| {
        move |tx: &mut mvcc::Transaction<'_, Serializable>| -> Result<bool> {
            let hero = tx.get::<Hero>(&cleo)?.expect("Cleo exists");
            let strength = hero.strength;
            let load = carried(tx, cleo)?;
            let weight = tx.get::<Item>(&item)?.map(|i| i.weight).unwrap_or(0);
            if load + weight > strength {
                return Ok(false);
            }
            tx.update::<Item>(&item, |i| i.owner = cleo)?;
            Ok(true)
        }
    };

    // Under Snapshot, both checks pass against a stale inventory.
    {
        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();
        let (load1, load2) = (carried(&mut t1, cleo)?, carried(&mut t2, cleo)?);
        println!(
            "  Snapshot:     both transactions see {load1}/{load2} carried, both think a 35kg helm fits"
        );
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        t2.commit()?;
        let over = carried(&mut db.begin(), cleo)?;
        println!(
            "  ✗ Cleo now carries {over} of a possible 60. Two legal transactions, one broken rule."
        );
    }

    // Put it back and try again at Serializable.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.update::<Item>(&23, |i| i.owner = GROUND).map(|_| ())
    })?;

    {
        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();
        // Both read the inventory, as before.
        let _ = carried(&mut t1, cleo)?;
        let _ = carried(&mut t2, cleo)?;
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        match t2.commit() {
            Err(e @ Error::SerializationFailure) => println!("  Serializable: second pickup {e}"),
            other => unreachable!("expected a serialization failure, got {other:?}"),
        }
        println!(
            "  ✓ Cleo carries {} — the second pickup read an inventory the first changed.",
            carried(&mut db.begin(), cleo)?
        );
    }

    // In real code you would not hand-roll that. `transaction_with` retries,
    // and the retry re-reads the inventory and correctly declines.
    let took_it = db.transaction_with::<Serializable, _, _>(pick_up(23))?;
    println!("  ...and on retry the second helm is refused on its merits: picked_up = {took_it}");

    // -----------------------------------------------------------------------
    act(4, "The party fills up — a phantom, not a conflict");

    db.transaction(|tx| {
        tx.insert(PartyMember {
            hero: ada,
            party: PARTY,
            role: "Vanguard".into(),
        })?;
        tx.insert(PartyMember {
            hero: bram,
            party: PARTY,
            role: "Shield".into(),
        })?;
        tx.insert(PartyMember {
            hero: cleo,
            party: PARTY,
            role: "Scout".into(),
        })
    })?;

    // Two newcomers apply at once, with three of four slots taken. Neither
    // writes a row the other wrote — they insert *different* rows — so there is
    // no write conflict to detect. What they collide on is a row that did not
    // exist when either of them counted.
    let (dara, finn) = (4u64, 5u64);
    db.transaction(|tx| {
        tx.insert(Hero {
            id: dara,
            name: "Dara".into(),
            gold: 40,
            strength: 80,
        })?;
        tx.insert(Hero {
            id: finn,
            name: "Finn".into(),
            gold: 40,
            strength: 80,
        })
    })?;

    let mut d = db.begin_with::<Serializable>();
    let mut f = db.begin_with::<Serializable>();
    println!(
        "  Dara counts {} members, Finn counts {} — both see a free slot",
        party_size(&mut d)?,
        party_size(&mut f)?
    );
    d.insert(PartyMember {
        hero: dara,
        party: PARTY,
        role: "Healer".into(),
    })?;
    f.insert(PartyMember {
        hero: finn,
        party: PARTY,
        role: "Healer".into(),
    })?;
    d.commit()?;
    match f.commit() {
        Err(e @ Error::SerializationFailure) => println!("  Finn: {e}"),
        other => unreachable!("expected a serialization failure, got {other:?}"),
    }
    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan_where::<PartyMember, _>(|m| m.party == PARTY)?
        .iter()
        .map(|m| (m.hero, m.role.clone()))
        .collect();
    drop(tx);
    roster.sort();
    let size = roster.len();
    let mut tx = db.begin();
    let listed: Vec<String> = roster
        .iter()
        .map(|(h, role)| Ok(format!("{} the {role}", name_of(&mut tx, *h)?)))
        .collect::<Result<_>>()?;
    drop(tx);
    println!("  party: {}", listed.join(", "));
    println!("  ✓ {size}/{PARTY_LIMIT} filled. Finn's insert became a phantom in Dara's count.");
    println!("    (`scan_where` hands the engine the predicate, so it can re-check it at commit.)");
    assert!(size <= PARTY_LIMIT);

    // -----------------------------------------------------------------------
    act(5, "A trade — two records, one atomic step");

    let before = total_gold(&db)?;

    let trade = |seller: u64, buyer: u64, item: u64, price: i64| {
        let db = Arc::clone(&db);
        move || -> Result<bool> {
            db.transaction_with::<Serializable, _, _>(|tx| {
                let buyer_gold = gold_of(tx, buyer)?;
                if buyer_gold < price {
                    return Ok(false);
                }
                let owned_by_seller = tx
                    .get::<Item>(&item)?
                    .map(|i| i.owner == seller)
                    .unwrap_or(false);
                if !owned_by_seller {
                    return Ok(false);
                }
                tx.update::<Hero>(&buyer, |h| h.gold -= price)?;
                tx.update::<Hero>(&seller, |h| h.gold += price)?;
                tx.update::<Item>(&item, |i| i.owner = buyer)?;
                Ok(true)
            })
        }
    };

    let sold = trade(bram, cleo, 21, 150)()?;
    println!("  Bram sells the Elven Cloak to Cleo for 150g: {sold}");
    let mut tx = db.begin();
    let cloak_owner = tx.get::<Item>(&21)?.expect("cloak").owner;
    println!(
        "  Bram {}g, Cleo {}g, cloak owner is {}",
        gold_of(&mut tx, bram)?,
        gold_of(&mut tx, cleo)?,
        name_of(&mut tx, cloak_owner)?
    );
    drop(tx);
    assert_eq!(total_gold(&db)?, before, "gold must be conserved");
    println!("  ✓ total gold unchanged at {before} — no step of that was separately visible");

    // -----------------------------------------------------------------------
    act(6, "The chronicler reads while the world moves");

    // A long report scanning the whole world while combat rewrites it. It never
    // blocks a writer, and its view never shifts underneath it.
    let mut chronicler = db.begin_with::<Snapshot>();
    let opening = total_gold(&db)?;
    let seen_first = chronicler.scan::<Hero>()?.len();

    for _ in 0..50 {
        db.transaction(|tx| {
            tx.update::<Hero>(&ada, |h| h.gold += 1)?;
            tx.update::<Hero>(&bram, |h| h.gold -= 1)?;
            Ok(())
        })?;
    }
    db.transaction(|tx| {
        tx.insert(Hero {
            id: 6,
            name: "Mira".into(),
            gold: 0,
            strength: 50,
        })
    })?;

    let chron_gold: i64 = chronicler.scan::<Hero>()?.iter().map(|h| h.gold).sum();
    let seen_after = chronicler.scan::<Hero>()?.len();
    let in_world = db.begin().scan::<Hero>()?.len();
    println!("  chronicler saw {seen_first} heroes at the start and still sees {seen_after};");
    println!("  the world now holds {in_world}, after 50 commits and one new arrival");
    println!(
        "  chronicler totals {chron_gold}g; the world now totals {}g",
        total_gold(&db)?
    );
    assert_eq!(chron_gold, opening, "a snapshot must not move");
    chronicler.commit()?;
    println!("  ✓ a reader that ran across 51 commits saw exactly one consistent world");

    // -----------------------------------------------------------------------
    act(7, "The raid — many adventurers at once");

    let stop = Arc::new(AtomicBool::new(false));
    let trades = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));
    let audits = Arc::new(AtomicU64::new(0));
    let opening = total_gold(&db)?;
    let heroes = [ada, bram, cleo, dara, finn];

    // An auditor that must never catch a trade half-applied.
    let auditor = {
        let (db, stop, audits) = (Arc::clone(&db), Arc::clone(&stop), Arc::clone(&audits));
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let total: i64 = tx.scan::<Hero>()?.iter().map(|h| h.gold).sum();
                assert_eq!(total, opening, "auditor saw a half-finished trade");
                audits.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    let start = Instant::now();
    let raiders: Vec<_> = (0..4)
        .map(|t| {
            let (db, trades, retries) =
                (Arc::clone(&db), Arc::clone(&trades), Arc::clone(&retries));
            thread::spawn(move || -> Result<()> {
                let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t + 1);
                for _ in 0..400 {
                    seed ^= seed << 13;
                    seed ^= seed >> 7;
                    seed ^= seed << 17;
                    let from = heroes[(seed % heroes.len() as u64) as usize];
                    let to = heroes[((seed >> 8) % heroes.len() as u64) as usize];
                    if from == to {
                        continue;
                    }

                    let mut attempts = 0u64;
                    // Serializable, because the payment depends on a balance we
                    // read — the same shape as Cleo's carry limit in Act 3.
                    db.transaction_with::<Serializable, _, _>(|tx| {
                        attempts += 1;
                        let purse = gold_of(tx, from)?;
                        if purse < 5 {
                            return Ok(());
                        }
                        tx.update::<Hero>(&from, |h| h.gold -= 5)?;
                        tx.update::<Hero>(&to, |h| h.gold += 5)?;
                        Ok(())
                    })?;
                    trades.fetch_add(1, Ordering::Relaxed);
                    retries.fetch_add(attempts - 1, Ordering::Relaxed);
                }
                Ok(())
            })
        })
        .collect();

    for r in raiders {
        r.join().expect("raider panicked")?;
    }
    let elapsed = start.elapsed();
    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    let done = trades.load(Ordering::Relaxed);
    println!("  {done} trades across 4 threads in {elapsed:.2?}");
    println!(
        "  {} audit sweeps completed alongside them, none blocked, none torn",
        audits.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.2} per trade) — contention on 5 heroes, handled by the engine",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / done.max(1) as f64
    );
    assert_eq!(total_gold(&db)?, opening, "gold was created or destroyed");
    println!("  ✓ total gold still {opening}");

    // -----------------------------------------------------------------------
    act(8, "Closing the ledger");

    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan::<Hero>()?
        .iter()
        .map(|h| (h.id, h.name.clone(), h.gold))
        .collect();
    roster.sort_by_key(|(id, _, _)| *id);
    drop(tx);

    println!(
        "  {:<6} {:>6} {:>5} {:>7}  inventory",
        "hero", "gold", "load", "worth"
    );
    let mut tx = db.begin();
    for (id, name, gold) in &roster {
        let held = tx.scan_index(Item::OWNER, *id..=*id)?;
        let load: i32 = held.iter().map(|i| i.weight).sum();
        let worth: i64 = held.iter().map(|i| i.value).sum();
        let names: Vec<String> = held
            .iter()
            .map(|i| match i.kind {
                Kind::Weapon => format!("[wpn] {}", i.name),
                Kind::Potion => format!("[pot] {}", i.name),
                Kind::Trinket => format!("[trk] {}", i.name),
            })
            .collect();
        let names = if names.is_empty() {
            "—".to_string()
        } else {
            names.join(", ")
        };
        println!("  {name:<6} {gold:>5}g {load:>4}kg {worth:>6}g  {names}");
    }

    // Potions anywhere in the world, found by predicate rather than by key —
    // and `kind` is an ordinary Rust enum the engine knows nothing about.
    let potions = tx.scan_where::<Item, _>(|i| i.kind == Kind::Potion)?.len();
    let sword = tx.get::<Item>(&20)?.expect("greatsword");
    let enchantment = sword
        .flavour
        .get("enchantment")
        .cloned()
        .unwrap_or_default();
    let forged_by = sword.flavour.get("forged_by").cloned().unwrap_or_default();
    drop(tx);
    // The rule Act 3 was about, checked rather than asserted in prose.
    let mut tx = db.begin();
    for (id, name, _) in &roster {
        let strength = tx.get::<Hero>(id)?.map(|h| h.strength).unwrap_or(0);
        let load = carried(&mut tx, *id)?;
        assert!(
            load <= strength,
            "{name} is over their carry limit: {load} > {strength}"
        );
    }
    drop(tx);

    println!("\n  potions in the world: {potions}");
    println!("  the greatsword is {enchantment}-enchanted, forged by {forged_by}");

    // Nothing is reclaimed while a transaction that could still reach it is
    // alive. This is the number to watch in a long-lived process: a forgotten
    // transaction pins it and version chains grow without limit.
    let stats = db.stats();
    println!(
        "\n  gc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );
    println!("  (a watermark that stops moving while writes continue is the leak to look for)");

    Ok(())
}

pub fn begin(&self) -> Transaction<'_, SnapshotLevel>

Begin a transaction at the default isolation level (snapshot isolation).

Examples found in repository

examples/game.rs (line 129)

fn total_gold(db: &Database) -> Result<i64> {
    let mut tx = db.begin();
    Ok(tx.scan::<Hero>()?.iter().map(|h| h.gold).sum())
}

fn act(n: u32, title: &str) {
    println!("\n\x1b[1m── Act {n}. {title}\x1b[0m");
}

// A narrative example: one linear script of acts, read top to bottom. Splitting
// it into helpers to satisfy the length and complexity budgets would make it
// harder to follow, which is the only thing this file is for.
#[allow(clippy::too_many_lines, clippy::cognitive_complexity)]
fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Hero>()?;
    db.register::<Item>()?;
    db.register::<PartyMember>()?;

    let (ada, bram, cleo) = (1u64, 2u64, 3u64);

    // -----------------------------------------------------------------------
    act(1, "The world is created");

    db.transaction(|tx| {
        tx.insert(Hero {
            id: ada,
            name: "Ada".into(),
            gold: 120,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: bram,
            name: "Bram".into(),
            gold: 80,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: cleo,
            name: "Cleo".into(),
            gold: 200,
            strength: 60,
        })?;

        let mut item = |id, owner, name: &str, weight, value, kind| {
            tx.insert(Item {
                id,
                owner,
                name: name.into(),
                weight,
                value,
                kind,
                flavour: HashMap::new(),
            })
        };
        item(10, ada, "Rusty Sword", 30, 15, Kind::Weapon)?;
        item(11, ada, "Health Potion", 5, 20, Kind::Potion)?;
        item(12, bram, "Oak Shield", 40, 35, Kind::Weapon)?;
        item(13, cleo, "Lucky Charm", 2, 90, Kind::Trinket)?;
        // Loot lying in the dungeon, owned by nobody.
        item(20, GROUND, "Flaming Greatsword", 55, 500, Kind::Weapon)?;
        item(21, GROUND, "Elven Cloak", 20, 240, Kind::Trinket)?;
        item(22, GROUND, "Iron Helm", 35, 60, Kind::Weapon)?;

        // Arbitrary per-item state, in a database row, with no schema to
        // declare and no serialisation format to agree on.
        tx.update::<Item>(&20, |i| {
            i.flavour.insert("enchantment".into(), "flame".into());
            i.flavour.insert("forged_by".into(), "Durin".into());
        })?;
        Ok(())
    })?;

    // A whole transaction that fails partway leaves nothing behind. The unique
    // index on `name` refuses the second Ada, and the gold change goes with it.
    let before = gold_of(&mut db.begin(), ada)?;
    let doomed = db.transaction(|tx| {
        tx.update::<Hero>(&ada, |h| h.gold += 10_000)?;
        tx.insert(Hero {
            id: 99,
            name: "Ada".into(),
            gold: 0,
            strength: 10,
        })
    });
    println!("  a duplicate hero name: {}", doomed.unwrap_err());
    println!(
        "  Ada's gold is still {} — the whole transaction rolled back, not just the insert",
        gold_of(&mut db.begin(), ada)?
    );
    assert_eq!(gold_of(&mut db.begin(), ada)?, before);

    let mut tx = db.begin();
    for hero in [ada, bram, cleo] {
        let name = name_of(&mut tx, hero)?;
        let carried = carried(&mut tx, hero)?;
        let items = inventory(&mut tx, hero)?;
        let names: Vec<_> = items.iter().map(|(_, n, _)| n.as_str()).collect();
        println!("  {name:<5} carries {carried:>3} — {}", names.join(", "));
    }
    drop(tx);

    // -----------------------------------------------------------------------
    act(2, "Two heroes reach for the same greatsword");

    // Both begin, both see the sword unclaimed. The first to write takes it;
    // the second is told immediately rather than being made to wait.
    let mut ada_grabs = db.begin_with::<Snapshot>();
    let mut bram_grabs = db.begin_with::<Snapshot>();

    ada_grabs.update::<Item>(&20, |i| i.owner = ada)?;

    match bram_grabs.update::<Item>(&20, |i| i.owner = bram) {
        Err(e @ Error::WriteConflict { .. }) => {
            println!("  Bram: {e}");
            println!(
                "  ...retriable: {} — he can try for something else",
                e.is_retriable()
            );
        }
        other => unreachable!("expected a conflict, got {other:?}"),
    }

    // A failed write does not end the transaction. Bram takes the cloak instead.
    bram_grabs.update::<Item>(&21, |i| i.owner = bram)?;
    ada_grabs.commit()?;
    bram_grabs.commit()?;
    println!("  Ada takes the greatsword, Bram takes the cloak. Nobody blocked.");

    // -----------------------------------------------------------------------
    act(3, "Cleo overloads herself — write skew, and the fix");

    // Cleo can carry 60 and is carrying 2. Either 35kg helm fits; both do not.
    // Each transaction reads her inventory, checks the total, and picks up a
    // *different* item — so the two writes never touch the same record, and
    // first-updater-wins has nothing to catch.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.insert(Item {
            id: 23,
            owner: GROUND,
            name: "Steel Helm".into(),
            weight: 35,
            value: 70,
            kind: Kind::Weapon,
            flavour: HashMap::new(),
        })
    })?;

    let pick_up = |item: u64| {
        move |tx: &mut mvcc::Transaction<'_, Serializable>| -> Result<bool> {
            let hero = tx.get::<Hero>(&cleo)?.expect("Cleo exists");
            let strength = hero.strength;
            let load = carried(tx, cleo)?;
            let weight = tx.get::<Item>(&item)?.map(|i| i.weight).unwrap_or(0);
            if load + weight > strength {
                return Ok(false);
            }
            tx.update::<Item>(&item, |i| i.owner = cleo)?;
            Ok(true)
        }
    };

    // Under Snapshot, both checks pass against a stale inventory.
    {
        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();
        let (load1, load2) = (carried(&mut t1, cleo)?, carried(&mut t2, cleo)?);
        println!(
            "  Snapshot:     both transactions see {load1}/{load2} carried, both think a 35kg helm fits"
        );
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        t2.commit()?;
        let over = carried(&mut db.begin(), cleo)?;
        println!(
            "  ✗ Cleo now carries {over} of a possible 60. Two legal transactions, one broken rule."
        );
    }

    // Put it back and try again at Serializable.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.update::<Item>(&23, |i| i.owner = GROUND).map(|_| ())
    })?;

    {
        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();
        // Both read the inventory, as before.
        let _ = carried(&mut t1, cleo)?;
        let _ = carried(&mut t2, cleo)?;
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        match t2.commit() {
            Err(e @ Error::SerializationFailure) => println!("  Serializable: second pickup {e}"),
            other => unreachable!("expected a serialization failure, got {other:?}"),
        }
        println!(
            "  ✓ Cleo carries {} — the second pickup read an inventory the first changed.",
            carried(&mut db.begin(), cleo)?
        );
    }

    // In real code you would not hand-roll that. `transaction_with` retries,
    // and the retry re-reads the inventory and correctly declines.
    let took_it = db.transaction_with::<Serializable, _, _>(pick_up(23))?;
    println!("  ...and on retry the second helm is refused on its merits: picked_up = {took_it}");

    // -----------------------------------------------------------------------
    act(4, "The party fills up — a phantom, not a conflict");

    db.transaction(|tx| {
        tx.insert(PartyMember {
            hero: ada,
            party: PARTY,
            role: "Vanguard".into(),
        })?;
        tx.insert(PartyMember {
            hero: bram,
            party: PARTY,
            role: "Shield".into(),
        })?;
        tx.insert(PartyMember {
            hero: cleo,
            party: PARTY,
            role: "Scout".into(),
        })
    })?;

    // Two newcomers apply at once, with three of four slots taken. Neither
    // writes a row the other wrote — they insert *different* rows — so there is
    // no write conflict to detect. What they collide on is a row that did not
    // exist when either of them counted.
    let (dara, finn) = (4u64, 5u64);
    db.transaction(|tx| {
        tx.insert(Hero {
            id: dara,
            name: "Dara".into(),
            gold: 40,
            strength: 80,
        })?;
        tx.insert(Hero {
            id: finn,
            name: "Finn".into(),
            gold: 40,
            strength: 80,
        })
    })?;

    let mut d = db.begin_with::<Serializable>();
    let mut f = db.begin_with::<Serializable>();
    println!(
        "  Dara counts {} members, Finn counts {} — both see a free slot",
        party_size(&mut d)?,
        party_size(&mut f)?
    );
    d.insert(PartyMember {
        hero: dara,
        party: PARTY,
        role: "Healer".into(),
    })?;
    f.insert(PartyMember {
        hero: finn,
        party: PARTY,
        role: "Healer".into(),
    })?;
    d.commit()?;
    match f.commit() {
        Err(e @ Error::SerializationFailure) => println!("  Finn: {e}"),
        other => unreachable!("expected a serialization failure, got {other:?}"),
    }
    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan_where::<PartyMember, _>(|m| m.party == PARTY)?
        .iter()
        .map(|m| (m.hero, m.role.clone()))
        .collect();
    drop(tx);
    roster.sort();
    let size = roster.len();
    let mut tx = db.begin();
    let listed: Vec<String> = roster
        .iter()
        .map(|(h, role)| Ok(format!("{} the {role}", name_of(&mut tx, *h)?)))
        .collect::<Result<_>>()?;
    drop(tx);
    println!("  party: {}", listed.join(", "));
    println!("  ✓ {size}/{PARTY_LIMIT} filled. Finn's insert became a phantom in Dara's count.");
    println!("    (`scan_where` hands the engine the predicate, so it can re-check it at commit.)");
    assert!(size <= PARTY_LIMIT);

    // -----------------------------------------------------------------------
    act(5, "A trade — two records, one atomic step");

    let before = total_gold(&db)?;

    let trade = |seller: u64, buyer: u64, item: u64, price: i64| {
        let db = Arc::clone(&db);
        move || -> Result<bool> {
            db.transaction_with::<Serializable, _, _>(|tx| {
                let buyer_gold = gold_of(tx, buyer)?;
                if buyer_gold < price {
                    return Ok(false);
                }
                let owned_by_seller = tx
                    .get::<Item>(&item)?
                    .map(|i| i.owner == seller)
                    .unwrap_or(false);
                if !owned_by_seller {
                    return Ok(false);
                }
                tx.update::<Hero>(&buyer, |h| h.gold -= price)?;
                tx.update::<Hero>(&seller, |h| h.gold += price)?;
                tx.update::<Item>(&item, |i| i.owner = buyer)?;
                Ok(true)
            })
        }
    };

    let sold = trade(bram, cleo, 21, 150)()?;
    println!("  Bram sells the Elven Cloak to Cleo for 150g: {sold}");
    let mut tx = db.begin();
    let cloak_owner = tx.get::<Item>(&21)?.expect("cloak").owner;
    println!(
        "  Bram {}g, Cleo {}g, cloak owner is {}",
        gold_of(&mut tx, bram)?,
        gold_of(&mut tx, cleo)?,
        name_of(&mut tx, cloak_owner)?
    );
    drop(tx);
    assert_eq!(total_gold(&db)?, before, "gold must be conserved");
    println!("  ✓ total gold unchanged at {before} — no step of that was separately visible");

    // -----------------------------------------------------------------------
    act(6, "The chronicler reads while the world moves");

    // A long report scanning the whole world while combat rewrites it. It never
    // blocks a writer, and its view never shifts underneath it.
    let mut chronicler = db.begin_with::<Snapshot>();
    let opening = total_gold(&db)?;
    let seen_first = chronicler.scan::<Hero>()?.len();

    for _ in 0..50 {
        db.transaction(|tx| {
            tx.update::<Hero>(&ada, |h| h.gold += 1)?;
            tx.update::<Hero>(&bram, |h| h.gold -= 1)?;
            Ok(())
        })?;
    }
    db.transaction(|tx| {
        tx.insert(Hero {
            id: 6,
            name: "Mira".into(),
            gold: 0,
            strength: 50,
        })
    })?;

    let chron_gold: i64 = chronicler.scan::<Hero>()?.iter().map(|h| h.gold).sum();
    let seen_after = chronicler.scan::<Hero>()?.len();
    let in_world = db.begin().scan::<Hero>()?.len();
    println!("  chronicler saw {seen_first} heroes at the start and still sees {seen_after};");
    println!("  the world now holds {in_world}, after 50 commits and one new arrival");
    println!(
        "  chronicler totals {chron_gold}g; the world now totals {}g",
        total_gold(&db)?
    );
    assert_eq!(chron_gold, opening, "a snapshot must not move");
    chronicler.commit()?;
    println!("  ✓ a reader that ran across 51 commits saw exactly one consistent world");

    // -----------------------------------------------------------------------
    act(7, "The raid — many adventurers at once");

    let stop = Arc::new(AtomicBool::new(false));
    let trades = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));
    let audits = Arc::new(AtomicU64::new(0));
    let opening = total_gold(&db)?;
    let heroes = [ada, bram, cleo, dara, finn];

    // An auditor that must never catch a trade half-applied.
    let auditor = {
        let (db, stop, audits) = (Arc::clone(&db), Arc::clone(&stop), Arc::clone(&audits));
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let total: i64 = tx.scan::<Hero>()?.iter().map(|h| h.gold).sum();
                assert_eq!(total, opening, "auditor saw a half-finished trade");
                audits.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    let start = Instant::now();
    let raiders: Vec<_> = (0..4)
        .map(|t| {
            let (db, trades, retries) =
                (Arc::clone(&db), Arc::clone(&trades), Arc::clone(&retries));
            thread::spawn(move || -> Result<()> {
                let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t + 1);
                for _ in 0..400 {
                    seed ^= seed << 13;
                    seed ^= seed >> 7;
                    seed ^= seed << 17;
                    let from = heroes[(seed % heroes.len() as u64) as usize];
                    let to = heroes[((seed >> 8) % heroes.len() as u64) as usize];
                    if from == to {
                        continue;
                    }

                    let mut attempts = 0u64;
                    // Serializable, because the payment depends on a balance we
                    // read — the same shape as Cleo's carry limit in Act 3.
                    db.transaction_with::<Serializable, _, _>(|tx| {
                        attempts += 1;
                        let purse = gold_of(tx, from)?;
                        if purse < 5 {
                            return Ok(());
                        }
                        tx.update::<Hero>(&from, |h| h.gold -= 5)?;
                        tx.update::<Hero>(&to, |h| h.gold += 5)?;
                        Ok(())
                    })?;
                    trades.fetch_add(1, Ordering::Relaxed);
                    retries.fetch_add(attempts - 1, Ordering::Relaxed);
                }
                Ok(())
            })
        })
        .collect();

    for r in raiders {
        r.join().expect("raider panicked")?;
    }
    let elapsed = start.elapsed();
    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    let done = trades.load(Ordering::Relaxed);
    println!("  {done} trades across 4 threads in {elapsed:.2?}");
    println!(
        "  {} audit sweeps completed alongside them, none blocked, none torn",
        audits.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.2} per trade) — contention on 5 heroes, handled by the engine",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / done.max(1) as f64
    );
    assert_eq!(total_gold(&db)?, opening, "gold was created or destroyed");
    println!("  ✓ total gold still {opening}");

    // -----------------------------------------------------------------------
    act(8, "Closing the ledger");

    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan::<Hero>()?
        .iter()
        .map(|h| (h.id, h.name.clone(), h.gold))
        .collect();
    roster.sort_by_key(|(id, _, _)| *id);
    drop(tx);

    println!(
        "  {:<6} {:>6} {:>5} {:>7}  inventory",
        "hero", "gold", "load", "worth"
    );
    let mut tx = db.begin();
    for (id, name, gold) in &roster {
        let held = tx.scan_index(Item::OWNER, *id..=*id)?;
        let load: i32 = held.iter().map(|i| i.weight).sum();
        let worth: i64 = held.iter().map(|i| i.value).sum();
        let names: Vec<String> = held
            .iter()
            .map(|i| match i.kind {
                Kind::Weapon => format!("[wpn] {}", i.name),
                Kind::Potion => format!("[pot] {}", i.name),
                Kind::Trinket => format!("[trk] {}", i.name),
            })
            .collect();
        let names = if names.is_empty() {
            "—".to_string()
        } else {
            names.join(", ")
        };
        println!("  {name:<6} {gold:>5}g {load:>4}kg {worth:>6}g  {names}");
    }

    // Potions anywhere in the world, found by predicate rather than by key —
    // and `kind` is an ordinary Rust enum the engine knows nothing about.
    let potions = tx.scan_where::<Item, _>(|i| i.kind == Kind::Potion)?.len();
    let sword = tx.get::<Item>(&20)?.expect("greatsword");
    let enchantment = sword
        .flavour
        .get("enchantment")
        .cloned()
        .unwrap_or_default();
    let forged_by = sword.flavour.get("forged_by").cloned().unwrap_or_default();
    drop(tx);
    // The rule Act 3 was about, checked rather than asserted in prose.
    let mut tx = db.begin();
    for (id, name, _) in &roster {
        let strength = tx.get::<Hero>(id)?.map(|h| h.strength).unwrap_or(0);
        let load = carried(&mut tx, *id)?;
        assert!(
            load <= strength,
            "{name} is over their carry limit: {load} > {strength}"
        );
    }
    drop(tx);

    println!("\n  potions in the world: {potions}");
    println!("  the greatsword is {enchantment}-enchanted, forged by {forged_by}");

    // Nothing is reclaimed while a transaction that could still reach it is
    // alive. This is the number to watch in a long-lived process: a forgotten
    // transaction pins it and version chains grow without limit.
    let stats = db.stats();
    println!(
        "\n  gc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );
    println!("  (a watermark that stops moving while writes continue is the leak to look for)");

    Ok(())
}

examples/basic.rs (line 54)

fn main() -> Result<()> {
    let db = Database::open(Config::in_memory())?;
    db.register::<Account>()?;

    // ---- insert -----------------------------------------------------------
    // `transaction` runs the closure, commits it, and retries it if it hits a
    // retriable conflict. Snapshot isolation by default.
    db.transaction(|tx| {
        tx.insert(Account {
            id: 1,
            owner: "ada".into(),
            branch: 10,
            balance: 500,
        })?;
        tx.insert(Account {
            id: 2,
            owner: "bob".into(),
            branch: 10,
            balance: 250,
        })?;
        tx.insert(Account {
            id: 3,
            owner: "cleo".into(),
            branch: 20,
            balance: 900,
        })?;
        Ok(())
    })?;

    // ---- read -------------------------------------------------------------
    let mut tx = db.begin();
    let ada = tx.get::<Account>(&1)?.expect("just inserted");
    println!("ada: branch {}, balance {}", ada.branch, ada.balance);

    // A read-only transaction has nothing to commit; dropping it rolls back,
    // which for a reader means simply releasing its snapshot.
    drop(tx);

    // ---- update -----------------------------------------------------------
    db.transaction(|tx| {
        tx.update::<Account>(&1, |a| a.balance -= 100)?;
        tx.update::<Account>(&2, |a| a.balance += 100)?;
        Ok(())
    })?;

    // ---- scan by primary key ----------------------------------------------
    let mut tx = db.begin();
    println!("\nall accounts:");
    for account in tx.scan::<Account>()? {
        println!(
            "  {:>4}  {:<6} branch {}  {:>5}",
            account.id, account.owner, account.branch, account.balance
        );
    }

    // ---- scan by secondary index ------------------------------------------
    println!("\nbranch 10:");
    for account in tx.scan_index(Account::BRANCH, 10u32..=10)? {
        println!("  {} ({})", account.owner, account.balance);
    }
    drop(tx);

    // ---- constraint violations --------------------------------------------
    let mut tx = db.begin();
    let duplicate = tx.insert(Account {
        id: 99,
        owner: "ada".into(),
        branch: 30,
        balance: 0,
    });
    println!("\nreusing owner 'ada': {}", duplicate.unwrap_err());
    drop(tx);

    // ---- delete -----------------------------------------------------------
    db.transaction(|tx| {
        let existed = tx.delete::<Account>(&3)?;
        println!("deleted cleo: {existed}");
        Ok(())
    })?;

    let mut tx = db.begin();
    println!(
        "cleo now: {:?}",
        tx.get::<Account>(&3)?.map(|r| r.to_owned())
    );

    Ok(())
}

examples/concurrent.rs (line 60)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Account>()?;

    db.transaction(|tx| {
        for id in 0..ACCOUNTS {
            tx.insert(Account {
                id,
                balance: STARTING_BALANCE,
            })?;
        }
        Ok(())
    })?;

    let total_before = total(&db)?;
    println!("starting total: {total_before}");

    let stop = Arc::new(AtomicBool::new(false));
    let reads = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));

    // ---- a reader that must never observe a torn transfer -----------------
    // Money moves between accounts constantly. Under snapshot isolation this
    // thread's scans must always sum to the same total: it either sees both
    // halves of a transfer or neither.
    let auditor = {
        let db = Arc::clone(&db);
        let stop = Arc::clone(&stop);
        let reads = Arc::clone(&reads);
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let sum: i64 = tx.scan::<Account>()?.iter().map(|a| a.balance).sum();
                assert_eq!(
                    sum,
                    ACCOUNTS as i64 * STARTING_BALANCE,
                    "a reader observed a partially applied transfer"
                );
                reads.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    // ---- writers ----------------------------------------------------------
    let start = Instant::now();
    let mut writers = Vec::new();
    for t in 0..TRANSFER_THREADS {
        let db = Arc::clone(&db);
        let retries = Arc::clone(&retries);
        writers.push(thread::spawn(move || -> Result<()> {
            // A cheap per-thread PRNG; no dependency needed for this.
            let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t as u64 + 1);
            let mut next = move || {
                seed ^= seed << 13;
                seed ^= seed >> 7;
                seed ^= seed << 17;
                seed
            };

            for _ in 0..TRANSFERS_PER_THREAD {
                let from = next() % ACCOUNTS;
                let to = next() % ACCOUNTS;
                if from == to {
                    continue;
                }

                let mut attempts = 0u64;
                // Serializable, because the transfer's *write* depends on a
                // balance it *read* — the write-skew shape. Snapshot isolation
                // would let two concurrent transfers both pass the check.
                db.transaction_with::<Serializable, _, _>(|tx| {
                    attempts += 1;
                    let balance = tx.get::<Account>(&from)?.map(|a| a.balance).unwrap_or(0);
                    if balance < 10 {
                        return Ok(());
                    }
                    tx.update::<Account>(&from, |a| a.balance -= 10)?;
                    tx.update::<Account>(&to, |a| a.balance += 10)?;
                    Ok(())
                })?;
                retries.fetch_add(attempts - 1, Ordering::Relaxed);
            }
            Ok(())
        }));
    }

    for w in writers {
        w.join().expect("writer panicked")?;
    }
    let elapsed = start.elapsed();

    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    // ---- results ----------------------------------------------------------
    let total_after = total(&db)?;
    let committed = (TRANSFER_THREADS * TRANSFERS_PER_THREAD) as u64;

    println!("final total:    {total_after}");
    assert_eq!(total_before, total_after, "money was created or destroyed");

    println!("\n{committed} transfers in {elapsed:.2?}");
    println!(
        "  {:.0} transfers/sec",
        committed as f64 / elapsed.as_secs_f64()
    );
    println!(
        "  {} concurrent audit scans completed, none of them blocked",
        reads.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.1} per transfer)",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / committed as f64
    );

    // The GC watermark should be advancing. If `active_transactions` climbs and
    // the watermark stalls, some transaction is being leaked — see engine::gc.
    let stats = db.stats();
    println!(
        "\ngc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );

    println!("\n✓ no torn reads, no lost updates, conservation held");

    // Slow readers are what pin the GC watermark. Left here as the shape of the
    // thing to watch for, not as something this example demonstrates going wrong.
    thread::sleep(Duration::from_millis(1));
    Ok(())
}

fn total(db: &Database) -> Result<i64> {
    let mut tx = db.begin();
    Ok(tx.scan::<Account>()?.iter().map(|a| a.balance).sum())
}

examples/isolation.rs (line 124)

fn main() -> Result<()> {
    let db = Database::open(Config::in_memory())?;
    db.register::<Doctor>()?;
    db.register::<Counter>()?;

    db.transaction(|tx| {
        tx.insert(Doctor {
            id: 1,
            name: "ada".into(),
            on_call: true,
        })?;
        tx.insert(Doctor {
            id: 2,
            name: "bob".into(),
            on_call: true,
        })?;
        tx.insert(Counter { id: 1, value: 0 })
    })?;

    // ------------------------------------------------------------------
    banner("1. Snapshot: a reader is unaffected by concurrent commits");
    {
        let mut reader = db.begin_with::<Snapshot>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        // Someone else commits, start to finish, while `reader` is open.
        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 42).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_eq!(before, after, "snapshot isolation must be stable");
        println!("  ✓ the snapshot held: readers never block and never change under you");
    }

    // ------------------------------------------------------------------
    banner("2. ReadCommitted: each statement sees a fresh snapshot");
    {
        let mut reader = db.begin_with::<ReadCommitted>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 99).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_ne!(
            before, after,
            "read committed should observe the new commit"
        );
        println!("  ✓ non-repeatable read — the tradeoff this level makes for cheapness");
    }

    // ------------------------------------------------------------------
    banner("3. Write-write conflict: first committer wins");
    {
        let mut first = db.begin_with::<Snapshot>();
        let mut second = db.begin_with::<Snapshot>();

        first.update::<Counter>(&1, |c| c.value += 1)?;
        first.commit()?;

        // `second` took its snapshot before `first` committed, so writing here
        // would silently drop `first`'s update.
        let result = second.update::<Counter>(&1, |c| c.value += 1);
        match result {
            Err(e) => println!(
                "  second transaction: {e}  (retriable: {})",
                e.is_retriable()
            ),
            Ok(_) => unreachable!("the stale write should have been rejected"),
        }
        println!("  ✓ no lost update");
    }

    // ------------------------------------------------------------------
    banner("4. Write skew: allowed under Snapshot");
    {
        // The rule: at least one doctor must stay on call. Each transaction
        // checks the rule, sees it satisfied, and takes a *different* doctor
        // off call — so there is no write-write conflict to catch them.
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();

        let t1_sees = t1.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t1.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let t2_sees = t2.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t2.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let names: Vec<String> = {
            let mut tx = db.begin();
            tx.scan::<Doctor>()?
                .iter()
                .map(|d| d.name.clone())
                .collect()
        };
        println!("  on the rota: {}", names.join(", "));
        println!("  t1 counts {t1_sees} on call, t2 counts {t2_sees} — both think it is safe");

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;
        t1.commit()?;
        t2.commit()?;

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✗ {remaining} doctors on call — the invariant is broken");
        println!("    Both transactions were individually legal. This is write skew,");
        println!("    and it is why Snapshot is not Serializable.");
    }

    // ------------------------------------------------------------------
    banner("5. Write skew: prevented under Serializable");
    {
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();

        // Both read both rows, exactly as before.
        let _ = t1.get::<Doctor>(&1)?;
        let _ = t1.get::<Doctor>(&2)?;
        let _ = t2.get::<Doctor>(&1)?;
        let _ = t2.get::<Doctor>(&2)?;

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;

        t1.commit()?;
        match t2.commit() {
            Err(e) => println!("  t2: {e}  (retriable: {})", e.is_retriable()),
            Ok(()) => unreachable!("t2 read a row t1 changed; it must not commit"),
        }

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✓ {remaining} doctor still on call — the invariant held");
        println!("    t2 read doctor 1, which t1 changed, so t2 could not be serialized.");
    }

    // ------------------------------------------------------------------
    banner("6. Retries are the normal way to use Serializable");
    {
        // `transaction_with` reruns the closure on a retriable failure, so the
        // abort in scenario 5 becomes invisible to the caller.
        let mut attempts = 0;
        db.transaction_with::<Serializable, _, _>(|tx| {
            attempts += 1;
            let ada = tx.get::<Doctor>(&1)?.unwrap().on_call;
            let bob = tx.get::<Doctor>(&2)?.unwrap().on_call;
            if ada && bob {
                tx.update::<Doctor>(&1, |d| d.on_call = false)?;
            }
            Ok(())
        })?;
        println!("  committed after {attempts} attempt(s)");
        println!("  ✓ write your transaction as if it runs alone; let the engine retry it");
    }

    banner("Summary");
    println!("  ReadCommitted   cheapest; non-repeatable reads and phantoms");
    println!("  RepeatableRead  stable snapshot; write skew possible");
    println!("  Snapshot        stable snapshot; write skew possible   ← default");
    println!("  Serializable    no anomalies; expect retriable aborts");

    Ok(())
}

pub fn begin_with<I: IsolationLevel>(&self) -> Transaction<'_, I>

Begin a transaction at a chosen isolation level.

The level is a type parameter, so the cost of the strongest never leaks into the weakest — see IsolationLevel.

let strict = db.begin_with::<Serializable>();
let cheap = db.begin_with::<ReadCommitted>();

Prefer Database::transaction_with unless you need manual control: this hands back a transaction you must commit yourself, and dropping it without committing rolls it back.

Examples found in repository

examples/isolation.rs (line 54)

fn main() -> Result<()> {
    let db = Database::open(Config::in_memory())?;
    db.register::<Doctor>()?;
    db.register::<Counter>()?;

    db.transaction(|tx| {
        tx.insert(Doctor {
            id: 1,
            name: "ada".into(),
            on_call: true,
        })?;
        tx.insert(Doctor {
            id: 2,
            name: "bob".into(),
            on_call: true,
        })?;
        tx.insert(Counter { id: 1, value: 0 })
    })?;

    // ------------------------------------------------------------------
    banner("1. Snapshot: a reader is unaffected by concurrent commits");
    {
        let mut reader = db.begin_with::<Snapshot>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        // Someone else commits, start to finish, while `reader` is open.
        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 42).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_eq!(before, after, "snapshot isolation must be stable");
        println!("  ✓ the snapshot held: readers never block and never change under you");
    }

    // ------------------------------------------------------------------
    banner("2. ReadCommitted: each statement sees a fresh snapshot");
    {
        let mut reader = db.begin_with::<ReadCommitted>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 99).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_ne!(
            before, after,
            "read committed should observe the new commit"
        );
        println!("  ✓ non-repeatable read — the tradeoff this level makes for cheapness");
    }

    // ------------------------------------------------------------------
    banner("3. Write-write conflict: first committer wins");
    {
        let mut first = db.begin_with::<Snapshot>();
        let mut second = db.begin_with::<Snapshot>();

        first.update::<Counter>(&1, |c| c.value += 1)?;
        first.commit()?;

        // `second` took its snapshot before `first` committed, so writing here
        // would silently drop `first`'s update.
        let result = second.update::<Counter>(&1, |c| c.value += 1);
        match result {
            Err(e) => println!(
                "  second transaction: {e}  (retriable: {})",
                e.is_retriable()
            ),
            Ok(_) => unreachable!("the stale write should have been rejected"),
        }
        println!("  ✓ no lost update");
    }

    // ------------------------------------------------------------------
    banner("4. Write skew: allowed under Snapshot");
    {
        // The rule: at least one doctor must stay on call. Each transaction
        // checks the rule, sees it satisfied, and takes a *different* doctor
        // off call — so there is no write-write conflict to catch them.
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();

        let t1_sees = t1.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t1.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let t2_sees = t2.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t2.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let names: Vec<String> = {
            let mut tx = db.begin();
            tx.scan::<Doctor>()?
                .iter()
                .map(|d| d.name.clone())
                .collect()
        };
        println!("  on the rota: {}", names.join(", "));
        println!("  t1 counts {t1_sees} on call, t2 counts {t2_sees} — both think it is safe");

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;
        t1.commit()?;
        t2.commit()?;

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✗ {remaining} doctors on call — the invariant is broken");
        println!("    Both transactions were individually legal. This is write skew,");
        println!("    and it is why Snapshot is not Serializable.");
    }

    // ------------------------------------------------------------------
    banner("5. Write skew: prevented under Serializable");
    {
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();

        // Both read both rows, exactly as before.
        let _ = t1.get::<Doctor>(&1)?;
        let _ = t1.get::<Doctor>(&2)?;
        let _ = t2.get::<Doctor>(&1)?;
        let _ = t2.get::<Doctor>(&2)?;

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;

        t1.commit()?;
        match t2.commit() {
            Err(e) => println!("  t2: {e}  (retriable: {})", e.is_retriable()),
            Ok(()) => unreachable!("t2 read a row t1 changed; it must not commit"),
        }

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✓ {remaining} doctor still on call — the invariant held");
        println!("    t2 read doctor 1, which t1 changed, so t2 could not be serialized.");
    }

    // ------------------------------------------------------------------
    banner("6. Retries are the normal way to use Serializable");
    {
        // `transaction_with` reruns the closure on a retriable failure, so the
        // abort in scenario 5 becomes invisible to the caller.
        let mut attempts = 0;
        db.transaction_with::<Serializable, _, _>(|tx| {
            attempts += 1;
            let ada = tx.get::<Doctor>(&1)?.unwrap().on_call;
            let bob = tx.get::<Doctor>(&2)?.unwrap().on_call;
            if ada && bob {
                tx.update::<Doctor>(&1, |d| d.on_call = false)?;
            }
            Ok(())
        })?;
        println!("  committed after {attempts} attempt(s)");
        println!("  ✓ write your transaction as if it runs alone; let the engine retry it");
    }

    banner("Summary");
    println!("  ReadCommitted   cheapest; non-repeatable reads and phantoms");
    println!("  RepeatableRead  stable snapshot; write skew possible");
    println!("  Snapshot        stable snapshot; write skew possible   ← default");
    println!("  Serializable    no anomalies; expect retriable aborts");

    Ok(())
}

examples/game.rs (line 235)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Hero>()?;
    db.register::<Item>()?;
    db.register::<PartyMember>()?;

    let (ada, bram, cleo) = (1u64, 2u64, 3u64);

    // -----------------------------------------------------------------------
    act(1, "The world is created");

    db.transaction(|tx| {
        tx.insert(Hero {
            id: ada,
            name: "Ada".into(),
            gold: 120,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: bram,
            name: "Bram".into(),
            gold: 80,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: cleo,
            name: "Cleo".into(),
            gold: 200,
            strength: 60,
        })?;

        let mut item = |id, owner, name: &str, weight, value, kind| {
            tx.insert(Item {
                id,
                owner,
                name: name.into(),
                weight,
                value,
                kind,
                flavour: HashMap::new(),
            })
        };
        item(10, ada, "Rusty Sword", 30, 15, Kind::Weapon)?;
        item(11, ada, "Health Potion", 5, 20, Kind::Potion)?;
        item(12, bram, "Oak Shield", 40, 35, Kind::Weapon)?;
        item(13, cleo, "Lucky Charm", 2, 90, Kind::Trinket)?;
        // Loot lying in the dungeon, owned by nobody.
        item(20, GROUND, "Flaming Greatsword", 55, 500, Kind::Weapon)?;
        item(21, GROUND, "Elven Cloak", 20, 240, Kind::Trinket)?;
        item(22, GROUND, "Iron Helm", 35, 60, Kind::Weapon)?;

        // Arbitrary per-item state, in a database row, with no schema to
        // declare and no serialisation format to agree on.
        tx.update::<Item>(&20, |i| {
            i.flavour.insert("enchantment".into(), "flame".into());
            i.flavour.insert("forged_by".into(), "Durin".into());
        })?;
        Ok(())
    })?;

    // A whole transaction that fails partway leaves nothing behind. The unique
    // index on `name` refuses the second Ada, and the gold change goes with it.
    let before = gold_of(&mut db.begin(), ada)?;
    let doomed = db.transaction(|tx| {
        tx.update::<Hero>(&ada, |h| h.gold += 10_000)?;
        tx.insert(Hero {
            id: 99,
            name: "Ada".into(),
            gold: 0,
            strength: 10,
        })
    });
    println!("  a duplicate hero name: {}", doomed.unwrap_err());
    println!(
        "  Ada's gold is still {} — the whole transaction rolled back, not just the insert",
        gold_of(&mut db.begin(), ada)?
    );
    assert_eq!(gold_of(&mut db.begin(), ada)?, before);

    let mut tx = db.begin();
    for hero in [ada, bram, cleo] {
        let name = name_of(&mut tx, hero)?;
        let carried = carried(&mut tx, hero)?;
        let items = inventory(&mut tx, hero)?;
        let names: Vec<_> = items.iter().map(|(_, n, _)| n.as_str()).collect();
        println!("  {name:<5} carries {carried:>3} — {}", names.join(", "));
    }
    drop(tx);

    // -----------------------------------------------------------------------
    act(2, "Two heroes reach for the same greatsword");

    // Both begin, both see the sword unclaimed. The first to write takes it;
    // the second is told immediately rather than being made to wait.
    let mut ada_grabs = db.begin_with::<Snapshot>();
    let mut bram_grabs = db.begin_with::<Snapshot>();

    ada_grabs.update::<Item>(&20, |i| i.owner = ada)?;

    match bram_grabs.update::<Item>(&20, |i| i.owner = bram) {
        Err(e @ Error::WriteConflict { .. }) => {
            println!("  Bram: {e}");
            println!(
                "  ...retriable: {} — he can try for something else",
                e.is_retriable()
            );
        }
        other => unreachable!("expected a conflict, got {other:?}"),
    }

    // A failed write does not end the transaction. Bram takes the cloak instead.
    bram_grabs.update::<Item>(&21, |i| i.owner = bram)?;
    ada_grabs.commit()?;
    bram_grabs.commit()?;
    println!("  Ada takes the greatsword, Bram takes the cloak. Nobody blocked.");

    // -----------------------------------------------------------------------
    act(3, "Cleo overloads herself — write skew, and the fix");

    // Cleo can carry 60 and is carrying 2. Either 35kg helm fits; both do not.
    // Each transaction reads her inventory, checks the total, and picks up a
    // *different* item — so the two writes never touch the same record, and
    // first-updater-wins has nothing to catch.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.insert(Item {
            id: 23,
            owner: GROUND,
            name: "Steel Helm".into(),
            weight: 35,
            value: 70,
            kind: Kind::Weapon,
            flavour: HashMap::new(),
        })
    })?;

    let pick_up = |item: u64| {
        move |tx: &mut mvcc::Transaction<'_, Serializable>| -> Result<bool> {
            let hero = tx.get::<Hero>(&cleo)?.expect("Cleo exists");
            let strength = hero.strength;
            let load = carried(tx, cleo)?;
            let weight = tx.get::<Item>(&item)?.map(|i| i.weight).unwrap_or(0);
            if load + weight > strength {
                return Ok(false);
            }
            tx.update::<Item>(&item, |i| i.owner = cleo)?;
            Ok(true)
        }
    };

    // Under Snapshot, both checks pass against a stale inventory.
    {
        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();
        let (load1, load2) = (carried(&mut t1, cleo)?, carried(&mut t2, cleo)?);
        println!(
            "  Snapshot:     both transactions see {load1}/{load2} carried, both think a 35kg helm fits"
        );
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        t2.commit()?;
        let over = carried(&mut db.begin(), cleo)?;
        println!(
            "  ✗ Cleo now carries {over} of a possible 60. Two legal transactions, one broken rule."
        );
    }

    // Put it back and try again at Serializable.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.update::<Item>(&23, |i| i.owner = GROUND).map(|_| ())
    })?;

    {
        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();
        // Both read the inventory, as before.
        let _ = carried(&mut t1, cleo)?;
        let _ = carried(&mut t2, cleo)?;
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        match t2.commit() {
            Err(e @ Error::SerializationFailure) => println!("  Serializable: second pickup {e}"),
            other => unreachable!("expected a serialization failure, got {other:?}"),
        }
        println!(
            "  ✓ Cleo carries {} — the second pickup read an inventory the first changed.",
            carried(&mut db.begin(), cleo)?
        );
    }

    // In real code you would not hand-roll that. `transaction_with` retries,
    // and the retry re-reads the inventory and correctly declines.
    let took_it = db.transaction_with::<Serializable, _, _>(pick_up(23))?;
    println!("  ...and on retry the second helm is refused on its merits: picked_up = {took_it}");

    // -----------------------------------------------------------------------
    act(4, "The party fills up — a phantom, not a conflict");

    db.transaction(|tx| {
        tx.insert(PartyMember {
            hero: ada,
            party: PARTY,
            role: "Vanguard".into(),
        })?;
        tx.insert(PartyMember {
            hero: bram,
            party: PARTY,
            role: "Shield".into(),
        })?;
        tx.insert(PartyMember {
            hero: cleo,
            party: PARTY,
            role: "Scout".into(),
        })
    })?;

    // Two newcomers apply at once, with three of four slots taken. Neither
    // writes a row the other wrote — they insert *different* rows — so there is
    // no write conflict to detect. What they collide on is a row that did not
    // exist when either of them counted.
    let (dara, finn) = (4u64, 5u64);
    db.transaction(|tx| {
        tx.insert(Hero {
            id: dara,
            name: "Dara".into(),
            gold: 40,
            strength: 80,
        })?;
        tx.insert(Hero {
            id: finn,
            name: "Finn".into(),
            gold: 40,
            strength: 80,
        })
    })?;

    let mut d = db.begin_with::<Serializable>();
    let mut f = db.begin_with::<Serializable>();
    println!(
        "  Dara counts {} members, Finn counts {} — both see a free slot",
        party_size(&mut d)?,
        party_size(&mut f)?
    );
    d.insert(PartyMember {
        hero: dara,
        party: PARTY,
        role: "Healer".into(),
    })?;
    f.insert(PartyMember {
        hero: finn,
        party: PARTY,
        role: "Healer".into(),
    })?;
    d.commit()?;
    match f.commit() {
        Err(e @ Error::SerializationFailure) => println!("  Finn: {e}"),
        other => unreachable!("expected a serialization failure, got {other:?}"),
    }
    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan_where::<PartyMember, _>(|m| m.party == PARTY)?
        .iter()
        .map(|m| (m.hero, m.role.clone()))
        .collect();
    drop(tx);
    roster.sort();
    let size = roster.len();
    let mut tx = db.begin();
    let listed: Vec<String> = roster
        .iter()
        .map(|(h, role)| Ok(format!("{} the {role}", name_of(&mut tx, *h)?)))
        .collect::<Result<_>>()?;
    drop(tx);
    println!("  party: {}", listed.join(", "));
    println!("  ✓ {size}/{PARTY_LIMIT} filled. Finn's insert became a phantom in Dara's count.");
    println!("    (`scan_where` hands the engine the predicate, so it can re-check it at commit.)");
    assert!(size <= PARTY_LIMIT);

    // -----------------------------------------------------------------------
    act(5, "A trade — two records, one atomic step");

    let before = total_gold(&db)?;

    let trade = |seller: u64, buyer: u64, item: u64, price: i64| {
        let db = Arc::clone(&db);
        move || -> Result<bool> {
            db.transaction_with::<Serializable, _, _>(|tx| {
                let buyer_gold = gold_of(tx, buyer)?;
                if buyer_gold < price {
                    return Ok(false);
                }
                let owned_by_seller = tx
                    .get::<Item>(&item)?
                    .map(|i| i.owner == seller)
                    .unwrap_or(false);
                if !owned_by_seller {
                    return Ok(false);
                }
                tx.update::<Hero>(&buyer, |h| h.gold -= price)?;
                tx.update::<Hero>(&seller, |h| h.gold += price)?;
                tx.update::<Item>(&item, |i| i.owner = buyer)?;
                Ok(true)
            })
        }
    };

    let sold = trade(bram, cleo, 21, 150)()?;
    println!("  Bram sells the Elven Cloak to Cleo for 150g: {sold}");
    let mut tx = db.begin();
    let cloak_owner = tx.get::<Item>(&21)?.expect("cloak").owner;
    println!(
        "  Bram {}g, Cleo {}g, cloak owner is {}",
        gold_of(&mut tx, bram)?,
        gold_of(&mut tx, cleo)?,
        name_of(&mut tx, cloak_owner)?
    );
    drop(tx);
    assert_eq!(total_gold(&db)?, before, "gold must be conserved");
    println!("  ✓ total gold unchanged at {before} — no step of that was separately visible");

    // -----------------------------------------------------------------------
    act(6, "The chronicler reads while the world moves");

    // A long report scanning the whole world while combat rewrites it. It never
    // blocks a writer, and its view never shifts underneath it.
    let mut chronicler = db.begin_with::<Snapshot>();
    let opening = total_gold(&db)?;
    let seen_first = chronicler.scan::<Hero>()?.len();

    for _ in 0..50 {
        db.transaction(|tx| {
            tx.update::<Hero>(&ada, |h| h.gold += 1)?;
            tx.update::<Hero>(&bram, |h| h.gold -= 1)?;
            Ok(())
        })?;
    }
    db.transaction(|tx| {
        tx.insert(Hero {
            id: 6,
            name: "Mira".into(),
            gold: 0,
            strength: 50,
        })
    })?;

    let chron_gold: i64 = chronicler.scan::<Hero>()?.iter().map(|h| h.gold).sum();
    let seen_after = chronicler.scan::<Hero>()?.len();
    let in_world = db.begin().scan::<Hero>()?.len();
    println!("  chronicler saw {seen_first} heroes at the start and still sees {seen_after};");
    println!("  the world now holds {in_world}, after 50 commits and one new arrival");
    println!(
        "  chronicler totals {chron_gold}g; the world now totals {}g",
        total_gold(&db)?
    );
    assert_eq!(chron_gold, opening, "a snapshot must not move");
    chronicler.commit()?;
    println!("  ✓ a reader that ran across 51 commits saw exactly one consistent world");

    // -----------------------------------------------------------------------
    act(7, "The raid — many adventurers at once");

    let stop = Arc::new(AtomicBool::new(false));
    let trades = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));
    let audits = Arc::new(AtomicU64::new(0));
    let opening = total_gold(&db)?;
    let heroes = [ada, bram, cleo, dara, finn];

    // An auditor that must never catch a trade half-applied.
    let auditor = {
        let (db, stop, audits) = (Arc::clone(&db), Arc::clone(&stop), Arc::clone(&audits));
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let total: i64 = tx.scan::<Hero>()?.iter().map(|h| h.gold).sum();
                assert_eq!(total, opening, "auditor saw a half-finished trade");
                audits.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    let start = Instant::now();
    let raiders: Vec<_> = (0..4)
        .map(|t| {
            let (db, trades, retries) =
                (Arc::clone(&db), Arc::clone(&trades), Arc::clone(&retries));
            thread::spawn(move || -> Result<()> {
                let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t + 1);
                for _ in 0..400 {
                    seed ^= seed << 13;
                    seed ^= seed >> 7;
                    seed ^= seed << 17;
                    let from = heroes[(seed % heroes.len() as u64) as usize];
                    let to = heroes[((seed >> 8) % heroes.len() as u64) as usize];
                    if from == to {
                        continue;
                    }

                    let mut attempts = 0u64;
                    // Serializable, because the payment depends on a balance we
                    // read — the same shape as Cleo's carry limit in Act 3.
                    db.transaction_with::<Serializable, _, _>(|tx| {
                        attempts += 1;
                        let purse = gold_of(tx, from)?;
                        if purse < 5 {
                            return Ok(());
                        }
                        tx.update::<Hero>(&from, |h| h.gold -= 5)?;
                        tx.update::<Hero>(&to, |h| h.gold += 5)?;
                        Ok(())
                    })?;
                    trades.fetch_add(1, Ordering::Relaxed);
                    retries.fetch_add(attempts - 1, Ordering::Relaxed);
                }
                Ok(())
            })
        })
        .collect();

    for r in raiders {
        r.join().expect("raider panicked")?;
    }
    let elapsed = start.elapsed();
    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    let done = trades.load(Ordering::Relaxed);
    println!("  {done} trades across 4 threads in {elapsed:.2?}");
    println!(
        "  {} audit sweeps completed alongside them, none blocked, none torn",
        audits.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.2} per trade) — contention on 5 heroes, handled by the engine",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / done.max(1) as f64
    );
    assert_eq!(total_gold(&db)?, opening, "gold was created or destroyed");
    println!("  ✓ total gold still {opening}");

    // -----------------------------------------------------------------------
    act(8, "Closing the ledger");

    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan::<Hero>()?
        .iter()
        .map(|h| (h.id, h.name.clone(), h.gold))
        .collect();
    roster.sort_by_key(|(id, _, _)| *id);
    drop(tx);

    println!(
        "  {:<6} {:>6} {:>5} {:>7}  inventory",
        "hero", "gold", "load", "worth"
    );
    let mut tx = db.begin();
    for (id, name, gold) in &roster {
        let held = tx.scan_index(Item::OWNER, *id..=*id)?;
        let load: i32 = held.iter().map(|i| i.weight).sum();
        let worth: i64 = held.iter().map(|i| i.value).sum();
        let names: Vec<String> = held
            .iter()
            .map(|i| match i.kind {
                Kind::Weapon => format!("[wpn] {}", i.name),
                Kind::Potion => format!("[pot] {}", i.name),
                Kind::Trinket => format!("[trk] {}", i.name),
            })
            .collect();
        let names = if names.is_empty() {
            "—".to_string()
        } else {
            names.join(", ")
        };
        println!("  {name:<6} {gold:>5}g {load:>4}kg {worth:>6}g  {names}");
    }

    // Potions anywhere in the world, found by predicate rather than by key —
    // and `kind` is an ordinary Rust enum the engine knows nothing about.
    let potions = tx.scan_where::<Item, _>(|i| i.kind == Kind::Potion)?.len();
    let sword = tx.get::<Item>(&20)?.expect("greatsword");
    let enchantment = sword
        .flavour
        .get("enchantment")
        .cloned()
        .unwrap_or_default();
    let forged_by = sword.flavour.get("forged_by").cloned().unwrap_or_default();
    drop(tx);
    // The rule Act 3 was about, checked rather than asserted in prose.
    let mut tx = db.begin();
    for (id, name, _) in &roster {
        let strength = tx.get::<Hero>(id)?.map(|h| h.strength).unwrap_or(0);
        let load = carried(&mut tx, *id)?;
        assert!(
            load <= strength,
            "{name} is over their carry limit: {load} > {strength}"
        );
    }
    drop(tx);

    println!("\n  potions in the world: {potions}");
    println!("  the greatsword is {enchantment}-enchanted, forged by {forged_by}");

    // Nothing is reclaimed while a transaction that could still reach it is
    // alive. This is the number to watch in a long-lived process: a forgotten
    // transaction pins it and version chains grow without limit.
    let stats = db.stats();
    println!(
        "\n  gc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );
    println!("  (a watermark that stops moving while writes continue is the leak to look for)");

    Ok(())
}

pub fn transaction<R, F>(&self, f: F) -> Result<R>

where F: FnMut(&mut Transaction<'_, SnapshotLevel>) -> Result<R>,

Run f in a transaction at snapshot isolation, retrying while it fails retriably, and commit.

This is the API most users should reach for. Under snapshot isolation, and especially serializable, an abort is a normal outcome rather than an error condition — every caller would otherwise write this loop.

f may run more than once, so it must not have side effects outside the transaction.

Split from Database::transaction_with rather than defaulting a type parameter, because Rust cannot infer a defaulted type parameter on a function — db.transaction(|tx| …) would not compile.

Examples found in repository

examples/basic.rs (lines 31-51)

fn main() -> Result<()> {
    let db = Database::open(Config::in_memory())?;
    db.register::<Account>()?;

    // ---- insert -----------------------------------------------------------
    // `transaction` runs the closure, commits it, and retries it if it hits a
    // retriable conflict. Snapshot isolation by default.
    db.transaction(|tx| {
        tx.insert(Account {
            id: 1,
            owner: "ada".into(),
            branch: 10,
            balance: 500,
        })?;
        tx.insert(Account {
            id: 2,
            owner: "bob".into(),
            branch: 10,
            balance: 250,
        })?;
        tx.insert(Account {
            id: 3,
            owner: "cleo".into(),
            branch: 20,
            balance: 900,
        })?;
        Ok(())
    })?;

    // ---- read -------------------------------------------------------------
    let mut tx = db.begin();
    let ada = tx.get::<Account>(&1)?.expect("just inserted");
    println!("ada: branch {}, balance {}", ada.branch, ada.balance);

    // A read-only transaction has nothing to commit; dropping it rolls back,
    // which for a reader means simply releasing its snapshot.
    drop(tx);

    // ---- update -----------------------------------------------------------
    db.transaction(|tx| {
        tx.update::<Account>(&1, |a| a.balance -= 100)?;
        tx.update::<Account>(&2, |a| a.balance += 100)?;
        Ok(())
    })?;

    // ---- scan by primary key ----------------------------------------------
    let mut tx = db.begin();
    println!("\nall accounts:");
    for account in tx.scan::<Account>()? {
        println!(
            "  {:>4}  {:<6} branch {}  {:>5}",
            account.id, account.owner, account.branch, account.balance
        );
    }

    // ---- scan by secondary index ------------------------------------------
    println!("\nbranch 10:");
    for account in tx.scan_index(Account::BRANCH, 10u32..=10)? {
        println!("  {} ({})", account.owner, account.balance);
    }
    drop(tx);

    // ---- constraint violations --------------------------------------------
    let mut tx = db.begin();
    let duplicate = tx.insert(Account {
        id: 99,
        owner: "ada".into(),
        branch: 30,
        balance: 0,
    });
    println!("\nreusing owner 'ada': {}", duplicate.unwrap_err());
    drop(tx);

    // ---- delete -----------------------------------------------------------
    db.transaction(|tx| {
        let existed = tx.delete::<Account>(&3)?;
        println!("deleted cleo: {existed}");
        Ok(())
    })?;

    let mut tx = db.begin();
    println!(
        "cleo now: {:?}",
        tx.get::<Account>(&3)?.map(|r| r.to_owned())
    );

    Ok(())
}

examples/concurrent.rs (lines 33-41)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Account>()?;

    db.transaction(|tx| {
        for id in 0..ACCOUNTS {
            tx.insert(Account {
                id,
                balance: STARTING_BALANCE,
            })?;
        }
        Ok(())
    })?;

    let total_before = total(&db)?;
    println!("starting total: {total_before}");

    let stop = Arc::new(AtomicBool::new(false));
    let reads = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));

    // ---- a reader that must never observe a torn transfer -----------------
    // Money moves between accounts constantly. Under snapshot isolation this
    // thread's scans must always sum to the same total: it either sees both
    // halves of a transfer or neither.
    let auditor = {
        let db = Arc::clone(&db);
        let stop = Arc::clone(&stop);
        let reads = Arc::clone(&reads);
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let sum: i64 = tx.scan::<Account>()?.iter().map(|a| a.balance).sum();
                assert_eq!(
                    sum,
                    ACCOUNTS as i64 * STARTING_BALANCE,
                    "a reader observed a partially applied transfer"
                );
                reads.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    // ---- writers ----------------------------------------------------------
    let start = Instant::now();
    let mut writers = Vec::new();
    for t in 0..TRANSFER_THREADS {
        let db = Arc::clone(&db);
        let retries = Arc::clone(&retries);
        writers.push(thread::spawn(move || -> Result<()> {
            // A cheap per-thread PRNG; no dependency needed for this.
            let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t as u64 + 1);
            let mut next = move || {
                seed ^= seed << 13;
                seed ^= seed >> 7;
                seed ^= seed << 17;
                seed
            };

            for _ in 0..TRANSFERS_PER_THREAD {
                let from = next() % ACCOUNTS;
                let to = next() % ACCOUNTS;
                if from == to {
                    continue;
                }

                let mut attempts = 0u64;
                // Serializable, because the transfer's *write* depends on a
                // balance it *read* — the write-skew shape. Snapshot isolation
                // would let two concurrent transfers both pass the check.
                db.transaction_with::<Serializable, _, _>(|tx| {
                    attempts += 1;
                    let balance = tx.get::<Account>(&from)?.map(|a| a.balance).unwrap_or(0);
                    if balance < 10 {
                        return Ok(());
                    }
                    tx.update::<Account>(&from, |a| a.balance -= 10)?;
                    tx.update::<Account>(&to, |a| a.balance += 10)?;
                    Ok(())
                })?;
                retries.fetch_add(attempts - 1, Ordering::Relaxed);
            }
            Ok(())
        }));
    }

    for w in writers {
        w.join().expect("writer panicked")?;
    }
    let elapsed = start.elapsed();

    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    // ---- results ----------------------------------------------------------
    let total_after = total(&db)?;
    let committed = (TRANSFER_THREADS * TRANSFERS_PER_THREAD) as u64;

    println!("final total:    {total_after}");
    assert_eq!(total_before, total_after, "money was created or destroyed");

    println!("\n{committed} transfers in {elapsed:.2?}");
    println!(
        "  {:.0} transfers/sec",
        committed as f64 / elapsed.as_secs_f64()
    );
    println!(
        "  {} concurrent audit scans completed, none of them blocked",
        reads.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.1} per transfer)",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / committed as f64
    );

    // The GC watermark should be advancing. If `active_transactions` climbs and
    // the watermark stalls, some transaction is being leaked — see engine::gc.
    let stats = db.stats();
    println!(
        "\ngc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );

    println!("\n✓ no torn reads, no lost updates, conservation held");

    // Slow readers are what pin the GC watermark. Left here as the shape of the
    // thing to watch for, not as something this example demonstrates going wrong.
    thread::sleep(Duration::from_millis(1));
    Ok(())
}

examples/isolation.rs (lines 37-49)

fn main() -> Result<()> {
    let db = Database::open(Config::in_memory())?;
    db.register::<Doctor>()?;
    db.register::<Counter>()?;

    db.transaction(|tx| {
        tx.insert(Doctor {
            id: 1,
            name: "ada".into(),
            on_call: true,
        })?;
        tx.insert(Doctor {
            id: 2,
            name: "bob".into(),
            on_call: true,
        })?;
        tx.insert(Counter { id: 1, value: 0 })
    })?;

    // ------------------------------------------------------------------
    banner("1. Snapshot: a reader is unaffected by concurrent commits");
    {
        let mut reader = db.begin_with::<Snapshot>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        // Someone else commits, start to finish, while `reader` is open.
        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 42).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_eq!(before, after, "snapshot isolation must be stable");
        println!("  ✓ the snapshot held: readers never block and never change under you");
    }

    // ------------------------------------------------------------------
    banner("2. ReadCommitted: each statement sees a fresh snapshot");
    {
        let mut reader = db.begin_with::<ReadCommitted>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 99).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_ne!(
            before, after,
            "read committed should observe the new commit"
        );
        println!("  ✓ non-repeatable read — the tradeoff this level makes for cheapness");
    }

    // ------------------------------------------------------------------
    banner("3. Write-write conflict: first committer wins");
    {
        let mut first = db.begin_with::<Snapshot>();
        let mut second = db.begin_with::<Snapshot>();

        first.update::<Counter>(&1, |c| c.value += 1)?;
        first.commit()?;

        // `second` took its snapshot before `first` committed, so writing here
        // would silently drop `first`'s update.
        let result = second.update::<Counter>(&1, |c| c.value += 1);
        match result {
            Err(e) => println!(
                "  second transaction: {e}  (retriable: {})",
                e.is_retriable()
            ),
            Ok(_) => unreachable!("the stale write should have been rejected"),
        }
        println!("  ✓ no lost update");
    }

    // ------------------------------------------------------------------
    banner("4. Write skew: allowed under Snapshot");
    {
        // The rule: at least one doctor must stay on call. Each transaction
        // checks the rule, sees it satisfied, and takes a *different* doctor
        // off call — so there is no write-write conflict to catch them.
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();

        let t1_sees = t1.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t1.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let t2_sees = t2.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t2.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let names: Vec<String> = {
            let mut tx = db.begin();
            tx.scan::<Doctor>()?
                .iter()
                .map(|d| d.name.clone())
                .collect()
        };
        println!("  on the rota: {}", names.join(", "));
        println!("  t1 counts {t1_sees} on call, t2 counts {t2_sees} — both think it is safe");

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;
        t1.commit()?;
        t2.commit()?;

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✗ {remaining} doctors on call — the invariant is broken");
        println!("    Both transactions were individually legal. This is write skew,");
        println!("    and it is why Snapshot is not Serializable.");
    }

    // ------------------------------------------------------------------
    banner("5. Write skew: prevented under Serializable");
    {
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();

        // Both read both rows, exactly as before.
        let _ = t1.get::<Doctor>(&1)?;
        let _ = t1.get::<Doctor>(&2)?;
        let _ = t2.get::<Doctor>(&1)?;
        let _ = t2.get::<Doctor>(&2)?;

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;

        t1.commit()?;
        match t2.commit() {
            Err(e) => println!("  t2: {e}  (retriable: {})", e.is_retriable()),
            Ok(()) => unreachable!("t2 read a row t1 changed; it must not commit"),
        }

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✓ {remaining} doctor still on call — the invariant held");
        println!("    t2 read doctor 1, which t1 changed, so t2 could not be serialized.");
    }

    // ------------------------------------------------------------------
    banner("6. Retries are the normal way to use Serializable");
    {
        // `transaction_with` reruns the closure on a retriable failure, so the
        // abort in scenario 5 becomes invisible to the caller.
        let mut attempts = 0;
        db.transaction_with::<Serializable, _, _>(|tx| {
            attempts += 1;
            let ada = tx.get::<Doctor>(&1)?.unwrap().on_call;
            let bob = tx.get::<Doctor>(&2)?.unwrap().on_call;
            if ada && bob {
                tx.update::<Doctor>(&1, |d| d.on_call = false)?;
            }
            Ok(())
        })?;
        println!("  committed after {attempts} attempt(s)");
        println!("  ✓ write your transaction as if it runs alone; let the engine retry it");
    }

    banner("Summary");
    println!("  ReadCommitted   cheapest; non-repeatable reads and phantoms");
    println!("  RepeatableRead  stable snapshot; write skew possible");
    println!("  Snapshot        stable snapshot; write skew possible   ← default");
    println!("  Serializable    no anomalies; expect retriable aborts");

    Ok(())
}

examples/game.rs (lines 152-199)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Hero>()?;
    db.register::<Item>()?;
    db.register::<PartyMember>()?;

    let (ada, bram, cleo) = (1u64, 2u64, 3u64);

    // -----------------------------------------------------------------------
    act(1, "The world is created");

    db.transaction(|tx| {
        tx.insert(Hero {
            id: ada,
            name: "Ada".into(),
            gold: 120,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: bram,
            name: "Bram".into(),
            gold: 80,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: cleo,
            name: "Cleo".into(),
            gold: 200,
            strength: 60,
        })?;

        let mut item = |id, owner, name: &str, weight, value, kind| {
            tx.insert(Item {
                id,
                owner,
                name: name.into(),
                weight,
                value,
                kind,
                flavour: HashMap::new(),
            })
        };
        item(10, ada, "Rusty Sword", 30, 15, Kind::Weapon)?;
        item(11, ada, "Health Potion", 5, 20, Kind::Potion)?;
        item(12, bram, "Oak Shield", 40, 35, Kind::Weapon)?;
        item(13, cleo, "Lucky Charm", 2, 90, Kind::Trinket)?;
        // Loot lying in the dungeon, owned by nobody.
        item(20, GROUND, "Flaming Greatsword", 55, 500, Kind::Weapon)?;
        item(21, GROUND, "Elven Cloak", 20, 240, Kind::Trinket)?;
        item(22, GROUND, "Iron Helm", 35, 60, Kind::Weapon)?;

        // Arbitrary per-item state, in a database row, with no schema to
        // declare and no serialisation format to agree on.
        tx.update::<Item>(&20, |i| {
            i.flavour.insert("enchantment".into(), "flame".into());
            i.flavour.insert("forged_by".into(), "Durin".into());
        })?;
        Ok(())
    })?;

    // A whole transaction that fails partway leaves nothing behind. The unique
    // index on `name` refuses the second Ada, and the gold change goes with it.
    let before = gold_of(&mut db.begin(), ada)?;
    let doomed = db.transaction(|tx| {
        tx.update::<Hero>(&ada, |h| h.gold += 10_000)?;
        tx.insert(Hero {
            id: 99,
            name: "Ada".into(),
            gold: 0,
            strength: 10,
        })
    });
    println!("  a duplicate hero name: {}", doomed.unwrap_err());
    println!(
        "  Ada's gold is still {} — the whole transaction rolled back, not just the insert",
        gold_of(&mut db.begin(), ada)?
    );
    assert_eq!(gold_of(&mut db.begin(), ada)?, before);

    let mut tx = db.begin();
    for hero in [ada, bram, cleo] {
        let name = name_of(&mut tx, hero)?;
        let carried = carried(&mut tx, hero)?;
        let items = inventory(&mut tx, hero)?;
        let names: Vec<_> = items.iter().map(|(_, n, _)| n.as_str()).collect();
        println!("  {name:<5} carries {carried:>3} — {}", names.join(", "));
    }
    drop(tx);

    // -----------------------------------------------------------------------
    act(2, "Two heroes reach for the same greatsword");

    // Both begin, both see the sword unclaimed. The first to write takes it;
    // the second is told immediately rather than being made to wait.
    let mut ada_grabs = db.begin_with::<Snapshot>();
    let mut bram_grabs = db.begin_with::<Snapshot>();

    ada_grabs.update::<Item>(&20, |i| i.owner = ada)?;

    match bram_grabs.update::<Item>(&20, |i| i.owner = bram) {
        Err(e @ Error::WriteConflict { .. }) => {
            println!("  Bram: {e}");
            println!(
                "  ...retriable: {} — he can try for something else",
                e.is_retriable()
            );
        }
        other => unreachable!("expected a conflict, got {other:?}"),
    }

    // A failed write does not end the transaction. Bram takes the cloak instead.
    bram_grabs.update::<Item>(&21, |i| i.owner = bram)?;
    ada_grabs.commit()?;
    bram_grabs.commit()?;
    println!("  Ada takes the greatsword, Bram takes the cloak. Nobody blocked.");

    // -----------------------------------------------------------------------
    act(3, "Cleo overloads herself — write skew, and the fix");

    // Cleo can carry 60 and is carrying 2. Either 35kg helm fits; both do not.
    // Each transaction reads her inventory, checks the total, and picks up a
    // *different* item — so the two writes never touch the same record, and
    // first-updater-wins has nothing to catch.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.insert(Item {
            id: 23,
            owner: GROUND,
            name: "Steel Helm".into(),
            weight: 35,
            value: 70,
            kind: Kind::Weapon,
            flavour: HashMap::new(),
        })
    })?;

    let pick_up = |item: u64| {
        move |tx: &mut mvcc::Transaction<'_, Serializable>| -> Result<bool> {
            let hero = tx.get::<Hero>(&cleo)?.expect("Cleo exists");
            let strength = hero.strength;
            let load = carried(tx, cleo)?;
            let weight = tx.get::<Item>(&item)?.map(|i| i.weight).unwrap_or(0);
            if load + weight > strength {
                return Ok(false);
            }
            tx.update::<Item>(&item, |i| i.owner = cleo)?;
            Ok(true)
        }
    };

    // Under Snapshot, both checks pass against a stale inventory.
    {
        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();
        let (load1, load2) = (carried(&mut t1, cleo)?, carried(&mut t2, cleo)?);
        println!(
            "  Snapshot:     both transactions see {load1}/{load2} carried, both think a 35kg helm fits"
        );
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        t2.commit()?;
        let over = carried(&mut db.begin(), cleo)?;
        println!(
            "  ✗ Cleo now carries {over} of a possible 60. Two legal transactions, one broken rule."
        );
    }

    // Put it back and try again at Serializable.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.update::<Item>(&23, |i| i.owner = GROUND).map(|_| ())
    })?;

    {
        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();
        // Both read the inventory, as before.
        let _ = carried(&mut t1, cleo)?;
        let _ = carried(&mut t2, cleo)?;
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        match t2.commit() {
            Err(e @ Error::SerializationFailure) => println!("  Serializable: second pickup {e}"),
            other => unreachable!("expected a serialization failure, got {other:?}"),
        }
        println!(
            "  ✓ Cleo carries {} — the second pickup read an inventory the first changed.",
            carried(&mut db.begin(), cleo)?
        );
    }

    // In real code you would not hand-roll that. `transaction_with` retries,
    // and the retry re-reads the inventory and correctly declines.
    let took_it = db.transaction_with::<Serializable, _, _>(pick_up(23))?;
    println!("  ...and on retry the second helm is refused on its merits: picked_up = {took_it}");

    // -----------------------------------------------------------------------
    act(4, "The party fills up — a phantom, not a conflict");

    db.transaction(|tx| {
        tx.insert(PartyMember {
            hero: ada,
            party: PARTY,
            role: "Vanguard".into(),
        })?;
        tx.insert(PartyMember {
            hero: bram,
            party: PARTY,
            role: "Shield".into(),
        })?;
        tx.insert(PartyMember {
            hero: cleo,
            party: PARTY,
            role: "Scout".into(),
        })
    })?;

    // Two newcomers apply at once, with three of four slots taken. Neither
    // writes a row the other wrote — they insert *different* rows — so there is
    // no write conflict to detect. What they collide on is a row that did not
    // exist when either of them counted.
    let (dara, finn) = (4u64, 5u64);
    db.transaction(|tx| {
        tx.insert(Hero {
            id: dara,
            name: "Dara".into(),
            gold: 40,
            strength: 80,
        })?;
        tx.insert(Hero {
            id: finn,
            name: "Finn".into(),
            gold: 40,
            strength: 80,
        })
    })?;

    let mut d = db.begin_with::<Serializable>();
    let mut f = db.begin_with::<Serializable>();
    println!(
        "  Dara counts {} members, Finn counts {} — both see a free slot",
        party_size(&mut d)?,
        party_size(&mut f)?
    );
    d.insert(PartyMember {
        hero: dara,
        party: PARTY,
        role: "Healer".into(),
    })?;
    f.insert(PartyMember {
        hero: finn,
        party: PARTY,
        role: "Healer".into(),
    })?;
    d.commit()?;
    match f.commit() {
        Err(e @ Error::SerializationFailure) => println!("  Finn: {e}"),
        other => unreachable!("expected a serialization failure, got {other:?}"),
    }
    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan_where::<PartyMember, _>(|m| m.party == PARTY)?
        .iter()
        .map(|m| (m.hero, m.role.clone()))
        .collect();
    drop(tx);
    roster.sort();
    let size = roster.len();
    let mut tx = db.begin();
    let listed: Vec<String> = roster
        .iter()
        .map(|(h, role)| Ok(format!("{} the {role}", name_of(&mut tx, *h)?)))
        .collect::<Result<_>>()?;
    drop(tx);
    println!("  party: {}", listed.join(", "));
    println!("  ✓ {size}/{PARTY_LIMIT} filled. Finn's insert became a phantom in Dara's count.");
    println!("    (`scan_where` hands the engine the predicate, so it can re-check it at commit.)");
    assert!(size <= PARTY_LIMIT);

    // -----------------------------------------------------------------------
    act(5, "A trade — two records, one atomic step");

    let before = total_gold(&db)?;

    let trade = |seller: u64, buyer: u64, item: u64, price: i64| {
        let db = Arc::clone(&db);
        move || -> Result<bool> {
            db.transaction_with::<Serializable, _, _>(|tx| {
                let buyer_gold = gold_of(tx, buyer)?;
                if buyer_gold < price {
                    return Ok(false);
                }
                let owned_by_seller = tx
                    .get::<Item>(&item)?
                    .map(|i| i.owner == seller)
                    .unwrap_or(false);
                if !owned_by_seller {
                    return Ok(false);
                }
                tx.update::<Hero>(&buyer, |h| h.gold -= price)?;
                tx.update::<Hero>(&seller, |h| h.gold += price)?;
                tx.update::<Item>(&item, |i| i.owner = buyer)?;
                Ok(true)
            })
        }
    };

    let sold = trade(bram, cleo, 21, 150)()?;
    println!("  Bram sells the Elven Cloak to Cleo for 150g: {sold}");
    let mut tx = db.begin();
    let cloak_owner = tx.get::<Item>(&21)?.expect("cloak").owner;
    println!(
        "  Bram {}g, Cleo {}g, cloak owner is {}",
        gold_of(&mut tx, bram)?,
        gold_of(&mut tx, cleo)?,
        name_of(&mut tx, cloak_owner)?
    );
    drop(tx);
    assert_eq!(total_gold(&db)?, before, "gold must be conserved");
    println!("  ✓ total gold unchanged at {before} — no step of that was separately visible");

    // -----------------------------------------------------------------------
    act(6, "The chronicler reads while the world moves");

    // A long report scanning the whole world while combat rewrites it. It never
    // blocks a writer, and its view never shifts underneath it.
    let mut chronicler = db.begin_with::<Snapshot>();
    let opening = total_gold(&db)?;
    let seen_first = chronicler.scan::<Hero>()?.len();

    for _ in 0..50 {
        db.transaction(|tx| {
            tx.update::<Hero>(&ada, |h| h.gold += 1)?;
            tx.update::<Hero>(&bram, |h| h.gold -= 1)?;
            Ok(())
        })?;
    }
    db.transaction(|tx| {
        tx.insert(Hero {
            id: 6,
            name: "Mira".into(),
            gold: 0,
            strength: 50,
        })
    })?;

    let chron_gold: i64 = chronicler.scan::<Hero>()?.iter().map(|h| h.gold).sum();
    let seen_after = chronicler.scan::<Hero>()?.len();
    let in_world = db.begin().scan::<Hero>()?.len();
    println!("  chronicler saw {seen_first} heroes at the start and still sees {seen_after};");
    println!("  the world now holds {in_world}, after 50 commits and one new arrival");
    println!(
        "  chronicler totals {chron_gold}g; the world now totals {}g",
        total_gold(&db)?
    );
    assert_eq!(chron_gold, opening, "a snapshot must not move");
    chronicler.commit()?;
    println!("  ✓ a reader that ran across 51 commits saw exactly one consistent world");

    // -----------------------------------------------------------------------
    act(7, "The raid — many adventurers at once");

    let stop = Arc::new(AtomicBool::new(false));
    let trades = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));
    let audits = Arc::new(AtomicU64::new(0));
    let opening = total_gold(&db)?;
    let heroes = [ada, bram, cleo, dara, finn];

    // An auditor that must never catch a trade half-applied.
    let auditor = {
        let (db, stop, audits) = (Arc::clone(&db), Arc::clone(&stop), Arc::clone(&audits));
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let total: i64 = tx.scan::<Hero>()?.iter().map(|h| h.gold).sum();
                assert_eq!(total, opening, "auditor saw a half-finished trade");
                audits.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    let start = Instant::now();
    let raiders: Vec<_> = (0..4)
        .map(|t| {
            let (db, trades, retries) =
                (Arc::clone(&db), Arc::clone(&trades), Arc::clone(&retries));
            thread::spawn(move || -> Result<()> {
                let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t + 1);
                for _ in 0..400 {
                    seed ^= seed << 13;
                    seed ^= seed >> 7;
                    seed ^= seed << 17;
                    let from = heroes[(seed % heroes.len() as u64) as usize];
                    let to = heroes[((seed >> 8) % heroes.len() as u64) as usize];
                    if from == to {
                        continue;
                    }

                    let mut attempts = 0u64;
                    // Serializable, because the payment depends on a balance we
                    // read — the same shape as Cleo's carry limit in Act 3.
                    db.transaction_with::<Serializable, _, _>(|tx| {
                        attempts += 1;
                        let purse = gold_of(tx, from)?;
                        if purse < 5 {
                            return Ok(());
                        }
                        tx.update::<Hero>(&from, |h| h.gold -= 5)?;
                        tx.update::<Hero>(&to, |h| h.gold += 5)?;
                        Ok(())
                    })?;
                    trades.fetch_add(1, Ordering::Relaxed);
                    retries.fetch_add(attempts - 1, Ordering::Relaxed);
                }
                Ok(())
            })
        })
        .collect();

    for r in raiders {
        r.join().expect("raider panicked")?;
    }
    let elapsed = start.elapsed();
    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    let done = trades.load(Ordering::Relaxed);
    println!("  {done} trades across 4 threads in {elapsed:.2?}");
    println!(
        "  {} audit sweeps completed alongside them, none blocked, none torn",
        audits.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.2} per trade) — contention on 5 heroes, handled by the engine",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / done.max(1) as f64
    );
    assert_eq!(total_gold(&db)?, opening, "gold was created or destroyed");
    println!("  ✓ total gold still {opening}");

    // -----------------------------------------------------------------------
    act(8, "Closing the ledger");

    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan::<Hero>()?
        .iter()
        .map(|h| (h.id, h.name.clone(), h.gold))
        .collect();
    roster.sort_by_key(|(id, _, _)| *id);
    drop(tx);

    println!(
        "  {:<6} {:>6} {:>5} {:>7}  inventory",
        "hero", "gold", "load", "worth"
    );
    let mut tx = db.begin();
    for (id, name, gold) in &roster {
        let held = tx.scan_index(Item::OWNER, *id..=*id)?;
        let load: i32 = held.iter().map(|i| i.weight).sum();
        let worth: i64 = held.iter().map(|i| i.value).sum();
        let names: Vec<String> = held
            .iter()
            .map(|i| match i.kind {
                Kind::Weapon => format!("[wpn] {}", i.name),
                Kind::Potion => format!("[pot] {}", i.name),
                Kind::Trinket => format!("[trk] {}", i.name),
            })
            .collect();
        let names = if names.is_empty() {
            "—".to_string()
        } else {
            names.join(", ")
        };
        println!("  {name:<6} {gold:>5}g {load:>4}kg {worth:>6}g  {names}");
    }

    // Potions anywhere in the world, found by predicate rather than by key —
    // and `kind` is an ordinary Rust enum the engine knows nothing about.
    let potions = tx.scan_where::<Item, _>(|i| i.kind == Kind::Potion)?.len();
    let sword = tx.get::<Item>(&20)?.expect("greatsword");
    let enchantment = sword
        .flavour
        .get("enchantment")
        .cloned()
        .unwrap_or_default();
    let forged_by = sword.flavour.get("forged_by").cloned().unwrap_or_default();
    drop(tx);
    // The rule Act 3 was about, checked rather than asserted in prose.
    let mut tx = db.begin();
    for (id, name, _) in &roster {
        let strength = tx.get::<Hero>(id)?.map(|h| h.strength).unwrap_or(0);
        let load = carried(&mut tx, *id)?;
        assert!(
            load <= strength,
            "{name} is over their carry limit: {load} > {strength}"
        );
    }
    drop(tx);

    println!("\n  potions in the world: {potions}");
    println!("  the greatsword is {enchantment}-enchanted, forged by {forged_by}");

    // Nothing is reclaimed while a transaction that could still reach it is
    // alive. This is the number to watch in a long-lived process: a forgotten
    // transaction pins it and version chains grow without limit.
    let stats = db.stats();
    println!(
        "\n  gc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );
    println!("  (a watermark that stops moving while writes continue is the leak to look for)");

    Ok(())
}

pub fn transaction_with<I, R, F>(&self, f: F) -> Result<R>

where I: IsolationLevel, F: FnMut(&mut Transaction<'_, I>) -> Result<R>,

Run f in a transaction at isolation level I, retrying while it fails retriably, and commit.

Reach for Serializable here when a transaction’s write depends on a value it merely read — a balance check, a capacity limit. That is the write-skew shape, and snapshot isolation does not catch it.

// Withdraw only if the balance covers it. The check and the write must
// be serializable together, or two concurrent transfers each see a
// sufficient balance and both withdraw.
let moved = db.transaction_with::<Serializable, _, _>(|tx| {
    let balance = tx.get::<Account>(&1)?.map_or(0, |a| a.balance);
    if balance < 50 {
        return Ok(false);
    }
    tx.update::<Account>(&1, |a| a.balance -= 50)?;
    tx.update::<Account>(&2, |a| a.balance += 50)?;
    Ok(true)
})?;

assert!(moved);

As with Database::transaction, f may run more than once, so it must not have side effects outside the transaction. Return the value and let the caller act on the committed result, as above.

Examples found in repository

examples/concurrent.rs (lines 100-109)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Account>()?;

    db.transaction(|tx| {
        for id in 0..ACCOUNTS {
            tx.insert(Account {
                id,
                balance: STARTING_BALANCE,
            })?;
        }
        Ok(())
    })?;

    let total_before = total(&db)?;
    println!("starting total: {total_before}");

    let stop = Arc::new(AtomicBool::new(false));
    let reads = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));

    // ---- a reader that must never observe a torn transfer -----------------
    // Money moves between accounts constantly. Under snapshot isolation this
    // thread's scans must always sum to the same total: it either sees both
    // halves of a transfer or neither.
    let auditor = {
        let db = Arc::clone(&db);
        let stop = Arc::clone(&stop);
        let reads = Arc::clone(&reads);
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let sum: i64 = tx.scan::<Account>()?.iter().map(|a| a.balance).sum();
                assert_eq!(
                    sum,
                    ACCOUNTS as i64 * STARTING_BALANCE,
                    "a reader observed a partially applied transfer"
                );
                reads.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    // ---- writers ----------------------------------------------------------
    let start = Instant::now();
    let mut writers = Vec::new();
    for t in 0..TRANSFER_THREADS {
        let db = Arc::clone(&db);
        let retries = Arc::clone(&retries);
        writers.push(thread::spawn(move || -> Result<()> {
            // A cheap per-thread PRNG; no dependency needed for this.
            let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t as u64 + 1);
            let mut next = move || {
                seed ^= seed << 13;
                seed ^= seed >> 7;
                seed ^= seed << 17;
                seed
            };

            for _ in 0..TRANSFERS_PER_THREAD {
                let from = next() % ACCOUNTS;
                let to = next() % ACCOUNTS;
                if from == to {
                    continue;
                }

                let mut attempts = 0u64;
                // Serializable, because the transfer's *write* depends on a
                // balance it *read* — the write-skew shape. Snapshot isolation
                // would let two concurrent transfers both pass the check.
                db.transaction_with::<Serializable, _, _>(|tx| {
                    attempts += 1;
                    let balance = tx.get::<Account>(&from)?.map(|a| a.balance).unwrap_or(0);
                    if balance < 10 {
                        return Ok(());
                    }
                    tx.update::<Account>(&from, |a| a.balance -= 10)?;
                    tx.update::<Account>(&to, |a| a.balance += 10)?;
                    Ok(())
                })?;
                retries.fetch_add(attempts - 1, Ordering::Relaxed);
            }
            Ok(())
        }));
    }

    for w in writers {
        w.join().expect("writer panicked")?;
    }
    let elapsed = start.elapsed();

    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    // ---- results ----------------------------------------------------------
    let total_after = total(&db)?;
    let committed = (TRANSFER_THREADS * TRANSFERS_PER_THREAD) as u64;

    println!("final total:    {total_after}");
    assert_eq!(total_before, total_after, "money was created or destroyed");

    println!("\n{committed} transfers in {elapsed:.2?}");
    println!(
        "  {:.0} transfers/sec",
        committed as f64 / elapsed.as_secs_f64()
    );
    println!(
        "  {} concurrent audit scans completed, none of them blocked",
        reads.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.1} per transfer)",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / committed as f64
    );

    // The GC watermark should be advancing. If `active_transactions` climbs and
    // the watermark stalls, some transaction is being leaked — see engine::gc.
    let stats = db.stats();
    println!(
        "\ngc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );

    println!("\n✓ no torn reads, no lost updates, conservation held");

    // Slow readers are what pin the GC watermark. Left here as the shape of the
    // thing to watch for, not as something this example demonstrates going wrong.
    thread::sleep(Duration::from_millis(1));
    Ok(())
}

examples/isolation.rs (lines 185-193)

fn main() -> Result<()> {
    let db = Database::open(Config::in_memory())?;
    db.register::<Doctor>()?;
    db.register::<Counter>()?;

    db.transaction(|tx| {
        tx.insert(Doctor {
            id: 1,
            name: "ada".into(),
            on_call: true,
        })?;
        tx.insert(Doctor {
            id: 2,
            name: "bob".into(),
            on_call: true,
        })?;
        tx.insert(Counter { id: 1, value: 0 })
    })?;

    // ------------------------------------------------------------------
    banner("1. Snapshot: a reader is unaffected by concurrent commits");
    {
        let mut reader = db.begin_with::<Snapshot>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        // Someone else commits, start to finish, while `reader` is open.
        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 42).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_eq!(before, after, "snapshot isolation must be stable");
        println!("  ✓ the snapshot held: readers never block and never change under you");
    }

    // ------------------------------------------------------------------
    banner("2. ReadCommitted: each statement sees a fresh snapshot");
    {
        let mut reader = db.begin_with::<ReadCommitted>();
        let before = reader.get::<Counter>(&1)?.unwrap().value;

        db.transaction(|tx| tx.update::<Counter>(&1, |c| c.value = 99).map(|_| ()))?;

        let after = reader.get::<Counter>(&1)?.unwrap().value;
        println!("  reader saw {before} before, {after} after a concurrent commit");
        assert_ne!(
            before, after,
            "read committed should observe the new commit"
        );
        println!("  ✓ non-repeatable read — the tradeoff this level makes for cheapness");
    }

    // ------------------------------------------------------------------
    banner("3. Write-write conflict: first committer wins");
    {
        let mut first = db.begin_with::<Snapshot>();
        let mut second = db.begin_with::<Snapshot>();

        first.update::<Counter>(&1, |c| c.value += 1)?;
        first.commit()?;

        // `second` took its snapshot before `first` committed, so writing here
        // would silently drop `first`'s update.
        let result = second.update::<Counter>(&1, |c| c.value += 1);
        match result {
            Err(e) => println!(
                "  second transaction: {e}  (retriable: {})",
                e.is_retriable()
            ),
            Ok(_) => unreachable!("the stale write should have been rejected"),
        }
        println!("  ✓ no lost update");
    }

    // ------------------------------------------------------------------
    banner("4. Write skew: allowed under Snapshot");
    {
        // The rule: at least one doctor must stay on call. Each transaction
        // checks the rule, sees it satisfied, and takes a *different* doctor
        // off call — so there is no write-write conflict to catch them.
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();

        let t1_sees = t1.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t1.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let t2_sees = t2.get::<Doctor>(&1)?.unwrap().on_call as u8
            + t2.get::<Doctor>(&2)?.unwrap().on_call as u8;
        let names: Vec<String> = {
            let mut tx = db.begin();
            tx.scan::<Doctor>()?
                .iter()
                .map(|d| d.name.clone())
                .collect()
        };
        println!("  on the rota: {}", names.join(", "));
        println!("  t1 counts {t1_sees} on call, t2 counts {t2_sees} — both think it is safe");

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;
        t1.commit()?;
        t2.commit()?;

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✗ {remaining} doctors on call — the invariant is broken");
        println!("    Both transactions were individually legal. This is write skew,");
        println!("    and it is why Snapshot is not Serializable.");
    }

    // ------------------------------------------------------------------
    banner("5. Write skew: prevented under Serializable");
    {
        db.transaction(|tx| {
            tx.update::<Doctor>(&1, |d| d.on_call = true)?;
            tx.update::<Doctor>(&2, |d| d.on_call = true).map(|_| ())
        })?;

        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();

        // Both read both rows, exactly as before.
        let _ = t1.get::<Doctor>(&1)?;
        let _ = t1.get::<Doctor>(&2)?;
        let _ = t2.get::<Doctor>(&1)?;
        let _ = t2.get::<Doctor>(&2)?;

        t1.update::<Doctor>(&1, |d| d.on_call = false)?;
        t2.update::<Doctor>(&2, |d| d.on_call = false)?;

        t1.commit()?;
        match t2.commit() {
            Err(e) => println!("  t2: {e}  (retriable: {})", e.is_retriable()),
            Ok(()) => unreachable!("t2 read a row t1 changed; it must not commit"),
        }

        let mut check = db.begin();
        let remaining = check.get::<Doctor>(&1)?.unwrap().on_call as u8
            + check.get::<Doctor>(&2)?.unwrap().on_call as u8;
        println!("  ✓ {remaining} doctor still on call — the invariant held");
        println!("    t2 read doctor 1, which t1 changed, so t2 could not be serialized.");
    }

    // ------------------------------------------------------------------
    banner("6. Retries are the normal way to use Serializable");
    {
        // `transaction_with` reruns the closure on a retriable failure, so the
        // abort in scenario 5 becomes invisible to the caller.
        let mut attempts = 0;
        db.transaction_with::<Serializable, _, _>(|tx| {
            attempts += 1;
            let ada = tx.get::<Doctor>(&1)?.unwrap().on_call;
            let bob = tx.get::<Doctor>(&2)?.unwrap().on_call;
            if ada && bob {
                tx.update::<Doctor>(&1, |d| d.on_call = false)?;
            }
            Ok(())
        })?;
        println!("  committed after {attempts} attempt(s)");
        println!("  ✓ write your transaction as if it runs alone; let the engine retry it");
    }

    banner("Summary");
    println!("  ReadCommitted   cheapest; non-repeatable reads and phantoms");
    println!("  RepeatableRead  stable snapshot; write skew possible");
    println!("  Snapshot        stable snapshot; write skew possible   ← default");
    println!("  Serializable    no anomalies; expect retriable aborts");

    Ok(())
}

examples/game.rs (line 336)

fn main() -> Result<()> {
    let db = Arc::new(Database::open(Config::in_memory())?);
    db.register::<Hero>()?;
    db.register::<Item>()?;
    db.register::<PartyMember>()?;

    let (ada, bram, cleo) = (1u64, 2u64, 3u64);

    // -----------------------------------------------------------------------
    act(1, "The world is created");

    db.transaction(|tx| {
        tx.insert(Hero {
            id: ada,
            name: "Ada".into(),
            gold: 120,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: bram,
            name: "Bram".into(),
            gold: 80,
            strength: 100,
        })?;
        tx.insert(Hero {
            id: cleo,
            name: "Cleo".into(),
            gold: 200,
            strength: 60,
        })?;

        let mut item = |id, owner, name: &str, weight, value, kind| {
            tx.insert(Item {
                id,
                owner,
                name: name.into(),
                weight,
                value,
                kind,
                flavour: HashMap::new(),
            })
        };
        item(10, ada, "Rusty Sword", 30, 15, Kind::Weapon)?;
        item(11, ada, "Health Potion", 5, 20, Kind::Potion)?;
        item(12, bram, "Oak Shield", 40, 35, Kind::Weapon)?;
        item(13, cleo, "Lucky Charm", 2, 90, Kind::Trinket)?;
        // Loot lying in the dungeon, owned by nobody.
        item(20, GROUND, "Flaming Greatsword", 55, 500, Kind::Weapon)?;
        item(21, GROUND, "Elven Cloak", 20, 240, Kind::Trinket)?;
        item(22, GROUND, "Iron Helm", 35, 60, Kind::Weapon)?;

        // Arbitrary per-item state, in a database row, with no schema to
        // declare and no serialisation format to agree on.
        tx.update::<Item>(&20, |i| {
            i.flavour.insert("enchantment".into(), "flame".into());
            i.flavour.insert("forged_by".into(), "Durin".into());
        })?;
        Ok(())
    })?;

    // A whole transaction that fails partway leaves nothing behind. The unique
    // index on `name` refuses the second Ada, and the gold change goes with it.
    let before = gold_of(&mut db.begin(), ada)?;
    let doomed = db.transaction(|tx| {
        tx.update::<Hero>(&ada, |h| h.gold += 10_000)?;
        tx.insert(Hero {
            id: 99,
            name: "Ada".into(),
            gold: 0,
            strength: 10,
        })
    });
    println!("  a duplicate hero name: {}", doomed.unwrap_err());
    println!(
        "  Ada's gold is still {} — the whole transaction rolled back, not just the insert",
        gold_of(&mut db.begin(), ada)?
    );
    assert_eq!(gold_of(&mut db.begin(), ada)?, before);

    let mut tx = db.begin();
    for hero in [ada, bram, cleo] {
        let name = name_of(&mut tx, hero)?;
        let carried = carried(&mut tx, hero)?;
        let items = inventory(&mut tx, hero)?;
        let names: Vec<_> = items.iter().map(|(_, n, _)| n.as_str()).collect();
        println!("  {name:<5} carries {carried:>3} — {}", names.join(", "));
    }
    drop(tx);

    // -----------------------------------------------------------------------
    act(2, "Two heroes reach for the same greatsword");

    // Both begin, both see the sword unclaimed. The first to write takes it;
    // the second is told immediately rather than being made to wait.
    let mut ada_grabs = db.begin_with::<Snapshot>();
    let mut bram_grabs = db.begin_with::<Snapshot>();

    ada_grabs.update::<Item>(&20, |i| i.owner = ada)?;

    match bram_grabs.update::<Item>(&20, |i| i.owner = bram) {
        Err(e @ Error::WriteConflict { .. }) => {
            println!("  Bram: {e}");
            println!(
                "  ...retriable: {} — he can try for something else",
                e.is_retriable()
            );
        }
        other => unreachable!("expected a conflict, got {other:?}"),
    }

    // A failed write does not end the transaction. Bram takes the cloak instead.
    bram_grabs.update::<Item>(&21, |i| i.owner = bram)?;
    ada_grabs.commit()?;
    bram_grabs.commit()?;
    println!("  Ada takes the greatsword, Bram takes the cloak. Nobody blocked.");

    // -----------------------------------------------------------------------
    act(3, "Cleo overloads herself — write skew, and the fix");

    // Cleo can carry 60 and is carrying 2. Either 35kg helm fits; both do not.
    // Each transaction reads her inventory, checks the total, and picks up a
    // *different* item — so the two writes never touch the same record, and
    // first-updater-wins has nothing to catch.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.insert(Item {
            id: 23,
            owner: GROUND,
            name: "Steel Helm".into(),
            weight: 35,
            value: 70,
            kind: Kind::Weapon,
            flavour: HashMap::new(),
        })
    })?;

    let pick_up = |item: u64| {
        move |tx: &mut mvcc::Transaction<'_, Serializable>| -> Result<bool> {
            let hero = tx.get::<Hero>(&cleo)?.expect("Cleo exists");
            let strength = hero.strength;
            let load = carried(tx, cleo)?;
            let weight = tx.get::<Item>(&item)?.map(|i| i.weight).unwrap_or(0);
            if load + weight > strength {
                return Ok(false);
            }
            tx.update::<Item>(&item, |i| i.owner = cleo)?;
            Ok(true)
        }
    };

    // Under Snapshot, both checks pass against a stale inventory.
    {
        let mut t1 = db.begin_with::<Snapshot>();
        let mut t2 = db.begin_with::<Snapshot>();
        let (load1, load2) = (carried(&mut t1, cleo)?, carried(&mut t2, cleo)?);
        println!(
            "  Snapshot:     both transactions see {load1}/{load2} carried, both think a 35kg helm fits"
        );
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        t2.commit()?;
        let over = carried(&mut db.begin(), cleo)?;
        println!(
            "  ✗ Cleo now carries {over} of a possible 60. Two legal transactions, one broken rule."
        );
    }

    // Put it back and try again at Serializable.
    db.transaction(|tx| {
        tx.update::<Item>(&22, |i| i.owner = GROUND)?;
        tx.update::<Item>(&23, |i| i.owner = GROUND).map(|_| ())
    })?;

    {
        let mut t1 = db.begin_with::<Serializable>();
        let mut t2 = db.begin_with::<Serializable>();
        // Both read the inventory, as before.
        let _ = carried(&mut t1, cleo)?;
        let _ = carried(&mut t2, cleo)?;
        t1.update::<Item>(&22, |i| i.owner = cleo)?;
        t2.update::<Item>(&23, |i| i.owner = cleo)?;
        t1.commit()?;
        match t2.commit() {
            Err(e @ Error::SerializationFailure) => println!("  Serializable: second pickup {e}"),
            other => unreachable!("expected a serialization failure, got {other:?}"),
        }
        println!(
            "  ✓ Cleo carries {} — the second pickup read an inventory the first changed.",
            carried(&mut db.begin(), cleo)?
        );
    }

    // In real code you would not hand-roll that. `transaction_with` retries,
    // and the retry re-reads the inventory and correctly declines.
    let took_it = db.transaction_with::<Serializable, _, _>(pick_up(23))?;
    println!("  ...and on retry the second helm is refused on its merits: picked_up = {took_it}");

    // -----------------------------------------------------------------------
    act(4, "The party fills up — a phantom, not a conflict");

    db.transaction(|tx| {
        tx.insert(PartyMember {
            hero: ada,
            party: PARTY,
            role: "Vanguard".into(),
        })?;
        tx.insert(PartyMember {
            hero: bram,
            party: PARTY,
            role: "Shield".into(),
        })?;
        tx.insert(PartyMember {
            hero: cleo,
            party: PARTY,
            role: "Scout".into(),
        })
    })?;

    // Two newcomers apply at once, with three of four slots taken. Neither
    // writes a row the other wrote — they insert *different* rows — so there is
    // no write conflict to detect. What they collide on is a row that did not
    // exist when either of them counted.
    let (dara, finn) = (4u64, 5u64);
    db.transaction(|tx| {
        tx.insert(Hero {
            id: dara,
            name: "Dara".into(),
            gold: 40,
            strength: 80,
        })?;
        tx.insert(Hero {
            id: finn,
            name: "Finn".into(),
            gold: 40,
            strength: 80,
        })
    })?;

    let mut d = db.begin_with::<Serializable>();
    let mut f = db.begin_with::<Serializable>();
    println!(
        "  Dara counts {} members, Finn counts {} — both see a free slot",
        party_size(&mut d)?,
        party_size(&mut f)?
    );
    d.insert(PartyMember {
        hero: dara,
        party: PARTY,
        role: "Healer".into(),
    })?;
    f.insert(PartyMember {
        hero: finn,
        party: PARTY,
        role: "Healer".into(),
    })?;
    d.commit()?;
    match f.commit() {
        Err(e @ Error::SerializationFailure) => println!("  Finn: {e}"),
        other => unreachable!("expected a serialization failure, got {other:?}"),
    }
    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan_where::<PartyMember, _>(|m| m.party == PARTY)?
        .iter()
        .map(|m| (m.hero, m.role.clone()))
        .collect();
    drop(tx);
    roster.sort();
    let size = roster.len();
    let mut tx = db.begin();
    let listed: Vec<String> = roster
        .iter()
        .map(|(h, role)| Ok(format!("{} the {role}", name_of(&mut tx, *h)?)))
        .collect::<Result<_>>()?;
    drop(tx);
    println!("  party: {}", listed.join(", "));
    println!("  ✓ {size}/{PARTY_LIMIT} filled. Finn's insert became a phantom in Dara's count.");
    println!("    (`scan_where` hands the engine the predicate, so it can re-check it at commit.)");
    assert!(size <= PARTY_LIMIT);

    // -----------------------------------------------------------------------
    act(5, "A trade — two records, one atomic step");

    let before = total_gold(&db)?;

    let trade = |seller: u64, buyer: u64, item: u64, price: i64| {
        let db = Arc::clone(&db);
        move || -> Result<bool> {
            db.transaction_with::<Serializable, _, _>(|tx| {
                let buyer_gold = gold_of(tx, buyer)?;
                if buyer_gold < price {
                    return Ok(false);
                }
                let owned_by_seller = tx
                    .get::<Item>(&item)?
                    .map(|i| i.owner == seller)
                    .unwrap_or(false);
                if !owned_by_seller {
                    return Ok(false);
                }
                tx.update::<Hero>(&buyer, |h| h.gold -= price)?;
                tx.update::<Hero>(&seller, |h| h.gold += price)?;
                tx.update::<Item>(&item, |i| i.owner = buyer)?;
                Ok(true)
            })
        }
    };

    let sold = trade(bram, cleo, 21, 150)()?;
    println!("  Bram sells the Elven Cloak to Cleo for 150g: {sold}");
    let mut tx = db.begin();
    let cloak_owner = tx.get::<Item>(&21)?.expect("cloak").owner;
    println!(
        "  Bram {}g, Cleo {}g, cloak owner is {}",
        gold_of(&mut tx, bram)?,
        gold_of(&mut tx, cleo)?,
        name_of(&mut tx, cloak_owner)?
    );
    drop(tx);
    assert_eq!(total_gold(&db)?, before, "gold must be conserved");
    println!("  ✓ total gold unchanged at {before} — no step of that was separately visible");

    // -----------------------------------------------------------------------
    act(6, "The chronicler reads while the world moves");

    // A long report scanning the whole world while combat rewrites it. It never
    // blocks a writer, and its view never shifts underneath it.
    let mut chronicler = db.begin_with::<Snapshot>();
    let opening = total_gold(&db)?;
    let seen_first = chronicler.scan::<Hero>()?.len();

    for _ in 0..50 {
        db.transaction(|tx| {
            tx.update::<Hero>(&ada, |h| h.gold += 1)?;
            tx.update::<Hero>(&bram, |h| h.gold -= 1)?;
            Ok(())
        })?;
    }
    db.transaction(|tx| {
        tx.insert(Hero {
            id: 6,
            name: "Mira".into(),
            gold: 0,
            strength: 50,
        })
    })?;

    let chron_gold: i64 = chronicler.scan::<Hero>()?.iter().map(|h| h.gold).sum();
    let seen_after = chronicler.scan::<Hero>()?.len();
    let in_world = db.begin().scan::<Hero>()?.len();
    println!("  chronicler saw {seen_first} heroes at the start and still sees {seen_after};");
    println!("  the world now holds {in_world}, after 50 commits and one new arrival");
    println!(
        "  chronicler totals {chron_gold}g; the world now totals {}g",
        total_gold(&db)?
    );
    assert_eq!(chron_gold, opening, "a snapshot must not move");
    chronicler.commit()?;
    println!("  ✓ a reader that ran across 51 commits saw exactly one consistent world");

    // -----------------------------------------------------------------------
    act(7, "The raid — many adventurers at once");

    let stop = Arc::new(AtomicBool::new(false));
    let trades = Arc::new(AtomicU64::new(0));
    let retries = Arc::new(AtomicU64::new(0));
    let audits = Arc::new(AtomicU64::new(0));
    let opening = total_gold(&db)?;
    let heroes = [ada, bram, cleo, dara, finn];

    // An auditor that must never catch a trade half-applied.
    let auditor = {
        let (db, stop, audits) = (Arc::clone(&db), Arc::clone(&stop), Arc::clone(&audits));
        thread::spawn(move || -> Result<()> {
            while !stop.load(Ordering::Relaxed) {
                let mut tx = db.begin();
                let total: i64 = tx.scan::<Hero>()?.iter().map(|h| h.gold).sum();
                assert_eq!(total, opening, "auditor saw a half-finished trade");
                audits.fetch_add(1, Ordering::Relaxed);
            }
            Ok(())
        })
    };

    let start = Instant::now();
    let raiders: Vec<_> = (0..4)
        .map(|t| {
            let (db, trades, retries) =
                (Arc::clone(&db), Arc::clone(&trades), Arc::clone(&retries));
            thread::spawn(move || -> Result<()> {
                let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t + 1);
                for _ in 0..400 {
                    seed ^= seed << 13;
                    seed ^= seed >> 7;
                    seed ^= seed << 17;
                    let from = heroes[(seed % heroes.len() as u64) as usize];
                    let to = heroes[((seed >> 8) % heroes.len() as u64) as usize];
                    if from == to {
                        continue;
                    }

                    let mut attempts = 0u64;
                    // Serializable, because the payment depends on a balance we
                    // read — the same shape as Cleo's carry limit in Act 3.
                    db.transaction_with::<Serializable, _, _>(|tx| {
                        attempts += 1;
                        let purse = gold_of(tx, from)?;
                        if purse < 5 {
                            return Ok(());
                        }
                        tx.update::<Hero>(&from, |h| h.gold -= 5)?;
                        tx.update::<Hero>(&to, |h| h.gold += 5)?;
                        Ok(())
                    })?;
                    trades.fetch_add(1, Ordering::Relaxed);
                    retries.fetch_add(attempts - 1, Ordering::Relaxed);
                }
                Ok(())
            })
        })
        .collect();

    for r in raiders {
        r.join().expect("raider panicked")?;
    }
    let elapsed = start.elapsed();
    stop.store(true, Ordering::Relaxed);
    auditor.join().expect("auditor panicked")?;

    let done = trades.load(Ordering::Relaxed);
    println!("  {done} trades across 4 threads in {elapsed:.2?}");
    println!(
        "  {} audit sweeps completed alongside them, none blocked, none torn",
        audits.load(Ordering::Relaxed)
    );
    println!(
        "  {} retries ({:.2} per trade) — contention on 5 heroes, handled by the engine",
        retries.load(Ordering::Relaxed),
        retries.load(Ordering::Relaxed) as f64 / done.max(1) as f64
    );
    assert_eq!(total_gold(&db)?, opening, "gold was created or destroyed");
    println!("  ✓ total gold still {opening}");

    // -----------------------------------------------------------------------
    act(8, "Closing the ledger");

    let mut tx = db.begin();
    let mut roster: Vec<_> = tx
        .scan::<Hero>()?
        .iter()
        .map(|h| (h.id, h.name.clone(), h.gold))
        .collect();
    roster.sort_by_key(|(id, _, _)| *id);
    drop(tx);

    println!(
        "  {:<6} {:>6} {:>5} {:>7}  inventory",
        "hero", "gold", "load", "worth"
    );
    let mut tx = db.begin();
    for (id, name, gold) in &roster {
        let held = tx.scan_index(Item::OWNER, *id..=*id)?;
        let load: i32 = held.iter().map(|i| i.weight).sum();
        let worth: i64 = held.iter().map(|i| i.value).sum();
        let names: Vec<String> = held
            .iter()
            .map(|i| match i.kind {
                Kind::Weapon => format!("[wpn] {}", i.name),
                Kind::Potion => format!("[pot] {}", i.name),
                Kind::Trinket => format!("[trk] {}", i.name),
            })
            .collect();
        let names = if names.is_empty() {
            "—".to_string()
        } else {
            names.join(", ")
        };
        println!("  {name:<6} {gold:>5}g {load:>4}kg {worth:>6}g  {names}");
    }

    // Potions anywhere in the world, found by predicate rather than by key —
    // and `kind` is an ordinary Rust enum the engine knows nothing about.
    let potions = tx.scan_where::<Item, _>(|i| i.kind == Kind::Potion)?.len();
    let sword = tx.get::<Item>(&20)?.expect("greatsword");
    let enchantment = sword
        .flavour
        .get("enchantment")
        .cloned()
        .unwrap_or_default();
    let forged_by = sword.flavour.get("forged_by").cloned().unwrap_or_default();
    drop(tx);
    // The rule Act 3 was about, checked rather than asserted in prose.
    let mut tx = db.begin();
    for (id, name, _) in &roster {
        let strength = tx.get::<Hero>(id)?.map(|h| h.strength).unwrap_or(0);
        let load = carried(&mut tx, *id)?;
        assert!(
            load <= strength,
            "{name} is over their carry limit: {load} > {strength}"
        );
    }
    drop(tx);

    println!("\n  potions in the world: {potions}");
    println!("  the greatsword is {enchantment}-enchanted, forged by {forged_by}");

    // Nothing is reclaimed while a transaction that could still reach it is
    // alive. This is the number to watch in a long-lived process: a forgotten
    // transaction pins it and version chains grow without limit.
    let stats = db.stats();
    println!(
        "\n  gc watermark {:?}, {} transactions still live",
        stats.watermark, stats.active_transactions
    );
    println!("  (a watermark that stops moving while writes continue is the leak to look for)");

    Ok(())
}