Struct Transaction 

pub struct Transaction<'db, I: IsolationLevel> { /* private fields */ }

A transaction at isolation level I.

I is a typestate: Transaction<'_, ReadCommitted> and Transaction<'_, Serializable> are different types with different generated code. See crate::core::isolation.

Implementations

impl<'db, I: IsolationLevel> Transaction<'db, I>

pub fn id(&self) -> TxnId

This transaction’s id, drawn from the same counter as commit timestamps. Stable for its whole life, and never reused.

pub fn snapshot(&self) -> Timestamp

The snapshot this transaction reads at.

pub fn get<T: Versioned>(&mut self, key: &T::Key) -> Result<Option<Ref<'_, T>>>

Read by primary key.

Returns a guard over the version in place rather than a copy. Reads take no locks and never block, whatever else is happening to the record.

Examples found in repository

examples/game.rs (line 111)

fn gold_of<I: mvcc::IsolationLevel>(tx: &mut mvcc::Transaction<'_, I>, hero: u64) -> Result<i64> {
    Ok(tx.get::<Hero>(&hero)?.map(|h| h.gold).unwrap_or(0))
}

fn name_of<I: mvcc::IsolationLevel>(
    tx: &mut mvcc::Transaction<'_, I>,
    hero: u64,
) -> Result<String> {
    Ok(tx
        .get::<Hero>(&hero)?
        .map(|h| h.name.clone())
        .unwrap_or_else(|| "?".into()))
}

fn party_size<I: mvcc::IsolationLevel>(tx: &mut mvcc::Transaction<'_, I>) -> Result<usize> {
    Ok(tx.scan_where::<PartyMember, _>(|m| m.party == PARTY)?.len())
}

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 55)

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 102)

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 55)

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 insert<T: Versioned>(&mut self, value: T) -> Result<()>

Insert a new record.

Fails with Error::DuplicateKey if the primary key already has a visible, non-deleted version, or if a unique index would be violated, and with Error::WriteConflict if another in-flight transaction is writing one of the same unique keys.

Examples found in repository

examples/basic.rs (lines 32-37)

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 35-38)

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 38-42)

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 153-158)

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 update<T: Versioned>( &mut self, key: &T::Key, f: impl FnOnce(&mut T), ) -> Result<bool>

Read-modify-write by primary key. Returns false if the record does not exist at this snapshot.

A closure rather than a &mut guard, deliberately: the engine needs to know exactly when mutation ends so it can install the version and update the indexes. A guard doing that work in Drop could not report a write conflict, because Drop has nowhere to return an error to.

Examples found in repository

examples/basic.rs (line 64)

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 106)

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 58)

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 194-197)

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 delete<T: Versioned>(&mut self, key: &T::Key) -> Result<bool>

Delete by primary key. Returns false if the record does not exist at this snapshot.

Installs a tombstone version rather than removing anything, so readers at older snapshots still see the record alive.

Examples found in repository

examples/basic.rs (line 99)

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(())
}

pub fn scan<T: Versioned>(&mut self) -> Result<Vec<Ref<'_, T>>>

Every record of T visible at this snapshot, in primary key order.

Equivalent to Transaction::scan_where with a predicate that accepts everything. Under Serializable that means the whole table becomes part of the read set, so any concurrent insert or update anywhere in it will abort this transaction. Prefer scan_where when you have a predicate: it is both faster to validate and far less likely to abort.

Examples found in repository

examples/game.rs (line 130)

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 72)

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 61)

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 125)

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 scan_where<T, P>(&mut self, predicate: P) -> Result<Vec<Ref<'_, T>>>

where T: Versioned, P: Fn(&T) -> bool + Send + Sync + 'static,

Every record of T visible at this snapshot that satisfies predicate, in primary key order.

The predicate is handed to the engine rather than applied by the caller afterwards, and that is the entire point under Serializable: the engine records it, re-evaluates it at commit, and aborts if the set of matching rows changed. That is what makes phantoms visible — see PredicateRead — and it is why the first form below is a materially stronger statement than the second, even though both return the same rows:

