From af71da4dbccb6fe871505ccf7f825ad81495f5c6 Mon Sep 17 00:00:00 2001 From: findolor <16416963+findolor@users.noreply.github.com> Date: Mon, 1 Jun 2026 08:39:56 +0000 Subject: [PATCH] fix: make vault balance materialization idempotent (#2594) ## Motivation Vault balances materialized in the local DB could become stale or incorrect when a sync window was replayed or overlapped. The running balance update used the existing materialized balance plus the current window delta, so a bad baseline could survive and replayed windows were not fully idempotent. ## Solution - Recompute affected running vault balances from all `derived_vault_deltas` up to the target block instead of adding the current window onto `running_vault_balances`. - Compute vault balance-change prefix balances from prior derived deltas rather than the materialized running balance table. - Delete balance-change rows in the refreshed block window before reinserting them. - Add SQLite regression coverage for replaying the same window and recomputing a later window from all derived deltas. ## Checks By submitting this for review, I'm confirming I've done the following: - [x] made this PR as small as possible - [x] unit-tested any new functionality - [ ] linked any relevant issues or PRs - [ ] included screenshots (if this involves a front-end change) Verified locally: - `COMMIT_SHA=local-test cargo test -p raindex_common local_db::query::upsert_vault_balances --lib` - `COMMIT_SHA=local-test cargo test -p raindex_common local_db::pipeline::adapters::apply --lib` ## Summary by CodeRabbit * **Bug Fixes** * Fixed vault balance computation to include all derived deltas, ensuring accurate empty-window and snapshot results. * **Refactor** * Reworked balance upsert flow to process only affected keys and avoid double-counting when reprocessing windows. * **Tests** * Updated tests and integration scenarios to validate batch ordering and balance correctness. * **Chores** * Updated registry URL used by the webapp. --- .../src/local_db/pipeline/adapters/apply.rs | 9 +- .../insert_balance_changes.sql | 20 +- .../insert_running_balances.sql | 100 +++------ .../query/upsert_vault_balances/mod.rs | 195 +++++++++++++++++- packages/webapp/src/lib/constants.ts | 2 +- 5 files changed, 234 insertions(+), 92 deletions(-) diff --git a/crates/common/src/local_db/pipeline/adapters/apply.rs b/crates/common/src/local_db/pipeline/adapters/apply.rs index 2c7b0bb1d2..62230348ea 100644 --- a/crates/common/src/local_db/pipeline/adapters/apply.rs +++ b/crates/common/src/local_db/pipeline/adapters/apply.rs @@ -375,14 +375,19 @@ mod tests { // Expect derived state refreshes plus the watermark and ANALYZE when there is no work. let texts: Vec<_> = batch.statements().iter().map(|s| s.sql().trim()).collect(); - assert_eq!(texts.len(), 8); + assert_eq!(texts.len(), 9); assert!(texts .iter() .any(|s| s.contains("DELETE FROM derived_vault_deltas"))); assert!(texts .iter() .any(|s| s.contains("INSERT OR REPLACE INTO derived_vault_deltas"))); - assert!(texts.iter().any(|s| s.contains("vault_balance_changes"))); + assert!(texts + .iter() + .any(|s| s.contains("DELETE FROM vault_balance_changes"))); + assert!(texts + .iter() + .any(|s| s.contains("INSERT OR IGNORE INTO vault_balance_changes"))); assert!(texts .iter() .any(|s| s.contains("INSERT OR REPLACE INTO running_vault_balances"))); diff --git a/crates/common/src/local_db/query/upsert_vault_balances/insert_balance_changes.sql b/crates/common/src/local_db/query/upsert_vault_balances/insert_balance_changes.sql index 709580f144..2d72434e14 100644 --- a/crates/common/src/local_db/query/upsert_vault_balances/insert_balance_changes.sql +++ b/crates/common/src/local_db/query/upsert_vault_balances/insert_balance_changes.sql @@ -22,17 +22,23 @@ filtered AS ( PARTITION BY vd.chain_id, vd.raindex_address, vd.owner, vd.token, vd.vault_id ORDER BY vd.block_number, vd.log_index ) AS rn, - COALESCE(rvb.balance, FLOAT_ZERO_HEX()) AS prefix_balance + COALESCE( + ( + SELECT FLOAT_SUM(prev.delta ORDER BY prev.block_number, prev.log_index) + FROM derived_vault_deltas prev + WHERE prev.chain_id = vd.chain_id + AND prev.raindex_address = vd.raindex_address + AND prev.owner = vd.owner + AND prev.token = vd.token + AND prev.vault_id = vd.vault_id + AND prev.block_number < p.start_block + ), + FLOAT_ZERO_HEX() + ) AS prefix_balance FROM derived_vault_deltas vd JOIN params p ON p.chain_id = vd.chain_id AND p.raindex_address = vd.raindex_address - LEFT JOIN running_vault_balances rvb - ON rvb.chain_id = vd.chain_id - AND rvb.raindex_address = vd.raindex_address - AND rvb.owner = vd.owner - AND rvb.token = vd.token - AND rvb.vault_id = vd.vault_id WHERE vd.block_number BETWEEN p.start_block AND p.end_block ), ordered AS ( diff --git a/crates/common/src/local_db/query/upsert_vault_balances/insert_running_balances.sql b/crates/common/src/local_db/query/upsert_vault_balances/insert_running_balances.sql index 39ed22b9b4..4f2946e6dc 100644 --- a/crates/common/src/local_db/query/upsert_vault_balances/insert_running_balances.sql +++ b/crates/common/src/local_db/query/upsert_vault_balances/insert_running_balances.sql @@ -1,18 +1,34 @@ -WITH latest_blocks AS ( - SELECT +WITH affected_keys AS ( + SELECT DISTINCT chain_id, raindex_address, owner, token, - vault_id, - MAX(block_number) AS last_block + vault_id FROM derived_vault_deltas vd WHERE vd.chain_id = ?1 AND vd.raindex_address = ?2 AND vd.block_number BETWEEN ?3 AND ?4 - GROUP BY chain_id, raindex_address, owner, token, vault_id ), -delta_batches AS ( +latest_blocks AS ( + SELECT + vd.chain_id, + vd.raindex_address, + vd.owner, + vd.token, + vd.vault_id, + MAX(vd.block_number) AS last_block + FROM derived_vault_deltas vd + JOIN affected_keys ak + ON ak.chain_id = vd.chain_id + AND ak.raindex_address = vd.raindex_address + AND ak.owner = vd.owner + AND ak.token = vd.token + AND ak.vault_id = vd.vault_id + WHERE vd.block_number <= ?4 + GROUP BY vd.chain_id, vd.raindex_address, vd.owner, vd.token, vd.vault_id +), +aggregated AS ( SELECT vd.chain_id, vd.raindex_address, @@ -22,7 +38,7 @@ delta_batches AS ( COALESCE( FLOAT_SUM(vd.delta ORDER BY vd.block_number, vd.log_index), FLOAT_ZERO_HEX() - ) AS balance_delta, + ) AS balance, lb.last_block, ( SELECT MAX(vd2.log_index) @@ -41,63 +57,8 @@ delta_batches AS ( AND lb.owner = vd.owner AND lb.token = vd.token AND lb.vault_id = vd.vault_id - WHERE vd.chain_id = ?1 - AND vd.raindex_address = ?2 - AND vd.block_number BETWEEN ?3 AND ?4 + WHERE vd.block_number <= ?4 GROUP BY vd.chain_id, vd.raindex_address, vd.owner, vd.token, vd.vault_id, lb.last_block -), -existing_matching AS ( - SELECT - mvb.chain_id, - mvb.raindex_address, - mvb.owner, - mvb.token, - mvb.vault_id, - mvb.balance AS balance_value, - mvb.last_block, - mvb.last_log_index - FROM running_vault_balances mvb - JOIN delta_batches db - ON db.chain_id = mvb.chain_id - AND db.raindex_address = mvb.raindex_address - AND db.owner = mvb.owner - AND db.token = mvb.token - AND db.vault_id = mvb.vault_id -), -combined AS ( - SELECT - chain_id, - raindex_address, - owner, - token, - vault_id, - balance_delta AS contribution, - last_block, - last_log_index - FROM delta_batches - UNION ALL - SELECT - chain_id, - raindex_address, - owner, - token, - vault_id, - balance_value AS contribution, - last_block, - last_log_index - FROM existing_matching -), -aggregated AS ( - SELECT - chain_id, - raindex_address, - owner, - token, - vault_id, - COALESCE(FLOAT_SUM(contribution), FLOAT_ZERO_HEX()) AS balance, - MAX(last_block) AS last_block - FROM combined - GROUP BY chain_id, raindex_address, owner, token, vault_id ) INSERT OR REPLACE INTO running_vault_balances ( chain_id, @@ -118,17 +79,6 @@ SELECT a.vault_id, a.balance, a.last_block, - ( - SELECT c.last_log_index - FROM combined c - WHERE c.chain_id = a.chain_id - AND c.raindex_address = a.raindex_address - AND c.owner = a.owner - AND c.token = a.token - AND c.vault_id = a.vault_id - AND c.last_block = a.last_block - ORDER BY c.last_log_index DESC - LIMIT 1 - ) AS last_log_index, + a.last_log_index, (CAST(strftime('%s', 'now') AS INTEGER) * 1000) AS updated_at FROM aggregated a; diff --git a/crates/common/src/local_db/query/upsert_vault_balances/mod.rs b/crates/common/src/local_db/query/upsert_vault_balances/mod.rs index 02debc726a..9c184cd84b 100644 --- a/crates/common/src/local_db/query/upsert_vault_balances/mod.rs +++ b/crates/common/src/local_db/query/upsert_vault_balances/mod.rs @@ -18,7 +18,30 @@ pub fn upsert_vault_balances_batch( end_block, ); let running_stmt = build_stmt(UPSERT_RUNNING_SQL, raindex_id, start_block, end_block); - SqlStatementBatch::from(vec![change_stmt, running_stmt]) + SqlStatementBatch::from(vec![ + delete_balance_changes_stmt(raindex_id, start_block, end_block), + change_stmt, + running_stmt, + ]) +} + +fn delete_balance_changes_stmt( + raindex_id: &RaindexIdentifier, + start_block: u64, + end_block: u64, +) -> SqlStatement { + SqlStatement::new_with_params( + "DELETE FROM vault_balance_changes +WHERE chain_id = ?1 + AND raindex_address = ?2 + AND block_number BETWEEN ?3 AND ?4", + [ + SqlValue::from(raindex_id.chain_id), + SqlValue::from(raindex_id.raindex_address), + SqlValue::from(start_block), + SqlValue::from(end_block), + ], + ) } fn build_stmt( @@ -47,7 +70,7 @@ mod tests { fn batch_binds_all_params() { let raindex_id = RaindexIdentifier::new(111, Address::from([0x11u8; 20])); let batch = upsert_vault_balances_batch(&raindex_id, 100, 200); - assert_eq!(batch.len(), 2); + assert_eq!(batch.len(), 3); for stmt in batch.statements() { assert_eq!(stmt.params().len(), 4); assert_eq!(stmt.params()[0], SqlValue::U64(111)); @@ -68,8 +91,9 @@ mod tests { .iter() .map(|s| s.sql().to_lowercase()) .collect(); - assert!(sql[0].contains("insert or ignore into vault_balance_changes")); - assert!(sql[1].contains("insert or replace into running_vault_balances")); + assert!(sql[0].contains("delete from vault_balance_changes")); + assert!(sql[1].contains("insert or ignore into vault_balance_changes")); + assert!(sql[2].contains("insert or replace into running_vault_balances")); } #[test] @@ -93,7 +117,7 @@ mod tests { #[test] fn running_stmt_includes_zero_balance_batches() { let batch = upsert_vault_balances_batch(&RaindexIdentifier::new(4, Address::ZERO), 0, 0); - let sql = batch.statements()[1].sql().to_lowercase(); + let sql = batch.statements()[2].sql().to_lowercase(); assert!( !sql.contains("having not float_is_zero"), "should not filter out zero balance batches" @@ -103,14 +127,171 @@ mod tests { #[test] fn running_stmt_uses_float_sum_for_updates() { let batch = upsert_vault_balances_batch(&RaindexIdentifier::new(5, Address::ZERO), 0, 1); - let sql = batch.statements()[1].sql().to_lowercase(); + let sql = batch.statements()[2].sql().to_lowercase(); assert!( sql.contains("insert or replace into running_vault_balances"), "missing INSERT OR REPLACE clause" ); assert!( - sql.contains("coalesce(float_sum"), + sql.contains("float_sum"), "missing FLOAT_SUM aggregation in query" ); } + + #[cfg(not(target_family = "wasm"))] + mod sqlite_integration { + use super::*; + use crate::local_db::query::create_tables::CREATE_TABLES_SQL; + use rain_math_float::Float; + use rusqlite::{params, Connection}; + + const CHAIN_ID: u32 = 8453; + const RAINDEX: &str = "0xe522cb4a5fcb2eb31a52ff41a4653d85a4fd7c9d"; + const OWNER: &str = "0xa9c16673f65ae808688cb18952afe3d9658c808f"; + const TOKEN: &str = "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913"; + const VAULT: &str = "0x000000000000000000000000000000000000000000000000000000000000fab4"; + + fn setup_conn() -> Connection { + let conn = Connection::open_in_memory().expect("open sqlite"); + crate::local_db::functions::register_all(&conn).expect("register local db functions"); + conn.execute_batch(CREATE_TABLES_SQL) + .expect("create tables"); + conn + } + + fn raindex_id() -> RaindexIdentifier { + RaindexIdentifier::new(CHAIN_ID, RAINDEX.parse().expect("valid raindex")) + } + + fn float(value: &str) -> String { + Float::parse(value.to_string()) + .expect("valid float") + .as_hex() + } + + fn sqlvalue_to_rusqlite(v: SqlValue) -> rusqlite::types::Value { + match v { + SqlValue::Text(t) => rusqlite::types::Value::Text(t), + SqlValue::I64(i) => rusqlite::types::Value::Integer(i), + SqlValue::U64(u) => rusqlite::types::Value::Integer(u as i64), + SqlValue::Null => rusqlite::types::Value::Null, + } + } + + fn execute_stmt(conn: &Connection, stmt: &SqlStatement) { + let params = stmt + .params() + .iter() + .cloned() + .map(sqlvalue_to_rusqlite) + .collect::>(); + conn.execute(stmt.sql(), rusqlite::params_from_iter(params)) + .expect("execute SQL statement"); + } + + fn execute_balance_batch(conn: &Connection, start_block: u64, end_block: u64) { + for stmt in + upsert_vault_balances_batch(&raindex_id(), start_block, end_block).statements() + { + execute_stmt(conn, stmt); + } + } + + fn seed_delta( + conn: &Connection, + block_number: u64, + log_index: u64, + kind: &str, + amount: &str, + ) { + conn.execute( + "INSERT INTO derived_vault_deltas ( + chain_id, raindex_address, transaction_hash, log_index, block_number, + block_timestamp, owner, kind, token, vault_id, delta + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", + params![ + CHAIN_ID, + RAINDEX, + format!("0x{:064x}", log_index), + log_index, + block_number, + block_number + 1000, + OWNER, + kind, + TOKEN, + VAULT, + float(amount) + ], + ) + .expect("insert derived delta"); + } + + fn running_balance(conn: &Connection) -> String { + conn.query_row( + "SELECT balance FROM running_vault_balances + WHERE chain_id = ?1 + AND raindex_address = ?2 + AND owner = ?3 + AND token = ?4 + AND vault_id = ?5", + params![CHAIN_ID, RAINDEX, OWNER, TOKEN, VAULT], + |row| row.get::<_, String>(0), + ) + .expect("running balance") + } + + fn change_running_balance(conn: &Connection, block_number: u64) -> String { + conn.query_row( + "SELECT running_balance FROM vault_balance_changes + WHERE chain_id = ?1 + AND raindex_address = ?2 + AND owner = ?3 + AND token = ?4 + AND vault_id = ?5 + AND block_number = ?6", + params![CHAIN_ID, RAINDEX, OWNER, TOKEN, VAULT, block_number], + |row| row.get::<_, String>(0), + ) + .expect("change running balance") + } + + #[test] + fn replaying_same_window_does_not_double_count_running_balance() { + let conn = setup_conn(); + seed_delta(&conn, 10, 1, "DEPOSIT", "10"); + + execute_balance_batch(&conn, 10, 10); + assert_eq!(running_balance(&conn), float("10")); + assert_eq!(change_running_balance(&conn, 10), float("10")); + + execute_balance_batch(&conn, 10, 10); + assert_eq!(running_balance(&conn), float("10")); + assert_eq!(change_running_balance(&conn, 10), float("10")); + assert_eq!( + conn.query_row("SELECT COUNT(*) FROM vault_balance_changes", [], |row| { + row.get::<_, i64>(0) + }) + .expect("change count"), + 1 + ); + } + + #[test] + fn replaying_later_window_recomputes_from_all_derived_deltas() { + let conn = setup_conn(); + seed_delta(&conn, 10, 1, "DEPOSIT", "10"); + seed_delta(&conn, 20, 2, "WITHDRAW", "-10"); + + execute_balance_batch(&conn, 10, 10); + assert_eq!(running_balance(&conn), float("10")); + + execute_balance_batch(&conn, 20, 20); + assert_eq!(running_balance(&conn), float("0")); + assert_eq!(change_running_balance(&conn, 20), float("0")); + + execute_balance_batch(&conn, 20, 20); + assert_eq!(running_balance(&conn), float("0")); + assert_eq!(change_running_balance(&conn, 20), float("0")); + } + } } diff --git a/packages/webapp/src/lib/constants.ts b/packages/webapp/src/lib/constants.ts index 9dfbb223b7..f9e65b27a5 100644 --- a/packages/webapp/src/lib/constants.ts +++ b/packages/webapp/src/lib/constants.ts @@ -1,2 +1,2 @@ export const REGISTRY_URL = - "https://raw.githubusercontent.com/rainlanguage/rain.strategies/96007c4d5a06cad390b67112a55f8a892dad29fc/registry"; + "https://raw.githubusercontent.com/rainlanguage/rain.strategies/85c5d10a507f9759b936ae6327e06c2d759cee55/registry";