fix(account_set): activate eventually_consistent flag for async balance rollup

[bodymindarts]

Summary

• Fix two bugs in NewAccountBuilder where eventually_consistent() set the wrong field and into_values() hardcoded it to false
• Add recalculate_balances() to AccountSets service that aggregates member account balances into account set balance snapshots for async rollup
• Add eventually_consistent field to NewAccountSet builder so account sets can opt in to deferred balance computation
• Include integration test proving end-to-end flow: EC account set skips inline balance update, recalculate_balances() materializes correct aggregate

Test plan

• Integration test eventually_consistent_balances verifies:
  • Account set created with eventually_consistent: true
  • Transaction posts without updating account set balance inline
  • recalculate_balances() produces correct aggregated balance
• Existing balances test continues to pass (non-EC account sets unaffected)
• nix flake check passes (fmt, clippy -D warnings, nextest)

🤖 Generated with Claude Code

---

Note

High Risk
High risk because it changes core balance aggregation/concurrency behavior (new advisory locking, watermarking via seq, and batch recalculation paths) and introduces a schema change affecting balance history/current balance writes.

Overview
Enables eventually-consistent account sets whose balances are no longer updated inline during posting; instead, balances are materialized via new AccountSets::recalculate_balances* APIs (single, batch, and deep/descendant) that rebuild account-set snapshots from member cala_balance_history deltas.

Fixes propagation of the eventually_consistent flag through NewAccount/NewAccountSet, adds new validation that membership changes are rejected once a member has any balance history, and adds pagination/query support for listing EC account set ids.

Reworks balance persistence to support async rollups: adds cala_balance_history.seq (with index/sequence) and cala_current_balances.latest_seq watermarking, updates advisory-lock strategy for posters vs recalculators, and extends effective-balance logic to delete/rebuild cumulative effective snapshots from the earliest affected effective date. Includes extensive new integration tests covering EC behavior, batching/deep recalc, idempotency, and error cases.

^{Reviewed by Cursor Bugbot for commit 0d7e524. Bugbot is set up for automated code reviews on this repo. Configure here.}

[github-actions]

📊 Performance Report

Commit: 0d7e524
Updated: 2026-04-09 14:43:33 UTC

Cala Performance Benchmark Results (non-representative)

Criterion Benchmark Results (single-threaded)

Benchmark	Time per Run	Throughput	% vs Baseline
post_simple_transaction	7.423ms	134 tx/s	0 (baseline)
post_and_recalculate_ec_account_set	18.901ms	52 tx/s	-154.0%
post_and_batch_recalculate_ec_account_set	13.156ms	76 tx/s	-77.0%
post_multi_layer_transaction	7.674ms	130 tx/s	-3.0%
post_simple_transaction_with_effective_balances	9.713ms	102 tx/s	-30.0%
post_simple_transaction_with_skipped_velocity	6.670ms	149 tx/s	+10.0%
post_simple_transaction_with_velocity	8.424ms	118 tx/s	-13.0%
post_simple_transaction_with_hit_velocity	4.200ms	238 tx/s	+43.0%
post_simple_transaction_with_one_account_set	6.809ms	146 tx/s	+8.0%
post_simple_transaction_with_five_account_sets	8.046ms	124 tx/s	-8.0%
post_simple_transaction_with_ec_account_set	6.565ms	152 tx/s	+11.0%

Load Testing Results (parallel-execution)

Scenario	tx/s
1 parallel	93.38
2 parallel	127.08
5 parallel	163.21
10 parallel	164.76
20 parallel	160.73
2 contention	125.60
5 contention	142.75
2 acct_sets	115.29
5 acct_sets	125.80

---

Note: Performance results may vary based on system resources and database state.

Last updated by commit 0d7e524

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Watermark advancement in add_member skips member history
  • For EC account sets, add/remove member now preserves the pre-change set watermark and restores latest_seq after summary snapshot writes so pending member history cannot be skipped.

Or push these changes by commenting:

@cursor push e5ccc8171a

Preview (e5ccc8171a)

