Extract the affected-row count contract into one commit-path primitive

The dumb-storage contract from ADR-0003 — apply a primitive, compare its
affected-row count to the rows targeted, and on a short count re-read the
end-state to tell idempotent replay from genuine failure — was hand-rolled four
times: once as verify_postings and then re-inlined after each of
deactivate_postings, insert_postings, and store_transfer in finalize_envelope.
A reader had to re-derive the same count reasoning at every write, and three of
the four copies had drifted into always re-reading, discarding the count fast
path that verify_postings already had.

Move the contract into a single apply_and_verify helper and rebuild
verify_postings as a thin posting-state adapter over it. finalize_envelope now
reads as a flat deactivate -> insert -> store -> event sequence, each step
folding its end-state check into one call, and the three finalize writes gain
the count fast path for free. The reserve step is unchanged. The count and
end-state logic now has one focused unit test instead of being exercised only
through full commit paths.

crates/kuatia/src/ledger/commit.rs

@@ -26,7 +26,7 @@ use kuatia_storage::store::EnvelopeRecord;
 use super::envelope_saga::*;
 use super::{Ledger, now_millis};
 use crate::error::LedgerError;
-use crate::saga::{FinalizeInput, LedgerCtx, ReserveInput};
+use crate::saga::{FinalizeInput, LedgerCtx, ReserveInput, apply_and_verify, verify_postings};
 
 /// Phase of an in-flight commit, persisted with the write-ahead record so
 /// recovery knows whether validation has completed.
@@ -404,17 +404,18 @@ impl Ledger {
         // guard: `deactivate_postings(Some(rid))` only removes rows we reserved,
         // so any consumed id still active or reserved by another saga leaves the
         // "all spent" check failing.
-        self.store
+        let spent = self
+            .store
             .deactivate_postings(consumes, Some(reservation))
             .await?;
-        if !consumes.is_empty() {
-            let after = self.store.get_posting_states(consumes).await?;
-            if after.len() != consumes.len() || after.iter().any(|s| *s != PostingState::Spent) {
-                return Err(LedgerError::Store(StoreError::Internal(
-                    "finalize: consumed postings not all spent (contended or not reserved by this saga)".into(),
-                )));
-            }
-        }
+        verify_postings(
+            self.store.as_ref(),
+            consumes,
+            spent,
+            |s| *s == PostingState::Spent,
+            "finalize: consume reserved postings",
+        )
+        .await?;
 
         // Created postings, derived deterministically from the envelope.
         let created: Vec<Posting> = envelope
@@ -433,15 +434,16 @@ impl Ledger {
                 )
             })
             .collect();
-        self.store.insert_postings(&created).await?;
-        if !created.is_empty() {
-            let ids: Vec<PostingId> = created.iter().map(|p| p.id).collect();
-            if self.store.get_postings(&ids).await?.len() != created.len() {
-                return Err(LedgerError::Store(StoreError::Internal(
-                    "finalize: created postings missing after insert".into(),
-                )));
-            }
-        }
+        let inserted = self.store.insert_postings(&created).await?;
+        let created_ids: Vec<PostingId> = created.iter().map(|p| p.id).collect();
+        verify_postings(
+            self.store.as_ref(),
+            &created_ids,
+            inserted,
+            |s| *s != PostingState::Missing,
+            "finalize: insert created postings",
+        )
+        .await?;
 
         // Index both created and consumed owners.
         let mut involved: Vec<AccountId> = created.iter().map(|p| p.owner).collect();
