Give the balance concept a single home

An account's balance for an asset was derived three ways across two files with
the convergence between them guaranteed only by prose and tests. The
authoritative live-posting sum (compute_balance) and the per-subaccount listing
lived in balance.rs; the cached read and its transfer-delta fold lived in
projection.rs. balance()'s cold path just delegated across the file boundary to
its sibling, and balances() bypassed the cache while balance() used it, so the
same logical value was served two ways with no shared entry point.

Fold projection.rs into balance.rs so the concept is one module. Keep the
public names (balance, balances, compute_balance, append_cache_point) so no
call site changes; make the fold/spawn helpers private. State the core contract
once on the module that owns both strategies: the cached read always equals the
authoritative live-posting sum at rest. Route balances() through the everyday
cached balance() so every balance read goes through one path, retiring the
singular/plural inconsistency.

crates/kuatia/src/ledger.rs

@@ -30,7 +30,6 @@ mod balance;
 mod commit;
 mod lifecycle;
 mod pending;
-mod projection;
 mod query;
 mod transition;
 

crates/kuatia/src/ledger/balance.rs

@@ -1,13 +1,34 @@
-//! Per-subaccount balance queries.
+//! Balances: the authoritative live-posting sum, the everyday cached read, and
+//! the append-only cache points that keep that read fast (ADR-0012, ADR-0019).
 //!
-//! Balances are always computed in Rust from the live (active or reserved)
-//! postings and are never summed across subaccounts (ADR-0012).
+//! One concept, two computation strategies that must agree:
+//!
+//! - [`compute_balance`](Ledger::compute_balance) is **authoritative**: the sum
+//!   of the live (active or reserved) postings for one `(account, asset)`, always
+//!   recomputed from the source of truth. It is what validation reads, what the
+//!   projector folds into snapshots, and what reconciliation checks against.
+//! - [`balance`](Ledger::balance) is the **everyday read**: the closest cache
+//!   point (at or before now) plus the folded tail of transfers committed after
+//!   its watermark.
+//!
+//! The module's core contract: summing every one of an account's transfer deltas
+//! equals its live-posting sum, so **`balance` always equals `compute_balance` at
+//! rest**. Cache points are a pure optimization that shortens the tail;
+//! correctness never depends on them, and reconciliation re-derives the
+//! authoritative value to check against a cache point.
+//!
+//! Balances are always computed in Rust with checked arithmetic and are never
+//! summed across subaccounts (ADR-0012).
+
+use std::collections::HashSet;
+use std::sync::Arc;
 
 use tracing::instrument;
 
-use kuatia_core::{AccountId, AssetId, Cent, PostingFilter};
+use kuatia_core::{AccountId, AssetId, Cent, PostingFilter, PostingId};
+use kuatia_storage::store::{EnvelopeRecord, Store, TransferQuery};
 
-use super::Ledger;
+use super::{Ledger, now_millis};
 use crate::error::LedgerError;
 
 /// A single subaccount's balance for one asset. Balances are always reported
@@ -20,6 +41,138 @@ pub struct SubAccountBalance {
     pub value: Cent,
 }
 