diff --git a/cala-ledger/src/account_set/mod.rs b/cala-ledger/src/account_set/mod.rs
--- a/cala-ledger/src/account_set/mod.rs
+++ b/cala-ledger/src/account_set/mod.rs
@@ -168,6 +168,15 @@
         member: impl Into<AccountSetMemberId>,
     ) -> Result<AccountSet, AccountSetError> {
         let member = member.into();
+        let target_account_id = AccountId::from(account_set_id);
+        let target_account = self
+            .accounts
+            .find_all_in_op::<Account>(&mut *op, &[target_account_id])
+            .await?
+            .remove(&target_account_id)
+            .ok_or(AccountSetError::CouldNotFindById(account_set_id))?;
+        let is_ec = target_account.values().config.eventually_consistent;
+
         let (time, parents, account_set, member_id) = match member {
             AccountSetMemberId::Account(id) => {
                 tracing::Span::current().record("is_account", true);
@@ -207,12 +216,23 @@
             }
         };
 
+        let ec_watermark = if is_ec {
+            self.balances
+                .load_balance_watermark_in_op(
+                    op,
+                    account_set.values().journal_id,
+                    target_account_id,
+                )
+                .await?
+        } else {
+            None
+        };
+
         let balances = self
             .balances
             .find_balances_for_update(op, account_set.values().journal_id, member_id)
             .await?;
 
-        let target_account_id = AccountId::from(&account_set.id());
         let mut entries = Vec::new();
         for balance in balances.into_values() {
             entries_for_add_balance(&mut entries, target_account_id, balance);
@@ -223,13 +243,6 @@
         }
         let entries = self.entries.create_all_in_op(op, entries).await?;
 
-        let target_account = self
-            .accounts
-            .find_all_in_op::<Account>(&mut *op, &[target_account_id])
-            .await?
-            .remove(&target_account_id)
-            .ok_or(AccountSetError::CouldNotFindById(account_set_id))?;
-        let is_ec = target_account.values().config.eventually_consistent;
         let entries_for_ec = if is_ec { Some(entries.clone()) } else { None };
 
         let mappings = std::iter::once((target_account_id, parents)).collect();
@@ -252,6 +265,7 @@
                     target_account_id,
                     &ec_entries,
                     time,
+                    ec_watermark,
                 )
                 .await?;
         }
@@ -281,6 +295,15 @@
         member: impl Into<AccountSetMemberId>,
     ) -> Result<AccountSet, AccountSetError> {
         let member = member.into();
+        let target_account_id = AccountId::from(account_set_id);
+        let target_account = self
+            .accounts
+            .find_all_in_op::<Account>(&mut *op, &[target_account_id])
+            .await?
+            .remove(&target_account_id)
+            .ok_or(AccountSetError::CouldNotFindById(account_set_id))?;
+        let is_ec = target_account.values().config.eventually_consistent;
+
         let (time, parents, account_set, member_id) = match member {
             AccountSetMemberId::Account(id) => {
                 let set = self.repo.find_by_id_in_op(&mut *op, account_set_id).await?;
@@ -314,12 +337,23 @@
             }
         };
 
+        let ec_watermark = if is_ec {
+            self.balances
+                .load_balance_watermark_in_op(
+                    op,
+                    account_set.values().journal_id,
+                    target_account_id,
+                )
+                .await?
+        } else {
+            None
+        };
+
         let balances = self
             .balances
             .find_balances_for_update(op, account_set.values().journal_id, member_id)
             .await?;
 
-        let target_account_id = AccountId::from(&account_set.id());
         let mut entries = Vec::new();
         for balance in balances.into_values() {
             entries_for_remove_balance(&mut entries, target_account_id, balance);
@@ -330,13 +364,6 @@
         }
         let entries = self.entries.create_all_in_op(op, entries).await?;
 
-        let target_account = self
-            .accounts
-            .find_all_in_op::<Account>(&mut *op, &[target_account_id])
-            .await?
-            .remove(&target_account_id)
-            .ok_or(AccountSetError::CouldNotFindById(account_set_id))?;
-        let is_ec = target_account.values().config.eventually_consistent;
         let entries_for_ec = if is_ec { Some(entries.clone()) } else { None };
 
         let mappings = std::iter::once((target_account_id, parents)).collect();
@@ -359,6 +386,7 @@
                     target_account_id,
                     &ec_entries,
                     time,
+                    ec_watermark,
                 )
                 .await?;
         }

diff --git a/cala-ledger/src/balance/mod.rs b/cala-ledger/src/balance/mod.rs
--- a/cala-ledger/src/balance/mod.rs
+++ b/cala-ledger/src/balance/mod.rs
@@ -170,6 +170,20 @@
             .await
     }
 
+    pub(crate) async fn load_balance_watermark_in_op(
+        &self,
+        op: &mut impl es_entity::AtomicOperation,
+        journal_id: JournalId,
+        account_id: AccountId,
+    ) -> Result<Option<i64>, BalanceError> {
+        Ok(self
+            .repo
+            .load_account_set_balances_batch(op, journal_id, &[account_id])
+            .await?
+            .remove(&account_id)
+            .and_then(|(_, watermark)| watermark))
+    }
+
     #[instrument(name = "cala_ledger.balance.update_balance_for_account_in_op", skip(self, op, entries), fields(journal_id = %journal_id, account_id = %account_id))]
     pub(crate) async fn update_balance_for_account_in_op(
         &self,
@@ -178,6 +192,7 @@
         account_id: AccountId,
         entries: &[EntryValues],
         created_at: DateTime<Utc>,
+        watermark: Option<i64>,
     ) -> Result<(), BalanceError> {
         let current = self
             .repo
@@ -195,6 +210,9 @@
         self.repo
             .insert_new_snapshots(op, journal_id, new_balances)
             .await?;
+        self.repo
+            .set_latest_seq_for_account(op, journal_id, account_id, watermark)
+            .await?;
         Ok(())
     }
 

diff --git a/cala-ledger/src/balance/repo.rs b/cala-ledger/src/balance/repo.rs
--- a/cala-ledger/src/balance/repo.rs
+++ b/cala-ledger/src/balance/repo.rs
@@ -304,6 +304,31 @@
         Ok(())
     }
 
