|
|
@@ -0,0 +1,469 @@
|
|
|
+//! Balance-projection correctness (ADR-0019): the projection-aware read must
|
|
|
+//! always equal the authoritative live-posting sum, and the projector must
|
|
|
+//! advance the snapshot without changing the answer.
|
|
|
+
|
|
|
+#![allow(missing_docs)]
|
|
|
+
|
|
|
+use std::sync::Arc;
|
|
|
+
|
|
|
+use kuatia::ledger::Ledger;
|
|
|
+use kuatia::mem_store::InMemoryStore;
|
|
|
+use kuatia_core::*;
|
|
|
+
|
|
|
+fn usd() -> AssetId {
|
|
|
+ AssetId::new(1)
|
|
|
+}
|
|
|
+
|
|
|
+fn account(id: i64) -> AccountId {
|
|
|
+ AccountId::new(id)
|
|
|
+}
|
|
|
+
|
|
|
+fn external() -> AccountId {
|
|
|
+ AccountId::new(99)
|
|
|
+}
|
|
|
+
|
|
|
+async fn no_overdraft(ledger: &Arc<Ledger>, id: i64) {
|
|
|
+ ledger
|
|
|
+ .store()
|
|
|
+ .create_account(Account::debit_must_not_exceed_credit(account(id)))
|
|
|
+ .await
|
|
|
+ .unwrap();
|
|
|
+}
|
|
|
+
|
|
|
+async fn overdraft(ledger: &Arc<Ledger>, id: i64) {
|
|
|
+ ledger
|
|
|
+ .store()
|
|
|
+ .create_account(Account::new(account(id)))
|
|
|
+ .await
|
|
|
+ .unwrap();
|
|
|
+}
|
|
|
+
|
|
|
+async fn deposit(ledger: &Arc<Ledger>, to: i64, amount: i64) {
|
|
|
+ let transfer = TransferBuilder::new()
|
|
|
+ .deposit(account(to), usd(), Cent::from(amount), external())
|
|
|
+ .unwrap()
|
|
|
+ .build();
|
|
|
+ ledger.commit(transfer).await.unwrap();
|
|
|
+}
|
|
|
+
|
|
|
+/// Current time as Unix milliseconds, the same clock the ledger stamps commits
|
|
|
+/// and cache-point watermarks with, so a test can pin a watermark to "now".
|
|
|
+fn unix_millis_now() -> i64 {
|
|
|
+ std::time::SystemTime::now()
|
|
|
+ .duration_since(std::time::UNIX_EPOCH)
|
|
|
+ .unwrap()
|
|
|
+ .as_millis() as i64
|
|
|
+}
|
|
|
+
|
|
|
+async fn pay(ledger: &Arc<Ledger>, from: i64, to: i64, amount: i64) {
|
|
|
+ let transfer = TransferBuilder::new()
|
|
|
+ .pay(account(from), account(to), usd(), Cent::from(amount))
|
|
|
+ .build();
|
|
|
+ ledger.commit(transfer).await.unwrap();
|
|
|
+}
|
|
|
+
|
|
|
+/// Assert the projection-aware read equals the authoritative live-posting sum
|
|
|
+/// for every account, at every asset it might hold.
|
|
|
+async fn assert_projection_matches(ledger: &Arc<Ledger>, ids: &[i64]) {
|
|
|
+ for &id in ids {
|
|
|
+ let authoritative = ledger.compute_balance(&account(id), &usd()).await.unwrap();
|
|
|
+ let projected = ledger.balance(&account(id), &usd()).await.unwrap();
|
|
|
+ assert_eq!(
|
|
|
+ projected, authoritative,
|
|
|
+ "projected balance for account {id} diverged from the live sum"
|
|
|
+ );
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/// With no projector run at all (every projection absent, so the read folds the
|
|
|
+/// whole history), the projection-aware read still equals the live sum through
|
|
|
+/// deposits, change-making pays, and an overdraft offset posting.
|
|
|
+#[tokio::test]
|
|
|
+async fn projected_balance_matches_live_sum_without_refresh() {
|
|
|
+ let ledger = Arc::new(Ledger::new(InMemoryStore::new()));
|
|
|
+ no_overdraft(&ledger, 1).await;
|
|
|
+ no_overdraft(&ledger, 2).await;
|
|
|
+ overdraft(&ledger, 10).await;
|
|
|
+ overdraft(&ledger, 99).await;
|
|
|
+
|
|
|
+ deposit(&ledger, 1, 1000).await;
|
|
|
+ // Several pays fragment account 1 into many change postings.
|
|
|
+ pay(&ledger, 1, 2, 150).await;
|
|
|
+ pay(&ledger, 1, 2, 70).await;
|
|
|
+ pay(&ledger, 1, 2, 30).await;
|
|
|
+ pay(&ledger, 2, 1, 40).await;
|
|
|
+ // Overdraft: account 10 (empty balance) pays into a negative offset posting.
|
|
|
+ pay(&ledger, 10, 2, 500).await;
|
|
|
+
|
|
|
+ assert_projection_matches(&ledger, &[1, 2, 10, 99]).await;
|
|
|
+}
|
|
|
+
|
|
|
+/// Appending a cache point stores the balance directly, and the read still equals
|
|
|
+/// the live sum. Appending only after all commits keeps this deterministic: grace
|
|
|
+/// 0 puts the watermark at "now", so every committed transfer folds into the cache
|
|
|
+/// point with an empty tail.
|
|
|
+#[tokio::test]
|
|
|
+async fn append_cache_point_snapshots_balance_and_preserves_answer() {
|
|
|
+ let ledger = Arc::new(Ledger::new(InMemoryStore::new()).with_projection_grace_ms(0));
|
|
|
+ no_overdraft(&ledger, 1).await;
|
|
|
+ no_overdraft(&ledger, 2).await;
|
|
|
+ overdraft(&ledger, 99).await;
|
|
|
+
|
|
|
+ deposit(&ledger, 1, 1000).await;
|
|
|
+ pay(&ledger, 1, 2, 250).await;
|
|
|
+ pay(&ledger, 1, 2, 100).await;
|
|
|
+
|
|
|
+ ledger
|
|
|
+ .append_cache_point(&account(1), &usd())
|
|
|
+ .await
|
|
|
+ .unwrap();
|
|
|
+
|
|
|
+ // The cache point holds account 1's balance (1000 - 250 - 100) directly.
|
|
|
+ let cache_point = ledger
|
|
|
+ .store()
|
|
|
+ .get_closest_balance_projection(&account(1), &usd(), i64::MAX)
|
|
|
+ .await
|
|
|
+ .unwrap()
|
|
|
+ .expect("a cache point exists after append");
|
|
|
+ assert_eq!(cache_point.balance, Cent::from(650));
|
|
|
+ assert_eq!(
|
|
|
+ ledger.balance(&account(1), &usd()).await.unwrap(),
|
|
|
+ Cent::from(650)
|
|
|
+ );
|
|
|
+ assert_projection_matches(&ledger, &[1, 2]).await;
|
|
|
+}
|
|
|
+
|
|
|
+/// Commit never writes a cache point: after commits with no read, none exists.
|
|
|
+#[tokio::test]
|
|
|
+async fn commit_does_not_write_cache_point() {
|
|
|
+ let ledger = Arc::new(Ledger::new(InMemoryStore::new()));
|
|
|
+ no_overdraft(&ledger, 1).await;
|
|
|
+ no_overdraft(&ledger, 2).await;
|
|
|
+ overdraft(&ledger, 99).await;
|
|
|
+
|
|
|
+ deposit(&ledger, 1, 1000).await;
|
|
|
+ pay(&ledger, 1, 2, 250).await;
|
|
|
+
|
|
|
+ // No read has happened, so the lazy append never fired.
|
|
|
+ assert!(
|
|
|
+ ledger
|
|
|
+ .store()
|
|
|
+ .get_closest_balance_projection(&account(1), &usd(), i64::MAX)
|
|
|
+ .await
|
|
|
+ .unwrap()
|
|
|
+ .is_none()
|
|
|
+ );
|
|
|
+}
|
|
|
+
|
|
|
+/// A read appends a cache point once `snapshot_interval` credits/debits have
|
|
|
+/// accrued (the append is spawned in the background; on the current-thread test
|
|
|
+/// runtime it runs when we yield).
|
|
|
+#[tokio::test]
|
|
|
+async fn read_appends_cache_point_after_interval() {
|
|
|
+ let ledger = Arc::new(
|
|
|
+ Ledger::new(InMemoryStore::new())
|
|
|
+ .with_projection_grace_ms(0)
|
|
|
+ .with_snapshot_interval(1),
|
|
|
+ );
|
|
|
+ no_overdraft(&ledger, 1).await;
|
|
|
+ no_overdraft(&ledger, 2).await;
|
|
|
+ overdraft(&ledger, 99).await;
|
|
|
+
|
|
|
+ deposit(&ledger, 1, 1000).await;
|
|
|
+ pay(&ledger, 1, 2, 250).await;
|
|
|
+
|
|
|
+ // This read folds >= 1 credit/debit for account 1, so it spawns an append.
|
|
|
+ let _ = ledger.balance(&account(1), &usd()).await.unwrap();
|
|
|
+
|
|
|
+ // Let the background append run, then confirm a cache point exists.
|
|
|
+ let mut appeared = false;
|
|
|
+ for _ in 0..1000 {
|
|
|
+ tokio::task::yield_now().await;
|
|
|
+ if ledger
|
|
|
+ .store()
|
|
|
+ .get_closest_balance_projection(&account(1), &usd(), i64::MAX)
|
|
|
+ .await
|
|
|
+ .unwrap()
|
|
|
+ .is_some()
|
|
|
+ {
|
|
|
+ appeared = true;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ assert!(
|
|
|
+ appeared,
|
|
|
+ "a read past the interval should append a cache point"
|
|
|
+ );
|
|
|
+ assert_eq!(
|
|
|
+ ledger.balance(&account(1), &usd()).await.unwrap(),
|
|
|
+ ledger.compute_balance(&account(1), &usd()).await.unwrap()
|
|
|
+ );
|
|
|
+}
|
|
|
+
|
|
|
+/// A read below the interval never appends a cache point: the background append
|
|
|
+/// gates on new credits/debits, so a low-activity account accrues no rows (this
|
|
|
+/// is what keeps a hot, frequently-read account from appending near-duplicates).
|
|
|
+#[tokio::test]
|
|
|
+async fn read_below_interval_appends_nothing() {
|
|
|
+ let ledger = Arc::new(
|
|
|
+ Ledger::new(InMemoryStore::new())
|
|
|
+ .with_projection_grace_ms(0)
|
|
|
+ .with_snapshot_interval(1_000),
|
|
|
+ );
|
|
|
+ no_overdraft(&ledger, 1).await;
|
|
|
+ no_overdraft(&ledger, 2).await;
|
|
|
+ overdraft(&ledger, 99).await;
|
|
|
+
|
|
|
+ deposit(&ledger, 1, 1000).await;
|
|
|
+ pay(&ledger, 1, 2, 250).await;
|
|
|
+
|
|
|
+ // A handful of credits/debits, far below the 1000 interval.
|
|
|
+ let bal = ledger.balance(&account(1), &usd()).await.unwrap();
|
|
|
+ assert_eq!(
|
|
|
+ bal,
|
|
|
+ ledger.compute_balance(&account(1), &usd()).await.unwrap()
|
|
|
+ );
|
|
|
+
|
|
|
+ // Even after the background task has every chance to run, no cache point was
|
|
|
+ // appended (the append gates on >= interval new credits/debits).
|
|
|
+ for _ in 0..1000 {
|
|
|
+ tokio::task::yield_now().await;
|
|
|
+ }
|
|
|
+ assert!(
|
|
|
+ ledger
|
|
|
+ .store()
|
|
|
+ .get_closest_balance_projection(&account(1), &usd(), i64::MAX)
|
|
|
+ .await
|
|
|
+ .unwrap()
|
|
|
+ .is_none(),
|
|
|
+ "a below-interval read must not append a cache point"
|
|
|
+ );
|
|
|
+}
|
|
|
+
|
|
|
+/// Snapshot plus a genuinely non-empty tail. A cache point is pinned to the
|
|
|
+/// boundary between two commit batches, then more commits land strictly after
|
|
|
+/// its watermark, so the read must fold `snapshot + tail`: not a whole-history
|
|
|
+/// live sum (no snapshot) and not a snapshot that already holds everything (empty
|
|
|
+/// tail). This is the watermark boundary the grace-0 and no-snapshot tests never
|
|
|
+/// reach; an off-by-one at `watermark + 1` would drop or double-count a
|
|
|
+/// tail transfer here.
|
|
|
+#[tokio::test]
|
|
|
+async fn snapshot_plus_nonempty_tail_matches_live_sum() {
|
|
|
+ let ledger = Arc::new(Ledger::new(InMemoryStore::new()));
|
|
|
+ no_overdraft(&ledger, 1).await;
|
|
|
+ no_overdraft(&ledger, 2).await;
|
|
|
+ overdraft(&ledger, 99).await;
|
|
|
+
|
|
|
+ // Batch 1: the snapshot will cover exactly this.
|
|
|
+ deposit(&ledger, 1, 1000).await;
|
|
|
+ pay(&ledger, 1, 2, 250).await;
|
|
|
+
|
|
|
+ // Pin a cache point at the batch-1/batch-2 boundary. Its balance is the
|
|
|
+ // authoritative batch-1 sum (folded by the ledger, not by hand); its watermark
|
|
|
+ // is a time at or after every batch-1 commit. Appended directly so the
|
|
|
+ // watermark is controlled, independent of the lazy read-path grace.
|
|
|
+ let snapshot = ledger.compute_balance(&account(1), &usd()).await.unwrap();
|
|
|
+ let watermark = unix_millis_now();
|
|
|
+ ledger
|
|
|
+ .store()
|
|
|
+ .append_balance_projection(&account(1), &usd(), snapshot, watermark)
|
|
|
+ .await
|
|
|
+ .unwrap();
|
|
|
+
|
|
|
+ // Cross the millisecond boundary so every batch-2 commit is stamped strictly
|
|
|
+ // after the watermark and lands in the tail, never the snapshot.
|
|
|
+ tokio::time::sleep(std::time::Duration::from_millis(3)).await;
|
|
|
+
|
|
|
+ // Batch 2: folded onto the snapshot as a non-empty tail (both credits and
|
|
|
+ // debits for account 1).
|
|
|
+ pay(&ledger, 1, 2, 100).await;
|
|
|
+ pay(&ledger, 2, 1, 40).await;
|
|
|
+ deposit(&ledger, 1, 500).await;
|
|
|
+
|
|
|
+ // The read is snapshot + folded tail, and still equals the live sum.
|
|
|
+ let projected = ledger.balance(&account(1), &usd()).await.unwrap();
|
|
|
+ let authoritative = ledger.compute_balance(&account(1), &usd()).await.unwrap();
|
|
|
+ assert_eq!(
|
|
|
+ projected, authoritative,
|
|
|
+ "snapshot + non-empty tail diverged from the live sum"
|
|
|
+ );
|
|
|
+ // Guard the test itself: the snapshot must have been partial, so the tail
|
|
|
+ // actually carried value. Otherwise this silently degrades to the empty-tail
|
|
|
+ // case the other tests already cover.
|
|
|
+ assert_ne!(
|
|
|
+ snapshot, authoritative,
|
|
|
+ "the snapshot already held the full balance; the tail was empty"
|
|
|
+ );
|
|
|
+ assert_projection_matches(&ledger, &[1, 2]).await;
|
|
|
+}
|
|
|
+
|
|
|
+/// Deterministic xorshift64 PRNG so the property test is reproducible.
|
|
|
+struct Rng(u64);
|
|
|
+impl Rng {
|
|
|
+ fn next(&mut self) -> u64 {
|
|
|
+ let mut x = self.0;
|
|
|
+ x ^= x << 13;
|
|
|
+ x ^= x >> 7;
|
|
|
+ x ^= x << 17;
|
|
|
+ self.0 = x;
|
|
|
+ x
|
|
|
+ }
|
|
|
+ fn below(&mut self, n: u64) -> u64 {
|
|
|
+ self.next() % n
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/// Property test: across a long random sequence of every UTXO-shaped operation
|
|
|
+/// (deposits, change-making pays, overdraft offsets, multi-asset, subaccounts,
|
|
|
+/// and reversals), the projection-aware read equals the authoritative live-posting
|
|
|
+/// sum for every (account, asset) after every step. This is the empirical form of
|
|
|
+/// the telescoping argument: whole-posting spends plus change-as-new-posting make
|
|
|
+/// `snapshot + tail` fold exactly to the live set, in every shape the ledger can
|
|
|
+/// produce.
|
|
|
+#[tokio::test]
|
|
|
+async fn projection_matches_live_sum_across_random_utxo_history() {
|
|
|
+ // Default grace (60s) over a sub-second run keeps every watermark before all
|
|
|
+ // commits, so no cache point is appended and each read folds the full tail
|
|
|
+ // (equal to the live sum). That makes the run deterministic and free of
|
|
|
+ // same-millisecond append races; the invariant holds with or without a cache
|
|
|
+ // point. The snapshot-plus-non-empty-tail path (a real snapshot with commits
|
|
|
+ // folded on top) is covered deterministically by
|
|
|
+ // `snapshot_plus_nonempty_tail_matches_live_sum`.
|
|
|
+ let ledger = Arc::new(Ledger::new(InMemoryStore::new()).with_snapshot_interval(4));
|
|
|
+
|
|
|
+ // Accounts under test, including two subaccounts and both overdraft kinds.
|
|
|
+ let no_overdraft_ids = [
|
|
|
+ AccountId::new(1),
|
|
|
+ AccountId::new(2),
|
|
|
+ AccountId::new(3),
|
|
|
+ AccountId::with_sub(1, 7),
|
|
|
+ ];
|
|
|
+ let overdraft_ids = [
|
|
|
+ AccountId::new(10),
|
|
|
+ AccountId::new(11),
|
|
|
+ AccountId::with_sub(11, 3),
|
|
|
+ ];
|
|
|
+ for id in no_overdraft_ids {
|
|
|
+ ledger
|
|
|
+ .store()
|
|
|
+ .create_account(Account::debit_must_not_exceed_credit(id))
|
|
|
+ .await
|
|
|
+ .unwrap();
|
|
|
+ }
|
|
|
+ for id in overdraft_ids {
|
|
|
+ ledger
|
|
|
+ .store()
|
|
|
+ .create_account(Account::new(id))
|
|
|
+ .await
|
|
|
+ .unwrap();
|
|
|
+ }
|
|
|
+ let ext = external();
|
|
|
+ ledger
|
|
|
+ .store()
|
|
|
+ .create_account(Account::new(ext))
|
|
|
+ .await
|
|
|
+ .unwrap();
|
|
|
+
|
|
|
+ let accounts: Vec<AccountId> = no_overdraft_ids
|
|
|
+ .iter()
|
|
|
+ .chain(overdraft_ids.iter())
|
|
|
+ .copied()
|
|
|
+ .collect();
|
|
|
+ let assets = [AssetId::new(1), AssetId::new(2)];
|
|
|
+
|
|
|
+ let mut rng = Rng(0x9e3779b97f4a7c15);
|
|
|
+ let mut receipts: Vec<EnvelopeId> = Vec::new();
|
|
|
+
|
|
|
+ for _ in 0..300 {
|
|
|
+ let asset = assets[rng.below(assets.len() as u64) as usize];
|
|
|
+ match rng.below(5) {
|
|
|
+ // Deposit into a random account.
|
|
|
+ 0 => {
|
|
|
+ let to = accounts[rng.below(accounts.len() as u64) as usize];
|
|
|
+ let amount = 1 + rng.below(500) as i64;
|
|
|
+ let t = TransferBuilder::new()
|
|
|
+ .deposit(to, asset, Cent::from(amount), ext)
|
|
|
+ .unwrap()
|
|
|
+ .build();
|
|
|
+ if let Ok(r) = ledger.commit(t).await {
|
|
|
+ receipts.push(r.transfer_id);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // Withdraw from a random account to the boundary.
|
|
|
+ 1 => {
|
|
|
+ let from = accounts[rng.below(accounts.len() as u64) as usize];
|
|
|
+ let amount = 1 + rng.below(200) as i64;
|
|
|
+ let t = TransferBuilder::new()
|
|
|
+ .withdraw(from, asset, Cent::from(amount), ext)
|
|
|
+ .build();
|
|
|
+ if let Ok(r) = ledger.commit(t).await {
|
|
|
+ receipts.push(r.transfer_id);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // Reverse a previously committed transfer.
|
|
|
+ 2 if !receipts.is_empty() => {
|
|
|
+ let id = receipts[rng.below(receipts.len() as u64) as usize];
|
|
|
+ if let Ok(r) = ledger.reverse(&id).await {
|
|
|
+ receipts.push(r.transfer_id);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // Pay between two random accounts (change / overdraft / cross-subaccount).
|
|
|
+ _ => {
|
|
|
+ let from = accounts[rng.below(accounts.len() as u64) as usize];
|
|
|
+ let to = accounts[rng.below(accounts.len() as u64) as usize];
|
|
|
+ if from == to {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ let amount = 1 + rng.below(300) as i64;
|
|
|
+ let t = TransferBuilder::new()
|
|
|
+ .pay(from, to, asset, Cent::from(amount))
|
|
|
+ .build();
|
|
|
+ if let Ok(r) = ledger.commit(t).await {
|
|
|
+ receipts.push(r.transfer_id);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // Invariant: at rest after every step, the projection-aware read equals
|
|
|
+ // the authoritative live-posting sum for every (account, asset).
|
|
|
+ for account in accounts.iter().chain(std::iter::once(&ext)) {
|
|
|
+ for asset in &assets {
|
|
|
+ let authoritative = ledger.compute_balance(account, asset).await.unwrap();
|
|
|
+ let projected = ledger.balance(account, asset).await.unwrap();
|
|
|
+ assert_eq!(
|
|
|
+ projected, authoritative,
|
|
|
+ "projected != live sum for {account:?} / {asset:?}"
|
|
|
+ );
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/// Concurrent commits from one funded account: after the dust settles, the
|
|
|
+/// projection-aware read agrees with the live sum for every participant.
|
|
|
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
|
+async fn projection_matches_under_concurrent_commits() {
|
|
|
+ let ledger = Arc::new(Ledger::new(InMemoryStore::new()));
|
|
|
+ no_overdraft(&ledger, 1).await;
|
|
|
+ for id in 2..=9 {
|
|
|
+ no_overdraft(&ledger, id).await;
|
|
|
+ }
|
|
|
+ overdraft(&ledger, 99).await;
|
|
|
+ deposit(&ledger, 1, 1000).await;
|
|
|
+
|
|
|
+ let mut handles = Vec::new();
|
|
|
+ for payee in 2..=9 {
|
|
|
+ let ledger = Arc::clone(&ledger);
|
|
|
+ handles.push(tokio::spawn(async move {
|
|
|
+ let transfer = TransferBuilder::new()
|
|
|
+ .pay(account(1), account(payee), usd(), Cent::from(10))
|
|
|
+ .build();
|
|
|
+ let _ = ledger.commit(transfer).await;
|
|
|
+ }));
|
|
|
+ }
|
|
|
+ for h in handles {
|
|
|
+ h.await.unwrap();
|
|
|
+ }
|
|
|
+
|
|
|
+ assert_projection_matches(&ledger, &[1, 2, 3, 4, 5, 6, 7, 8, 9, 99]).await;
|
|
|
+}
|