diff --git a/packages/rs-platform-wallet-storage/SCHEMA.md b/packages/rs-platform-wallet-storage/SCHEMA.md index 899d88a0c7f..882825c11c8 100644 --- a/packages/rs-platform-wallet-storage/SCHEMA.md +++ b/packages/rs-platform-wallet-storage/SCHEMA.md @@ -357,6 +357,12 @@ SQL lookups without blob decoding. pair. Sentinel `0` / `zeroblob(32)` default for account types without that axis (including the provider key-material types). - FK: `wallet_id → wallets(wallet_id) ON DELETE CASCADE`. +- `dashpay_covered_from` (V020): the height a `'dashpay_receiving'` account + is covered from in the DashPay coreHeight backfill record (DIP-15 §12.6, + dashpay/platform#4302); NULL when the record does not cover it. A CHECK + limits it to `'dashpay_receiving'` rows. The record's extent lives on + [`core_sync_state`](#core_sync_state); the account upsert never touches + this column. ### `pending_contact_crypto` @@ -452,7 +458,19 @@ finalized. Rows are removed when the transaction becomes confirmed. One row per wallet, holding monotonically-advancing SPV sync watermarks and the last applied ChainLock. `last_processed_height` and `synced_height` are -NULL until the first block is processed. `last_applied_chain_lock` is NULL +NULL until the first block is processed. The one exception to the monotonic +max is a `synced_height` marked `CoreChangeSet::synced_height_is_rewind`: the +DashPay coreHeight backfill lowers the cursor deliberately, and the +backfill record committed in the same transaction vouches for the lowered +value, so a rewind overwrites the stored cursor. + +`dashpay_backfill_floor` / `dashpay_backfill_rewound_from` (V020) are that +record's extent: the lowest height the backfill rewound the cursor to and the +cursor it rewound from. Both set or both NULL (CHECK). Its covered receiving +accounts are `account_registrations.dashpay_covered_from`. Every write +replaces the whole record; an empty record, or one none of whose accounts has +a row, sets both NULL and clears every `dashpay_covered_from`. NULL restores +the empty record, which upstream answers by re-running the backfill. `last_applied_chain_lock` is NULL until a ChainLock has been applied and flushed. `chainlock_height` (V007) mirrors `CoreChangeSet::last_applied_chain_lock` @@ -541,7 +559,10 @@ pending). The request columns hold a bincode-encoded `ContactRequest`; `accepted_accounts` holds a bincode-encoded `Vec`. `payment_channel_broken` is nullable and read as false when NULL; it is set when external-account registration permanently fails for a contact and -cleared on a superseding rotation. +cleared on a superseding rotation. `external_account_reference` (V020) is the +outbound-account marker — the account reference the contact's outbound +account was last built for — written only on established rows, NULL when +absent; a NULL marker makes the next sweep rebuild that account once. - PK: `(wallet_id, owner_id, contact_id)`. - FK: `wallet_id → wallets(wallet_id) ON DELETE CASCADE`. @@ -850,3 +871,4 @@ table-rebuild migration, as V004 does. | V017 | `V017__identity_scan_state.rs` | Adds `identity_scan_states` (one row per wallet: the last gap-limit identity-scan verdict — `complete`, `probed_from`/`probed_through`, `unlocated_gap`) and `identity_scan_failed_indices` (indices probed without an answer, cascading from the verdict row via `wallet_id`). Purely additive; an upgraded database reads back "no verdict recorded" for every wallet until the next scan (dashpay/platform#4365). | | V018 | `V018__identity_hard_delete.rs` | Retires identity tombstoning. Adds `cascade_children_on_identity_delete` (brooms `identity_keys` / `contacts` / `ignored_senders` / `pending_contact_crypto` by the deleted identity id, covering the rows no live FK reaches) plus its access-path indexes `idx_contacts_owner`, `idx_ignored_senders_owner`, and `idx_pending_contact_crypto_owner`; purges every already-tombstoned identity and its dependents; drops `identities.tombstoned`. | | V019 | `V019__asset_lock_operation_id.rs` | Adds the nullable `asset_locks.operation_id BLOB` column (CHECK: NULL or exactly 16 bytes) that correlates a wallet-drain operation with its funding outpoint; the lifecycle blob is untouched. Adds the unique partial index `asset_locks_wallet_operation_id` on `(wallet_id, operation_id) WHERE operation_id IS NOT NULL` and the `asset_locks_operation_id_immutable` BEFORE UPDATE trigger, which aborts any change to a non-NULL `operation_id`. Purely additive; existing rows read back NULL. | +| V020 | `V020__dashpay_backfill.rs` | Adds the DashPay coreHeight backfill record as nullable columns: `core_sync_state.dashpay_backfill_floor` / `dashpay_backfill_rewound_from` (its extent, both set or both NULL) and `account_registrations.dashpay_covered_from` (per covered `'dashpay_receiving'` account), plus nullable `contacts.external_account_reference` (outbound-account marker). Purely additive; an upgraded database reads back "no backfill recorded" and no marker for every wallet, so the first launch re-runs the backfill once (dashpay/platform#4302). | diff --git a/packages/rs-platform-wallet-storage/migrations/V020__dashpay_backfill.rs b/packages/rs-platform-wallet-storage/migrations/V020__dashpay_backfill.rs new file mode 100644 index 00000000000..ff8eb59a6dc --- /dev/null +++ b/packages/rs-platform-wallet-storage/migrations/V020__dashpay_backfill.rs @@ -0,0 +1,46 @@ +//! Persist the DashPay coreHeight backfill record and the outbound-account +//! marker as plain columns on the tables they describe. +//! +//! The backfill record is the durable half of the guard that stops the +//! DIP-15 §12.6 rescan from re-lowering the cursor it lowers. Without it the +//! guard lived only in memory while the lowered cursor was durable, so every +//! launch restored a cursor inside the climb, saw no guard, and rewound again +//! (dashpay/platform#4302). +//! +//! - `core_sync_state.dashpay_backfill_floor` / `dashpay_backfill_rewound_from`: +//! the backfill's extent, beside the `synced_height` it lowered. Both set or +//! both NULL. +//! - `account_registrations.dashpay_covered_from`: the height a DashPay +//! receiving account is covered from. A covered entry names exactly one +//! receiving account (owner, contact, index), so its height is a property of +//! that account's row; only `dashpay_receiving` rows may carry one. +//! - `contacts.external_account_reference`: the account reference a contact's +//! outbound account was last built for. A restored marker spares the +//! cold-start rebuild of that account; `NULL` restores "no marker", which +//! rebuilds it once — the behaviour every row had before this column existed. +//! +//! Purely additive: every existing row reads back NULL, which upstream treats +//! as "no backfill recorded" and re-runs once — slow, never lossy. + +pub fn migration() -> String { + "\ +ALTER TABLE core_sync_state ADD COLUMN dashpay_backfill_floor INTEGER + CHECK (dashpay_backfill_floor IS NULL + OR dashpay_backfill_floor BETWEEN 0 AND 4294967295); + +ALTER TABLE core_sync_state ADD COLUMN dashpay_backfill_rewound_from INTEGER + CHECK ((dashpay_backfill_rewound_from IS NULL) = (dashpay_backfill_floor IS NULL) + AND (dashpay_backfill_rewound_from IS NULL + OR dashpay_backfill_rewound_from BETWEEN 0 AND 4294967295)); + +ALTER TABLE account_registrations ADD COLUMN dashpay_covered_from INTEGER + CHECK (dashpay_covered_from IS NULL + OR (account_type = 'dashpay_receiving' + AND dashpay_covered_from BETWEEN 0 AND 4294967295)); + +ALTER TABLE contacts ADD COLUMN external_account_reference INTEGER + CHECK (external_account_reference IS NULL + OR external_account_reference BETWEEN 0 AND 4294967295); +" + .to_string() +} diff --git a/packages/rs-platform-wallet-storage/src/sqlite/persister.rs b/packages/rs-platform-wallet-storage/src/sqlite/persister.rs index 15634f6293f..a0ae1fd73ad 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/persister.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/persister.rs @@ -1303,6 +1303,22 @@ impl PlatformWalletPersistence for SqlitePersister { self.config.flush_mode == FlushMode::Immediate } + /// Every `store` writes the record in the same transaction as the cursor + /// it travels with (`core_sync_state` and `account_registrations` + /// columns, see `schema::dashpay_backfill`), and an empty record clears + /// the stored one. Durable coverage additionally needs inline rounds, so + /// in [`FlushMode::Manual`] `persists_backfill_coverage` still reads + /// `false` through [`store_commits_inline`](Self::store_commits_inline). + fn stores_dashpay_backfill(&self) -> bool { + true + } + + /// [`LoadPolicy::Recovery`] serves a degraded projection and refuses + /// every write ([`ensure_writable`](Self::ensure_writable)). + fn refuses_writes(&self) -> bool { + self.config.load_policy == LoadPolicy::Recovery + } + fn persistence_capabilities(&self) -> PersistenceCapabilities { // Every `flush_inner` applies the complete changeset in one SQLite // transaction. The current schema also has lossless token balances, @@ -1517,7 +1533,17 @@ impl PlatformWalletPersistence for SqlitePersister { /// # } /// ``` fn load(&self) -> Result { - let conn = self.conn().map_err(PersistenceError::from)?; + let mut guard = self.conn().map_err(PersistenceError::from)?; + // One read snapshot for the whole load. The mutex excludes only this + // process, so without it another process could commit between two + // reads: a rewound cursor plus its backfill coverage landing after + // the cursor read but before the coverage read would hand back the + // old cursor beside coverage that suppresses the rescan of the range + // it skipped. A deferred transaction takes no lock until its first + // read, which in WAL mode pins the snapshot every later read sees. + let conn = guard + .transaction() + .map_err(|e| PersistenceError::from(WalletStorageError::from(e)))?; let ctx = LoadCtx::new(self.config.load_policy); // Cleared up front so a failed load never leaves the previous load's // snapshot behind, masquerading as this one's verdict. @@ -1607,6 +1633,9 @@ impl PlatformWalletPersistence for SqlitePersister { unimplemented_rows = degradation.unimplemented_rows, "load() summary" ); + // Read-only, so committing only releases the snapshot. + conn.commit() + .map_err(|e| PersistenceError::from(WalletStorageError::from(e)))?; self.replace_load_degradation(degradation); Ok(state) } @@ -1851,6 +1880,8 @@ fn load_one_wallet( )) })?; } + let dashpay_backfill = schema::dashpay_backfill::load_for_wallet(conn, &wallet_id) + .map_err(PersistenceError::from)?; Ok(platform_wallet::changeset::ClientWalletStartState { wallet, wallet_info, @@ -1861,10 +1892,7 @@ fn load_one_wallet( // replay inert here, which is the behaviour this path had before the // field existed. unconfirmed_outgoing_txs: Vec::new(), - // Not persisted by this backend yet: the backfill record rides the - // FFI hosts' wallet row only. Empty keeps the rescan sweep on its - // pre-record behaviour here (a rewind per launch, never lossy). - dashpay_backfill: Default::default(), + dashpay_backfill, }) } @@ -2230,6 +2258,9 @@ fn apply_changeset_to_tx( if let Some(scan_state) = cs.identity_scan_state.as_ref() { schema::identity_scan_states::apply(tx, wallet_id, scan_state)?; } + if let Some(record) = cs.dashpay_backfill.as_ref() { + schema::dashpay_backfill::apply(tx, wallet_id, record)?; + } if let Some(balances) = cs.token_balances.as_ref() { schema::token_balances::apply(tx, wallet_id, balances)?; } diff --git a/packages/rs-platform-wallet-storage/src/sqlite/rehydrate.rs b/packages/rs-platform-wallet-storage/src/sqlite/rehydrate.rs index 270e0a436d0..ed69f2c95c6 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/rehydrate.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/rehydrate.rs @@ -290,8 +290,15 @@ pub fn apply_persisted_core_state( wallet_info.metadata.last_processed_height = wallet_info.metadata.last_processed_height.max(h); } + // The stored cursor is restored as written, not maxed against the seeded + // `birth_height - 1`: a DashPay backfill rewind can lower it below the + // wallet's birth (`reconcile_dashpay_rescan` passes the contact's + // checkpoint to `update_synced_height` unclamped), and the restored + // backfill record vouches for exactly that cursor. Taking the seed would + // skip the range between them for a contact the record calls covered. + // The FFI restore path assigns it the same way. if let Some(h) = core.synced_height { - wallet_info.metadata.synced_height = wallet_info.metadata.synced_height.max(h); + wallet_info.metadata.synced_height = h; } // Restore the highest applied chainlock when the persister carries it diff --git a/packages/rs-platform-wallet-storage/src/sqlite/schema/contacts.rs b/packages/rs-platform-wallet-storage/src/sqlite/schema/contacts.rs index c7d482995d7..a691b730a7f 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/schema/contacts.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/schema/contacts.rs @@ -16,6 +16,7 @@ use platform_wallet::wallet::platform_wallet::WalletId; use crate::sqlite::error::WalletStorageError; use crate::sqlite::load_ctx::{LoadCtx, LoadSite}; use crate::sqlite::schema::blob; +use crate::sqlite::util::safe_cast::i64_to_u32; use dpp::prelude::Identifier; use platform_wallet::changeset::{ @@ -211,8 +212,9 @@ pub fn apply( let mut stmt = tx.prepare_cached( "INSERT INTO contacts \ (wallet_id, owner_id, contact_id, state, outgoing_request, incoming_request, \ - alias, note, is_hidden, accepted_accounts, payment_channel_broken) \ - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) \ + alias, note, is_hidden, accepted_accounts, payment_channel_broken, \ + external_account_reference) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) \ ON CONFLICT(wallet_id, owner_id, contact_id) DO UPDATE SET \ state = excluded.state, \ outgoing_request = excluded.outgoing_request, \ @@ -221,7 +223,8 @@ pub fn apply( note = excluded.note, \ is_hidden = excluded.is_hidden, \ accepted_accounts = excluded.accepted_accounts, \ - payment_channel_broken = excluded.payment_channel_broken", + payment_channel_broken = excluded.payment_channel_broken, \ + external_account_reference = excluded.external_account_reference", )?; for (key, established) in &cs.established { let outgoing = blob::encode(&established.outgoing_request)?; @@ -239,6 +242,12 @@ pub fn apply( established.is_hidden as i64, accepted, established.payment_channel_broken as i64, + // Written as handed. `WalletPersister::store` strips it on a + // backend that cannot keep it beside its account + // (`persists_outbound_account_markers`), and the manager's + // load clears any restored marker it cannot trust, so a + // direct caller of this store is covered on the way back. + established.external_account_reference.map(i64::from), ])?; } } @@ -302,7 +311,8 @@ pub(crate) fn load_state( let mut stmt = conn.prepare( "SELECT owner_id, contact_id, state, outgoing_request, incoming_request, \ - alias, note, is_hidden, accepted_accounts, payment_channel_broken \ + alias, note, is_hidden, accepted_accounts, payment_channel_broken, \ + external_account_reference \ FROM contacts WHERE wallet_id = ?1", )?; let mut rows = stmt.query(params![wallet_id.as_slice()])?; @@ -324,6 +334,7 @@ pub(crate) fn load_state( is_hidden: row.get::<_, Option>(7)?.unwrap_or(0) != 0, accepted_accounts: row.get(8)?, payment_channel_broken: row.get::<_, Option>(9)?.unwrap_or(0) != 0, + external_account_reference: row.get(10)?, }; if let Err(err) = bucket_contact_row(&mut state, columns) { ctx.tolerate(LoadSite::ContactRow, err)?; @@ -345,6 +356,7 @@ struct ContactRowColumns { is_hidden: bool, accepted_accounts: Option>, payment_channel_broken: bool, + external_account_reference: Option, } /// File one read row into the bucket its `state` names, decoding the blobs @@ -365,6 +377,7 @@ fn bucket_contact_row( is_hidden, accepted_accounts, payment_channel_broken, + external_account_reference, } = columns; match contact_state_from_label(&label)? { @@ -413,11 +426,13 @@ fn bucket_contact_row( // no column for it, so it restores empty and re-derives // on the next contact-info sweep. contact_account_label: None, - // Rotation self-heal marker; this backend has no column - // for it, so it restores `None` — which conservatively - // forces the next sweep to re-verify (tear down + rebuild) - // the external account once, then re-stamp it. - external_account_reference: None, + // Rotation self-heal marker. `None` (a row written before + // the column existed, or by a backend that could not keep + // it) conservatively forces the next sweep to re-verify + // (tear down + rebuild) the external account once. + external_account_reference: external_account_reference + .map(|value| i64_to_u32("contacts.external_account_reference", value)) + .transpose()?, }, ); } diff --git a/packages/rs-platform-wallet-storage/src/sqlite/schema/core_state.rs b/packages/rs-platform-wallet-storage/src/sqlite/schema/core_state.rs index 4e1f02f6b8a..4b2ae0741ad 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/schema/core_state.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/schema/core_state.rs @@ -211,6 +211,7 @@ pub fn apply( wallet_id, cs.last_processed_height, cs.synced_height, + cs.synced_height_is_rewind, cl_bytes, chainlock_height, )?; @@ -811,6 +812,7 @@ fn upsert_sync_state( wallet_id: &WalletId, last_processed: Option, synced: Option, + synced_is_rewind: bool, chain_lock_bytes: Option>, chainlock: Option, ) -> Result<(), WalletStorageError> { @@ -834,7 +836,13 @@ fn upsert_sync_state( (Some(a), Some(b)) => Some(a.max(b)), (a, b) => a.or(b), }; + // A rewind is the one exception: the DashPay backfill lowers the cursor + // so the scanner re-walks a range with new addresses watched, and the + // backfill record committed in this same transaction vouches for the + // lowered value. Keeping the max would leave the record beside a cursor + // that never came down, so the next launch would skip the re-walk. let sy = match (current.1, synced) { + (_, Some(b)) if synced_is_rewind => Some(b), (Some(a), Some(b)) => Some(a.max(b)), (a, b) => a.or(b), }; diff --git a/packages/rs-platform-wallet-storage/src/sqlite/schema/dashpay_backfill.rs b/packages/rs-platform-wallet-storage/src/sqlite/schema/dashpay_backfill.rs new file mode 100644 index 00000000000..eb688025922 --- /dev/null +++ b/packages/rs-platform-wallet-storage/src/sqlite/schema/dashpay_backfill.rs @@ -0,0 +1,152 @@ +//! DashPay coreHeight backfill record writer + reader. +//! +//! See [`DashPayBackfillRecord`] for what a record claims. It has no table of +//! its own: the extent (`floor`, `rewound_from`) sits on the wallet's +//! `core_sync_state` row beside the cursor it lowered, and each covered entry +//! is `dashpay_covered_from` on the `account_registrations` row of the +//! receiving account it names. Every write carries the whole record, so the +//! writer replaces rather than folds, and an empty record is an invalidation: +//! it clears whatever is on file. + +use dpp::prelude::Identifier; +use key_wallet::account::AccountType; +use rusqlite::{params, Connection, OptionalExtension, Transaction}; + +use platform_wallet::changeset::{DashPayBackfillCoveredContact, DashPayBackfillRecord}; +use platform_wallet::wallet::platform_wallet::WalletId; + +use crate::sqlite::error::WalletStorageError; +use crate::sqlite::schema::accounts::{ + account_dashpay_ids, account_index, account_key_class, account_type_db_label, +}; +use crate::sqlite::schema::{blob, id32}; +use crate::sqlite::util::safe_cast::i64_to_u32; + +/// Replace this wallet's backfill record with `record`; an empty record (no +/// covered receiving accounts) only clears the stored one. It is how upstream +/// retracts coverage it can no longer vouch for, and a record with nothing +/// covered claims nothing anyway. +/// +/// Runs after the round's account registrations are written. An entry whose +/// receiving account has no `account_registrations` row is dropped: a contact +/// missing from the record is only backfilled again, while one wrongly present +/// would never be. When every entry is dropped the extent is cleared too. +pub fn apply( + tx: &Transaction<'_>, + wallet_id: &WalletId, + record: &DashPayBackfillRecord, +) -> Result<(), WalletStorageError> { + tx.prepare_cached( + "UPDATE account_registrations SET dashpay_covered_from = NULL \ + WHERE wallet_id = ?1 AND dashpay_covered_from IS NOT NULL", + )? + .execute(params![wallet_id.as_slice()])?; + let mut mark = tx.prepare_cached( + "UPDATE account_registrations SET dashpay_covered_from = ?7 \ + WHERE wallet_id = ?1 AND account_type = ?2 AND account_index = ?3 \ + AND key_class = ?4 AND user_identity_id = ?5 AND friend_identity_id = ?6", + )?; + let mut marked = 0; + for entry in record.covered() { + let account = receiving_account(entry); + let (user_identity_id, friend_identity_id) = account_dashpay_ids(&account); + marked += mark.execute(params![ + wallet_id.as_slice(), + account_type_db_label(&account), + i64::from(account_index(&account)), + i64::from(account_key_class(&account)), + &user_identity_id[..], + &friend_identity_id[..], + i64::from(entry.covered_from), + ])?; + } + let extent = (marked > 0).then(|| (i64::from(record.floor), i64::from(record.rewound_from))); + tx.prepare_cached( + "INSERT INTO core_sync_state \ + (wallet_id, dashpay_backfill_floor, dashpay_backfill_rewound_from) \ + VALUES (?1, ?2, ?3) \ + ON CONFLICT(wallet_id) DO UPDATE SET \ + dashpay_backfill_floor = excluded.dashpay_backfill_floor, \ + dashpay_backfill_rewound_from = excluded.dashpay_backfill_rewound_from", + )? + .execute(params![ + wallet_id.as_slice(), + extent.map(|(floor, _)| floor), + extent.map(|(_, rewound_from)| rewound_from), + ])?; + Ok(()) +} + +/// This wallet's backfill record, or the empty record when none is on file. +/// +/// Absence reads as "no backfill recorded", which upstream answers by +/// re-running the backfill — slow, never lossy — so a wallet that predates +/// these columns is left on exactly the behaviour it had. Covered accounts +/// without an extent read the same way; the writer never leaves that state. +/// +/// # Errors +/// +/// A height or index that does not narrow to `u32`, or an identity id that is +/// not 32 bytes, is structural and fatal in every load policy; the columns' +/// CHECK constraints make either one a sign the file was written with them +/// bypassed. +pub fn load_for_wallet( + conn: &Connection, + wallet_id: &WalletId, +) -> Result { + let extent: Option<(Option, Option)> = conn + .prepare_cached( + "SELECT dashpay_backfill_floor, dashpay_backfill_rewound_from \ + FROM core_sync_state WHERE wallet_id = ?1", + )? + .query_row(params![wallet_id.as_slice()], |row| { + Ok((row.get(0)?, row.get(1)?)) + }) + .optional()?; + let Some((Some(floor), Some(rewound_from))) = extent else { + return Ok(DashPayBackfillRecord::default()); + }; + let floor = i64_to_u32("core_sync_state.dashpay_backfill_floor", floor)?; + let rewound_from = i64_to_u32( + "core_sync_state.dashpay_backfill_rewound_from", + rewound_from, + )?; + + let mut stmt = conn.prepare_cached( + "SELECT account_index, length(user_identity_id), user_identity_id, \ + length(friend_identity_id), friend_identity_id, dashpay_covered_from \ + FROM account_registrations \ + WHERE wallet_id = ?1 AND dashpay_covered_from IS NOT NULL", + )?; + let mut rows = stmt.query(params![wallet_id.as_slice()])?; + let mut covered = Vec::new(); + while let Some(row) = rows.next()? { + let account_index = i64_to_u32("account_registrations.account_index", row.get(0)?)?; + blob::check_fixed_width(row.get(1)?, 32, "account_registrations.user_identity_id")?; + let owner: Vec = row.get(2)?; + blob::check_fixed_width(row.get(3)?, 32, "account_registrations.friend_identity_id")?; + let contact: Vec = row.get(4)?; + let covered_from = i64_to_u32("account_registrations.dashpay_covered_from", row.get(5)?)?; + covered.push(DashPayBackfillCoveredContact { + owner: Identifier::from(id32("account_registrations.user_identity_id", &owner)?), + contact: Identifier::from(id32("account_registrations.friend_identity_id", &contact)?), + account_index, + covered_from, + }); + } + Ok(DashPayBackfillRecord::from_entries( + floor, + rewound_from, + covered, + )) +} + +/// The receiving account a covered entry names, so its row is addressed +/// through the same typed-column helpers the account writer binds. +fn receiving_account(entry: &DashPayBackfillCoveredContact) -> AccountType { + AccountType::DashpayReceivingFunds { + index: entry.account_index, + user_identity_id: entry.owner.to_buffer(), + friend_identity_id: entry.contact.to_buffer(), + } +} diff --git a/packages/rs-platform-wallet-storage/src/sqlite/schema/mod.rs b/packages/rs-platform-wallet-storage/src/sqlite/schema/mod.rs index 88bff87e377..c1e5f7dc7ed 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/schema/mod.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/schema/mod.rs @@ -13,6 +13,7 @@ pub mod contacts; pub mod core_pool; pub mod core_state; pub mod dashpay; +pub mod dashpay_backfill; pub mod dpns_name_states; pub mod identities; pub mod identity_keys; diff --git a/packages/rs-platform-wallet-storage/src/sqlite/schema/versions.rs b/packages/rs-platform-wallet-storage/src/sqlite/schema/versions.rs index de2b418cae1..9a14dceb2b6 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/schema/versions.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/schema/versions.rs @@ -38,6 +38,7 @@ pub enum Domain { Invitations, DpnsNameStates, IdentityScanState, + DashPayBackfill, #[cfg(feature = "shielded")] ShieldedViewingKeys, } @@ -63,6 +64,7 @@ impl Domain { Domain::Invitations => "invitations", Domain::DpnsNameStates => "dpns_name_states", Domain::IdentityScanState => "identity_scan_states", + Domain::DashPayBackfill => "dashpay_backfill", #[cfg(feature = "shielded")] Domain::ShieldedViewingKeys => "shielded_viewing_keys", } @@ -71,7 +73,7 @@ impl Domain { /// Every domain, for coverage tests. #[cfg(any(test, feature = "__test-helpers"))] #[cfg(feature = "shielded")] - pub const ALL: [Domain; 17] = [ + pub const ALL: [Domain; 18] = [ Domain::Core, Domain::Identities, Domain::IdentityKeys, @@ -88,12 +90,13 @@ impl Domain { Domain::Invitations, Domain::DpnsNameStates, Domain::IdentityScanState, + Domain::DashPayBackfill, Domain::ShieldedViewingKeys, ]; /// Every domain, for coverage tests without shielded persistence. #[cfg(all(any(test, feature = "__test-helpers"), not(feature = "shielded")))] - pub const ALL: [Domain; 16] = [ + pub const ALL: [Domain; 17] = [ Domain::Core, Domain::Identities, Domain::IdentityKeys, @@ -110,6 +113,7 @@ impl Domain { Domain::Invitations, Domain::DpnsNameStates, Domain::IdentityScanState, + Domain::DashPayBackfill, ]; } @@ -144,10 +148,7 @@ pub fn touched_domains(cs: &PlatformWalletChangeSet) -> Vec { invitations, dpns_name_states, identity_scan_state, - // Not persisted by this backend yet (see `ClientWalletStartState:: - // dashpay_backfill` on the load side): the record rides the FFI - // hosts' wallet row only, so it touches no domain here. - dashpay_backfill: _, + dashpay_backfill, #[cfg(feature = "shielded")] shielded, #[cfg(not(feature = "shielded"))] @@ -214,6 +215,11 @@ pub fn touched_domains(cs: &PlatformWalletChangeSet) -> Vec { if identity_scan_state.is_some() { out.push(Domain::IdentityScanState); } + // Every write is the whole record and an empty one clears the stored + // record, so presence alone is a change, the same as the scan verdict. + if dashpay_backfill.is_some() { + out.push(Domain::DashPayBackfill); + } #[cfg(feature = "shielded")] if shielded .as_ref() diff --git a/packages/rs-platform-wallet-storage/tests/integration/main.rs b/packages/rs-platform-wallet-storage/tests/integration/main.rs index 918d0c6bac2..c4c06c7c15f 100644 --- a/packages/rs-platform-wallet-storage/tests/integration/main.rs +++ b/packages/rs-platform-wallet-storage/tests/integration/main.rs @@ -32,6 +32,8 @@ mod sqlite_compile_time; mod sqlite_contacts_keys_rehydration; mod sqlite_core_pool_writer; mod sqlite_core_state_reader; +mod sqlite_dashpay_backfill; +mod sqlite_dashpay_backfill_relaunch; mod sqlite_dashpay_overlay_contract; mod sqlite_delete_buffer_reconcile; mod sqlite_delete_cross_process_exclusion; diff --git a/packages/rs-platform-wallet-storage/tests/integration/sqlite_contacts_keys_rehydration.rs b/packages/rs-platform-wallet-storage/tests/integration/sqlite_contacts_keys_rehydration.rs index 948d4a981c4..bf4a8b39122 100644 --- a/packages/rs-platform-wallet-storage/tests/integration/sqlite_contacts_keys_rehydration.rs +++ b/packages/rs-platform-wallet-storage/tests/integration/sqlite_contacts_keys_rehydration.rs @@ -238,6 +238,55 @@ fn tc3_established_contact_restores_onto_identity() { assert!(managed.dashpay().incoming_contact_requests().is_empty()); } +/// The outbound-account marker survives a reopen, so a contact whose +/// outbound account was already built for its current account reference is +/// not torn down and rebuilt on every cold start. +#[test] +fn tc3b_established_contact_restores_its_outbound_account_marker() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0xCB); + ensure_wallet_meta(&persister, &w); + let owner = Identifier::from([0x11; 32]); + let contact = Identifier::from([0x22; 32]); + let contact_state = EstablishedContact { + external_account_reference: Some(3), + ..established(owner, contact) + }; + let mut established_map = BTreeMap::new(); + established_map.insert( + SentContactRequestKey { + owner_id: owner, + recipient_id: contact, + }, + contact_state.clone(), + ); + persister + .store( + w, + PlatformWalletChangeSet { + identities: Some(id_changeset([wallet_identity_entry(owner, w, 0)])), + contacts: Some(ContactChangeSet { + established: established_map, + ..Default::default() + }), + ..Default::default() + }, + ) + .unwrap(); + drop(persister); + + let state = reopen(&path).load().expect("load"); + let managed = &state.wallets[&w].identity_manager.wallet_identities[&w][&0]; + assert_eq!( + managed + .dashpay() + .established_contacts() + .get(&contact) + .and_then(|c| c.external_account_reference), + Some(3) + ); +} + /// Pending sent and incoming requests restore with correct /// directionality and map keys. #[test] diff --git a/packages/rs-platform-wallet-storage/tests/integration/sqlite_dashpay_backfill.rs b/packages/rs-platform-wallet-storage/tests/integration/sqlite_dashpay_backfill.rs new file mode 100644 index 00000000000..b2558a043c6 --- /dev/null +++ b/packages/rs-platform-wallet-storage/tests/integration/sqlite_dashpay_backfill.rs @@ -0,0 +1,806 @@ +//! Cross-launch persistence of the DashPay coreHeight backfill record. +//! +//! dashpay/platform#4302: the guard that stops the DIP-15 §12.6 rescan from +//! re-lowering the cursor lived only in memory while the lowered cursor was +//! durable, so every launch restored a cursor inside the climb and rewound +//! again. These tests drive the public [`PlatformWalletPersistence`] surface +//! and assert on what `load()` hands back, plus the stored cursor the record +//! vouches for. The record has no table: its extent is two `core_sync_state` +//! columns and each covered entry is `dashpay_covered_from` on the receiving +//! account's `account_registrations` row. + +use crate::common; + +use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +use common::{ + ensure_wallet_meta, fresh_persister, fresh_persister_with_mode, store_and_flush, wid, +}; +use dpp::prelude::Identifier; +use key_wallet::account::AccountType; +use key_wallet::bip32::ExtendedPubKey; +use platform_wallet::changeset::{ + persists_backfill_coverage, AccountRegistrationEntry, ClientWalletStartState, CoreChangeSet, + DashPayBackfillCoveredContact, DashPayBackfillRecord, IdentityChangeSet, IdentityEntry, + PlatformWalletChangeSet, PlatformWalletPersistence, SentContactRequestKey, +}; +use platform_wallet::wallet::identity::{ContactRequest, IdentityStatus}; +use platform_wallet::wallet::platform_wallet::WalletId; +use platform_wallet_storage::sqlite::migrations as mig; +use platform_wallet_storage::sqlite::schema::{blob, contacts, dashpay_backfill}; +use platform_wallet_storage::{ + FlushMode, LoadCtx, SqlitePersister, SqlitePersisterConfig, WalletStorageError, +}; +use rusqlite::trace::{TraceEvent, TraceEventCodes}; +use rusqlite::{params, Connection, ErrorCode, OptionalExtension}; + +fn covered(owner: u8, contact: u8, account_index: u32, from: u32) -> DashPayBackfillCoveredContact { + DashPayBackfillCoveredContact { + owner: Identifier::from([owner; 32]), + contact: Identifier::from([contact; 32]), + account_index, + covered_from: from, + } +} + +fn record( + floor: u32, + rewound_from: u32, + entries: Vec, +) -> DashPayBackfillRecord { + DashPayBackfillRecord::from_entries(floor, rewound_from, entries) +} + +/// A distinct extended public key per `seed` byte. +fn xpub(seed: u8) -> ExtendedPubKey { + use key_wallet::wallet::initialization::WalletAccountCreationOptions; + use key_wallet::wallet::Wallet; + Wallet::from_seed_bytes( + [seed; 64], + key_wallet::Network::Testnet, + WalletAccountCreationOptions::Default, + ) + .expect("wallet") + .accounts + .all_accounts() + .first() + .expect("at least one account") + .account_xpub +} + +/// The receiving account a covered entry names, registered under `xpub_seed`. +fn receiving_account( + entry: &DashPayBackfillCoveredContact, + xpub_seed: u8, +) -> AccountRegistrationEntry { + AccountRegistrationEntry { + account_type: AccountType::DashpayReceivingFunds { + index: entry.account_index, + user_identity_id: entry.owner.to_buffer(), + friend_identity_id: entry.contact.to_buffer(), + }, + account_xpub: xpub(xpub_seed), + } +} + +/// The record plus the receiving accounts it covers, registered in the same +/// round as the wallet does before it can cover one. +fn record_cs(record: DashPayBackfillRecord) -> PlatformWalletChangeSet { + PlatformWalletChangeSet { + account_registrations: record + .covered() + .iter() + .map(|entry| receiving_account(entry, 0x01)) + .collect(), + dashpay_backfill: Some(record), + ..Default::default() + } +} + +fn cursor_cs(height: u32, rewind: bool) -> PlatformWalletChangeSet { + PlatformWalletChangeSet { + core: Some(CoreChangeSet { + synced_height: Some(height), + synced_height_is_rewind: rewind, + ..Default::default() + }), + ..Default::default() + } +} + +fn stage(persister: &SqlitePersister, wallet_id: WalletId, cs: PlatformWalletChangeSet) { + persister.store(wallet_id, cs).expect("store"); +} + +fn stored_cursor(persister: &SqlitePersister, wallet_id: &WalletId) -> Option { + persister + .lock_conn_for_test() + .query_row( + "SELECT synced_height FROM core_sync_state WHERE wallet_id = ?1", + params![wallet_id.as_slice()], + |row| row.get(0), + ) + .expect("core_sync_state row") +} + +/// What is on file for the record: its stored extent (`None` when unset or +/// when the wallet has no `core_sync_state` row) and how many account rows +/// carry a covered height. +fn stored_record_parts( + persister: &SqlitePersister, + wallet_id: &WalletId, +) -> (Option<(i64, i64)>, i64) { + let conn = persister.lock_conn_for_test(); + let extent: Option<(Option, Option)> = conn + .query_row( + "SELECT dashpay_backfill_floor, dashpay_backfill_rewound_from \ + FROM core_sync_state WHERE wallet_id = ?1", + params![wallet_id.as_slice()], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .optional() + .expect("read extent"); + let covered = conn + .query_row( + "SELECT COUNT(*) FROM account_registrations \ + WHERE wallet_id = ?1 AND dashpay_covered_from IS NOT NULL", + params![wallet_id.as_slice()], + |row| row.get(0), + ) + .expect("count covered accounts"); + let extent = match extent { + Some((Some(floor), Some(rewound_from))) => Some((floor, rewound_from)), + _ => None, + }; + (extent, covered) +} + +/// Reopen the database at `path` and return what `load()` restored for +/// `wallet_id`. The seeding persister must already be dropped — the +/// process-wide open-path registry refuses a second live persister on a path. +fn reload_wallet(path: &Path, wallet_id: &WalletId) -> ClientWalletStartState { + let persister = + SqlitePersister::open(SqlitePersisterConfig::new(path)).expect("reopen persister"); + let mut state = persister.load().expect("load"); + state + .wallets + .remove(wallet_id) + .expect("the seeded wallet must come back from load()") +} + +/// A transient SQLite failure, so a failed flush restores its buffer for retry. +fn busy_error() -> WalletStorageError { + WalletStorageError::Sqlite(rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: ErrorCode::DatabaseBusy, + extended_code: rusqlite::ffi::SQLITE_BUSY, + }, + Some("database is busy".into()), + )) +} + +fn identity_entry(id: Identifier, wallet_id: WalletId, index: u32) -> IdentityEntry { + IdentityEntry { + id, + balance: 1_000, + revision: 1, + identity_index: Some(index), + last_updated_balance_block_time: None, + last_synced_keys_block_time: None, + dpns_names: Vec::new(), + contested_dpns_names: Vec::new(), + status: IdentityStatus::Active, + wallet_id: Some(wallet_id), + dashpay_profile: None, + dashpay_payments: Default::default(), + contact_profiles: Default::default(), + ignored_senders: Default::default(), + } +} + +#[test] +fn should_hold_backfill_coverage_durably_only_when_store_commits_inline() { + let (immediate, _tmp, _path) = fresh_persister(); + assert!(immediate.stores_dashpay_backfill()); + assert!( + persists_backfill_coverage(&immediate), + "an Immediate-mode store is the commit, so coverage may be durable" + ); + + let (manual, _tmp, _path) = fresh_persister_with_mode(FlushMode::Manual); + assert!( + manual.stores_dashpay_backfill(), + "a Manual-mode store can still write (and so invalidate) a record" + ); + assert!( + !persists_backfill_coverage(&manual), + "a buffered round can still be dropped at flush, so coverage stays session-only" + ); +} + +/// The regression #4302 names: a backfill recorded in one process must still +/// be known in the next, so it resumes instead of rewinding again. +#[test] +fn should_restore_the_backfill_record_across_a_reopen() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0x51); + ensure_wallet_meta(&persister, &w); + let written = record( + 1_000, + 5_000, + vec![covered(0x01, 0x02, 0, 1_200), covered(0x01, 0x03, 1, 1_000)], + ); + store_and_flush(&persister, w, record_cs(written.clone())); + drop(persister); + + let restored = reload_wallet(&path, &w).dashpay_backfill; + + assert_eq!( + restored, written, + "the record must survive the reopen exactly" + ); + assert_eq!( + restored.covered_from( + &Identifier::from([0x01; 32]), + &Identifier::from([0x03; 32]), + 1 + ), + Some(1_000) + ); +} + +/// Absence reads as "no backfill recorded", which upstream answers by +/// re-running the backfill — the behaviour every wallet had before. +#[test] +fn should_restore_an_empty_record_for_a_wallet_that_never_backfilled() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0x52); + ensure_wallet_meta(&persister, &w); + drop(persister); + + assert!(reload_wallet(&path, &w).dashpay_backfill.is_empty()); +} + +/// Covered accounts with no extent beside them are not a record the writer +/// ever leaves; the reader takes them as "no record", which only costs a +/// re-run, rather than inventing an extent. +#[test] +fn should_read_coverage_without_an_extent_as_no_record() { + let mut conn = Connection::open_in_memory().expect("open in-memory db"); + mig::run(&mut conn).expect("migrate"); + let w = wid(0x5D); + conn.execute( + "INSERT INTO wallets (wallet_id, network, birth_height) VALUES (?1, 'testnet', 0)", + params![w.as_slice()], + ) + .expect("insert wallet"); + conn.execute( + "INSERT INTO account_registrations \ + (wallet_id, account_type, account_index, user_identity_id, friend_identity_id, \ + account_xpub_bytes, dashpay_covered_from) \ + VALUES (?1, 'dashpay_receiving', 0, ?2, ?3, x'00', 1000)", + params![ + w.as_slice(), + [0x01u8; 32].as_slice(), + [0x02u8; 32].as_slice() + ], + ) + .expect("insert covered account"); + + assert!(dashpay_backfill::load_for_wallet(&conn, &w) + .expect("backfill load") + .is_empty()); +} + +/// A covered entry is stored on its receiving account's row, so an entry whose +/// account has no row is dropped — re-armed, never lost the other way. When +/// nothing is left the extent goes too, and the record reads back empty. +#[test] +fn should_drop_coverage_for_a_receiving_account_with_no_row() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0x60); + ensure_wallet_meta(&persister, &w); + let registered = covered(0x01, 0x02, 0, 1_200); + let unregistered = covered(0x01, 0x09, 0, 1_000); + store_and_flush( + &persister, + w, + PlatformWalletChangeSet { + account_registrations: vec![receiving_account(®istered, 0x01)], + dashpay_backfill: Some(record(1_000, 5_000, vec![registered, unregistered])), + ..Default::default() + }, + ); + assert_eq!( + stored_record_parts(&persister, &w), + (Some((1_000, 5_000)), 1) + ); + + store_and_flush( + &persister, + w, + PlatformWalletChangeSet { + dashpay_backfill: Some(record(900, 5_000, vec![unregistered])), + ..Default::default() + }, + ); + assert_eq!(stored_record_parts(&persister, &w), (None, 0)); + drop(persister); + + assert!(reload_wallet(&path, &w).dashpay_backfill.is_empty()); +} + +/// Re-registering the receiving account upserts its row; the covered height +/// on that row must survive, or the next launch would backfill again. A receiving account's xpub derives from +/// the wallet's seed at `index'/owner/contact` — the row's key — so the +/// re-registration carries the same xpub. +#[test] +fn should_keep_coverage_when_its_receiving_account_is_re_registered() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0x61); + ensure_wallet_meta(&persister, &w); + let entry = covered(0x01, 0x02, 0, 1_200); + let written = record(1_000, 5_000, vec![entry]); + store_and_flush(&persister, w, record_cs(written.clone())); + store_and_flush( + &persister, + w, + PlatformWalletChangeSet { + account_registrations: vec![receiving_account(&entry, 0x01)], + ..Default::default() + }, + ); + drop(persister); + + assert_eq!(reload_wallet(&path, &w).dashpay_backfill, written); +} + +/// Every write carries the whole record: an account missing from the new one +/// must not survive from the old one, or it would never be backfilled again. +#[test] +fn should_replace_the_stored_record_wholesale() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0x53); + ensure_wallet_meta(&persister, &w); + store_and_flush( + &persister, + w, + record_cs(record( + 1_000, + 5_000, + vec![covered(0x01, 0x02, 0, 1_200), covered(0x01, 0x03, 0, 1_100)], + )), + ); + let replacement = record(900, 6_000, vec![covered(0x01, 0x02, 0, 900)]); + store_and_flush(&persister, w, record_cs(replacement.clone())); + drop(persister); + + assert_eq!(reload_wallet(&path, &w).dashpay_backfill, replacement); +} + +/// An empty record is how upstream retracts coverage it can no longer vouch +/// for; it has to clear the stored record. +#[test] +fn should_clear_the_stored_record_when_an_empty_one_is_written() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0x54); + ensure_wallet_meta(&persister, &w); + store_and_flush( + &persister, + w, + record_cs(record(1_000, 5_000, vec![covered(0x01, 0x02, 0, 1_000)])), + ); + store_and_flush(&persister, w, record_cs(DashPayBackfillRecord::default())); + assert_eq!(stored_record_parts(&persister, &w), (None, 0)); + drop(persister); + + assert!(reload_wallet(&path, &w).dashpay_backfill.is_empty()); +} + +/// The rewind the record vouches for must actually lower the stored cursor; +/// an ordinary advance below the stored cursor must still lose to the max. +#[test] +fn should_lower_the_stored_cursor_only_on_a_rewind() { + let (persister, _tmp, _path) = fresh_persister(); + let w = wid(0x55); + ensure_wallet_meta(&persister, &w); + + store_and_flush(&persister, w, cursor_cs(5_000, false)); + store_and_flush(&persister, w, cursor_cs(4_000, false)); + assert_eq!( + stored_cursor(&persister, &w), + Some(5_000), + "an advance keeps the monotonic max" + ); + + store_and_flush(&persister, w, cursor_cs(1_000, true)); + assert_eq!( + stored_cursor(&persister, &w), + Some(1_000), + "a rewind overwrites the stored cursor" + ); + + store_and_flush(&persister, w, cursor_cs(1_500, false)); + assert_eq!( + stored_cursor(&persister, &w), + Some(1_500), + "the climb after a rewind advances from the lowered cursor" + ); +} + +/// The round the rescan writes — the lowered cursor plus the record that +/// vouches for it — lands as one unit, and a relaunch reads both back. +#[test] +fn should_restore_the_rewound_cursor_beside_its_record() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0x56); + ensure_wallet_meta(&persister, &w); + store_and_flush(&persister, w, cursor_cs(5_000, false)); + let written = record(1_000, 5_000, vec![covered(0x01, 0x02, 0, 1_000)]); + store_and_flush( + &persister, + w, + PlatformWalletChangeSet { + core: cursor_cs(1_000, true).core, + ..record_cs(written.clone()) + }, + ); + drop(persister); + + let wallet = reload_wallet(&path, &w); + assert_eq!(wallet.dashpay_backfill, written); + assert_eq!( + wallet.wallet_info.metadata.synced_height, 1_000, + "a relaunch must resume the climb from the lowered cursor" + ); +} + +/// The concurrent write [`commit_rewind_mid_load`] makes: where it goes, and +/// what it reported once it ran. A trace callback is a plain `fn`, so its +/// state is a static; only this file's snapshot test arms it. +static MID_LOAD_WRITE: Mutex = Mutex::new(MidLoadWrite::Disarmed); + +enum MidLoadWrite { + Disarmed, + Armed(PathBuf, WalletId), + Done(Result<(), String>), +} + +/// Trace callback on the loading connection. On the first `asset_locks` +/// read — after `load()` has read the cursor, before it reads the coverage — +/// another connection commits a rewind to 1 000 and the coverage that +/// vouches for it, as a second process would. Errors are reported through +/// the static, since rusqlite swallows a panic raised inside the callback. +fn commit_rewind_mid_load(event: TraceEvent<'_>) { + let TraceEvent::Stmt(_, sql) = event else { + return; + }; + if !sql.contains("FROM asset_locks") { + return; + } + let mut state = MID_LOAD_WRITE.lock().unwrap_or_else(|p| p.into_inner()); + let MidLoadWrite::Armed(path, wallet_id) = &*state else { + return; + }; + let result = Connection::open(path) + .and_then(|mut other| { + let tx = other.transaction()?; + tx.execute( + "UPDATE core_sync_state SET synced_height = 1000, \ + dashpay_backfill_floor = 1000, dashpay_backfill_rewound_from = 5000 \ + WHERE wallet_id = ?1", + params![wallet_id.as_slice()], + )?; + tx.execute( + "UPDATE account_registrations SET dashpay_covered_from = 1000 \ + WHERE wallet_id = ?1 AND account_type = 'dashpay_receiving'", + params![wallet_id.as_slice()], + )?; + tx.commit() + }) + .map_err(|e| e.to_string()); + *state = MidLoadWrite::Done(result); +} + +/// `load()` reads the cursor and the coverage that suppresses the rescan +/// below it from one snapshot. Another process committing a rewind plus its +/// coverage between those two reads must not hand back the old cursor beside +/// the new coverage: the contact would read as covered, the rescan would be +/// skipped, and the range below the old cursor never checked. +#[test] +fn should_load_the_cursor_and_its_coverage_from_one_snapshot() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0x5D); + ensure_wallet_meta(&persister, &w); + let entry = covered(0x01, 0x02, 0, 1_000); + store_and_flush( + &persister, + w, + PlatformWalletChangeSet { + account_registrations: vec![receiving_account(&entry, 0x01)], + core: cursor_cs(5_000, false).core, + ..Default::default() + }, + ); + + *MID_LOAD_WRITE.lock().unwrap() = MidLoadWrite::Armed(path.clone(), w); + persister.lock_conn_for_test().trace_v2( + TraceEventCodes::SQLITE_TRACE_STMT, + Some(commit_rewind_mid_load), + ); + let mut state = persister.load().expect("load"); + persister + .lock_conn_for_test() + .trace_v2(TraceEventCodes::empty(), None); + + match std::mem::replace(&mut *MID_LOAD_WRITE.lock().unwrap(), MidLoadWrite::Disarmed) { + MidLoadWrite::Done(result) => result.expect("the concurrent rewind must commit"), + _ => panic!("load() must have read asset_locks, where the rewind is injected"), + } + assert_eq!( + stored_cursor(&persister, &w), + Some(1_000), + "the concurrent rewind is on disk" + ); + let wallet = state.wallets.remove(&w).expect("the seeded wallet"); + assert_eq!( + ( + wallet.wallet_info.metadata.synced_height, + wallet.dashpay_backfill + ), + (5_000, DashPayBackfillRecord::default()), + "load() must see the state from before the rewind, cursor and coverage alike" + ); +} + +/// A buffered record is last-write-wins like every other record write, so an +/// invalidation staged after a record leaves nothing on disk at flush. +#[test] +fn should_flush_only_the_last_buffered_record_in_manual_mode() { + let (persister, _tmp, path) = fresh_persister_with_mode(FlushMode::Manual); + let w = wid(0x57); + ensure_wallet_meta(&persister, &w); + stage( + &persister, + w, + record_cs(record(1_000, 5_000, vec![covered(0x01, 0x02, 0, 1_000)])), + ); + stage(&persister, w, record_cs(DashPayBackfillRecord::default())); + persister.flush(w).expect("flush"); + drop(persister); + + assert!(reload_wallet(&path, &w).dashpay_backfill.is_empty()); +} + +/// The record is wallet-scoped state, so deleting the wallet takes it with it. +#[test] +fn should_drop_the_backfill_record_when_its_wallet_is_deleted() { + let (persister, _tmp, _path) = fresh_persister(); + let w = wid(0x58); + ensure_wallet_meta(&persister, &w); + store_and_flush( + &persister, + w, + record_cs(record(1_000, 5_000, vec![covered(0x01, 0x02, 0, 1_000)])), + ); + + persister.delete_wallet(w).expect("delete_wallet"); + + assert_eq!(stored_record_parts(&persister, &w), (None, 0)); +} + +/// A buffered rewind has to survive the advances merged around it: the +/// advance before it is replaced, the climb after it starts from it. +#[test] +fn should_flush_a_buffered_rewind_merged_between_advances() { + let (persister, _tmp, _path) = fresh_persister_with_mode(FlushMode::Manual); + let w = wid(0x59); + ensure_wallet_meta(&persister, &w); + stage(&persister, w, cursor_cs(5_000, false)); + persister.flush(w).expect("seed flush"); + + stage(&persister, w, cursor_cs(6_000, false)); + stage(&persister, w, cursor_cs(1_000, true)); + stage(&persister, w, cursor_cs(1_500, false)); + persister.flush(w).expect("flush"); + + assert_eq!(stored_cursor(&persister, &w), Some(1_500)); +} + +/// A failed flush puts its changeset back in front of whatever arrived +/// meanwhile, so the rewind keeps its place on both sides of the restore. +#[test] +fn should_keep_a_rewind_in_order_across_a_failed_flush() { + for (restored, arrived, expected) in [ + // The restored round is the rewind; the climb lands after it. + ((1_000, true), (1_200, false), 1_200), + // The restored round is an advance; the rewind replaces it. + ((6_000, false), (1_000, true), 1_000), + ] { + let (persister, _tmp, _path) = fresh_persister_with_mode(FlushMode::Manual); + let w = wid(0x5A); + ensure_wallet_meta(&persister, &w); + stage(&persister, w, cursor_cs(5_000, false)); + persister.flush(w).expect("seed flush"); + + stage(&persister, w, cursor_cs(restored.0, restored.1)); + persister.force_next_flush_to_fail(busy_error()); + persister.flush(w).expect_err("injected transient failure"); + stage(&persister, w, cursor_cs(arrived.0, arrived.1)); + persister.flush(w).expect("retry"); + + assert_eq!( + stored_cursor(&persister, &w), + Some(expected), + "restored {restored:?}, then {arrived:?}" + ); + } +} + +/// An identity removed and re-added inside one buffer splits it into +/// segments that apply in order: a rewind staged in the earlier segment must +/// land before the advance staged in the later one. +#[test] +fn should_apply_a_rewind_and_its_climb_in_segment_order() { + let (persister, _tmp, _path) = fresh_persister_with_mode(FlushMode::Manual); + let w = wid(0x5B); + ensure_wallet_meta(&persister, &w); + let owner = Identifier::from([0x61; 32]); + let mut upsert = IdentityChangeSet::default(); + upsert.identities.insert(owner, identity_entry(owner, w, 0)); + let mut removal = IdentityChangeSet::default(); + removal.removed.insert(owner); + + stage( + &persister, + w, + PlatformWalletChangeSet { + identities: Some(upsert.clone()), + ..cursor_cs(5_000, false) + }, + ); + persister.flush(w).expect("seed flush"); + + stage( + &persister, + w, + PlatformWalletChangeSet { + identities: Some(removal), + ..cursor_cs(1_000, true) + }, + ); + stage( + &persister, + w, + PlatformWalletChangeSet { + identities: Some(upsert), + ..cursor_cs(1_200, false) + }, + ); + persister.flush(w).expect("flush"); + + assert_eq!(stored_cursor(&persister, &w), Some(1_200)); +} + +fn contact_request(sender: Identifier, recipient: Identifier) -> ContactRequest { + ContactRequest { + sender_id: sender, + recipient_id: recipient, + sender_key_index: 1, + recipient_key_index: 2, + account_reference: 3, + encrypted_account_label: None, + encrypted_public_key: vec![9, 9, 9], + auto_accept_proof: None, + core_height_created_at: 42, + created_at: 7, + } +} + +/// An upgraded database holds facts no earlier schema could: every +/// established contact reads back with no outbound marker (one rebuild) and +/// every wallet with no backfill recorded (one re-run). +#[test] +fn should_upgrade_a_pre_v020_database_to_no_marker_and_no_backfill() { + let mut conn = Connection::open_in_memory().expect("open in-memory db"); + conn.pragma_update(None, "foreign_keys", true) + .expect("enable foreign keys"); + mig::runner() + .set_target(refinery::Target::Version(19)) + .run(&mut conn) + .expect("migrate to the last pre-V020 schema"); + + let w = wid(0x5C); + let owner = Identifier::from([0x71; 32]); + let contact = Identifier::from([0x72; 32]); + conn.execute( + "INSERT INTO wallets (wallet_id, network, birth_height) VALUES (?1, 'testnet', 0)", + params![w.as_slice()], + ) + .expect("insert wallet"); + conn.execute( + "INSERT INTO contacts \ + (wallet_id, owner_id, contact_id, state, outgoing_request, incoming_request) \ + VALUES (?1, ?2, ?3, 'established', ?4, ?5)", + params![ + w.as_slice(), + owner.as_slice(), + contact.as_slice(), + blob::encode(&contact_request(owner, contact)).expect("encode outgoing"), + blob::encode(&contact_request(contact, owner)).expect("encode incoming"), + ], + ) + .expect("insert established contact"); + conn.execute( + "INSERT INTO core_sync_state (wallet_id, synced_height) VALUES (?1, 5000)", + params![w.as_slice()], + ) + .expect("insert sync state"); + conn.execute( + "INSERT INTO account_registrations \ + (wallet_id, account_type, account_index, user_identity_id, friend_identity_id, \ + account_xpub_bytes) \ + VALUES (?1, 'dashpay_receiving', 0, ?2, ?3, x'00')", + params![w.as_slice(), owner.as_slice(), contact.as_slice()], + ) + .expect("insert receiving account"); + + mig::run(&mut conn).expect("migrate to the newest version"); + + let state = + contacts::load_state_for_test(&conn, &w, &LoadCtx::strict()).expect("contacts load"); + let restored = state + .established + .get(&SentContactRequestKey { + owner_id: owner, + recipient_id: contact, + }) + .expect("the established contact survives the upgrade"); + assert_eq!(restored.external_account_reference, None); + assert!(dashpay_backfill::load_for_wallet(&conn, &w) + .expect("backfill load") + .is_empty()); +} + +/// A backfill rewind can lower the cursor below the wallet's birth: the +/// rescan hands the contact's checkpoint to `update_synced_height` unclamped. +/// The relaunch must restore that cursor, not the `birth_height - 1` seed, +/// or the range between them is skipped for a contact the record calls +/// covered. +#[test] +fn should_restore_a_rewound_cursor_below_the_wallet_birth() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0x5E); + persister + .lock_conn_for_test() + .execute( + "INSERT INTO wallets (wallet_id, network, birth_height) VALUES (?1, 'testnet', 501)", + params![w.as_slice()], + ) + .expect("insert wallet born at 501"); + store_and_flush(&persister, w, cursor_cs(1_000, false)); + store_and_flush( + &persister, + w, + PlatformWalletChangeSet { + core: cursor_cs(100, true).core, + ..record_cs(record(100, 1_000, vec![covered(0x01, 0x02, 0, 100)])) + }, + ); + drop(persister); + + let wallet = reload_wallet(&path, &w); + assert_eq!( + wallet.wallet_info.metadata.synced_height, 100, + "the stored rewind must win over the birth seed" + ); + assert_eq!( + wallet.dashpay_backfill.covered_from( + &Identifier::from([0x01; 32]), + &Identifier::from([0x02; 32]), + 0 + ), + Some(100) + ); +} diff --git a/packages/rs-platform-wallet-storage/tests/integration/sqlite_dashpay_backfill_relaunch.rs b/packages/rs-platform-wallet-storage/tests/integration/sqlite_dashpay_backfill_relaunch.rs new file mode 100644 index 00000000000..9bef24dadd4 --- /dev/null +++ b/packages/rs-platform-wallet-storage/tests/integration/sqlite_dashpay_backfill_relaunch.rs @@ -0,0 +1,808 @@ +//! End-to-end relaunch of the DashPay coreHeight backfill over SQLite. +//! +//! `sqlite_dashpay_backfill.rs` pins the persister boundary: what `store()` +//! writes and `load()` hands back. These tests run the real +//! [`PlatformWalletManager`] over a real [`SqlitePersister`] in +//! [`FlushMode::Immediate`] — its own wallet-event adapter commits every +//! cursor advance, the reconcile stores its record, `load_from_persistor()` +//! rebuilds the wallet — and close and reopen the database file between +//! sessions, so what is asserted is what a relaunching process sees +//! (dashpay/platform#4302). +//! +//! The established contacts are applied in memory and stored as a +//! `ContactChangeSet` round — what the contact-request sync does once both +//! requests are fetched, which is the only step here that would otherwise +//! need the network. Every account, identity and cursor row comes from the +//! manager's own writers. + +use crate::common; + +use std::path::Path; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use common::secure_tempdir; +use dashcore::Network; +use dpp::identity::{Identity, IdentityV0}; +use dpp::prelude::Identifier; +use key_wallet::account::account_collection::DashpayAccountKey; +use key_wallet::managed_account::managed_account_trait::ManagedAccountTrait; +use key_wallet::mnemonic::Mnemonic; +use key_wallet::wallet::initialization::WalletAccountCreationOptions; +use key_wallet::wallet::managed_wallet_info::wallet_info_interface::WalletInfoInterface; +use platform_wallet::changeset::{ + ContactChangeSet, PlatformWalletChangeSet, PlatformWalletPersistence, SentContactRequestKey, +}; +use platform_wallet::events::EventHandler; +use platform_wallet::key_wallet_manager::WalletInterface; +use platform_wallet::test_support::{ + encrypt_contact_xpub_for_test, test_platform_wallet_manager_with, +}; +use platform_wallet::wallet::identity::ExternalAccountRegistration; +use platform_wallet::wallet::platform_wallet::WalletId; +use platform_wallet::{ + derive_contact_xpub, ContactRequest, EstablishedContact, PlatformEventHandler, PlatformWallet, + PlatformWalletManager, +}; +use platform_wallet_storage::{FlushMode, LoadPolicy, SqlitePersister, SqlitePersisterConfig}; +use rusqlite::{params, Connection}; + +/// The mnemonic `test_platform_wallet_manager_with` builds its wallet from. +const TEST_MNEMONIC: &str = "abandon abandon abandon abandon abandon abandon \ + abandon abandon abandon abandon abandon about"; + +const OWNER: [u8; 32] = [0xAA; 32]; +const CONTACT: [u8; 32] = [0xBB; 32]; + +/// The contact's request height, below the cursor: a backfill is owed. +const REQUEST_HEIGHT: u32 = 100; +/// The cursor when the contact is established. +const TIP: u32 = 1_000; +/// Where the rescan has climbed to when the process dies. +const MID_CLIMB: u32 = 400; + +struct NoopHandler; +impl EventHandler for NoopHandler {} +impl PlatformEventHandler for NoopHandler {} + +fn open(path: &Path) -> Arc { + Arc::new( + SqlitePersister::open( + SqlitePersisterConfig::new(path).with_flush_mode(FlushMode::Immediate), + ) + .expect("open persister"), + ) +} + +/// A fresh process: a new manager over `persister`, restored from it. +async fn relaunch(persister: &Arc) -> Arc> { + let sdk = Arc::new(dash_sdk::SdkBuilder::new_mock().build().expect("mock sdk")); + let handler: Arc = Arc::new(NoopHandler); + let manager = Arc::new(PlatformWalletManager::new( + sdk, + Arc::clone(persister), + handler, + )); + manager + .load_from_persistor() + .await + .expect("the SQLite file must load into the manager"); + manager +} + +/// End a session: stop the manager's workers, unregister the wallet and +/// release every handle on the database, so the next session can open it +/// (the process-wide open-path registry refuses a second live persister on +/// one file). +/// +/// The wallet is removed from the manager (memory only; nothing is deleted +/// from the file) before the drop: a registered `PlatformWallet` is reachable +/// from itself through its SPV broadcaster -> `SpvRuntime` -> event manager -> +/// balance/spend handlers -> the shared `wallets` map, so dropping the +/// manager with the wallet still registered never frees the wallet, nor the +/// persister handles it carries. +async fn close( + manager: Arc>, + wallet: Arc, + persister: Arc, +) { + let report = manager.shutdown().await; + assert!( + report.all_clean(), + "shutdown must join every worker: {report:?}" + ); + let removed = manager + .remove_wallet(&wallet.wallet_id()) + .await + .expect("unregister the wallet"); + drop(removed); + drop(wallet); + drop(manager); + // Give any adapter task still unwinding its claim a chance to drop it. + let deadline = Instant::now() + Duration::from_secs(10); + while Arc::strong_count(&persister) > 1 && Instant::now() < deadline { + tokio::task::yield_now().await; + } + assert_eq!( + Arc::strong_count(&persister), + 1, + "every manager-side handle on the persister must be gone at close" + ); + drop(persister); +} + +fn seed_wallet() -> key_wallet::wallet::Wallet { + let seed = Mnemonic::from_phrase(TEST_MNEMONIC) + .expect("mnemonic") + .to_seed(""); + key_wallet::wallet::Wallet::from_seed_bytes( + seed, + Network::Testnet, + WalletAccountCreationOptions::None, + ) + .expect("seed wallet") +} + +fn bare_identity(id: Identifier) -> Identity { + Identity::V0(IdentityV0 { + id, + public_keys: Default::default(), + balance: 0, + revision: 0, + }) +} + +fn stored_cursor(persister: &SqlitePersister, wallet_id: &WalletId) -> Option { + persister + .lock_conn_for_test() + .query_row( + "SELECT synced_height FROM core_sync_state WHERE wallet_id = ?1", + params![wallet_id.as_slice()], + |row| row.get::<_, Option>(0), + ) + .ok() + .flatten() +} + +/// Receiving accounts the stored record covers, counted only when its extent +/// is on file beside them — what the loader needs to restore a record. +fn stored_covered_accounts(persister: &SqlitePersister, wallet_id: &WalletId) -> i64 { + persister + .lock_conn_for_test() + .query_row( + "SELECT COUNT(*) FROM account_registrations AS a \ + JOIN core_sync_state AS s ON s.wallet_id = a.wallet_id \ + WHERE a.wallet_id = ?1 AND a.dashpay_covered_from IS NOT NULL \ + AND s.dashpay_backfill_floor IS NOT NULL", + params![wallet_id.as_slice()], + |row| row.get(0), + ) + .expect("count covered receiving accounts") +} + +/// Advance the scan the way the SPV engine does — the wallet manager's +/// `update_wallet_synced_height`, which emits `SyncHeightAdvanced` to the +/// manager's own adapter — and wait until the adapter has committed it. +async fn advance_and_commit(wallet: &PlatformWallet, persister: &SqlitePersister, height: u32) { + let wallet_id = wallet.wallet_id(); + wallet + .wallet_manager() + .write() + .await + .update_wallet_synced_height(&wallet_id, height); + let deadline = Instant::now() + Duration::from_secs(10); + while stored_cursor(persister, &wallet_id) != Some(height) { + assert!( + Instant::now() < deadline, + "the adapter never committed the advance to {height} (stored: {:?})", + stored_cursor(persister, &wallet_id) + ); + tokio::task::yield_now().await; + } +} + +/// Add the owner identity through the manager's writer, then establish the +/// `(owner, contact)` relationship with both requests at `height`: in memory, +/// and on disk as the contact sync's round would store it. +async fn establish_contact( + wallet: &PlatformWallet, + persister: &SqlitePersister, + owner: Identifier, + contact: Identifier, + height: u32, + incoming_account_reference: u32, + incoming_encrypted_xpub: Vec, +) -> EstablishedContact { + let wallet_id = wallet.wallet_id(); + let established = EstablishedContact::new( + contact, + ContactRequest::new(owner, contact, 0, 0, 0, vec![0u8; 96], height, 0), + ContactRequest::new( + contact, + owner, + 0, + 0, + incoming_account_reference, + incoming_encrypted_xpub, + height, + 0, + ), + ); + { + let mut state = wallet.state_mut().await; + if state.identity_manager.managed_identity(&owner).is_none() { + state + .identity_manager + .add_identity(bare_identity(owner), 0, wallet_id, wallet.persister()) + .expect("add the owner identity"); + } + state + .identity_manager + .managed_identity_mut(&owner) + .expect("owner managed") + .apply_established_contact(established.clone()); + } + let mut contacts = ContactChangeSet::default(); + contacts.established.insert( + SentContactRequestKey { + owner_id: owner, + recipient_id: contact, + }, + established.clone(), + ); + persister + .store( + wallet_id, + PlatformWalletChangeSet { + contacts: Some(contacts), + ..Default::default() + }, + ) + .expect("store the established contact"); + established +} + +async fn synced_height(wallet: &PlatformWallet) -> u32 { + wallet.state().await.core_wallet.synced_height() +} + +/// dashpay/platform#4302 on the SQLite backend, end to end. The rewind's +/// record and the lowered cursor reach the file; the scan climbs part-way; +/// the process dies. The next process restores the cursor inside the climb +/// and must resume from it rather than rewind again. Then the control: the +/// same file with its backfill record cleared rewinds again, so it is +/// the persisted record — not anything else the file restores — that stops +/// the second rewind. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn should_resume_the_backfill_from_the_persisted_cursor_after_a_sqlite_relaunch() { + let tmp = secure_tempdir().expect("tempdir"); + let path = tmp.path().join("wallet.db"); + let owner = Identifier::from(OWNER); + let contact = Identifier::from(CONTACT); + + // ── Session one: establish, rewind, climb part-way, die. ── + let persister = open(&path); + let (manager, wallet_id) = test_platform_wallet_manager_with(Arc::clone(&persister)).await; + let wallet = manager.get_wallet(&wallet_id).await.expect("wallet"); + let receiving = derive_contact_xpub(&seed_wallet(), Network::Testnet, 0, &owner, &contact) + .expect("derive receiving xpub") + .xpub; + wallet + .identity() + .dashpay() + .register_contact_account(&owner, &contact, 0, receiving) + .await + .expect("register the receival account"); + establish_contact( + &wallet, + &persister, + owner, + contact, + REQUEST_HEIGHT, + 0, + vec![0u8; 96], + ) + .await; + // The rewind path serves accounts on file before the record existed; the + // record must not already know this one. + wallet + .state_mut() + .await + .dashpay_backfill + .retain(|o, c, _| (o, c) != (&owner, &contact)); + advance_and_commit(&wallet, &persister, TIP).await; + + assert_eq!( + wallet + .identity() + .dashpay() + .reconcile_dashpay_rescan() + .await + .expect("first pass"), + Some(REQUEST_HEIGHT), + "a contact funded below the cursor rewinds it to the request height" + ); + assert_eq!(synced_height(&wallet).await, REQUEST_HEIGHT); + assert_eq!( + stored_cursor(&persister, &wallet_id), + Some(REQUEST_HEIGHT), + "the rewind must overwrite the stored cursor (synced_height_is_rewind)" + ); + assert_eq!(stored_covered_accounts(&persister, &wallet_id), 1); + { + let mut state = persister.load().expect("load"); + let record = state + .wallets + .remove(&wallet_id) + .expect("wallet on file") + .dashpay_backfill; + assert_eq!((record.floor, record.rewound_from), (REQUEST_HEIGHT, TIP)); + assert_eq!( + record.covered_from(&owner, &contact, 0), + Some(REQUEST_HEIGHT) + ); + } + + advance_and_commit(&wallet, &persister, MID_CLIMB).await; + close(manager, wallet, persister).await; + + // ── Session two: the next launch resumes instead of rewinding. ── + let persister = open(&path); + let manager = relaunch(&persister).await; + let wallet = manager + .get_wallet(&wallet_id) + .await + .expect("restored wallet"); + assert_eq!( + synced_height(&wallet).await, + MID_CLIMB, + "the cursor comes back inside the climb" + ); + assert!( + wallet + .state() + .await + .core_wallet + .accounts + .dashpay_receival_accounts + .contains_key(&DashpayAccountKey { + index: 0, + user_identity_id: OWNER, + friend_identity_id: CONTACT, + }), + "the receival account the record covers must be restored" + ); + assert_eq!( + wallet + .identity() + .dashpay() + .reconcile_dashpay_rescan() + .await + .expect("pass after relaunch"), + None, + "a contact the persisted record covers must not be rewound for again" + ); + assert_eq!( + synced_height(&wallet).await, + MID_CLIMB, + "the scan resumes from the restored cursor" + ); + assert_eq!(stored_cursor(&persister, &wallet_id), Some(MID_CLIMB)); + assert!(wallet.state().await.dashpay_backfill.is_pending(MID_CLIMB)); + close(manager, wallet, persister).await; + + // ── Control: the same file without its record rewinds again. ── + { + let conn = Connection::open(&path).expect("open the closed file"); + let cleared = conn + .execute( + "UPDATE account_registrations SET dashpay_covered_from = NULL \ + WHERE wallet_id = ?1 AND dashpay_covered_from IS NOT NULL", + params![wallet_id.as_slice()], + ) + .expect("clear the covered accounts"); + assert_eq!( + cleared, 1, + "the record's covered account must have been on file" + ); + conn.execute( + "UPDATE core_sync_state \ + SET dashpay_backfill_floor = NULL, dashpay_backfill_rewound_from = NULL \ + WHERE wallet_id = ?1", + params![wallet_id.as_slice()], + ) + .expect("clear the record's extent"); + } + let persister = open(&path); + let manager = relaunch(&persister).await; + let wallet = manager + .get_wallet(&wallet_id) + .await + .expect("restored wallet"); + assert_eq!(synced_height(&wallet).await, MID_CLIMB); + assert_eq!( + wallet + .identity() + .dashpay() + .reconcile_dashpay_rescan() + .await + .expect("pass without a record"), + Some(REQUEST_HEIGHT), + "without the persisted record the relaunch rewinds again — the #4302 defect" + ); + close(manager, wallet, persister).await; +} + +/// Session one of the recovery tests: establish a contact below the cursor +/// and let the rescan rewind for it, so its coverage is on file, then close. +async fn seed_rewound_coverage(path: &Path) -> WalletId { + let owner = Identifier::from(OWNER); + let contact = Identifier::from(CONTACT); + let persister = open(path); + let (manager, wallet_id) = test_platform_wallet_manager_with(Arc::clone(&persister)).await; + let wallet = manager.get_wallet(&wallet_id).await.expect("wallet"); + let receiving = derive_contact_xpub(&seed_wallet(), Network::Testnet, 0, &owner, &contact) + .expect("derive receiving xpub") + .xpub; + wallet + .identity() + .dashpay() + .register_contact_account(&owner, &contact, 0, receiving) + .await + .expect("register the receival account"); + establish_contact( + &wallet, + &persister, + owner, + contact, + REQUEST_HEIGHT, + 0, + vec![0u8; 96], + ) + .await; + wallet + .state_mut() + .await + .dashpay_backfill + .retain(|o, c, _| (o, c) != (&owner, &contact)); + advance_and_commit(&wallet, &persister, TIP).await; + assert_eq!( + wallet + .identity() + .dashpay() + .reconcile_dashpay_rescan() + .await + .expect("rewind"), + Some(REQUEST_HEIGHT) + ); + assert_eq!(stored_covered_accounts(&persister, &wallet_id), 1); + close(manager, wallet, persister).await; + wallet_id +} + +/// Reopen `path` read-only, as a user rescuing a damaged database does. +fn open_recovery(path: &Path, mode: FlushMode) -> Arc { + Arc::new( + SqlitePersister::open( + SqlitePersisterConfig::new(path) + .with_flush_mode(mode) + .with_load_policy(LoadPolicy::Recovery), + ) + .expect("open in recovery"), + ) +} + +/// Whether the published wallet carries coverage for the seeded contact. +async fn carries_coverage(wallet: &PlatformWallet) -> bool { + wallet + .state() + .await + .dashpay_backfill + .covered_from(&Identifier::from(OWNER), &Identifier::from(CONTACT), 0) + .is_some() +} + +/// A recovery reopen publishes its degraded projection even when it skips a +/// receiving account the stored record covers. The manager drops coverage for +/// accounts the wallet does not hold and would write the smaller record back +/// before publishing; a recovery persister refuses every write, so the drop +/// stays in memory and `load_from_persistor` serves the read-only wallet. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn should_publish_a_recovery_load_that_skips_a_covered_receiving_account() { + let tmp = secure_tempdir().expect("tempdir"); + let path = tmp.path().join("wallet.db"); + let wallet_id = seed_rewound_coverage(&path).await; + { + let conn = Connection::open(&path).expect("open the closed file"); + let drifted = conn + .execute( + "UPDATE account_registrations SET account_index = 7 \ + WHERE wallet_id = ?1 AND account_type = 'dashpay_receiving'", + params![wallet_id.as_slice()], + ) + .expect("drift the receiving account's row"); + assert_eq!(drifted, 1); + } + + let persister = open_recovery(&path, FlushMode::Immediate); + let manager = relaunch(&persister).await; + let wallet = manager + .get_wallet(&wallet_id) + .await + .expect("restored wallet"); + assert!( + wallet + .state() + .await + .core_wallet + .accounts + .dashpay_receival_accounts + .is_empty(), + "the drifted receiving account is skipped" + ); + assert!( + wallet.state().await.dashpay_backfill.is_empty(), + "no coverage survives for the account the wallet does not hold" + ); + assert_eq!( + stored_covered_accounts(&persister, &wallet_id), + 1, + "recovery leaves the file as it was" + ); + close(manager, wallet, persister).await; +} + +/// A Manual-mode backend cannot hold coverage durably, so the manager gives +/// the wallet none and would invalidate the host's record before publishing. +/// Read-only, that invalidation is skipped rather than failing the load. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn should_publish_a_manual_recovery_load_over_stored_coverage() { + let tmp = secure_tempdir().expect("tempdir"); + let path = tmp.path().join("wallet.db"); + let wallet_id = seed_rewound_coverage(&path).await; + + let persister = open_recovery(&path, FlushMode::Manual); + let manager = relaunch(&persister).await; + let wallet = manager + .get_wallet(&wallet_id) + .await + .expect("restored wallet"); + assert!( + !carries_coverage(&wallet).await, + "a backend that cannot hold coverage durably gives the wallet none" + ); + assert_eq!( + stored_covered_accounts(&persister, &wallet_id), + 1, + "recovery leaves the file as it was" + ); + close(manager, wallet, persister).await; +} + +/// Reloading a wallet the same manager published before makes the manager +/// replace the host's record, since the earlier wallet might have changed it. +/// Read-only, nothing could have, so the reload publishes without the write. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn should_reload_a_removed_wallet_from_a_recovery_backend() { + let tmp = secure_tempdir().expect("tempdir"); + let path = tmp.path().join("wallet.db"); + let wallet_id = seed_rewound_coverage(&path).await; + + let persister = open_recovery(&path, FlushMode::Immediate); + let manager = relaunch(&persister).await; + let wallet = manager + .get_wallet(&wallet_id) + .await + .expect("restored wallet"); + assert!(carries_coverage(&wallet).await); + drop(manager.remove_wallet(&wallet_id).await.expect("remove")); + drop(wallet); + + manager + .load_from_persistor() + .await + .expect("a second load of the same file must not need a write"); + let wallet = manager + .get_wallet(&wallet_id) + .await + .expect("reloaded wallet"); + assert!( + carries_coverage(&wallet).await, + "the reloaded wallet keeps the coverage on file" + ); + assert_eq!(stored_covered_accounts(&persister, &wallet_id), 1); + close(manager, wallet, persister).await; +} + +/// The outbound-account marker survives a SQLite relaunch together with the +/// account it names. Registering the outbound (`DashpayExternalAccount`) +/// account binds the marker in the same round as the account and its pool; +/// after a reopen the contact carries the marker (so the startup sweep does +/// not rebuild the account) AND the account is there with the same xpub and +/// pool — a marker restored without its account would suppress the rebuild +/// of an account that no longer exists. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn should_restore_the_outbound_account_marker_with_its_account_after_a_sqlite_relaunch() { + let tmp = secure_tempdir().expect("tempdir"); + let path = tmp.path().join("wallet.db"); + let owner = Identifier::from(OWNER); + let contact = Identifier::from(CONTACT); + let key = DashpayAccountKey { + index: 0, + user_identity_id: OWNER, + friend_identity_id: CONTACT, + }; + const ACCOUNT_REFERENCE: u32 = 7; + let shared_key = [0x55u8; 32]; + + // ── Session one: establish, then build the outbound account. ── + let persister = open(&path); + let (manager, wallet_id) = test_platform_wallet_manager_with(Arc::clone(&persister)).await; + let wallet = manager.get_wallet(&wallet_id).await.expect("wallet"); + // The contact's receiving xpub for us, as their request carries it. + let theirs = derive_contact_xpub(&seed_wallet(), Network::Testnet, 0, &contact, &owner) + .expect("derive the contact's xpub"); + let encrypted = encrypt_contact_xpub_for_test(&shared_key, &theirs.compact_xpub()); + establish_contact( + &wallet, + &persister, + owner, + contact, + REQUEST_HEIGHT, + ACCOUNT_REFERENCE, + encrypted.clone(), + ) + .await; + let registration = wallet + .identity() + .dashpay() + .register_external_contact_account( + &owner, + &bare_identity(contact), + &encrypted, + zeroize::Zeroizing::new(shared_key), + ) + .await + .expect("register the outbound account"); + assert_eq!(registration, ExternalAccountRegistration::Built); + + let (xpub_before, pool_before) = { + let state = wallet.state().await; + let established = state + .identity_manager + .managed_identity(&owner) + .expect("owner") + .dashpay() + .established_contacts() + .get(&contact) + .expect("contact") + .clone(); + assert_eq!( + established.external_account_reference, + Some(ACCOUNT_REFERENCE), + "an inline, atomic backend gets the marker in the registration round" + ); + let managed = state + .core_wallet + .accounts + .dashpay_external_accounts + .get(&key) + .expect("outbound account registered"); + let pool: Vec<_> = managed + .managed_account_type() + .address_pools() + .first() + .expect("outbound pool") + .addresses + .values() + .map(|info| info.address.clone()) + .collect(); + drop(state); + let wm = wallet.wallet_manager().read().await; + let xpub = wm + .get_wallet(&wallet_id) + .expect("wallet") + .accounts + .dashpay_external_accounts + .get(&key) + .expect("outbound account on the wallet") + .account_xpub; + (xpub, pool) + }; + assert!(!pool_before.is_empty(), "the outbound pool is pre-derived"); + { + let conn = persister.lock_conn_for_test(); + let marker: Option = conn + .query_row( + "SELECT external_account_reference FROM contacts \ + WHERE owner_id = ?1 AND contact_id = ?2", + params![&OWNER[..], &CONTACT[..]], + |row| row.get(0), + ) + .expect("contact row"); + assert_eq!(marker, Some(i64::from(ACCOUNT_REFERENCE))); + } + close(manager, wallet, persister).await; + + // ── Session two: the marker and its account both come back. ── + let persister = open(&path); + let manager = relaunch(&persister).await; + let wallet = manager + .get_wallet(&wallet_id) + .await + .expect("restored wallet"); + { + let state = wallet.state().await; + let established = state + .identity_manager + .managed_identity(&owner) + .expect("owner restored") + .dashpay() + .established_contacts() + .get(&contact) + .expect("established contact restored") + .clone(); + assert_eq!( + established.external_account_reference, + Some(ACCOUNT_REFERENCE), + "the persisted marker must survive the relaunch" + ); + assert_eq!( + established.external_account_reference, + Some(established.incoming_request.account_reference), + "the marker names the live request, so the sweep does not rebuild" + ); + let managed = state + .core_wallet + .accounts + .dashpay_external_accounts + .get(&key) + .expect("the outbound account the marker names must be restored"); + let pool: Vec<_> = managed + .managed_account_type() + .address_pools() + .first() + .expect("outbound pool restored") + .addresses + .values() + .map(|info| info.address.clone()) + .collect(); + assert_eq!(pool, pool_before, "the restored pool is the one built"); + } + { + let wm = wallet.wallet_manager().read().await; + let account = wm + .get_wallet(&wallet_id) + .expect("wallet") + .accounts + .dashpay_external_accounts + .get(&key) + .expect("the outbound account's xpub must be restored on the wallet"); + assert_eq!(account.account_xpub, xpub_before); + // The account is rebuilt from the 69-byte compact form, which omits + // the BIP32 metadata, so compare the key material it derives from. + assert_eq!( + ( + account.account_xpub.public_key, + account.account_xpub.chain_code + ), + (theirs.xpub.public_key, theirs.xpub.chain_code), + "the restored account derives from the contact's xpub" + ); + } + // The startup sweep would now take the account as built: re-registering + // finds it rather than rebuilding it. + let again = wallet + .identity() + .dashpay() + .register_external_contact_account( + &owner, + &bare_identity(contact), + &encrypted, + zeroize::Zeroizing::new(shared_key), + ) + .await + .expect("re-register after relaunch"); + assert_eq!(again, ExternalAccountRegistration::AlreadyExisted); + close(manager, wallet, persister).await; +} 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..ff3ea1fc21d 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 @@ -1547,11 +1547,13 @@ fn tc_p4_012_load_query_count_bounded() { "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` - // ⇒ shared must itself be a small constant, not growing with N. + // `platform_addrs::load_all` scans, plus the `BEGIN` / `COMMIT` of the + // one read snapshot the whole load runs in. `count_one = shared + + // per_wallet` ⇒ shared must itself be a small constant, not growing + // with N. let shared = count_one - per_wallet; assert!( - (1..=8).contains(&shared), + (1..=10).contains(&shared), "shared load() overhead must be a small constant, got {shared} \ (N=1 → {count_one}, per-wallet → {per_wallet})" ); diff --git a/packages/rs-platform-wallet-storage/tests/integration/sqlite_schema_pinning.rs b/packages/rs-platform-wallet-storage/tests/integration/sqlite_schema_pinning.rs index b4ef31e3c31..c7629f96bef 100644 --- a/packages/rs-platform-wallet-storage/tests/integration/sqlite_schema_pinning.rs +++ b/packages/rs-platform-wallet-storage/tests/integration/sqlite_schema_pinning.rs @@ -15,13 +15,13 @@ use platform_wallet_storage::sqlite::{migrations as mig, schema::versions::Domai /// Golden `(version, name)` fingerprint of the frozen migration set. Bump /// deliberately only when adding/removing/renaming a migration file. const EXPECTED_ID_FINGERPRINT: &str = - "5f6d87af51ff949ecf5058c9e1086dc7624f5d6955164e19be33bc460132776b"; + "2aafbd0ccf1f8e98b23f7cdf78cd5c652551699ec7f6de11908290a943d9593d"; /// Golden content-level fingerprint over every migration's rendered SQL. /// Bump it only when ADDING a migration file; a body change on an already /// applied migration is a defect, not a golden to refresh. const EXPECTED_SQL_FINGERPRINT: &str = - "3a3e9b2df9b1dde2664ae086a6bd53a30b57208024662b917698184761f14515"; + "d0d20516126dcaf31511ec36f0820d3617b9a81983f5400b8a777b02031cbc00"; /// The migrations merged `v4.2-dev` already ships. Refinery keys /// `refinery_schema_history` by version and validates an applied migration's diff --git a/packages/rs-platform-wallet-storage/tests/integration/sqlite_version_bump.rs b/packages/rs-platform-wallet-storage/tests/integration/sqlite_version_bump.rs index 73d2437db0b..bd562669a04 100644 --- a/packages/rs-platform-wallet-storage/tests/integration/sqlite_version_bump.rs +++ b/packages/rs-platform-wallet-storage/tests/integration/sqlite_version_bump.rs @@ -272,6 +272,21 @@ fn single_domain_changeset(domain: Domain) -> PlatformWalletChangeSet { use platform_wallet::changeset::IdentityScanStateEntry; cs.identity_scan_state = Some(IdentityScanStateEntry::incomplete(0, 5, vec![1])); } + Domain::DashPayBackfill => { + use platform_wallet::changeset::{ + DashPayBackfillCoveredContact, DashPayBackfillRecord, + }; + cs.dashpay_backfill = Some(DashPayBackfillRecord::from_entries( + 10, + 20, + vec![DashPayBackfillCoveredContact { + owner: Identifier::from([0x11; 32]), + contact: Identifier::from([0x12; 32]), + account_index: 0, + covered_from: 10, + }], + )); + } #[cfg(feature = "shielded")] Domain::ShieldedViewingKeys => { let mut shielded = ShieldedChangeSet::default(); @@ -384,7 +399,7 @@ fn every_domain_maps_and_isolates() { use std::collections::BTreeSet; assert_eq!( Domain::ALL.len(), - if cfg!(feature = "shielded") { 17 } else { 16 }, + if cfg!(feature = "shielded") { 18 } else { 17 }, "every compiled persistence domain must be covered" ); let mut covered = BTreeSet::new(); diff --git a/packages/rs-platform-wallet/src/changeset/changeset.rs b/packages/rs-platform-wallet/src/changeset/changeset.rs index 8b8f8418fe2..5c9d94b00fc 100644 --- a/packages/rs-platform-wallet/src/changeset/changeset.rs +++ b/packages/rs-platform-wallet/src/changeset/changeset.rs @@ -175,8 +175,8 @@ pub struct CoreChangeSet { /// rewind(max(a, b))` — which is associative, so how a backend groups /// rounds into accumulations never changes the result. Meaningless while /// `synced_height` is `None`. Hosts that write the cursor they are handed - /// honour it directly; the SQLite backend keeps its monotonic max and - /// does not store the backfill record, so it neither needs nor sees it. + /// honour it directly; the SQLite backend keeps its monotonic max for + /// advances and overwrites its stored cursor only on a rewind. #[cfg_attr(feature = "serde", serde(default))] pub synced_height_is_rewind: bool, diff --git a/packages/rs-platform-wallet/src/changeset/traits.rs b/packages/rs-platform-wallet/src/changeset/traits.rs index 11e5d2c58a7..c91e69e947f 100644 --- a/packages/rs-platform-wallet/src/changeset/traits.rs +++ b/packages/rs-platform-wallet/src/changeset/traits.rs @@ -315,6 +315,19 @@ pub trait PlatformWalletPersistence: Send + Sync { false } + /// Whether this backend serves a read-only view and refuses every + /// [`store`](Self::store) and [`flush`](Self::flush) — a storage backend + /// opened to recover a damaged database, for example. + /// + /// Loading such a backend must not need a write to succeed: housekeeping + /// a load would otherwise write back (replacing a stale backfill record) + /// is applied in memory only and left on the host for a writable load. + /// + /// The default is `false`: a backend that accepts writes. + fn refuses_writes(&self) -> bool { + false + } + /// Compatibility summary for older invitation callers. It is true when the /// backend attests atomic changesets plus durably persisted invitation rows /// and asset-lock funding indices. This does not attest restart hydration; diff --git a/packages/rs-platform-wallet/src/manager/mod.rs b/packages/rs-platform-wallet/src/manager/mod.rs index 6ea8ba9b5fd..d21487d1026 100644 --- a/packages/rs-platform-wallet/src/manager/mod.rs +++ b/packages/rs-platform-wallet/src/manager/mod.rs @@ -748,6 +748,21 @@ impl PlatformWalletManager

{ if !replace { return Ok(record); } + // A read-only backend refuses the replacement, and failing here would + // fail the whole load. The wallet still carries only `record`; the + // host keeps its stale one until a writable load replaces it. + if self.persister.refuses_writes() { + tracing::info!( + wallet_id = %hex::encode(wallet_id), + covered = record.covered().len(), + dropped, + predecessor, + durable, + "the host's DashPay backfill record is left as it is: the backend \ + is read-only" + ); + return Ok(record); + } let host_cursor = durable_cursors .get(&wallet_id) .map_or(snapshot_cursor, |entry| entry.height); diff --git a/packages/rs-platform-wallet/src/test_support.rs b/packages/rs-platform-wallet/src/test_support.rs index 0dbca35e9b1..94ea1c7aa83 100644 --- a/packages/rs-platform-wallet/src/test_support.rs +++ b/packages/rs-platform-wallet/src/test_support.rs @@ -720,6 +720,16 @@ where (manager, wallet_id) } +/// Encrypt a contact's 69-byte DIP-15 compact receiving xpub under +/// `shared_key` into the 96-byte `encryptedPublicKey` wire form of a contact +/// request (fixed IV), so tests in crates without a `platform-encryption` +/// dependency can drive +/// [`DashPayView::register_external_contact_account`](crate::wallet::identity::DashPayView::register_external_contact_account) +/// offline. Test scaffolding only: the fixed IV is not fit for real use. +pub fn encrypt_contact_xpub_for_test(shared_key: &[u8; 32], compact_xpub: &[u8]) -> Vec { + platform_encryption::encrypt_extended_public_key(shared_key, &[0x11u8; 16], compact_xpub) +} + /// Canonical all-`abandon` BIP-39 test vector. Fixed (not /// `TestWalletContext::new_random`) so every key it derives is a stable golden — /// which is what lets the signed-message tests pin an RFC6979-deterministic diff --git a/packages/rs-platform-wallet/src/wallet/identity/mod.rs b/packages/rs-platform-wallet/src/wallet/identity/mod.rs index 01079577e1b..2f443219e19 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/mod.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/mod.rs @@ -29,7 +29,9 @@ pub use crypto::{ derive_contact_payment_addresses, derive_contact_xpub, pubkey_binds_expected_key_data, unmask_account_reference, ContactXpubData, DEFAULT_CONTACT_GAP_LIMIT, }; -pub use network::{DashPayView, IdentityWallet}; +pub use network::{ + DashPayView, ExternalAccountRegistration, IdentityWallet, RegisterExternalError, +}; pub use state::{ BlockTime, DashPayState, IdentityLocation, IdentityManager, ManagedIdentity, RegistrationIndex, }; diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs b/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs index 8e121a95005..399f96c3e58 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs @@ -93,6 +93,7 @@ pub use contact_requests::{ AutoAcceptProofSource, ContactCryptoProvider, ContactInfoOpened, ContactInfoSealed, ContactSyncReport, }; +pub use contacts::{ExternalAccountRegistration, RegisterExternalError}; pub use dashpay_view::DashPayView; pub use discovery::IdentityDiscoveryOptions; pub use dpns::{ContestContender, ContestVoteState, ContestWinner};