1use std::collections::HashMap;
16use std::sync::Arc;
17use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
18use std::thread;
19use std::time::Instant;
20
21use mvcc::{Config, Database, Error, Mvcc, Result, Serializable, Snapshot};
22
23#[derive(Mvcc, Clone, Debug)]
28#[mvcc(table = "heroes")]
29struct Hero {
30 #[mvcc(primary_key)]
31 id: u64,
32 #[mvcc(index(unique))]
34 name: String,
35 gold: i64,
36 strength: i32,
38}
39
40const GROUND: u64 = 0;
43
44#[derive(Mvcc, Clone, Debug)]
45#[mvcc(table = "items")]
46struct Item {
47 #[mvcc(primary_key)]
48 id: u64,
49 #[mvcc(index)]
52 owner: u64,
53 name: String,
54 weight: i32,
55 value: i64,
56 kind: Kind,
59 flavour: HashMap<String, String>,
62}
63
64#[derive(Clone, Copy, Debug, PartialEq, Eq)]
65enum Kind {
66 Weapon,
67 Potion,
68 Trinket,
69}
70
71#[derive(Mvcc, Clone, Debug)]
72#[mvcc(table = "party_members")]
73struct PartyMember {
74 #[mvcc(primary_key)]
75 hero: u64,
76 #[mvcc(index)]
77 party: u64,
78 role: String,
79}
80
81const PARTY: u64 = 1;
82const PARTY_LIMIT: usize = 4;
83
84fn inventory<I: mvcc::IsolationLevel>(
90 tx: &mut mvcc::Transaction<'_, I>,
91 hero: u64,
92) -> Result<Vec<(u64, String, i32)>> {
93 Ok(tx
94 .scan_index(Item::OWNER, hero..=hero)?
95 .iter()
96 .map(|i| (i.id, i.name.clone(), i.weight))
97 .collect())
98}
99
100fn carried<I: mvcc::IsolationLevel>(tx: &mut mvcc::Transaction<'_, I>, hero: u64) -> Result<i32> {
103 Ok(tx
104 .scan_index(Item::OWNER, hero..=hero)?
105 .iter()
106 .map(|i| i.weight)
107 .sum())
108}
109
110fn gold_of<I: mvcc::IsolationLevel>(tx: &mut mvcc::Transaction<'_, I>, hero: u64) -> Result<i64> {
111 Ok(tx.get::<Hero>(&hero)?.map(|h| h.gold).unwrap_or(0))
112}
113
114fn name_of<I: mvcc::IsolationLevel>(
115 tx: &mut mvcc::Transaction<'_, I>,
116 hero: u64,
117) -> Result<String> {
118 Ok(tx
119 .get::<Hero>(&hero)?
120 .map(|h| h.name.clone())
121 .unwrap_or_else(|| "?".into()))
122}
123
124fn party_size<I: mvcc::IsolationLevel>(tx: &mut mvcc::Transaction<'_, I>) -> Result<usize> {
125 Ok(tx.scan_where::<PartyMember, _>(|m| m.party == PARTY)?.len())
126}
127
128fn total_gold(db: &Database) -> Result<i64> {
129 let mut tx = db.begin();
130 Ok(tx.scan::<Hero>()?.iter().map(|h| h.gold).sum())
131}
132
133fn act(n: u32, title: &str) {
134 println!("\n\x1b[1m── Act {n}. {title}\x1b[0m");
135}
136
137#[allow(clippy::too_many_lines, clippy::cognitive_complexity)]
141fn main() -> Result<()> {
142 let db = Arc::new(Database::open(Config::in_memory())?);
143 db.register::<Hero>()?;
144 db.register::<Item>()?;
145 db.register::<PartyMember>()?;
146
147 let (ada, bram, cleo) = (1u64, 2u64, 3u64);
148
149 act(1, "The world is created");
151
152 db.transaction(|tx| {
153 tx.insert(Hero {
154 id: ada,
155 name: "Ada".into(),
156 gold: 120,
157 strength: 100,
158 })?;
159 tx.insert(Hero {
160 id: bram,
161 name: "Bram".into(),
162 gold: 80,
163 strength: 100,
164 })?;
165 tx.insert(Hero {
166 id: cleo,
167 name: "Cleo".into(),
168 gold: 200,
169 strength: 60,
170 })?;
171
172 let mut item = |id, owner, name: &str, weight, value, kind| {
173 tx.insert(Item {
174 id,
175 owner,
176 name: name.into(),
177 weight,
178 value,
179 kind,
180 flavour: HashMap::new(),
181 })
182 };
183 item(10, ada, "Rusty Sword", 30, 15, Kind::Weapon)?;
184 item(11, ada, "Health Potion", 5, 20, Kind::Potion)?;
185 item(12, bram, "Oak Shield", 40, 35, Kind::Weapon)?;
186 item(13, cleo, "Lucky Charm", 2, 90, Kind::Trinket)?;
187 item(20, GROUND, "Flaming Greatsword", 55, 500, Kind::Weapon)?;
189 item(21, GROUND, "Elven Cloak", 20, 240, Kind::Trinket)?;
190 item(22, GROUND, "Iron Helm", 35, 60, Kind::Weapon)?;
191
192 tx.update::<Item>(&20, |i| {
195 i.flavour.insert("enchantment".into(), "flame".into());
196 i.flavour.insert("forged_by".into(), "Durin".into());
197 })?;
198 Ok(())
199 })?;
200
201 let before = gold_of(&mut db.begin(), ada)?;
204 let doomed = db.transaction(|tx| {
205 tx.update::<Hero>(&ada, |h| h.gold += 10_000)?;
206 tx.insert(Hero {
207 id: 99,
208 name: "Ada".into(),
209 gold: 0,
210 strength: 10,
211 })
212 });
213 println!(" a duplicate hero name: {}", doomed.unwrap_err());
214 println!(
215 " Ada's gold is still {} — the whole transaction rolled back, not just the insert",
216 gold_of(&mut db.begin(), ada)?
217 );
218 assert_eq!(gold_of(&mut db.begin(), ada)?, before);
219
220 let mut tx = db.begin();
221 for hero in [ada, bram, cleo] {
222 let name = name_of(&mut tx, hero)?;
223 let carried = carried(&mut tx, hero)?;
224 let items = inventory(&mut tx, hero)?;
225 let names: Vec<_> = items.iter().map(|(_, n, _)| n.as_str()).collect();
226 println!(" {name:<5} carries {carried:>3} — {}", names.join(", "));
227 }
228 drop(tx);
229
230 act(2, "Two heroes reach for the same greatsword");
232
233 let mut ada_grabs = db.begin_with::<Snapshot>();
236 let mut bram_grabs = db.begin_with::<Snapshot>();
237
238 ada_grabs.update::<Item>(&20, |i| i.owner = ada)?;
239
240 match bram_grabs.update::<Item>(&20, |i| i.owner = bram) {
241 Err(e @ Error::WriteConflict { .. }) => {
242 println!(" Bram: {e}");
243 println!(
244 " ...retriable: {} — he can try for something else",
245 e.is_retriable()
246 );
247 }
248 other => unreachable!("expected a conflict, got {other:?}"),
249 }
250
251 bram_grabs.update::<Item>(&21, |i| i.owner = bram)?;
253 ada_grabs.commit()?;
254 bram_grabs.commit()?;
255 println!(" Ada takes the greatsword, Bram takes the cloak. Nobody blocked.");
256
257 act(3, "Cleo overloads herself — write skew, and the fix");
259
260 db.transaction(|tx| {
265 tx.update::<Item>(&22, |i| i.owner = GROUND)?;
266 tx.insert(Item {
267 id: 23,
268 owner: GROUND,
269 name: "Steel Helm".into(),
270 weight: 35,
271 value: 70,
272 kind: Kind::Weapon,
273 flavour: HashMap::new(),
274 })
275 })?;
276
277 let pick_up = |item: u64| {
278 move |tx: &mut mvcc::Transaction<'_, Serializable>| -> Result<bool> {
279 let hero = tx.get::<Hero>(&cleo)?.expect("Cleo exists");
280 let strength = hero.strength;
281 let load = carried(tx, cleo)?;
282 let weight = tx.get::<Item>(&item)?.map(|i| i.weight).unwrap_or(0);
283 if load + weight > strength {
284 return Ok(false);
285 }
286 tx.update::<Item>(&item, |i| i.owner = cleo)?;
287 Ok(true)
288 }
289 };
290
291 {
293 let mut t1 = db.begin_with::<Snapshot>();
294 let mut t2 = db.begin_with::<Snapshot>();
295 let (load1, load2) = (carried(&mut t1, cleo)?, carried(&mut t2, cleo)?);
296 println!(
297 " Snapshot: both transactions see {load1}/{load2} carried, both think a 35kg helm fits"
298 );
299 t1.update::<Item>(&22, |i| i.owner = cleo)?;
300 t2.update::<Item>(&23, |i| i.owner = cleo)?;
301 t1.commit()?;
302 t2.commit()?;
303 let over = carried(&mut db.begin(), cleo)?;
304 println!(
305 " ✗ Cleo now carries {over} of a possible 60. Two legal transactions, one broken rule."
306 );
307 }
308
309 db.transaction(|tx| {
311 tx.update::<Item>(&22, |i| i.owner = GROUND)?;
312 tx.update::<Item>(&23, |i| i.owner = GROUND).map(|_| ())
313 })?;
314
315 {
316 let mut t1 = db.begin_with::<Serializable>();
317 let mut t2 = db.begin_with::<Serializable>();
318 let _ = carried(&mut t1, cleo)?;
320 let _ = carried(&mut t2, cleo)?;
321 t1.update::<Item>(&22, |i| i.owner = cleo)?;
322 t2.update::<Item>(&23, |i| i.owner = cleo)?;
323 t1.commit()?;
324 match t2.commit() {
325 Err(e @ Error::SerializationFailure) => println!(" Serializable: second pickup {e}"),
326 other => unreachable!("expected a serialization failure, got {other:?}"),
327 }
328 println!(
329 " ✓ Cleo carries {} — the second pickup read an inventory the first changed.",
330 carried(&mut db.begin(), cleo)?
331 );
332 }
333
334 let took_it = db.transaction_with::<Serializable, _, _>(pick_up(23))?;
337 println!(" ...and on retry the second helm is refused on its merits: picked_up = {took_it}");
338
339 act(4, "The party fills up — a phantom, not a conflict");
341
342 db.transaction(|tx| {
343 tx.insert(PartyMember {
344 hero: ada,
345 party: PARTY,
346 role: "Vanguard".into(),
347 })?;
348 tx.insert(PartyMember {
349 hero: bram,
350 party: PARTY,
351 role: "Shield".into(),
352 })?;
353 tx.insert(PartyMember {
354 hero: cleo,
355 party: PARTY,
356 role: "Scout".into(),
357 })
358 })?;
359
360 let (dara, finn) = (4u64, 5u64);
365 db.transaction(|tx| {
366 tx.insert(Hero {
367 id: dara,
368 name: "Dara".into(),
369 gold: 40,
370 strength: 80,
371 })?;
372 tx.insert(Hero {
373 id: finn,
374 name: "Finn".into(),
375 gold: 40,
376 strength: 80,
377 })
378 })?;
379
380 let mut d = db.begin_with::<Serializable>();
381 let mut f = db.begin_with::<Serializable>();
382 println!(
383 " Dara counts {} members, Finn counts {} — both see a free slot",
384 party_size(&mut d)?,
385 party_size(&mut f)?
386 );
387 d.insert(PartyMember {
388 hero: dara,
389 party: PARTY,
390 role: "Healer".into(),
391 })?;
392 f.insert(PartyMember {
393 hero: finn,
394 party: PARTY,
395 role: "Healer".into(),
396 })?;
397 d.commit()?;
398 match f.commit() {
399 Err(e @ Error::SerializationFailure) => println!(" Finn: {e}"),
400 other => unreachable!("expected a serialization failure, got {other:?}"),
401 }
402 let mut tx = db.begin();
403 let mut roster: Vec<_> = tx
404 .scan_where::<PartyMember, _>(|m| m.party == PARTY)?
405 .iter()
406 .map(|m| (m.hero, m.role.clone()))
407 .collect();
408 drop(tx);
409 roster.sort();
410 let size = roster.len();
411 let mut tx = db.begin();
412 let listed: Vec<String> = roster
413 .iter()
414 .map(|(h, role)| Ok(format!("{} the {role}", name_of(&mut tx, *h)?)))
415 .collect::<Result<_>>()?;
416 drop(tx);
417 println!(" party: {}", listed.join(", "));
418 println!(" ✓ {size}/{PARTY_LIMIT} filled. Finn's insert became a phantom in Dara's count.");
419 println!(" (`scan_where` hands the engine the predicate, so it can re-check it at commit.)");
420 assert!(size <= PARTY_LIMIT);
421
422 act(5, "A trade — two records, one atomic step");
424
425 let before = total_gold(&db)?;
426
427 let trade = |seller: u64, buyer: u64, item: u64, price: i64| {
428 let db = Arc::clone(&db);
429 move || -> Result<bool> {
430 db.transaction_with::<Serializable, _, _>(|tx| {
431 let buyer_gold = gold_of(tx, buyer)?;
432 if buyer_gold < price {
433 return Ok(false);
434 }
435 let owned_by_seller = tx
436 .get::<Item>(&item)?
437 .map(|i| i.owner == seller)
438 .unwrap_or(false);
439 if !owned_by_seller {
440 return Ok(false);
441 }
442 tx.update::<Hero>(&buyer, |h| h.gold -= price)?;
443 tx.update::<Hero>(&seller, |h| h.gold += price)?;
444 tx.update::<Item>(&item, |i| i.owner = buyer)?;
445 Ok(true)
446 })
447 }
448 };
449
450 let sold = trade(bram, cleo, 21, 150)()?;
451 println!(" Bram sells the Elven Cloak to Cleo for 150g: {sold}");
452 let mut tx = db.begin();
453 let cloak_owner = tx.get::<Item>(&21)?.expect("cloak").owner;
454 println!(
455 " Bram {}g, Cleo {}g, cloak owner is {}",
456 gold_of(&mut tx, bram)?,
457 gold_of(&mut tx, cleo)?,
458 name_of(&mut tx, cloak_owner)?
459 );
460 drop(tx);
461 assert_eq!(total_gold(&db)?, before, "gold must be conserved");
462 println!(" ✓ total gold unchanged at {before} — no step of that was separately visible");
463
464 act(6, "The chronicler reads while the world moves");
466
467 let mut chronicler = db.begin_with::<Snapshot>();
470 let opening = total_gold(&db)?;
471 let seen_first = chronicler.scan::<Hero>()?.len();
472
473 for _ in 0..50 {
474 db.transaction(|tx| {
475 tx.update::<Hero>(&ada, |h| h.gold += 1)?;
476 tx.update::<Hero>(&bram, |h| h.gold -= 1)?;
477 Ok(())
478 })?;
479 }
480 db.transaction(|tx| {
481 tx.insert(Hero {
482 id: 6,
483 name: "Mira".into(),
484 gold: 0,
485 strength: 50,
486 })
487 })?;
488
489 let chron_gold: i64 = chronicler.scan::<Hero>()?.iter().map(|h| h.gold).sum();
490 let seen_after = chronicler.scan::<Hero>()?.len();
491 let in_world = db.begin().scan::<Hero>()?.len();
492 println!(" chronicler saw {seen_first} heroes at the start and still sees {seen_after};");
493 println!(" the world now holds {in_world}, after 50 commits and one new arrival");
494 println!(
495 " chronicler totals {chron_gold}g; the world now totals {}g",
496 total_gold(&db)?
497 );
498 assert_eq!(chron_gold, opening, "a snapshot must not move");
499 chronicler.commit()?;
500 println!(" ✓ a reader that ran across 51 commits saw exactly one consistent world");
501
502 act(7, "The raid — many adventurers at once");
504
505 let stop = Arc::new(AtomicBool::new(false));
506 let trades = Arc::new(AtomicU64::new(0));
507 let retries = Arc::new(AtomicU64::new(0));
508 let audits = Arc::new(AtomicU64::new(0));
509 let opening = total_gold(&db)?;
510 let heroes = [ada, bram, cleo, dara, finn];
511
512 let auditor = {
514 let (db, stop, audits) = (Arc::clone(&db), Arc::clone(&stop), Arc::clone(&audits));
515 thread::spawn(move || -> Result<()> {
516 while !stop.load(Ordering::Relaxed) {
517 let mut tx = db.begin();
518 let total: i64 = tx.scan::<Hero>()?.iter().map(|h| h.gold).sum();
519 assert_eq!(total, opening, "auditor saw a half-finished trade");
520 audits.fetch_add(1, Ordering::Relaxed);
521 }
522 Ok(())
523 })
524 };
525
526 let start = Instant::now();
527 let raiders: Vec<_> = (0..4)
528 .map(|t| {
529 let (db, trades, retries) =
530 (Arc::clone(&db), Arc::clone(&trades), Arc::clone(&retries));
531 thread::spawn(move || -> Result<()> {
532 let mut seed = 0x9e37_79b9_7f4a_7c15u64 ^ (t + 1);
533 for _ in 0..400 {
534 seed ^= seed << 13;
535 seed ^= seed >> 7;
536 seed ^= seed << 17;
537 let from = heroes[(seed % heroes.len() as u64) as usize];
538 let to = heroes[((seed >> 8) % heroes.len() as u64) as usize];
539 if from == to {
540 continue;
541 }
542
543 let mut attempts = 0u64;
544 db.transaction_with::<Serializable, _, _>(|tx| {
547 attempts += 1;
548 let purse = gold_of(tx, from)?;
549 if purse < 5 {
550 return Ok(());
551 }
552 tx.update::<Hero>(&from, |h| h.gold -= 5)?;
553 tx.update::<Hero>(&to, |h| h.gold += 5)?;
554 Ok(())
555 })?;
556 trades.fetch_add(1, Ordering::Relaxed);
557 retries.fetch_add(attempts - 1, Ordering::Relaxed);
558 }
559 Ok(())
560 })
561 })
562 .collect();
563
564 for r in raiders {
565 r.join().expect("raider panicked")?;
566 }
567 let elapsed = start.elapsed();
568 stop.store(true, Ordering::Relaxed);
569 auditor.join().expect("auditor panicked")?;
570
571 let done = trades.load(Ordering::Relaxed);
572 println!(" {done} trades across 4 threads in {elapsed:.2?}");
573 println!(
574 " {} audit sweeps completed alongside them, none blocked, none torn",
575 audits.load(Ordering::Relaxed)
576 );
577 println!(
578 " {} retries ({:.2} per trade) — contention on 5 heroes, handled by the engine",
579 retries.load(Ordering::Relaxed),
580 retries.load(Ordering::Relaxed) as f64 / done.max(1) as f64
581 );
582 assert_eq!(total_gold(&db)?, opening, "gold was created or destroyed");
583 println!(" ✓ total gold still {opening}");
584
585 act(8, "Closing the ledger");
587
588 let mut tx = db.begin();
589 let mut roster: Vec<_> = tx
590 .scan::<Hero>()?
591 .iter()
592 .map(|h| (h.id, h.name.clone(), h.gold))
593 .collect();
594 roster.sort_by_key(|(id, _, _)| *id);
595 drop(tx);
596
597 println!(
598 " {:<6} {:>6} {:>5} {:>7} inventory",
599 "hero", "gold", "load", "worth"
600 );
601 let mut tx = db.begin();
602 for (id, name, gold) in &roster {
603 let held = tx.scan_index(Item::OWNER, *id..=*id)?;
604 let load: i32 = held.iter().map(|i| i.weight).sum();
605 let worth: i64 = held.iter().map(|i| i.value).sum();
606 let names: Vec<String> = held
607 .iter()
608 .map(|i| match i.kind {
609 Kind::Weapon => format!("[wpn] {}", i.name),
610 Kind::Potion => format!("[pot] {}", i.name),
611 Kind::Trinket => format!("[trk] {}", i.name),
612 })
613 .collect();
614 let names = if names.is_empty() {
615 "—".to_string()
616 } else {
617 names.join(", ")
618 };
619 println!(" {name:<6} {gold:>5}g {load:>4}kg {worth:>6}g {names}");
620 }
621
622 let potions = tx.scan_where::<Item, _>(|i| i.kind == Kind::Potion)?.len();
625 let sword = tx.get::<Item>(&20)?.expect("greatsword");
626 let enchantment = sword
627 .flavour
628 .get("enchantment")
629 .cloned()
630 .unwrap_or_default();
631 let forged_by = sword.flavour.get("forged_by").cloned().unwrap_or_default();
632 drop(tx);
633 let mut tx = db.begin();
635 for (id, name, _) in &roster {
636 let strength = tx.get::<Hero>(id)?.map(|h| h.strength).unwrap_or(0);
637 let load = carried(&mut tx, *id)?;
638 assert!(
639 load <= strength,
640 "{name} is over their carry limit: {load} > {strength}"
641 );
642 }
643 drop(tx);
644
645 println!("\n potions in the world: {potions}");
646 println!(" the greatsword is {enchantment}-enchanted, forged by {forged_by}");
647
648 let stats = db.stats();
652 println!(
653 "\n gc watermark {:?}, {} transactions still live",
654 stats.watermark, stats.active_transactions
655 );
656 println!(" (a watermark that stops moving while writes continue is the leak to look for)");
657
658 Ok(())
659}