From 1e407fc4fe0929483757b45f05350c39089dccf0 Mon Sep 17 00:00:00 2001 From: Lukasz Klimek <842586+lklimek@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:07:51 +0000 Subject: [PATCH] fix(wallet-storage): load platform addresses in the wallet snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Read Platform checkpoints, registrations and balances within the existing per-wallet SQLite transaction used for Core reconstruction. Publish both halves only after success, while preserving typed corruption handling, raw address counts and Strict/Recovery isolation. Cover an actual external SQLite writer deterministically, atomic omission of failed wallets, and unattributable corrupt wallet IDs. Account for the three targeted Platform reads in the bounded per-wallet query budget. 🤖 Co-authored by [Claudius the Magnificent](https://github.com/lklimek/claudius) AI Agent Co-Authored-By: Claude GPT-6 Astra --- .../src/sqlite/persister.rs | 64 ++-- .../src/sqlite/schema/accounts.rs | 4 +- .../src/sqlite/schema/platform_addrs.rs | 97 ++++-- .../integration/sqlite_load_reconstruction.rs | 300 +++++++++++++++++- 4 files changed, 382 insertions(+), 83 deletions(-) diff --git a/packages/rs-platform-wallet-storage/src/sqlite/persister.rs b/packages/rs-platform-wallet-storage/src/sqlite/persister.rs index 7b46cef2c55..b600191d93d 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/persister.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/persister.rs @@ -9,8 +9,8 @@ use rusqlite::{Connection, OptionalExtension, Transaction}; use dpp::prelude::Identifier; use platform_wallet::changeset::changeset::SweepBatch; use platform_wallet::changeset::{ - ClientStartState, IdentityChangeSet, PersistenceCapabilities, PersistenceError, - PlatformWalletChangeSet, PlatformWalletPersistence, + ClientStartState, ClientWalletStartState, IdentityChangeSet, PersistenceCapabilities, + PersistenceError, PlatformWalletChangeSet, PlatformWalletPersistence, }; use platform_wallet::wallet::identity::ManagedIdentity; use platform_wallet::wallet::platform_wallet::WalletId; @@ -1533,46 +1533,26 @@ impl PlatformWalletPersistence for SqlitePersister { .map_err(PersistenceError::from)?; } - let addrs_all = schema::platform_addrs::load_all(&conn).map_err(PersistenceError::from)?; + schema::platform_addrs::validate_wallet_ids(&conn).map_err(PersistenceError::from)?; let mut addresses_loaded: usize = 0; - // Wallets whose platform-address rows could not be read. They are - // already counted here, so the loop below skips them rather than - // rebuilding a wallet whose platform balance would be understated. - let mut unreadable: std::collections::BTreeSet = - std::collections::BTreeSet::new(); - for (wallet_id, entry) in addrs_all { - let (addrs, count) = match entry { - Ok(entry) => entry, - Err(original) => { - degrade_whole_wallet(&ctx, wallet_id, &original) - .map_err(|_| PersistenceError::from(original))?; - unreadable.insert(wallet_id); - continue; - } - }; - // Skip a wallet with no platform state at all (no addresses, - // no registrations, all sync watermarks zero). - if count > 0 - || !addrs.per_account.is_empty() - || addrs.sync_height > 0 - || addrs.sync_timestamp > 0 - || addrs.last_known_recent_block > 0 - { - addresses_loaded += count; - state.platform_addresses.insert(wallet_id, addrs); - } - } // Per-wallet keyless rehydration payload; the manager rebuilds each // wallet watch-only and derives signing keys later on demand. let wallet_ids = schema::wallets::list_ids(&conn).map_err(PersistenceError::from)?; let wallets_seen = wallet_ids.len(); for wallet_id in wallet_ids { - if unreadable.contains(&wallet_id) { - continue; - } match load_one_wallet(&conn, wallet_id, &ctx, self.config.load_policy) { - Ok(wallet_state) => { + Ok((wallet_state, (addrs, count))) => { + // Publish both halves only after the wallet's transaction succeeds. + if count > 0 + || !addrs.per_account.is_empty() + || addrs.sync_height > 0 + || addrs.sync_timestamp > 0 + || addrs.last_known_recent_block > 0 + { + addresses_loaded += count; + state.platform_addresses.insert(wallet_id, addrs); + } state.wallets.insert(wallet_id, wallet_state); } // The isolation boundary. One wallet's failure is recorded @@ -1689,11 +1669,13 @@ fn load_one_wallet( wallet_id: WalletId, ctx: &LoadCtx, policy: LoadPolicy, -) -> Result { +) -> Result<(ClientWalletStartState, schema::platform_addrs::LoadAllEntry), PersistenceError> { let tx = rusqlite::Transaction::new_unchecked(conn, rusqlite::TransactionBehavior::Immediate) .map_err(WalletStorageError::from) .map_err(PersistenceError::from)?; let counted_before = ctx.counts_checkpoint(); + let platform_state = + schema::platform_addrs::load_entry(&tx, &wallet_id).map_err(PersistenceError::from)?; let (mut state, sweeps) = load_wallet_snapshot(&tx, wallet_id, ctx)?; if !sweeps.is_empty() { schema::core_state::apply_replay_sweeps(&tx, &wallet_id, sweeps) @@ -1722,7 +1704,7 @@ fn load_one_wallet( } .map_err(WalletStorageError::from) .map_err(PersistenceError::from)?; - Ok(state) + Ok((state, platform_state)) } fn load_wallet_snapshot( @@ -1974,16 +1956,6 @@ fn record_wallet_degradation( Ok(()) } -/// [`record_wallet_degradation`] for a failure that is already a typed -/// storage error, so its own kind tag is the cause. -fn degrade_whole_wallet( - ctx: &LoadCtx, - wallet_id: WalletId, - original: &WalletStorageError, -) -> Result<(), WalletStorageError> { - record_wallet_degradation(ctx, wallet_id, original.error_kind_str(), original) -} - /// The kind tag of the typed storage error inside a persistence error. /// /// The boundary attributes a dropped wallet to its cause, and the cause is diff --git a/packages/rs-platform-wallet-storage/src/sqlite/schema/accounts.rs b/packages/rs-platform-wallet-storage/src/sqlite/schema/accounts.rs index d7ab594edcd..ab8a1fd06f0 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/schema/accounts.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/schema/accounts.rs @@ -84,7 +84,6 @@ fn decode_platform_payment_row( /// Every `platform_payment` registration for one wallet, decoded into /// `(account_index, xpub)`. -#[cfg(any(test, feature = "__test-helpers"))] pub(crate) fn list_platform_payment_registrations( conn: &Connection, wallet_id: &WalletId, @@ -109,8 +108,7 @@ pub(crate) fn list_platform_payment_registrations( /// Bulk variant of [`list_platform_payment_registrations`]: every /// wallet's `platform_payment` registrations in one scan, grouped by -/// `wallet_id`. Used by `load()` to avoid a per-wallet registrations -/// query. +/// `wallet_id`. pub(crate) fn all_platform_payment_registrations( conn: &Connection, ) -> Result< diff --git a/packages/rs-platform-wallet-storage/src/sqlite/schema/platform_addrs.rs b/packages/rs-platform-wallet-storage/src/sqlite/schema/platform_addrs.rs index 3d8e777e2d7..56a565b7020 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/schema/platform_addrs.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/schema/platform_addrs.rs @@ -112,36 +112,28 @@ pub struct PlatformAddressRow { pub funds: AddressFunds, } -#[cfg(any(test, feature = "__test-helpers"))] +/// Decode every address row for one wallet, including unregistered accounts. pub fn list_per_wallet( conn: &Connection, wallet_id: &WalletId, ) -> Result, WalletStorageError> { let mut stmt = conn.prepare( - "SELECT account_index, address_index, address, balance, nonce, as_of_height \ + "SELECT account_index, address_index, length(address), address, balance, nonce, as_of_height \ FROM platform_addresses WHERE wallet_id = ?1 \ ORDER BY account_index, address_index, address", )?; - let rows = stmt.query_map(params![wallet_id.as_slice()], |row| { - Ok(( - row.get::<_, i64>(0)?, - row.get::<_, i64>(1)?, - row.get::<_, Vec>(2)?, - row.get::<_, i64>(3)?, - row.get::<_, i64>(4)?, - row.get::<_, i64>(5)?, - )) - })?; + let mut rows = stmt.query(params![wallet_id.as_slice()])?; let mut out = Vec::new(); - for r in rows { - let (account_index, address_index, address_bytes, balance, nonce, as_of_height) = r?; + while let Some(row) = rows.next()? { + blob::check_fixed_width(row.get::<_, i64>(2)?, 20, "platform_addresses.address")?; + let address_bytes: Vec = row.get(3)?; out.push(decode_address_row( - account_index, - address_index, + row.get(0)?, + row.get(1)?, &address_bytes, - balance, - nonce, - as_of_height, + row.get(4)?, + row.get(5)?, + row.get(6)?, )?); } Ok(out) @@ -196,14 +188,19 @@ fn account_state_from_rows( /// Build `PlatformAddressSyncStartState` for one wallet: the /// network-scoped sync watermark plus the per-account committed state /// reconstructed from registrations + address rows. -/// -/// `load()` uses the grouped [`load_all`] path; this per-wallet form is -/// retained for this crate's integration tests. #[cfg(any(test, feature = "__test-helpers"))] pub fn load_state( conn: &Connection, wallet_id: &WalletId, ) -> Result { + load_entry(conn, wallet_id).map(|(state, _)| state) +} + +/// Read one wallet inside the caller's snapshot, retaining the raw address count. +pub(crate) fn load_entry( + conn: &Connection, + wallet_id: &WalletId, +) -> Result { let row: Option<(i64, i64, i64)> = conn .query_row( "SELECT sync_height, sync_timestamp, last_known_recent_block \ @@ -222,12 +219,54 @@ pub fn load_state( }; let registrations = accounts::list_platform_payment_registrations(conn, wallet_id)?; let address_rows = list_per_wallet(conn, wallet_id)?; - Ok(PlatformAddressSyncStartState { - per_account: build_per_account(®istrations, &address_rows), - sync_height: h, - sync_timestamp: t, - last_known_recent_block: r, - }) + Ok(( + PlatformAddressSyncStartState { + per_account: build_per_account(®istrations, &address_rows), + sync_height: h, + sync_timestamp: t, + last_known_recent_block: r, + }, + address_rows.len(), + )) +} + +/// Reject unattributable corrupt IDs before per-wallet isolation skips them. +pub(crate) fn validate_wallet_ids(conn: &Connection) -> Result<(), WalletStorageError> { + for (sql, column) in [ + ( + "SELECT length(wallet_id), wallet_id FROM platform_address_sync \ + WHERE length(wallet_id) != 32 OR typeof(wallet_id) != 'blob' LIMIT 1", + "platform_addresses.wallet_id", + ), + ( + "SELECT length(wallet_id), wallet_id FROM platform_addresses \ + WHERE length(wallet_id) != 32 OR typeof(wallet_id) != 'blob' LIMIT 1", + "platform_addresses.wallet_id", + ), + ( + "SELECT length(wallet_id), wallet_id FROM account_registrations \ + WHERE account_type = 'platform_payment' \ + AND (length(wallet_id) != 32 OR typeof(wallet_id) != 'blob') LIMIT 1", + "account_registrations.wallet_id", + ), + ] { + let mut stmt = conn.prepare(sql)?; + let mut rows = stmt.query([])?; + if let Some(row) = rows.next()? { + let length: i64 = row.get(0)?; + if column == "account_registrations.wallet_id" { + blob::check_fixed_width(length, 32, column)?; + } else if length != 32 { + return Err(WalletStorageError::InvalidWalletIdLength { + column, + actual: usize::try_from(length).unwrap_or(usize::MAX), + }); + } + let bytes: Vec = row.get(1)?; + super::id32(column, &bytes)?; + } + } + Ok(()) } /// Total `platform_addresses` row count per wallet — used by tests @@ -279,7 +318,7 @@ fn fail_wallet(out: &mut PerWallet, wallet_id: WalletId, err: WalletStorag } } -/// Bulk reader for `load()`. Cost is a fixed number of grouped scans — +/// Bulk reader. Cost is a fixed number of grouped scans — /// one over `platform_address_sync`, one over `platform_addresses`, and /// one over the `platform_payment` `account_registrations` — regardless /// of wallet count, rather than a per-wallet fan-out. diff --git a/packages/rs-platform-wallet-storage/tests/integration/sqlite_load_reconstruction.rs b/packages/rs-platform-wallet-storage/tests/integration/sqlite_load_reconstruction.rs index 4206b1db3bc..ed1f01c6bdb 100644 --- a/packages/rs-platform-wallet-storage/tests/integration/sqlite_load_reconstruction.rs +++ b/packages/rs-platform-wallet-storage/tests/integration/sqlite_load_reconstruction.rs @@ -7,14 +7,19 @@ use crate::common; +use std::sync::{Arc, Mutex}; + use common::{ensure_wallet_meta, fresh_persister, wid}; use dash_sdk::platform::address_sync::AddressFunds; use key_wallet::PlatformP2PKHAddress; use platform_wallet::changeset::{ - PlatformAddressBalanceEntry, PlatformAddressChangeSet, PlatformWalletChangeSet, - PlatformWalletPersistence, + AccountRegistrationEntry, CoreChangeSet, PersistenceError, PlatformAddressBalanceEntry, + PlatformAddressChangeSet, PlatformWalletChangeSet, PlatformWalletPersistence, +}; +use platform_wallet_storage::{ + LoadPolicy, SqlitePersister, SqlitePersisterConfig, WalletStorageError, }; -use platform_wallet_storage::WalletStorageError; +use rusqlite::hooks::{AuthAction, Authorization}; fn entry( wallet_id: [u8; 32], @@ -78,6 +83,290 @@ fn test_xpub() -> key_wallet::bip32::ExtendedPubKey { ).unwrap()).unwrap() } +fn replace_snapshot_dataset( + conn: &rusqlite::Connection, + wallet: &[u8; 32], +) -> rusqlite::Result<()> { + let tx = rusqlite::Transaction::new_unchecked(conn, rusqlite::TransactionBehavior::Immediate)?; + let registration = AccountRegistrationEntry { + account_type: key_wallet::account::AccountType::PlatformPayment { + account: 8, + key_class: 0, + }, + account_xpub: test_xpub(), + }; + let payload = + bincode::serde::encode_to_vec(®istration, bincode::config::standard()).unwrap(); + tx.execute("UPDATE platform_address_sync SET sync_height = 200, sync_timestamp = 200, last_known_recent_block = 200 WHERE wallet_id = ?1", [wallet.as_slice()])?; + tx.execute("UPDATE platform_addresses SET account_index = 8, balance = 200, nonce = 2, as_of_height = 200 WHERE wallet_id = ?1", [wallet.as_slice()])?; + tx.execute("UPDATE account_registrations SET account_index = 8, account_xpub_bytes = ?2 WHERE wallet_id = ?1 AND account_type = 'platform_payment'", rusqlite::params![wallet.as_slice(), payload])?; + tx.execute( + "UPDATE core_sync_state SET synced_height = 200 WHERE wallet_id = ?1", + [wallet.as_slice()], + )?; + tx.commit() +} + +fn platform_snapshot_with_external_writer(policy: LoadPolicy) { + let (persister, _tmp, path) = fresh_persister(); + let wallet = wid(0x63); + ensure_wallet_meta(&persister, &wallet); + persister + .store( + wallet, + PlatformWalletChangeSet { + account_registrations: vec![AccountRegistrationEntry { + account_type: key_wallet::account::AccountType::PlatformPayment { + account: 7, + key_class: 0, + }, + account_xpub: test_xpub(), + }], + platform_addresses: Some(PlatformAddressChangeSet { + addresses: vec![PlatformAddressBalanceEntry { + wallet_id: wallet, + account_index: 7, + address_index: 0, + address: PlatformP2PKHAddress::new([0x63; 20]), + funds: AddressFunds { + balance: 100, + nonce: 1, + as_of_height: 100, + }, + }], + sync_height: Some(100), + sync_timestamp: Some(100), + last_known_recent_block: Some(100), + }), + core: Some(CoreChangeSet { + synced_height: Some(100), + ..Default::default() + }), + ..Default::default() + }, + ) + .unwrap(); + drop(persister); + let persister = + SqlitePersister::open(SqlitePersisterConfig::new(&path).with_load_policy(policy)).unwrap(); + let writer = rusqlite::Connection::open(&path).unwrap(); + writer.busy_timeout(std::time::Duration::ZERO).unwrap(); + let attempt = Arc::new(Mutex::new(None)); + let callback_attempt = Arc::clone(&attempt); + // This exact read follows the checkpoint read. No timing or production seam is needed. + persister + .lock_conn_for_test() + .authorizer(Some(move |ctx: rusqlite::hooks::AuthContext<'_>| { + if matches!( + ctx.action, + AuthAction::Read { + table_name: "platform_addresses", + column_name: "balance" + } + ) { + let mut attempted = callback_attempt.lock().unwrap(); + if attempted.is_none() { + *attempted = Some(replace_snapshot_dataset(&writer, &wallet)); + } + } + Authorization::Allow + })) + .unwrap(); + let state = persister.load().unwrap(); + let platform = &state.platform_addresses[&wallet]; + let account_indices: Vec<_> = platform.per_account.keys().copied().collect(); + let balances: Vec<_> = platform + .per_account + .values() + .flat_map(|a| { + a.found() + .values() + .map(|f| (f.balance, f.nonce, f.as_of_height)) + }) + .collect(); + assert_eq!( + ( + platform.sync_height, + platform.sync_timestamp, + platform.last_known_recent_block, + account_indices, + balances, + state.wallets[&wallet].wallet_info.metadata.synced_height + ), + (100, 100, 100, vec![7], vec![(100, 1, 100)], 100), + "Platform and Core must come from the same wallet snapshot" + ); + assert!( + matches!(attempt.lock().unwrap().as_ref().expect("Platform read must trigger writer"), Err(rusqlite::Error::SqliteFailure(e, _)) if e.code == rusqlite::ErrorCode::DatabaseBusy) + ); + persister + .lock_conn_for_test() + .authorizer(None::) -> Authorization>) + .unwrap(); + replace_snapshot_dataset(&rusqlite::Connection::open(&path).unwrap(), &wallet).unwrap(); + let state = persister.load().unwrap(); + let platform = &state.platform_addresses[&wallet]; + assert_eq!( + ( + platform.sync_height, + platform.sync_timestamp, + platform.last_known_recent_block + ), + (200, 200, 200) + ); + assert_eq!( + platform.per_account.keys().copied().collect::>(), + vec![8] + ); + assert_eq!( + platform.per_account[&8].found()[&PlatformP2PKHAddress::new([0x63; 20])], + AddressFunds { + balance: 200, + nonce: 2, + as_of_height: 200 + } + ); + assert_eq!( + state.wallets[&wallet].wallet_info.metadata.synced_height, + 200 + ); +} + +#[test] +fn should_load_platform_and_core_from_one_snapshot_in_strict_mode() { + platform_snapshot_with_external_writer(LoadPolicy::Strict); +} + +#[test] +fn should_load_platform_and_core_from_one_snapshot_in_recovery_mode() { + platform_snapshot_with_external_writer(LoadPolicy::Recovery); +} + +#[test] +fn should_omit_platform_state_when_core_wallet_reconstruction_fails() { + let (persister, _tmp, path) = fresh_persister(); + let healthy = wid(0x61); + let sick = wid(0x62); + for wallet in [healthy, sick] { + ensure_wallet_meta(&persister, &wallet); + persister + .store( + wallet, + PlatformWalletChangeSet { + platform_addresses: Some(PlatformAddressChangeSet { + sync_height: Some(100), + ..Default::default() + }), + ..Default::default() + }, + ) + .unwrap(); + } + drop(persister); + let persister = SqlitePersister::open( + SqlitePersisterConfig::new(&path).with_load_policy(LoadPolicy::Recovery), + ) + .unwrap(); + { + let conn = persister.lock_conn_for_test(); + // Simulate persisted corruption without bypassing the production loader. + conn.execute_batch("PRAGMA ignore_check_constraints = ON") + .unwrap(); + conn.execute( + "UPDATE wallets SET network = 'unknown' WHERE wallet_id = ?1", + [sick.as_slice()], + ) + .unwrap(); + conn.execute_batch("PRAGMA ignore_check_constraints = OFF") + .unwrap(); + } + let state = persister.load().unwrap(); + assert!(state.wallets.contains_key(&healthy)); + assert!(state.platform_addresses.contains_key(&healthy)); + assert!(!state.wallets.contains_key(&sick)); + assert!( + !state.platform_addresses.contains_key(&sick), + "a failed wallet must not publish its other half" + ); + assert!(persister + .last_load_degradation() + .wallets_degraded + .contains_key(&sick)); +} + +#[test] +fn should_reject_unattributable_platform_wallet_ids_in_both_load_policies() { + for policy in [LoadPolicy::Strict, LoadPolicy::Recovery] { + for table in [ + "platform_address_sync", + "platform_addresses", + "account_registrations", + ] { + let (persister, _tmp, path) = fresh_persister(); + let wallet = wid(0x64); + ensure_wallet_meta(&persister, &wallet); + persister + .store( + wallet, + PlatformWalletChangeSet { + account_registrations: vec![AccountRegistrationEntry { + account_type: key_wallet::account::AccountType::PlatformPayment { + account: 7, + key_class: 0, + }, + account_xpub: test_xpub(), + }], + platform_addresses: Some(PlatformAddressChangeSet { + addresses: vec![entry(wallet, 7, 0, 0x64)], + sync_height: Some(100), + ..Default::default() + }), + ..Default::default() + }, + ) + .unwrap(); + drop(persister); + let persister = + SqlitePersister::open(SqlitePersisterConfig::new(&path).with_load_policy(policy)) + .unwrap(); + { + let conn = persister.lock_conn_for_test(); + conn.execute_batch( + "PRAGMA foreign_keys = OFF; PRAGMA ignore_check_constraints = ON", + ) + .unwrap(); + conn.execute( + &format!("UPDATE {table} SET wallet_id = ?1"), + [[0x64; 31].as_slice()], + ) + .unwrap(); + conn.execute_batch( + "PRAGMA foreign_keys = ON; PRAGMA ignore_check_constraints = OFF", + ) + .unwrap(); + } + let PersistenceError::Backend { source, .. } = persister.load().unwrap_err() else { + panic!("typed storage error expected") + }; + let error = source.downcast_ref::().unwrap(); + if table == "account_registrations" { + assert!( + matches!(error, WalletStorageError::BlobDecode { .. }), + "{error:?}" + ); + } else { + assert!( + matches!( + error, + WalletStorageError::InvalidWalletIdLength { actual: 31, .. } + ), + "{error:?}" + ); + } + } + } +} + /// A `platform_payment` registration + platform_addresses for an account /// reconstruct into `load_state().per_account` keyed by the account /// index. Addresses for the account seed the per-account state. @@ -1542,12 +1831,13 @@ fn tc_p4_012_load_query_count_bounded() { (N=1 → {count_one}, N=10 → {count_ten}, delta → {delta})" ); let per_wallet = delta / 9; + // Three Platform readers share the wallet transaction with the Core readers. assert!( - (1..=20).contains(&per_wallet), + (1..=23).contains(&per_wallet), "per-wallet statement count must be small + bounded, got {per_wallet}" ); // Shared (wallet-count-independent) overhead: the `list_ids` + - // `platform_addrs::load_all` scans. `count_one = shared + per_wallet` + // Platform wallet-ID validation. `count_one = shared + per_wallet` // ⇒ shared must itself be a small constant, not growing with N. let shared = count_one - per_wallet; assert!(