+    pub(crate) async fn set_latest_seq_for_account(
+        &self,
+        op: &mut impl es_entity::AtomicOperation,
+        journal_id: JournalId,
+        account_id: AccountId,
+        latest_seq: Option<i64>,
+    ) -> Result<(), BalanceError> {
+        if let Some(latest_seq) = latest_seq {
+            sqlx::query(
+                r#"
+                UPDATE cala_current_balances
+                SET latest_seq = LEAST(latest_seq, $3::bigint)
+                WHERE journal_id = $1
+                  AND account_id = $2
+                "#,
+            )
+            .bind(uuid::Uuid::from(journal_id))
+            .bind(uuid::Uuid::from(account_id))
+            .bind(latest_seq)
+            .execute(op.as_executor())
+            .await?;
+        }
+        Ok(())
+    }
+
     #[instrument(
         name = "balance.load_all_for_update",
         skip(self, op),

diff --git a/cala-ledger/tests/account_set.rs b/cala-ledger/tests/account_set.rs
--- a/cala-ledger/tests/account_set.rs
+++ b/cala-ledger/tests/account_set.rs
@@ -1112,6 +1112,113 @@
 }
 
 #[tokio::test]
+async fn eventually_consistent_second_member_add_keeps_existing_member_history(
+) -> anyhow::Result<()> {
+    let btc: Currency = "BTC".parse().unwrap();
+
+    let pool = helpers::init_pool().await?;
+    let cala_config = CalaLedgerConfig::builder()
+        .pool(pool)
+        .exec_migrations(false)
+        .build()?;
+    let cala = CalaLedger::init(cala_config).await?;
+
+    let journal = cala
+        .journals()
+        .create(helpers::test_journal())
+        .await
+        .unwrap();
+
+    let (sender, first_member) = helpers::test_accounts();
+    let (_, second_member) = helpers::test_accounts();
+    let sender_account = cala.accounts().create(sender).await.unwrap();
+    let first_member_account = cala.accounts().create(first_member).await.unwrap();
+    let second_member_account = cala.accounts().create(second_member).await.unwrap();
+
+    let tx_code = Alphanumeric.sample_string(&mut rand::rng(), 32);
+    cala.tx_templates()
+        .create(helpers::currency_conversion_template(&tx_code))
+        .await
+        .unwrap();
+
+    let inline_set = NewAccountSet::builder()
+        .id(AccountSetId::new())
+        .name("Inline Set")
+        .journal_id(journal.id())
+        .build()
+        .unwrap();
+    let inline_set = cala.account_sets().create(inline_set).await.unwrap();
+
+    let ec_set = NewAccountSet::builder()
+        .id(AccountSetId::new())
+        .name("EC Set")
+        .journal_id(journal.id())
+        .eventually_consistent(true)
+        .build()
+        .unwrap();
+    let ec_set = cala.account_sets().create(ec_set).await.unwrap();
+
+    // First member joins with no existing balance.
+    cala.account_sets()
+        .add_member(inline_set.id(), first_member_account.id())
+        .await
+        .unwrap();
+    cala.account_sets()
+        .add_member(ec_set.id(), first_member_account.id())
+        .await
+        .unwrap();
+
+    // Transactions for first member remain unreconciled on the EC set.
+    for _ in 0..2 {
+        let mut params = Params::new();
+        params.insert("journal_id", journal.id().to_string());
+        params.insert("sender", sender_account.id());
+        params.insert("recipient", first_member_account.id());
+        cala.post_transaction(TransactionId::new(), &tx_code, params)
+            .await
+            .unwrap();
+    }
+
+    // Give second member pre-existing balance before joining both sets.
+    for _ in 0..2 {
+        let mut params = Params::new();
+        params.insert("journal_id", journal.id().to_string());
+        params.insert("sender", sender_account.id());
+        params.insert("recipient", second_member_account.id());
+        cala.post_transaction(TransactionId::new(), &tx_code, params)
+            .await
+            .unwrap();
+    }
+
+    cala.account_sets()
+        .add_member(inline_set.id(), second_member_account.id())
+        .await
+        .unwrap();
+    cala.account_sets()
+        .add_member(ec_set.id(), second_member_account.id())
+        .await
+        .unwrap();
+
+    cala.account_sets()
+        .recalculate_balances(ec_set.id())
+        .await
+        .unwrap();
+
+    let inline_bal = cala
+        .balances()
+        .find(journal.id(), inline_set.id(), btc)
+        .await?;
+    let ec_bal = cala.balances().find(journal.id(), ec_set.id(), btc).await?;
+    assert_eq!(
+        inline_bal.settled(),
+        ec_bal.settled(),
+        "EC set should not lose pre-existing member deltas when adding another member",
+    );
+
+    Ok(())
+}
+
+#[tokio::test]
 async fn list_eventually_consistent_ids() -> anyhow::Result<()> {
     let pool = helpers::init_pool().await?;
     let cala_config = CalaLedgerConfig::builder()

_{This Bugbot Autofix run was free. To enable autofix for future PRs, go to the Cursor dashboard.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 262a20b. Configure here.}