db.transaction_with::<Serializable, _, _>(|tx| {
    // The engine holds the predicate and re-runs it at commit, so an
    // overdrawn account appearing concurrently aborts this transaction.
    let overdrawn = tx.scan_where::<Account, _>(|a| a.balance < 0)?;
    assert_eq!(overdrawn.len(), 1);

    // Same rows, weaker claim: this says only "I read the whole table",
    // so *any* concurrent write to it aborts, and no phantom is named.
    let same: Vec<_> = tx
        .scan::<Account>()?
        .into_iter()
        .filter(|a| a.balance < 0)
        .collect();
    assert_eq!(same.len(), 1);
    Ok(())
})?;

predicate must be Send + Sync + 'static because it is retained for the transaction’s lifetime and re-run at commit. It should be pure: it will be called more than once, and on rows this call never returns.

Examples found in repository

examples/game.rs (line 125)

fn party_size<I: mvcc::IsolationLevel>(tx: &mut mvcc::Transaction<'_, I>) -> Result<usize> {
    Ok(tx.scan_where::<PartyMember, _>(|m| m.party == PARTY)?.len())
}

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(())
}

pub fn scan_index<T, K, R>( &mut self, index: Index<T, K>, range: R, ) -> Result<Vec<Ref<'_, T>>>

where T: Versioned, K: Encodable, R: RangeBounds<K>,

Range scan over a secondary index, in index key order.

The index is the associated const the derive emitted for the field — #[mvcc(index)] owner: u64 gives Item::OWNER — so a wrong name does not compile, and neither does a range of the wrong type. Naming an index by string would make both runtime outcomes instead: a bad name an error, a mistyped range an empty result.

use mvcc::{Config, Database, Mvcc};

#[derive(Mvcc, Clone)]
struct Item {
    #[mvcc(primary_key)] id: u64,
    #[mvcc(index)] owner: u64,
}

let db = Database::open(Config::in_memory())?;
db.register::<Item>()?;
let mut tx = db.begin();
let owned = tx.scan_index(Item::OWNER, 1u64..=3)?;
assert!(owned.is_empty());

Scanning that index with a range of the wrong type does not compile:

// `owner` is a `u64`, so a `String` range has nothing to unify with.
tx.scan_index(Item::OWNER, "a".to_string().."b".to_string()).unwrap();

Under Serializable the range is recorded and re-scanned at commit, so a concurrent insert into the range aborts this transaction while one outside it does not.

Examples found in repository

examples/game.rs (line 94)

fn inventory<I: mvcc::IsolationLevel>(
    tx: &mut mvcc::Transaction<'_, I>,
    hero: u64,
) -> Result<Vec<(u64, String, i32)>> {
    Ok(tx
        .scan_index(Item::OWNER, hero..=hero)?
        .iter()
        .map(|i| (i.id, i.name.clone(), i.weight))
        .collect())
}

/// How much a hero is carrying. Uses the same index read, so under
/// `Serializable` it is *this* that a concurrent pickup invalidates.
fn carried<I: mvcc::IsolationLevel>(tx: &mut mvcc::Transaction<'_, I>, hero: u64) -> Result<i32> {
    Ok(tx
        .scan_index(Item::OWNER, hero..=hero)?
        .iter()
        .map(|i| i.weight)
        .sum())
}

fn gold_of<I: mvcc::IsolationLevel>(tx: &mut mvcc::Transaction<'_, I>, hero: u64) -> Result<i64> {
    Ok(tx.get::<Hero>(&hero)?.map(|h| h.gold).unwrap_or(0))
}

fn name_of<I: mvcc::IsolationLevel>(
    tx: &mut mvcc::Transaction<'_, I>,
    hero: u64,
) -> Result<String> {
    Ok(tx
        .get::<Hero>(&hero)?
        .map(|h| h.name.clone())
        .unwrap_or_else(|| "?".into()))
}

fn party_size<I: mvcc::IsolationLevel>(tx: &mut mvcc::Transaction<'_, I>) -> Result<usize> {
    Ok(tx.scan_where::<PartyMember, _>(|m| m.party == PARTY)?.len())
}

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 81)

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(())
}

pub fn commit(self) -> Result<()>

Commit.

Consumes the transaction, so use-after-commit is a compile error rather than a runtime check.

Examples found in repository

examples/isolation.rs (line 90)

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 253)

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 abort(self)

Abort and discard every write.

Infallible: rollback unlinks version pointers, which cannot fail.

Trait Implementations

impl<'db, I: IsolationLevel> Drop for Transaction<'db, I>

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.