@@ -450,7 +452,8 @@ impl Ledger {
         involved.dedup();
 
         let receipt = Receipt { transfer_id: tid };
-        self.store
+        let stored = self
+            .store
             .store_transfer(
                 EnvelopeRecord {
                     envelope: envelope.clone(),
@@ -460,11 +463,10 @@ impl Ledger {
                 &involved,
             )
             .await?;
-        if self.store.get_transfer(&tid).await?.is_none() {
-            return Err(LedgerError::Store(StoreError::Internal(
-                "finalize: transfer record missing after store".into(),
-            )));
-        }
+        apply_and_verify(stored, 1, "finalize: store transfer record", || async {
+            Ok(self.store.get_transfer(&tid).await?.is_some())
+        })
+        .await?;
 
         self.append_committed_event(tid).await?;
         Ok(receipt)

crates/kuatia/src/saga.rs

@@ -48,35 +48,56 @@ fn internal(message: impl Into<String>) -> LedgerError {
     LedgerError::Store(StoreError::Internal(message.into()))
 }
 
-/// Interpret a dumb primitive's affected-row `count` against the `ids` it
-/// targeted. `count == ids.len()` is success. A short count is acceptable only if
-/// the shortfall is already in the desired end-state — a prior attempt (or this
-/// saga, replayed by recovery) already applied it — verified by reading the
-/// postings and checking `ok`. Otherwise it is a genuine failure (contended or
-/// concurrently modified) and the saga compensates.
-async fn verify_postings(
-    store: &dyn Store,
-    ids: &[PostingId],
+/// The single home of the ADR-0003 affected-row count contract, used after every
+/// dumb write primitive in the commit path.
+///
+/// Interpret a primitive's affected-row `count` against the number of rows it
+/// `target`ed. `count == target` is success. A short count is acceptable only if
+/// the desired end-state already holds (a prior attempt, or this saga replayed by
+/// recovery, already applied it), which `verify` re-reads and reports as a bool.
+/// Otherwise it is a genuine failure (contended or concurrently modified) and the
+/// caller compensates.
+pub(crate) async fn apply_and_verify<F, Fut>(
     count: u64,
-    ok: impl Fn(&PostingState) -> bool,
+    target: usize,
     what: &str,
-) -> Result<(), LedgerError> {
-    if count == ids.len() as u64 {
+    verify: F,
+) -> Result<(), LedgerError>
+where
+    F: FnOnce() -> Fut,
+    Fut: std::future::Future<Output = Result<bool, LedgerError>>,
+{
+    if count == target as u64 {
         return Ok(());
     }
-    let states = store
-        .get_posting_states(ids)
-        .await
-        .map_err(LedgerError::Store)?;
-    if states.len() == ids.len() && states.iter().all(&ok) {
+    if verify().await? {
         return Ok(());
     }
     Err(internal(format!(
-        "{what}: storage applied {count}/{} rows and the end-state is not satisfied",
-        ids.len()
+        "{what}: storage applied {count}/{target} rows and the end-state is not satisfied"
     )))
 }
 
+/// Apply the count contract to a posting primitive whose end-state is a property
+/// of the targeted postings: a short count is idempotent-safe only when every
+/// targeted posting already satisfies `ok`.
+pub(crate) async fn verify_postings(
+    store: &dyn Store,
+    ids: &[PostingId],
+    count: u64,
+    ok: impl Fn(&PostingState) -> bool,
+    what: &str,
+) -> Result<(), LedgerError> {
+    apply_and_verify(count, ids.len(), what, || async {
+        let states = store
+            .get_posting_states(ids)
+            .await
+            .map_err(LedgerError::Store)?;
+        Ok(states.len() == ids.len() && states.iter().all(&ok))
+    })
+    .await
+}
+
 // ---------------------------------------------------------------------------
 // Saga context -- carries the ledger handle + state between steps
 // ---------------------------------------------------------------------------
@@ -396,3 +417,54 @@ impl Step<LedgerCtx, LedgerError> for DepositMovementStep {
         compensate_last_receipt(ctx).await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::cell::Cell;
+
+    #[tokio::test]
+    async fn full_count_is_ok_without_re_reading() {
+        let verified = Cell::new(false);
+        let result = apply_and_verify(3, 3, "reserve", || {
+            verified.set(true);
+            async { Ok(true) }
+        })
+        .await;
+        assert!(result.is_ok());
+        assert!(
+            !verified.get(),
+            "a full count must not re-read the end-state"
+        );
+    }
+
+    #[tokio::test]
+    async fn short_count_is_ok_when_end_state_already_holds() {
+        // Idempotent replay: a prior attempt applied the shortfall.
+        let result = apply_and_verify(2, 3, "reserve", || async { Ok(true) }).await;
+        assert!(result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn short_count_is_internal_error_when_end_state_missing() {
+        let result = apply_and_verify(2, 3, "reserve", || async { Ok(false) }).await;
+        assert!(matches!(
+            result,
+            Err(LedgerError::Store(StoreError::Internal(_)))
+        ));
+    }
+
+    #[tokio::test]
+    async fn verify_error_propagates() {
+        let result = apply_and_verify(0, 1, "store", || async {
+            Err(LedgerError::Store(StoreError::Internal(
+                "read failed".into(),
+            )))
+        })
+        .await;
+        assert!(matches!(
+            result,
+            Err(LedgerError::Store(StoreError::Internal(_)))
+        ));
+    }
+}