+// ---------------------------------------------------------------------------
+// Cache-point internals (ADR-0019): fold committed-transfer deltas onto a
+// snapshot. Private to this module; the read/reconcile entry points are on
+// `Ledger` below.
+// ---------------------------------------------------------------------------
+
+/// Fold the per-`(account, asset)` delta of a set of committed transfers,
+/// `Σ created(+) − Σ consumed(−)` restricted to postings owned by `account` in
+/// `asset`, and count how many such postings (credits + debits) were seen.
+/// Consumed postings are resolved from the immutable table (the envelope carries
+/// only their ids). All arithmetic is checked, in Rust.
+async fn fold_account_delta(
+    store: &dyn Store,
+    account: &AccountId,
+    asset: &AssetId,
+    records: &[EnvelopeRecord],
+) -> Result<(Cent, u64), LedgerError> {
+    let mut delta = Cent::ZERO;
+    let mut count: u64 = 0;
+
+    // Created side (credits): the envelope carries owner/asset/value directly.
+    for record in records {
+        for np in record.envelope.creates() {
+            if np.owner == *account && np.asset == *asset {
+                delta = delta.checked_add(np.value)?;
+                count += 1;
+            }
+        }
+    }
+
+    // Consumed side (debits): gather every consumed id across the window, resolve
+    // the postings in one batch, and subtract those owned by this (account, asset).
+    let mut consumed_ids: Vec<PostingId> = Vec::new();
+    let mut seen: HashSet<PostingId> = HashSet::new();
+    for record in records {
+        for id in record.envelope.consumes() {
+            if seen.insert(*id) {
+                consumed_ids.push(*id);
+            }
+        }
+    }
+    if !consumed_ids.is_empty() {
+        for posting in store.get_postings(&consumed_ids).await? {
+            if posting.owner == *account && posting.asset == *asset {
+                delta = delta.checked_sub(posting.value)?;
+                count += 1;
+            }
+        }
+    }
+
+    Ok((delta, count))
+}
+
+/// Load the committed transfers involving `account` with commit time in
+/// `[from_ts, to_ts)` (both optional), then fold their `(account, asset)` delta
+/// and credit/debit count. `from_ts == None` spans from the beginning;
+/// `to_ts == None` spans to the newest committed transfer.
+async fn fold_tail(
+    store: &dyn Store,
+    account: &AccountId,
+    asset: &AssetId,
+    from_ts: Option<i64>,
+    to_ts: Option<i64>,
+) -> Result<(Cent, u64), LedgerError> {
+    let query = TransferQuery {
+        account: Some(account.id),
+        sub: Some(account.sub),
+        from_ts,
+        to_ts,
+        ..Default::default()
+    };
+    let records = store.query_transfers(&query).await?.items;
+    fold_account_delta(store, account, asset, &records).await
+}
+
+/// Append a fresh cache point for `(account, asset)`: fold the window since the
+/// closest cache point's watermark up to `now − grace` onto its snapshot, and
+/// append the result. Only appends when that window has at least `min_new`
+/// credits/debits, so a hot account that is read constantly does not append a
+/// near-duplicate row on every read (`min_new == 0` forces an append). Append-only
+/// and best effort; a stale or duplicate append is harmless because a read takes
+/// the closest-at-or-before cache point.
+///
+/// `debounce_ms` is the storage-based single-flight guard. When the newest cache
+/// point already sits within `debounce_ms` below the target watermark, this
+/// returns before the fold, so a hot account read at high QPS does not fan out a
+/// full fold per read. The guard lives in the shared `balance_projection` rows,
+/// not in process memory, so it dedups across every ledger instance and survives
+/// a restart, and it needs no lease, lock, or CAS (ADR-0019). `debounce_ms == 0`
+/// disables it (the reconcile path forces an append regardless).
+async fn append_cache_point(
+    store: &dyn Store,
+    grace_ms: i64,
+    min_new: u64,
+    debounce_ms: i64,
+    account: &AccountId,
+    asset: &AssetId,
+) -> Result<(), LedgerError> {
+    let new_watermark = now_millis()?.saturating_sub(grace_ms);
+    let closest = store
+        .get_closest_balance_projection(account, asset, new_watermark)
+        .await?;
+    let (snapshot, from_ts) = match closest {
+        // Storage-based debounce: a cache point within `debounce_ms` below the
+        // target already shortens the tail enough, so skip the fold. This is what
+        // collapses a per-read fan-out into at most one fold per debounce window.
+        Some(p) if new_watermark.saturating_sub(p.watermark) < debounce_ms => return Ok(()),
+        // A cache point already covers this watermark: nothing to add. (Also the
+        // debounce == 0 stop, so an exact-or-newer watermark is never duplicated.)
+        Some(p) if p.watermark >= new_watermark => return Ok(()),
+        Some(p) => (p.balance, Some(p.watermark.saturating_add(1))),
+        None => (Cent::ZERO, None),
+    };
+    let (fold, count) = fold_tail(
+        store,
+        account,
+        asset,
+        from_ts,
+        Some(new_watermark.saturating_add(1)),
+    )
+    .await?;
+    // Not enough new activity since the closest cache point to earn a new row.
+    if count < min_new {
+        return Ok(());
+    }
+    let balance = snapshot.checked_add(fold)?;
+    store
+        .append_balance_projection(account, asset, balance, new_watermark)
+        .await?;
+    Ok(())
+}
+
 impl Ledger {
     /// The authoritative balance: the sum of the live (active or reserved)
     /// postings for one `(account, asset)`, computed in Rust. This bypasses the
@@ -45,11 +198,56 @@ impl Ledger {
         Ok(Cent::checked_sum(postings.iter().map(|p| p.value))?)
     }
 
+    /// The everyday balance read for one subaccount and asset (ADR-0019): the
+    /// closest cache point (at or before now) plus the folded tail of transfers
+    /// committed after its watermark. Always equal to
+    /// [`compute_balance`](Ledger::compute_balance) at rest, and faster once a
+    /// cache point keeps the tail short. With no cache point yet it returns the
+    /// authoritative live-posting sum directly (rather than folding the whole
+    /// history) and bootstraps a cache point in the background. Once enough
+    /// credits/debits have accrued since the closest cache point, it also appends
+    /// a new one in the background for later reads.
+    #[instrument(skip(self), name = "ledger.balance")]
+    pub async fn balance(&self, account: &AccountId, asset: &AssetId) -> Result<Cent, LedgerError> {
+        let now = now_millis()?;
+        let closest = self
+            .store
+            .get_closest_balance_projection(account, asset, now)
+            .await?;
+        match closest {
+            Some(p) => {
+                let (tail, count) = fold_tail(
+                    self.store(),
+                    account,
+                    asset,
+                    Some(p.watermark.saturating_add(1)),
+                    None,
+                )
+                .await?;
+                if count >= self.snapshot_interval {
+                    self.spawn_append(*account, *asset);
+                }
+                Ok(p.balance.checked_add(tail)?)
+            }
+            // No cache point: the authoritative live sum is O(live postings),
+            // cheaper than folding the whole history. Bootstrap a cache point in
+            // the background (the append itself gates on `snapshot_interval`, so a
+            // small account never actually appends) so later reads use the tail.
+            None => {
+                let balance = self.compute_balance(account, asset).await?;
+                self.spawn_append(*account, *asset);
+                Ok(balance)
+            }
+        }
+    }
+
     /// Report the per-subaccount balances of a base account for one asset.
     ///
-    /// One entry per non-closed subaccount. `sub == None` spans every
-    /// subaccount of `account`'s base id; `Some(s)` restricts to that one.
-    /// Balances are never summed across subaccounts (ADR-0012).
+    /// One entry per non-closed subaccount, each read through the everyday cached
+    /// [`balance`](Ledger::balance) so every balance read goes through one path.
+    /// `sub == None` spans every subaccount of `account`'s base id; `Some(s)`
+    /// restricts to that one. Balances are never summed across subaccounts
+    /// (ADR-0012).
     #[instrument(skip(self), name = "ledger.balances")]
     pub async fn balances(
         &self,
@@ -64,7 +262,7 @@ impl Ledger {
             {
                 continue;
             }
-            let value = self.compute_balance(&subaccount, asset).await?;
+            let value = self.balance(&subaccount, asset).await?;
             result.push(SubAccountBalance {
                 account: subaccount,
                 value,
@@ -94,4 +292,43 @@ impl Ledger {
         subs.sort();
         Ok(subs)
     }
+
+    /// Spawn a best-effort background append gated on `snapshot_interval` new
+    /// credits/debits. Uses only the store handle and config, so it needs no
+    /// `Arc<Self>` and can run from a `&self` read.
+    ///
+    /// Redundant spawns are cheap, not suppressed here: the storage-based debounce
+    /// inside [`append_cache_point`] (keyed on the newest shared cache-point row,
+    /// with `debounce_ms == grace`) returns before the fold when a recent cache
+    /// point already exists, so a hot account read at high QPS does at most one
+    /// fold per grace window, coordinated across every instance rather than in
+    /// this process's memory.
+    fn spawn_append(&self, account: AccountId, asset: AssetId) {
+        let store = Arc::clone(&self.store);
+        let grace = self.projection_grace_ms;
+        let min_new = self.snapshot_interval;
+        tokio::spawn(async move {
+            if let Err(err) =
+                append_cache_point(store.as_ref(), grace, min_new, grace, &account, &asset).await
+            {
+                // Best effort: a failed append only lengthens a later read's tail.
+                // Log it so a projection that silently stops advancing is visible.
+                tracing::warn!(?account, ?asset, error = %err, "balance projection append failed");
+            }
+        });
+    }
+
+    /// Append a cache point for one `(account, asset)` now (folding up to
+    /// `now − grace`), unconditionally. The read path appends lazily in the
+    /// background; this forces one (no debounce), exposed for tests and
+    /// reconciliation. Append-only: a repeat within the same grace-adjusted
+    /// millisecond is a no-op (the watermark is already covered), otherwise it
+    /// adds a fresh row.
+    pub async fn append_cache_point(
+        &self,
+        account: &AccountId,
+        asset: &AssetId,
+    ) -> Result<(), LedgerError> {
+        append_cache_point(self.store(), self.projection_grace_ms, 0, 0, account, asset).await
+    }
 }

crates/kuatia/src/ledger/projection.rs

@@ -1,231 +0,0 @@
-//! Balance cache points (ADR-0019).
-//!
-//! Balance is `snapshot + Σ(creates − consumes)` over committed transfers past
-//! the closest cache point's commit-time watermark. Summing every one of an
-//! account's transfer deltas equals its live-posting sum, so this read always
-//! agrees with [`compute_balance`](Ledger::compute_balance). Cache points are
-//! append-only: a read takes the one closest to (at or before) now and, once
-//! enough new credits/debits have accrued since it, appends a fresh one off the
-//! read path. They are a pure optimization that shortens the tail; correctness
-//! never depends on them.
-
-use std::collections::HashSet;
-use std::sync::Arc;
-
-use tracing::instrument;
-
-use kuatia_core::{AccountId, AssetId, Cent, PostingId};
-use kuatia_storage::store::{EnvelopeRecord, Store, TransferQuery};
-
-use super::{Ledger, now_millis};
-use crate::error::LedgerError;
-
-/// Fold the per-`(account, asset)` delta of a set of committed transfers,
-/// `Σ created(+) − Σ consumed(−)` restricted to postings owned by `account` in
-/// `asset`, and count how many such postings (credits + debits) were seen.
-/// Consumed postings are resolved from the immutable table (the envelope carries
-/// only their ids). All arithmetic is checked, in Rust.
-async fn fold_account_delta(
-    store: &dyn Store,
-    account: &AccountId,
-    asset: &AssetId,
-    records: &[EnvelopeRecord],
-) -> Result<(Cent, u64), LedgerError> {
-    let mut delta = Cent::ZERO;
-    let mut count: u64 = 0;
-
-    // Created side (credits): the envelope carries owner/asset/value directly.
-    for record in records {
-        for np in record.envelope.creates() {
-            if np.owner == *account && np.asset == *asset {
-                delta = delta.checked_add(np.value)?;
-                count += 1;
-            }
-        }
-    }
-
-    // Consumed side (debits): gather every consumed id across the window, resolve
-    // the postings in one batch, and subtract those owned by this (account, asset).
-    let mut consumed_ids: Vec<PostingId> = Vec::new();
-    let mut seen: HashSet<PostingId> = HashSet::new();
-    for record in records {
-        for id in record.envelope.consumes() {
-            if seen.insert(*id) {
-                consumed_ids.push(*id);
-            }
-        }
-    }
-    if !consumed_ids.is_empty() {
-        for posting in store.get_postings(&consumed_ids).await? {
-            if posting.owner == *account && posting.asset == *asset {
-                delta = delta.checked_sub(posting.value)?;
-                count += 1;
-            }
-        }
-    }
-
-    Ok((delta, count))
-}
-
-/// Load the committed transfers involving `account` with commit time in
-/// `[from_ts, to_ts)` (both optional), then fold their `(account, asset)` delta
-/// and credit/debit count. `from_ts == None` spans from the beginning;
-/// `to_ts == None` spans to the newest committed transfer.
-async fn fold_tail(
-    store: &dyn Store,
-    account: &AccountId,
-    asset: &AssetId,
-    from_ts: Option<i64>,
-    to_ts: Option<i64>,
-) -> Result<(Cent, u64), LedgerError> {
-    let query = TransferQuery {
-        account: Some(account.id),
-        sub: Some(account.sub),
-        from_ts,
-        to_ts,
-        ..Default::default()
-    };
-    let records = store.query_transfers(&query).await?.items;
-    fold_account_delta(store, account, asset, &records).await
-}
-
-/// Append a fresh cache point for `(account, asset)`: fold the window since the
-/// closest cache point's watermark up to `now − grace` onto its snapshot, and
-/// append the result. Only appends when that window has at least `min_new`
-/// credits/debits, so a hot account that is read constantly does not append a
-/// near-duplicate row on every read (`min_new == 0` forces an append). Append-only
-/// and best effort; a stale or duplicate append is harmless because a read takes
-/// the closest-at-or-before cache point.
-///
-/// `debounce_ms` is the storage-based single-flight guard. When the newest cache
-/// point already sits within `debounce_ms` below the target watermark, this
-/// returns before the fold, so a hot account read at high QPS does not fan out a
-/// full fold per read. The guard lives in the shared `balance_projection` rows,
-/// not in process memory, so it dedups across every ledger instance and survives
-/// a restart, and it needs no lease, lock, or CAS (ADR-0019). `debounce_ms == 0`
-/// disables it (the reconcile path forces an append regardless).
-async fn append_cache_point(
-    store: &dyn Store,
-    grace_ms: i64,
-    min_new: u64,
-    debounce_ms: i64,
-    account: &AccountId,
-    asset: &AssetId,
-) -> Result<(), LedgerError> {
-    let new_watermark = now_millis()?.saturating_sub(grace_ms);
-    let closest = store
-        .get_closest_balance_projection(account, asset, new_watermark)
-        .await?;
-    let (snapshot, from_ts) = match closest {
-        // Storage-based debounce: a cache point within `debounce_ms` below the
-        // target already shortens the tail enough, so skip the fold. This is what
-        // collapses a per-read fan-out into at most one fold per debounce window.
-        Some(p) if new_watermark.saturating_sub(p.watermark) < debounce_ms => return Ok(()),
-        // A cache point already covers this watermark: nothing to add. (Also the
-        // debounce == 0 stop, so an exact-or-newer watermark is never duplicated.)
-        Some(p) if p.watermark >= new_watermark => return Ok(()),
-        Some(p) => (p.balance, Some(p.watermark.saturating_add(1))),
-        None => (Cent::ZERO, None),
-    };
-    let (fold, count) = fold_tail(
-        store,
-        account,
-        asset,
-        from_ts,
-        Some(new_watermark.saturating_add(1)),
-    )
-    .await?;
-    // Not enough new activity since the closest cache point to earn a new row.
-    if count < min_new {
-        return Ok(());
-    }
-    let balance = snapshot.checked_add(fold)?;
-    store
-        .append_balance_projection(account, asset, balance, new_watermark)
-        .await?;
-    Ok(())
-}
-
-impl Ledger {
-    /// The everyday balance read for one subaccount and asset (ADR-0019): the
-    /// closest cache point (at or before now) plus the folded tail of transfers
-    /// committed after its watermark. Always equal to
-    /// [`compute_balance`](Ledger::compute_balance) at rest, and faster once a
-    /// cache point keeps the tail short. With no cache point yet it returns the
-    /// authoritative live-posting sum directly (rather than folding the whole
-    /// history) and bootstraps a cache point in the background. Once enough
-    /// credits/debits have accrued since the closest cache point, it also appends
-    /// a new one in the background for later reads.
-    #[instrument(skip(self), name = "ledger.balance")]
-    pub async fn balance(&self, account: &AccountId, asset: &AssetId) -> Result<Cent, LedgerError> {
-        let now = now_millis()?;
-        let closest = self
-            .store
-            .get_closest_balance_projection(account, asset, now)
-            .await?;
-        match closest {
-            Some(p) => {
-                let (tail, count) = fold_tail(
-                    self.store(),
-                    account,
-                    asset,
-                    Some(p.watermark.saturating_add(1)),
-                    None,
-                )
-                .await?;
-                if count >= self.snapshot_interval {
-                    self.spawn_append(*account, *asset);
-                }
-                Ok(p.balance.checked_add(tail)?)
-            }
-            // No cache point: the authoritative live sum is O(live postings),
-            // cheaper than folding the whole history. Bootstrap a cache point in
-            // the background (the append itself gates on `snapshot_interval`, so a
-            // small account never actually appends) so later reads use the tail.
-            None => {
-                let balance = self.compute_balance(account, asset).await?;
-                self.spawn_append(*account, *asset);
-                Ok(balance)
-            }
-        }
-    }
-
-    /// Spawn a best-effort background append gated on `snapshot_interval` new
-    /// credits/debits. Uses only the store handle and config, so it needs no
-    /// `Arc<Self>` and can run from a `&self` read.
-    ///
-    /// Redundant spawns are cheap, not suppressed here: the storage-based debounce
-    /// inside [`append_cache_point`] (keyed on the newest shared cache-point row,
-    /// with `debounce_ms == grace`) returns before the fold when a recent cache
-    /// point already exists, so a hot account read at high QPS does at most one
-    /// fold per grace window, coordinated across every instance rather than in
-    /// this process's memory.
-    fn spawn_append(&self, account: AccountId, asset: AssetId) {
-        let store = Arc::clone(&self.store);
-        let grace = self.projection_grace_ms;
-        let min_new = self.snapshot_interval;
-        tokio::spawn(async move {
-            if let Err(err) =
-                append_cache_point(store.as_ref(), grace, min_new, grace, &account, &asset).await
-            {
-                // Best effort: a failed append only lengthens a later read's tail.
-                // Log it so a projection that silently stops advancing is visible.
-                tracing::warn!(?account, ?asset, error = %err, "balance projection append failed");
-            }
-        });
-    }
-
-    /// Append a cache point for one `(account, asset)` now (folding up to
-    /// `now − grace`), unconditionally. The read path appends lazily in the
-    /// background; this forces one (no debounce), exposed for tests and
-    /// reconciliation. Append-only: a repeat within the same grace-adjusted
-    /// millisecond is a no-op (the watermark is already covered), otherwise it
-    /// adds a fresh row.
-    pub async fn append_cache_point(
-        &self,
-        account: &AccountId,
-        asset: &AssetId,
-    ) -> Result<(), LedgerError> {
-        append_cache_point(self.store(), self.projection_grace_ms, 0, 0, account, asset).await
-    }
-}