diff --git a/packages/rs-platform-wallet-ffi/src/persistence.rs b/packages/rs-platform-wallet-ffi/src/persistence.rs index 578d6d2d46f..4d614581c2f 100644 --- a/packages/rs-platform-wallet-ffi/src/persistence.rs +++ b/packages/rs-platform-wallet-ffi/src/persistence.rs @@ -759,6 +759,7 @@ impl PlatformWalletPersistence for FFIPersister { nonce: entry.funds.nonce, account_index: entry.account_index, address_index: entry.address_index, + as_of_height: entry.funds.as_of_height, }) .collect(); if !entries.is_empty() { @@ -3409,6 +3410,10 @@ fn build_wallet_start_state( dash_sdk::platform::address_sync::AddressFunds { nonce: persisted.nonce, balance: persisted.balance, + // Height pin round-trip: rows persisted before the pin + // existed load as 0 ("unknown provenance") and yield to + // the first pinned absolute — the self-healing path. + as_of_height: persisted.as_of_height, }, ); } diff --git a/packages/rs-platform-wallet-ffi/src/platform_address_types.rs b/packages/rs-platform-wallet-ffi/src/platform_address_types.rs index 356e22c3ef7..7d56610a1e7 100644 --- a/packages/rs-platform-wallet-ffi/src/platform_address_types.rs +++ b/packages/rs-platform-wallet-ffi/src/platform_address_types.rs @@ -232,6 +232,11 @@ pub struct AddressBalanceEntryFFI { pub account_index: u32, /// DIP-17 derivation index within the account. pub address_index: u32, + /// Platform block height `balance` is current as of — the height pin + /// (see `AddressFunds::as_of_height` in `dash-sdk`). Meaningful on the + /// persistence round-trip (persist callback → host storage → load); + /// pass 0 on request paths that only name outputs/amounts. + pub as_of_height: u64, } /// Parse output entries into the DPP-canonical `BTreeMap`. @@ -516,6 +521,7 @@ impl From<&platform_wallet::PlatformAddressChangeSet> for PlatformAddressChangeS nonce: entry.funds.nonce, account_index: entry.account_index, address_index: entry.address_index, + as_of_height: entry.funds.as_of_height, }) .collect(); @@ -553,6 +559,7 @@ mod tests { nonce: 0, account_index: 0, address_index: 0, + as_of_height: 0, }, AddressBalanceEntryFFI { address: dup, @@ -560,6 +567,7 @@ mod tests { nonce: 0, account_index: 0, address_index: 0, + as_of_height: 0, }, ]; @@ -694,6 +702,7 @@ mod tests { nonce: 0, account_index: 0, address_index: 0, + as_of_height: 0, }]; assert_eq!( unsafe { parse_outputs(out.as_ptr(), out.len()) } @@ -739,6 +748,7 @@ mod tests { nonce: 0, account_index: 0, address_index: 0, + as_of_height: 0, }, AddressBalanceEntryFFI { address: PlatformAddressFFI { @@ -749,6 +759,7 @@ mod tests { nonce: 0, account_index: 0, address_index: 0, + as_of_height: 0, }, ]; diff --git a/packages/rs-platform-wallet-ffi/src/platform_addresses/wallet.rs b/packages/rs-platform-wallet-ffi/src/platform_addresses/wallet.rs index c9351c282d3..d42a6dfb6a6 100644 --- a/packages/rs-platform-wallet-ffi/src/platform_addresses/wallet.rs +++ b/packages/rs-platform-wallet-ffi/src/platform_addresses/wallet.rs @@ -145,6 +145,7 @@ pub unsafe extern "C" fn platform_address_wallet_addresses_with_balances( nonce: 0, account_index: 0, address_index: 0, + as_of_height: 0, }) .collect::>() }); diff --git a/packages/rs-platform-wallet-storage/migrations/V002__address_height_pin.rs b/packages/rs-platform-wallet-storage/migrations/V002__address_height_pin.rs new file mode 100644 index 00000000000..415a124b81f --- /dev/null +++ b/packages/rs-platform-wallet-storage/migrations/V002__address_height_pin.rs @@ -0,0 +1,18 @@ +//! Add the balance height pin to `platform_addresses`. +//! +//! `as_of_height` mirrors `AddressFunds::as_of_height` (see `dash-sdk`'s +//! `address_sync` module): the Platform block height a persisted balance +//! is current **as of**. It is the reconciliation rule between +//! proof-attested absolutes and the recent/compacted balance-change +//! delta stream — a delta recorded at or below the pin is already +//! included in the absolute and must not be re-applied (the ADDR-09 +//! double-count). +//! +//! `DEFAULT 0` means "unknown provenance" for rows persisted before the +//! pin existed: every delta applies and any pinned absolute supersedes +//! them, which is exactly the pre-pin behavior plus self-healing. + +pub fn migration() -> String { + "ALTER TABLE platform_addresses ADD COLUMN as_of_height INTEGER NOT NULL DEFAULT 0;" + .to_string() +} diff --git a/packages/rs-platform-wallet-storage/src/sqlite/schema/platform_addrs.rs b/packages/rs-platform-wallet-storage/src/sqlite/schema/platform_addrs.rs index 95291ecf3b1..5b3d6998d40 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/schema/platform_addrs.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/schema/platform_addrs.rs @@ -24,13 +24,15 @@ pub fn apply( if !cs.addresses.is_empty() { let mut stmt = tx.prepare_cached( "INSERT INTO platform_addresses \ - (wallet_id, account_index, address_index, address, balance, nonce) \ - VALUES (?1, ?2, ?3, ?4, ?5, ?6) \ + (wallet_id, account_index, address_index, address, balance, nonce, \ + as_of_height) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) \ ON CONFLICT(wallet_id, address) DO UPDATE SET \ account_index = excluded.account_index, \ address_index = excluded.address_index, \ balance = excluded.balance, \ - nonce = excluded.nonce", + nonce = excluded.nonce, \ + as_of_height = excluded.as_of_height", )?; for entry in &cs.addresses { // The row is keyed by the outer `wallet_id`; an entry that @@ -51,6 +53,7 @@ pub fn apply( entry.address.as_bytes(), safe_cast::u64_to_i64("platform_addresses.balance", entry.funds.balance)?, i64::from(entry.funds.nonce), + safe_cast::u64_to_i64("platform_addresses.as_of_height", entry.funds.as_of_height)?, ])?; } } @@ -114,7 +117,7 @@ pub fn list_per_wallet( wallet_id: &WalletId, ) -> Result, WalletStorageError> { let mut stmt = conn.prepare( - "SELECT account_index, address_index, address, balance, nonce \ + "SELECT account_index, address_index, address, balance, nonce, as_of_height \ FROM platform_addresses WHERE wallet_id = ?1 \ ORDER BY account_index, address_index, address", )?; @@ -125,17 +128,19 @@ pub fn list_per_wallet( row.get::<_, Vec>(2)?, row.get::<_, i64>(3)?, row.get::<_, i64>(4)?, + row.get::<_, i64>(5)?, )) })?; let mut out = Vec::new(); for r in rows { - let (account_index, address_index, address_bytes, balance, nonce) = r?; + let (account_index, address_index, address_bytes, balance, nonce, as_of_height) = r?; out.push(decode_address_row( account_index, address_index, &address_bytes, balance, nonce, + as_of_height, )?); } Ok(out) @@ -323,7 +328,8 @@ fn all_address_rows( conn: &Connection, ) -> Result>, WalletStorageError> { let mut stmt = conn.prepare( - "SELECT wallet_id, account_index, address_index, address, balance, nonce \ + "SELECT wallet_id, account_index, address_index, address, balance, nonce, \ + as_of_height \ FROM platform_addresses ORDER BY wallet_id, account_index, address_index, address", )?; let rows = stmt.query_map([], |row| { @@ -334,11 +340,13 @@ fn all_address_rows( row.get::<_, Vec>(3)?, row.get::<_, i64>(4)?, row.get::<_, i64>(5)?, + row.get::<_, i64>(6)?, )) })?; let mut out: BTreeMap> = BTreeMap::new(); for r in rows { - let (wid_bytes, account_index, address_index, address_bytes, balance, nonce) = r?; + let (wid_bytes, account_index, address_index, address_bytes, balance, nonce, as_of_height) = + r?; let wallet_id = wallet_id_from_bytes(&wid_bytes)?; out.entry(wallet_id).or_default().push(decode_address_row( account_index, @@ -346,6 +354,7 @@ fn all_address_rows( &address_bytes, balance, nonce, + as_of_height, )?); } Ok(out) @@ -358,6 +367,7 @@ fn decode_address_row( address_bytes: &[u8], balance: i64, nonce: i64, + as_of_height: i64, ) -> Result { if address_bytes.len() != 20 { return Err(WalletStorageError::blob_decode( @@ -367,6 +377,7 @@ fn decode_address_row( let mut hash160 = [0u8; 20]; hash160.copy_from_slice(address_bytes); let balance = safe_cast::i64_to_u64("platform_addresses.balance", balance)?; + let as_of_height = safe_cast::i64_to_u64("platform_addresses.as_of_height", as_of_height)?; let nonce = u32::try_from(nonce).map_err(|_| WalletStorageError::IntegerOverflow { field: "platform_addresses.nonce", value: nonce as u64, @@ -388,7 +399,11 @@ fn decode_address_row( account_index, address_index, address: PlatformP2PKHAddress::new(hash160), - funds: AddressFunds { balance, nonce }, + funds: AddressFunds { + balance, + nonce, + as_of_height, + }, }) } diff --git a/packages/rs-platform-wallet-storage/tests/sqlite_load_reconstruction.rs b/packages/rs-platform-wallet-storage/tests/sqlite_load_reconstruction.rs index 97cfaf974a7..db12cfef2e1 100644 --- a/packages/rs-platform-wallet-storage/tests/sqlite_load_reconstruction.rs +++ b/packages/rs-platform-wallet-storage/tests/sqlite_load_reconstruction.rs @@ -35,6 +35,7 @@ fn entry( funds: AddressFunds { balance: address_index as u64 * 100, nonce: address_index, + as_of_height: address_index as u64 * 1_000, }, } } @@ -149,7 +150,8 @@ fn load_state_reconstructs_per_account_from_registration_and_addresses() { account.found().get(&addr0), Some(&AddressFunds { balance: 0, - nonce: 0 + nonce: 0, + as_of_height: 0, }), "address 0 funds must match the seeded entry" ); @@ -157,7 +159,8 @@ fn load_state_reconstructs_per_account_from_registration_and_addresses() { account.found().get(&addr1), Some(&AddressFunds { balance: 100, - nonce: 1 + nonce: 1, + as_of_height: 1_000, }), "address 1 funds must match the seeded entry" ); diff --git a/packages/rs-platform-wallet-storage/tests/sqlite_persist_roundtrip.rs b/packages/rs-platform-wallet-storage/tests/sqlite_persist_roundtrip.rs index 3004b16fc88..d2fd05b4a3d 100644 --- a/packages/rs-platform-wallet-storage/tests/sqlite_persist_roundtrip.rs +++ b/packages/rs-platform-wallet-storage/tests/sqlite_persist_roundtrip.rs @@ -295,6 +295,7 @@ fn tc009_platform_address_roundtrip() { funds: AddressFunds { nonce: 1, balance: 500, + as_of_height: 111, }, }, PlatformAddressBalanceEntry { @@ -305,6 +306,7 @@ fn tc009_platform_address_roundtrip() { funds: AddressFunds { nonce: 2, balance: 1500, + as_of_height: 222, }, }, ]; diff --git a/packages/rs-platform-wallet-storage/tests/sqlite_structural_hardening.rs b/packages/rs-platform-wallet-storage/tests/sqlite_structural_hardening.rs index eb36c856894..2da12f8c0b5 100644 --- a/packages/rs-platform-wallet-storage/tests/sqlite_structural_hardening.rs +++ b/packages/rs-platform-wallet-storage/tests/sqlite_structural_hardening.rs @@ -80,6 +80,7 @@ fn platform_addr_mixed_wallet_rejected() { funds: AddressFunds { nonce: 0, balance: 0, + as_of_height: 0, }, }], ..Default::default() diff --git a/packages/rs-platform-wallet/src/changeset/changeset.rs b/packages/rs-platform-wallet/src/changeset/changeset.rs index d4c589323ce..8dc2c705427 100644 --- a/packages/rs-platform-wallet/src/changeset/changeset.rs +++ b/packages/rs-platform-wallet/src/changeset/changeset.rs @@ -1498,7 +1498,11 @@ mod tests { let addr1 = PlatformP2PKHAddress::new([1u8; 20]); let addr2 = PlatformP2PKHAddress::new([2u8; 20]); - let funds = |balance, nonce| AddressFunds { balance, nonce }; + let funds = |balance, nonce| AddressFunds { + balance, + nonce, + as_of_height: 0, + }; let entry = |address_index, address, funds| PlatformAddressBalanceEntry { wallet_id, account_index: 0, diff --git a/packages/rs-platform-wallet/src/changeset/serde_adapters.rs b/packages/rs-platform-wallet/src/changeset/serde_adapters.rs index 330fab55c80..23d163ecd88 100644 --- a/packages/rs-platform-wallet/src/changeset/serde_adapters.rs +++ b/packages/rs-platform-wallet/src/changeset/serde_adapters.rs @@ -55,7 +55,7 @@ pub mod asset_lock_funding_type { } /// Adapter for `AddressFunds` (re-exported from `dash-sdk`; no serde -/// derive there). Encodes the two scalar fields side-by-side. +/// derive there). Encodes the scalar fields side-by-side. pub mod address_funds { use super::*; @@ -63,6 +63,11 @@ pub mod address_funds { struct Wire { nonce: AddressNonce, balance: Credits, + /// Height pin (see `AddressFunds::as_of_height`). Defaults to 0 + /// ("unknown provenance") when decoding blobs persisted before + /// the pin existed. + #[serde(default)] + as_of_height: u64, } pub fn serialize( @@ -72,6 +77,7 @@ pub mod address_funds { Wire { nonce: value.nonce, balance: value.balance, + as_of_height: value.as_of_height, } .serialize(serializer) } @@ -83,6 +89,7 @@ pub mod address_funds { Ok(AddressFunds { nonce: w.nonce, balance: w.balance, + as_of_height: w.as_of_height, }) } } diff --git a/packages/rs-platform-wallet/src/wallet/apply.rs b/packages/rs-platform-wallet/src/wallet/apply.rs index 01668a762f9..6100e19d57c 100644 --- a/packages/rs-platform-wallet/src/wallet/apply.rs +++ b/packages/rs-platform-wallet/src/wallet/apply.rs @@ -636,7 +636,11 @@ mod tests { let p2pkh2 = PlatformP2PKHAddress::new([20u8; 20]); use dash_sdk::platform::address_sync::AddressFunds; - let funds = |balance, nonce| AddressFunds { balance, nonce }; + let funds = |balance, nonce| AddressFunds { + balance, + nonce, + as_of_height: 0, + }; let wallet_id: crate::wallet::platform_wallet::WalletId = [0u8; 32]; let entry = |address_index, address, funds| crate::PlatformAddressBalanceEntry { wallet_id, @@ -1799,6 +1803,7 @@ mod tests { funds: dash_sdk::platform::address_sync::AddressFunds { balance: 1_000, nonce: 0, + as_of_height: 0, }, }); diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/register_from_addresses.rs b/packages/rs-platform-wallet/src/wallet/identity/network/register_from_addresses.rs index 710ce8566ff..df7c1456d0f 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/register_from_addresses.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/register_from_addresses.rs @@ -87,7 +87,7 @@ impl IdentityWallet { identity_signer: &IS, input_address_signer: &AS, settings: Option, - ) -> Result<(Identity, dash_sdk::query_types::AddressInfos), PlatformWalletError> { + ) -> Result<(Identity, dash_sdk::query_types::AddressInfos, u64), PlatformWalletError> { if inputs.is_empty() { return Err(PlatformWalletError::InvalidIdentityData( "At least one input address is required".to_string(), @@ -97,7 +97,7 @@ impl IdentityWallet { // Route through the auto-fetching SDK variant so the caller // doesn't need to maintain its own nonce cache — Platform is // always the source of truth at submit time. - let (mut registered_identity, address_infos) = identity + let (mut registered_identity, address_infos, proof_height) = identity .put_with_address_funding_fetching_nonces( &self.sdk, inputs, @@ -175,8 +175,8 @@ impl IdentityWallet { // The spent platform-address balances are reconciled by the // composite `PlatformWallet::register_from_addresses`, which routes - // the returned `AddressInfos` through the platform-address wallet's - // shared reconciliation seam. - Ok((identity, address_infos)) + // the returned `AddressInfos` (pinned at `proof_height`) through + // the platform-address wallet's shared reconciliation seam. + Ok((identity, address_infos, proof_height)) } } diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/top_up_from_addresses.rs b/packages/rs-platform-wallet/src/wallet/identity/network/top_up_from_addresses.rs index e9b4a124eb8..da15c81cd8d 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/top_up_from_addresses.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/top_up_from_addresses.rs @@ -57,7 +57,7 @@ impl IdentityWallet { inputs: BTreeMap, address_signer: &S, settings: Option, - ) -> Result<(AddressInfos, Credits), PlatformWalletError> { + ) -> Result<(AddressInfos, Credits, u64), PlatformWalletError> { let identity = { let wm = self.wallet_manager.read().await; let info = wm.get_wallet_info(&self.wallet_id).ok_or_else(|| { @@ -72,7 +72,7 @@ impl IdentityWallet { .ok_or(PlatformWalletError::IdentityNotFound(*identity_id))? }; - let (address_infos, new_balance) = identity + let (address_infos, new_balance, proof_height) = identity .top_up_from_addresses(&self.sdk, inputs, address_signer, settings) .await .map_err(|e| { @@ -110,6 +110,6 @@ impl IdentityWallet { // composite `PlatformWallet::top_up_from_addresses`, which routes // the returned `AddressInfos` through the platform-address wallet's // shared reconciliation seam. - Ok((address_infos, new_balance)) + Ok((address_infos, new_balance, proof_height)) } } diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/transfer_to_addresses.rs b/packages/rs-platform-wallet/src/wallet/identity/network/transfer_to_addresses.rs index 5d1ada2dbd4..bbed9c22a44 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/transfer_to_addresses.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/transfer_to_addresses.rs @@ -82,7 +82,7 @@ impl IdentityWallet { recipient_addresses: BTreeMap, signer: &S, settings: Option, - ) -> Result<(dash_sdk::query_types::AddressInfos, Credits), PlatformWalletError> + ) -> Result<(dash_sdk::query_types::AddressInfos, Credits, u64), PlatformWalletError> where S: Signer + Send + Sync, { @@ -100,7 +100,7 @@ impl IdentityWallet { .ok_or(PlatformWalletError::IdentityNotFound(*identity_id))? }; - let (address_infos, new_balance) = identity + let (address_infos, new_balance, proof_height) = identity .transfer_credits_to_addresses( &self.sdk, recipient_addresses, @@ -143,6 +143,6 @@ impl IdentityWallet { // composite `PlatformWallet::transfer_credits_to_addresses_with_external_signer`, // which routes the returned `AddressInfos` through the // platform-address wallet's shared reconciliation seam. - Ok((address_infos, new_balance)) + Ok((address_infos, new_balance, proof_height)) } } diff --git a/packages/rs-platform-wallet/src/wallet/platform_addresses/fund_from_asset_lock.rs b/packages/rs-platform-wallet/src/wallet/platform_addresses/fund_from_asset_lock.rs index f58d054be98..0658ef74e56 100644 --- a/packages/rs-platform-wallet/src/wallet/platform_addresses/fund_from_asset_lock.rs +++ b/packages/rs-platform-wallet/src/wallet/platform_addresses/fund_from_asset_lock.rs @@ -202,7 +202,9 @@ impl PlatformAddressWallet { // cache, and IS-lock rejection triggers an IS→CL upgrade on // the same outpoint. let proof_out_point = out_point_from_proof(&proof); - let address_infos = match submit_with_cl_height_retry(settings, |s| { + // `proof_height` is the broadcast proof's committed block — the + // height pin for the reconciled absolutes below. + let (address_infos, proof_height) = match submit_with_cl_height_retry(settings, |s| { addresses.top_up_with_signers( &self.sdk, proof.clone(), @@ -296,17 +298,13 @@ impl PlatformAddressWallet { // persistence hiccup shouldn't mask that. // // ADDR-09: every recipient of an asset-lock top-up is credited via - // an on-chain `AddBalanceToAddress` DELTA, so the whole recipient - // set goes into the seam's `credited_outputs` gate. Committing - // their proof-attested ABSOLUTE balances while the incremental - // watermark stayed stale would let the next incremental BLAST pass - // re-apply the delta on top → `X + X = 2X`, the ADDR-09 - // double-count. The seam invalidates the watermark inside its own - // critical section (see `reconcile_address_infos` and - // `PlatformPaymentAddressProvider::invalidate_sync_watermark`), - // forcing the next pass to full-scan-reconcile — the automated - // equivalent of the manual Sync-tab "Clear" + "Sync Now". - let credited_outputs = super::credited_outputs_set(addresses.keys()); + // an on-chain `AddBalanceToAddress` DELTA at exactly this proof's + // block height. The committed absolutes carry `proof_height` as + // their height pin (`AddressFunds::as_of_height`), so the sync's + // apply loops drop that delta (and any older one) instead of + // re-applying it on top → no `X + X = 2X` double-count, on + // incremental AND full-scan passes alike. + // // Use the persistence-reporting variant: marking the lock // `Consumed` below is irreversible, so it MUST be gated on the // reconciled balances actually reaching disk. `persisted` is @@ -317,7 +315,7 @@ impl PlatformAddressWallet { let (cs, persisted) = self .reconcile_address_infos_with_persistence( &address_infos, - &credited_outputs, + proof_height, "fund from asset lock", ) .await; diff --git a/packages/rs-platform-wallet/src/wallet/platform_addresses/mod.rs b/packages/rs-platform-wallet/src/wallet/platform_addresses/mod.rs index 2910b38aff2..e8ca3d92704 100644 --- a/packages/rs-platform-wallet/src/wallet/platform_addresses/mod.rs +++ b/packages/rs-platform-wallet/src/wallet/platform_addresses/mod.rs @@ -1,11 +1,10 @@ //! DIP-17 platform payment address wallet and provider. -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::BTreeMap; use dpp::address_funds::PlatformAddress; use dpp::fee::Credits; pub use dpp::prelude::AddressNonce; -use key_wallet::PlatformP2PKHAddress; #[cfg(doc)] use crate::PlatformWalletError; @@ -43,27 +42,6 @@ where .ok_or(crate::PlatformWalletError::InputSumOverflow) } -/// Collect the P2PKH members of an address iterator into the set shape -/// [`PlatformAddressWallet::reconcile_address_infos`] takes for its -/// `credited_outputs` parameter — the addresses a transition credits via -/// an on-chain `AddBalanceToAddress` DELTA (transfer outputs, asset-lock -/// top-up recipients, identity-registration change, identity→address -/// credit-transfer recipients), as opposed to inputs, which are recorded -/// as absolute `SetBalanceToAddress` ops. Non-P2PKH addresses are skipped: -/// wallet-owned platform-payment addresses are always P2PKH, so a non-P2PKH -/// output can never be an owned address the seam would reconcile. -pub(crate) fn credited_outputs_set<'a>( - addresses: impl IntoIterator, -) -> BTreeSet { - addresses - .into_iter() - .filter_map(|addr| match addr { - PlatformAddress::P2pkh(hash) => Some(PlatformP2PKHAddress::new(*hash)), - _ => None, - }) - .collect() -} - pub use provider::{ PerAccountPlatformAddressState, PerWalletPlatformAddressState, PlatformAddressTag, }; diff --git a/packages/rs-platform-wallet/src/wallet/platform_addresses/provider.rs b/packages/rs-platform-wallet/src/wallet/platform_addresses/provider.rs index c64bfaa11cd..b141af762ae 100644 --- a/packages/rs-platform-wallet/src/wallet/platform_addresses/provider.rs +++ b/packages/rs-platform-wallet/src/wallet/platform_addresses/provider.rs @@ -551,68 +551,6 @@ impl PlatformPaymentAddressProvider { } } - /// Zero the incremental-sync watermark ONLY, so the next - /// `sync_balances` takes the full-scan branch — WITHOUT dropping the - /// cached `found` seed (unlike [`reset_sync_state`](Self::reset_sync_state), - /// which is the "Clear" flow). - /// - /// WHY (ADDR-09): every flow that credits a wallet-owned address via - /// an on-chain `AddBalanceToAddress` DELTA (`AddToCredits` in Drive's - /// recent-address-balance-changes tree) — asset-lock top-up recipients, - /// same-wallet transfer outputs, identity-registration change, - /// identity→address credit-transfer recipients — reconciles the - /// proof-attested ABSOLUTE balance `X` into both the managed account - /// and the provider's committed `found` seed. If the next pass ran - /// INCREMENTALLY it would seed `result.found` from `current_balances()` - /// (already `X`) and then re-apply that recent `AddToCredits(X)` delta - /// from the stale watermark, landing at `X + X = 2X` — the ADDR-09 - /// double-count. An optimistic absolute write is fundamentally - /// inconsistent with incremental delta re-application, so we force the - /// very next pass to full-scan-reconcile. The single call site is the - /// gate inside `PlatformAddressWallet::reconcile_address_infos`, which - /// fires when a committed entry matches the caller-declared - /// `credited_outputs` set — inside the seam's provider-lock critical - /// section, so no sync pass can interleave between the seed commit and - /// this invalidation with the stale watermark. - /// - /// DURABILITY: in-memory only. The persisted sync watermark cannot be - /// reset to 0 through the normal changeset — - /// `PlatformAddressChangeSet::merge` combines `sync_height` with - /// `.max()` and the FFI persister only fires `on_persist_sync_state_fn` - /// when a component is `> 0` — so a durable zero would have to fight - /// both the merge and the `> 0` gate. Instead we rely on the in-session - /// BLAST cadence (~15s): the next pass full-scans, reconciles to `X`, - /// and persists a correct FORWARD watermark, so durable state - /// self-corrects within ~15s. The only residual gap is an app kill - /// inside that window; a restart then resumes incremental sync from the - /// stale persisted watermark and the double-count could briefly - /// reappear until the next full rescan (or a manual Clear). That narrow - /// window is accepted rather than over-engineering a durable - /// invalidation against the `.max()` merge / `> 0` gate. - /// - /// With `sync_timestamp == 0`, [`last_sync_timestamp`](Self::last_sync_timestamp) - /// returns `None`, which makes `sync_address_balances` choose the - /// full-scan branch: `result.found` is re-seeded ABSOLUTELY from the - /// tree (the `found` seed is only consulted on the incremental branch, - /// which is skipped), and incremental catch-up runs from the fresh - /// full-scan checkpoint rather than the stale height, so no recent - /// delta is re-applied. `last_known_recent_block` is zeroed too since - /// catch-up reads it as its recent-tree boundary. - /// - /// The `found` seed is deliberately KEPT (not cleared): a full scan - /// ignores it as a seed, and preserving it means display and - /// `auto_select_inputs` budgeting keep the just-applied balance `X` - /// visible during the ~15s until the reconciling scan completes, - /// instead of the momentary zero `reset_sync_state` would show. - /// - /// This is the in-memory equivalent of the manual Sync-tab - /// "Clear" + "Sync Now" that also fixes the double-count. - pub(crate) fn invalidate_sync_watermark(&mut self) { - self.sync_height = 0; - self.sync_timestamp = 0; - self.last_known_recent_block = 0; - } - /// Diagnostic snapshot counts used by the read-only memory /// explorer surface on /// [`crate::manager::PlatformWalletManager::platform_address_provider_state_blocking`]. @@ -883,7 +821,27 @@ impl AddressProvider for PlatformPaymentAddressProvider { }; let PerAccountInSyncPlatformAddressState { found, absent } = &mut account_scratch; absent.retain(|addr| !found.contains_key(addr)); - account_state.found.extend(account_scratch.found); + // Height-pin freshness on the merge: a pass that ran + // against a lagging node can stage a stale-but-valid + // absolute for a row a fresher reconcile already + // committed (its pin is older). Keeping the fresher + // committed entry mirrors `commit_reconciliation`'s + // rule AND `compute_address_balance_diff`'s persist + // guard, so the in-memory seed and the durable rows + // never diverge over which of the two is truth. + for (addr, incoming) in account_scratch.found { + match account_state.found.get(&addr) { + Some(existing) if existing.as_of_height > incoming.as_of_height => {} + _ => { + account_state.found.insert(addr, incoming); + } + } + } + // Absence carries no per-entry height, so a stale pass's + // removal is NOT pin-guarded here; the persist diff skips + // the durable zero for fresher-pinned rows, and the next + // pass reconstructs the in-memory entry from a full + // replay (base 0, pin 0 → every delta applies). for absent_addr in &account_scratch.absent { account_state.found.remove(absent_addr); } @@ -930,6 +888,11 @@ impl AddressProvider for PlatformPaymentAddressProvider { /// (e.g. a fully consumed input). Pure and lock-free so every caller's /// translation is unit-testable. /// +/// `as_of_height` is the proof's block height: every produced entry is an +/// absolute attested at that height, so its funds carry it as the height +/// pin (see `AddressFunds::as_of_height`) — including removals, which are +/// equally height-attested statements. +/// /// Callers supply the resolver so every reconciliation path can resolve /// through the provider's persisted `index <-> address` bijection — /// covering addresses restored from disk that are no longer in a live @@ -940,6 +903,7 @@ pub(crate) fn build_address_balance_entries( wallet_id: WalletId, resolve_index: impl Fn(&PlatformP2PKHAddress) -> Option<(u32, AddressIndex)>, address_infos: &AddressInfos, + as_of_height: u64, ) -> Vec { let mut entries = Vec::new(); for (addr, maybe_info) in address_infos.iter() { @@ -954,10 +918,12 @@ pub(crate) fn build_address_balance_entries( Some(ai) => AddressFunds { balance: ai.balance, nonce: ai.nonce, + as_of_height, }, None => AddressFunds { balance: 0, nonce: 0, + as_of_height, }, }; entries.push(PlatformAddressBalanceEntry { @@ -1012,16 +978,24 @@ impl PlatformPaymentAddressProvider { /// Pool-resolved addresses are merged into the bijection so /// `current_balances` can yield their committed funds. /// - /// Freshness guard, per resolved entry: - /// * zero funds (balance 0, nonce 0 — the address was removed from - /// Platform state, e.g. a fully consumed input) is an authoritative - /// removal: always applied, and the address is dropped from `found` - /// (mirroring the sync's absent handling); - /// * otherwise, an entry whose nonce is *below* the committed seed's - /// nonce is stale — a concurrent sync pass or later transition - /// already committed fresher state — and is dropped; + /// Freshness guard, per resolved entry — height-pin authority (see + /// `AddressFunds::as_of_height`): + /// * an entry whose pin is *below* the committed seed's pin is stale — + /// a sync pass or later transition already committed state attested + /// at a later block — and is dropped. This applies to removals too: + /// an older removal proof must not clobber a newer re-credit. + /// * on equal pins (same block), the nonce breaks the tie — it only + /// advances on outgoing ops, so it can order same-block states but + /// not receive-only states across blocks (the pin does that); /// * entries identical to the committed seed are dropped as no-ops. /// + /// Zero funds (balance 0, nonce 0 — the address was removed from + /// Platform state, e.g. a fully consumed input) that survive the guard + /// drop the address from `found`, mirroring the sync's absent handling. + /// + /// `as_of_height` is the proof's block height and becomes the pin on + /// every committed entry. + /// /// Callers must hold the provider write lock (i.e. call through /// `&mut self`) across this commit AND the managed-account balance /// write that follows, so a background sync — which holds the same @@ -1031,6 +1005,7 @@ impl PlatformPaymentAddressProvider { wallet_id: &WalletId, address_infos: &AddressInfos, pool_indexes: &BTreeMap, + as_of_height: u64, ) -> ReconciliationOutcome { let mut outcome = ReconciliationOutcome::default(); let Some(wallet_state) = self.per_wallet.get_mut(wallet_id) else { @@ -1058,6 +1033,7 @@ impl PlatformPaymentAddressProvider { .filter(|(account_index, _)| wallet_state.contains_key(account_index)) }, address_infos, + as_of_height, ); outcome.resolved = resolved_entries.len(); @@ -1075,21 +1051,28 @@ impl PlatformPaymentAddressProvider { continue; }; let existing = state.found.get(&entry.address).copied(); - // Zero funds = the address no longer exists in Platform state. - // That removal is attested by the proof, so it bypasses the - // nonce guard (the pre-spend seed necessarily has a lower - // nonce than "gone"). + // Zero funds = the address no longer exists in Platform state + // (e.g. a fully consumed input); survivors of the freshness + // guard drop the address from `found` below. let is_removal = entry.funds.balance == 0 && entry.funds.nonce == 0; - if !is_removal { - if let Some(existing) = existing { - if existing.nonce > entry.funds.nonce { - outcome.stale_skipped += 1; - continue; - } - if existing == entry.funds { - outcome.unchanged_skipped += 1; - continue; - } + if let Some(existing) = existing { + // Height-pin authority: a committed absolute pinned at a + // later block supersedes this proof — removals included + // (an older removal must not clobber a newer re-credit). + // Equal pins (same block) fall back to the nonce, which + // orders same-block outgoing ops. Legacy pin-0 rows lose + // to any pinned proof, which is the self-healing path for + // state persisted before the pin existed. + let stale = existing.as_of_height > entry.funds.as_of_height + || (existing.as_of_height == entry.funds.as_of_height + && existing.nonce > entry.funds.nonce); + if stale { + outcome.stale_skipped += 1; + continue; + } + if existing == entry.funds { + outcome.unchanged_skipped += 1; + continue; } } // Derivation-index conflict: `entry.address` isn't yet in the @@ -1179,7 +1162,19 @@ mod tests { } fn funds(balance: u64, nonce: u32) -> AddressFunds { - AddressFunds { balance, nonce } + AddressFunds { + balance, + nonce, + as_of_height: 0, + } + } + + fn funds_at(balance: u64, nonce: u32, as_of_height: u64) -> AddressFunds { + AddressFunds { + balance, + nonce, + as_of_height, + } } /// Regression for the top-up reconciliation: a spent platform address @@ -1215,6 +1210,7 @@ mod tests { WALLET, |p2pkh| bimap.get_by_right(p2pkh).map(|&idx| (ACCOUNT, idx)), &address_infos, + 42, ); assert_eq!( @@ -1270,6 +1266,7 @@ mod tests { WALLET, |p2pkh| bimap.get_by_right(p2pkh).map(|&idx| (ACCOUNT, idx)), &address_infos, + 42, ); assert_eq!(entries.len(), 1, "external recipient must be filtered out"); @@ -1343,6 +1340,54 @@ mod tests { .insert(addr); } + /// Stage `addr` as found with `f` in the in-sync scratch — the shape + /// `on_address_found` produces (without the wallet-manager write). + fn stage_found( + provider: &mut PlatformPaymentAddressProvider, + addr: PlatformP2PKHAddress, + f: AddressFunds, + ) { + provider + .per_wallet_in_sync + .entry(WALLET) + .or_default() + .entry(ACCOUNT) + .or_default() + .found + .insert(addr, f); + } + + /// `sync_finished`'s scratch merge applies height-pin freshness: a + /// pass that ran against a lagging node stages a stale-but-valid + /// absolute (older pin) for a row a fresher reconcile committed — + /// the committed entry must survive, mirroring the persist diff's + /// guard so memory and disk agree. A same-or-newer pin still lands. + #[tokio::test] + async fn sync_finished_keeps_fresher_pinned_committed_entry() { + let addr = p2pkh(1); + // Committed by the reconcile seam at the funding proof height. + let mut provider = + provider_with_one_funded_address(addr, funds_at(9_985_071_720, 0, 379_731)); + + // A lagging pass stages the pre-funding absolute at an older pin. + stage_found(&mut provider, addr, funds_at(0, 0, 379_728)); + provider.sync_finished().await; + + let seed: Vec<_> = provider.current_balances().collect(); + assert_eq!( + seed[0].2, + funds_at(9_985_071_720, 0, 379_731), + "a stale-pinned scratch entry must not clobber a fresher \ + committed row" + ); + + // A genuinely newer pass replaces it. + stage_found(&mut provider, addr, funds_at(5, 1, 379_740)); + provider.sync_finished().await; + let seed: Vec<_> = provider.current_balances().collect(); + assert_eq!(seed[0].2, funds_at(5, 1, 379_740)); + } + /// `sync_finished` must drop an address proven absent this pass from /// the committed `found` map so it stops seeding the next pass and /// `current_balances()` no longer yields it. This is the core of the @@ -1685,7 +1730,7 @@ mod tests { ); wallet - .reconcile_address_infos(&address_infos, &BTreeSet::new(), "test top-up") + .reconcile_address_infos(&address_infos, 42, "test top-up") .await; // The reconciliation must have PERSISTED the decremented entry — the @@ -1721,20 +1766,22 @@ mod tests { assert_eq!(seed.len(), 1); assert_eq!( seed[0].2, - funds(5, 4), - "committed found seed must carry the reconciled funds" + funds_at(5, 4, 42), + "committed found seed must carry the reconciled funds pinned \ + at the proof height" ); } - /// ADDR-09 watermark gate, credit side: when the seam COMMITS an entry - /// for an address the caller declared as a delta-credited output, it - /// must invalidate the incremental sync watermark INSIDE its critical - /// section (forcing the next BLAST pass to full-scan-reconcile instead - /// of re-applying the on-chain `AddToCredits` delta on top of the - /// just-committed absolute seed) — while KEEPING the reconciled `found` - /// seed visible for display / input budgeting. + /// ADDR-09, credit side: a committed credit is pinned at the proof + /// height (`AddressFunds::as_of_height`), which is what stops the + /// sync's delta replay from re-applying the on-chain `AddToCredits` + /// on top of the just-committed absolute — so the seam no longer + /// needs to invalidate the incremental watermark (the old gate, + /// which forced a full rescan yet could not protect the rescan + /// itself from the same replay). The fast incremental cadence is + /// preserved for every flow. #[tokio::test] - async fn reconcile_invalidates_watermark_when_credited_output_committed() { + async fn reconcile_pins_committed_credit_and_keeps_watermark() { use dash_sdk::query_types::AddressInfo; let recorder = Arc::new(CapturingPersister::default()); @@ -1761,41 +1808,45 @@ mod tests { balance: 600, }), ); - let credited_outputs: BTreeSet = [addr].into_iter().collect(); - let cs = wallet - .reconcile_address_infos(&address_infos, &credited_outputs, "test credit") + .reconcile_address_infos(&address_infos, 42, "test credit") .await; assert_eq!(cs.addresses.len(), 1, "the credit must be committed"); + assert_eq!( + cs.addresses[0].funds, + funds_at(600, 1, 42), + "the persisted entry must carry the proof-height pin — the \ + sync's replay gate reads it to drop the on-chain credit delta" + ); let guard = wallet.provider.read().await; let provider = guard.as_ref().expect("provider present"); assert_eq!( provider.last_sync_timestamp(), - None, - "committed credited output must zero the watermark so the next \ - pass takes the full-scan branch (the ADDR-09 gate)" + Some(20), + "a committed credit must keep the incremental watermark — the \ + pin, not a forced full rescan, is what prevents the ADDR-09 \ + double-count" ); - assert_eq!(provider.last_sync_height(), 0); - assert_eq!(provider.last_known_recent_block(), 0); - // The reconciled seed survives the invalidation. + assert_eq!(provider.last_sync_height(), 10); + assert_eq!(provider.last_known_recent_block(), 30); let seed: Vec<_> = provider.current_balances().collect(); assert_eq!(seed.len(), 1); assert_eq!( seed[0].2, - funds(600, 1), - "invalidation must not drop the just-reconciled found seed" + funds_at(600, 1, 42), + "the committed found seed must carry the pinned funds" ); } - /// ADDR-09 watermark gate, drain side: a reconciliation that only - /// touches INPUT addresses (empty `credited_outputs` — e.g. an - /// external-recipient transfer or a withdrawal) must PRESERVE the - /// incremental watermark. Inputs are recorded on-chain as absolute - /// `SetBalanceToAddress` ops, idempotent under incremental re-apply, - /// so forcing a full scan would only burn the fast cadence. + /// Drain side: an input-only reconciliation (e.g. an + /// external-recipient transfer or a withdrawal) keeps the incremental + /// watermark, exactly as before the pin existed. Inputs are recorded + /// on-chain as absolute `SetBalanceToAddress` ops; the pin makes them + /// (and any future change output) replay-safe without touching the + /// sync cadence. #[tokio::test] - async fn reconcile_keeps_watermark_without_credited_outputs() { + async fn reconcile_keeps_watermark_on_input_only_drain() { use dash_sdk::query_types::AddressInfo; let recorder = Arc::new(CapturingPersister::default()); @@ -1821,7 +1872,7 @@ mod tests { ); let cs = wallet - .reconcile_address_infos(&address_infos, &BTreeSet::new(), "test drain") + .reconcile_address_infos(&address_infos, 42, "test drain") .await; assert_eq!(cs.addresses.len(), 1, "the drain must be committed"); @@ -1836,23 +1887,27 @@ mod tests { assert_eq!(provider.last_known_recent_block(), 30); } - /// ADDR-09 watermark gate keys on entries actually COMMITTED, not - /// merely requested: a credited output whose proof entry matches the - /// committed seed exactly (`unchanged_skipped` — a background sync - /// already applied this credit and advanced the watermark past it) - /// must NOT trigger invalidation. + /// No-op skip: a proof entry identical to the committed seed — + /// including the pin — is dropped (`unchanged_skipped`) instead of + /// re-committed, avoiding persister churn when a background sync + /// already applied this credit at the same height. #[tokio::test] - async fn reconcile_keeps_watermark_when_credited_output_not_committed() { + async fn reconcile_skips_entry_identical_to_committed_seed() { use dash_sdk::query_types::AddressInfo; let recorder = Arc::new(CapturingPersister::default()); let (wallet, wallet_manager) = reconcile_seam_wallet(recorder).await; - // Seed already carries the post-credit state (600, nonce 1) — the + // Seed already carries the post-credit state (600, nonce 1) + // pinned at the same height this reconcile will use — the // background sync applied the credit before this reconcile ran. let addr = p2pkh(0x11); - let mut provider = - provider_tracking_address(Arc::clone(&wallet_manager), WALLET, addr, funds(600, 1)); + let mut provider = provider_tracking_address( + Arc::clone(&wallet_manager), + WALLET, + addr, + funds_at(600, 1, 42), + ); provider.set_stored_sync_state(10, 20, 30); *wallet.provider.write().await = Some(provider); @@ -1866,10 +1921,8 @@ mod tests { balance: 600, }), ); - let credited_outputs: BTreeSet = [addr].into_iter().collect(); - let cs = wallet - .reconcile_address_infos(&address_infos, &credited_outputs, "test unchanged") + .reconcile_address_infos(&address_infos, 42, "test unchanged") .await; assert!( cs.addresses.is_empty(), @@ -1881,8 +1934,7 @@ mod tests { assert_eq!( provider.last_sync_timestamp(), Some(20), - "a credit the sync already applied (and advanced the watermark \ - past) must not force a full rescan" + "a no-op reconcile must leave the sync watermark untouched" ); } @@ -1909,7 +1961,7 @@ mod tests { }), ); - let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &BTreeMap::new()); + let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &BTreeMap::new(), 0); assert_eq!(outcome.resolved, 1); assert_eq!(outcome.stale_skipped, 1); @@ -1958,12 +2010,12 @@ mod tests { // Fully-consumed input: Drive elides the info → removal entry. address_infos.insert(removed_addr, None); - let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &pool_indexes); + let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &pool_indexes, 42); // The removal is emitted so the durable persister writes the zero. assert_eq!(outcome.entries.len(), 1, "removal survives the guard"); assert_eq!(outcome.entries[0].address, removed); - assert_eq!(outcome.entries[0].funds, funds(0, 0)); + assert_eq!(outcome.entries[0].funds, funds_at(0, 0, 42)); let state = provider .per_wallet @@ -2020,7 +2072,7 @@ mod tests { }), ); - let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &pool_indexes); + let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &pool_indexes, 42); assert!( outcome.entries.is_empty(), @@ -2162,7 +2214,7 @@ mod tests { }), ); - let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &BTreeMap::new()); + let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &BTreeMap::new(), 0); assert_eq!(outcome.resolved, 1); assert_eq!(outcome.unchanged_skipped, 1); @@ -2189,7 +2241,7 @@ mod tests { }), ); - let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &BTreeMap::new()); + let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &BTreeMap::new(), 0); assert_eq!(outcome.entries.len(), 1); assert_eq!(outcome.entries[0].funds, funds(50, 4)); @@ -2200,9 +2252,10 @@ mod tests { } /// Zero funds (balance 0, nonce 0) means the address was removed from - /// Platform state (fully consumed input). That removal is authoritative: - /// it bypasses the nonce guard and drops the address from the committed - /// `found` seed, mirroring the sync's absent handling. + /// Platform state (fully consumed input). Pinned at a height above the + /// committed seed's, the removal wins by height authority (nonces + /// cannot order a "gone" state) and drops the address from the + /// committed `found` seed, mirroring the sync's absent handling. #[test] fn commit_reconciliation_zero_funds_removes_from_found() { let addr = p2pkh(0x11); @@ -2213,10 +2266,10 @@ mod tests { // Drive elides the info for a fully consumed input. address_infos.insert(consumed, None); - let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &BTreeMap::new()); + let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &BTreeMap::new(), 42); assert_eq!(outcome.entries.len(), 1); - assert_eq!(outcome.entries[0].funds, funds(0, 0)); + assert_eq!(outcome.entries[0].funds, funds_at(0, 0, 42)); assert!( provider.current_balances().next().is_none(), "consumed address must be dropped from the found seed" @@ -2252,7 +2305,7 @@ mod tests { }), ); - let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &pool_indexes); + let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &pool_indexes, 42); assert_eq!( outcome.resolved, 0, @@ -2261,6 +2314,78 @@ mod tests { assert!(outcome.entries.is_empty()); } + /// Height authority: an absolute pinned at a LATER height is + /// authoritative even when it revises the balance DOWNWARD — this is + /// the ADDR-09 healing property. A poisoned legacy row (e.g. a + /// double-counted balance persisted before the pin existed, pin 0) + /// must yield to a proof-attested absolute at any real height. + #[test] + fn commit_reconciliation_later_pin_revises_balance_downward() { + use dash_sdk::query_types::AddressInfo; + + let addr = p2pkh(0x11); + // Poisoned pre-pin seed: 2X with "unknown provenance" (pin 0). + let mut provider = provider_with_one_funded_address(addr, funds(19_970_143_440, 0)); + + let credited = PlatformAddress::P2pkh([0x11; 20]); + let mut address_infos = AddressInfos::new(); + address_infos.insert( + credited, + Some(AddressInfo { + address: credited, + nonce: 0, + balance: 9_985_071_720, + }), + ); + + let outcome = + provider.commit_reconciliation(&WALLET, &address_infos, &BTreeMap::new(), 379_395); + + assert_eq!(outcome.entries.len(), 1, "the downward revision commits"); + assert_eq!(outcome.stale_skipped, 0); + let seed: Vec<_> = provider.current_balances().collect(); + assert_eq!( + seed[0].2, + funds_at(9_985_071_720, 0, 379_395), + "the later-pinned single-counted absolute replaces the 2x row" + ); + } + + /// Height authority, stale side: an absolute pinned BELOW the + /// committed seed's pin is stale — a sync pass or later transition + /// already committed state attested at a later block — and must be + /// dropped even though its nonce is not below the seed's (nonces + /// cannot order receive-only states across blocks). + #[test] + fn commit_reconciliation_drops_stale_height() { + use dash_sdk::query_types::AddressInfo; + + let addr = p2pkh(0x11); + let mut provider = provider_with_one_funded_address(addr, funds_at(1_000, 0, 100)); + + let credited = PlatformAddress::P2pkh([0x11; 20]); + let mut address_infos = AddressInfos::new(); + address_infos.insert( + credited, + Some(AddressInfo { + address: credited, + nonce: 0, + balance: 600, + }), + ); + + let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &BTreeMap::new(), 50); + + assert_eq!(outcome.stale_skipped, 1, "older-pinned absolute is stale"); + assert!(outcome.entries.is_empty()); + let seed: Vec<_> = provider.current_balances().collect(); + assert_eq!( + seed[0].2, + funds_at(1_000, 0, 100), + "the fresher committed funds survive" + ); + } + /// An address missing from the provider bijection (derived since the /// last sync, e.g. a fresh change address) resolves through the /// live-pool fallback, and the pair is merged into the bijection so @@ -2288,7 +2413,7 @@ mod tests { }), ); - let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &pool_indexes); + let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &pool_indexes, 42); assert_eq!(outcome.entries.len(), 1); assert_eq!( @@ -2303,7 +2428,7 @@ mod tests { .find(|(_, a, _)| *a == fresh) .expect("fresh address must appear in the found seed"); assert_eq!(fresh_row.0, (WALLET, ACCOUNT, 9)); - assert_eq!(fresh_row.2, funds(1_234, 0)); + assert_eq!(fresh_row.2, funds_at(1_234, 0, 42)); } /// `reset_sync_state` must zero the incremental watermark AND drop @@ -2339,49 +2464,4 @@ mod tests { "reset must drop the cached `found` seed" ); } - - /// ADDR-09: after an asset-lock top-up reconciles an absolute balance, - /// the fund path calls `invalidate_sync_watermark` to force the next - /// BLAST pass into full-scan mode. Unlike `reset_sync_state`, it must - /// zero all three watermark scalars (so `last_sync_timestamp()` returns - /// `None`, the full-scan trigger) WITHOUT dropping the freshly - /// reconciled `found` seed — display and input budgeting rely on the - /// balance staying visible until the reconciling scan completes. - #[tokio::test] - async fn invalidate_sync_watermark_forces_full_scan_keeps_seed() { - let addr = p2pkh(1); - let mut provider = provider_with_one_funded_address(addr, funds(294_627_247_940, 5)); - - // Simulate a wallet mid-incremental-sync: non-zero watermark and a - // populated balance seed (the just-reconciled top-up balance `X`). - provider.set_stored_sync_state(10, 20, 30); - assert_eq!(provider.last_sync_height(), 10); - assert_eq!(provider.last_sync_timestamp(), Some(20)); - assert_eq!(provider.last_known_recent_block(), 30); - assert_eq!(provider.current_balances().count(), 1); - - provider.invalidate_sync_watermark(); - - // Watermark fully zeroed → SDK drops back to full-scan mode. - assert_eq!(provider.last_sync_height(), 0); - assert_eq!( - provider.last_sync_timestamp(), - None, - "invalidated watermark must report no last-sync timestamp so the \ - next pass takes the full-scan branch (the ADDR-09 fix)" - ); - assert_eq!(provider.last_known_recent_block(), 0); - - // The reconciled `found` seed SURVIVES — a full scan ignores it as - // a seed, but keeping it means the balance `X` stays visible for - // display / input budgeting during the ~15s until the scan runs. - let seed: Vec<_> = provider.current_balances().collect(); - assert_eq!( - seed.len(), - 1, - "invalidate_sync_watermark must NOT drop the cached `found` seed" - ); - assert_eq!(seed[0].1, addr); - assert_eq!(seed[0].2, funds(294_627_247_940, 5)); - } } diff --git a/packages/rs-platform-wallet/src/wallet/platform_addresses/sync.rs b/packages/rs-platform-wallet/src/wallet/platform_addresses/sync.rs index b332e351898..b2ead28309e 100644 --- a/packages/rs-platform-wallet/src/wallet/platform_addresses/sync.rs +++ b/packages/rs-platform-wallet/src/wallet/platform_addresses/sync.rs @@ -41,13 +41,22 @@ pub(crate) fn compute_address_balance_diff( before: &BTreeMap, found: &BTreeMap<(PlatformAddressTag, PlatformP2PKHAddress), AddressFunds>, absent: &BTreeSet<(PlatformAddressTag, PlatformP2PKHAddress)>, + absent_as_of_height: u64, ) -> Vec { let mut entries = Vec::new(); // 1. Found-and-changed. for (&(tag, p2pkh), &funds) in found { - if before.get(&tag) == Some(&funds) { - continue; + if let Some(existing) = before.get(&tag) { + // Skip when unchanged, or when the scanned pin is OLDER than + // the committed pin — a stale-but-valid proof from a lagging + // node must not clobber a row a fresher reconcile committed + // (mirrors `commit_reconciliation`'s height-freshness rule, + // and `sync_finished`'s scratch merge applies the same guard + // so memory and disk stay in agreement). + if existing == &funds || existing.as_of_height > funds.as_of_height { + continue; + } } let (wallet_id, account_index, address_index) = tag; entries.push(PlatformAddressBalanceEntry { @@ -70,10 +79,17 @@ pub(crate) fn compute_address_balance_diff( // Only emit when we actually had non-default cached funds for the // address. An address that was already empty (or never cached) // doesn't need a zeroing write. - let had_funds = before - .get(&tag) - .is_some_and(|f| f.balance != 0 || f.nonce != 0); - if !had_funds { + let Some(existing) = before.get(&tag) else { + continue; + }; + if existing.balance == 0 && existing.nonce == 0 { + continue; + } + // A stale absence proof must not zero a row pinned by a fresher + // proof or transition reconcile: on a lagging node an address + // funded seconds ago legitimately does not exist yet at the scan + // checkpoint, but the committed row is newer truth. + if existing.as_of_height > absent_as_of_height { continue; } let (wallet_id, account_index, address_index) = tag; @@ -82,10 +98,14 @@ pub(crate) fn compute_address_balance_diff( account_index, address_index, address: p2pkh, - // Absent in state ⇒ no balance and no nonce. Reset both. + // Absent in state ⇒ no balance and no nonce. Reset both. The + // absence was proven by the tree scan, so the zero is pinned + // at the scan's checkpoint height (`absent_as_of_height`) — + // a later re-credit carries a higher pin and wins. funds: AddressFunds { balance: 0, nonce: 0, + as_of_height: absent_as_of_height, }, }); } @@ -139,7 +159,14 @@ impl PlatformAddressWallet { // proven absent this pass that previously carried cached funds. // The latter is what zeroes a stale balance after a chain reset. let mut cs = PlatformAddressChangeSet { - addresses: compute_address_balance_diff(&before, &result.found, &result.absent), + addresses: compute_address_balance_diff( + &before, + &result.found, + &result.absent, + // Absence is only ever proven by the trunk/branch scan, + // so its checkpoint height pins the zeroing entries. + result.checkpoint_height, + ), ..Default::default() }; if result.new_sync_height > 0 { @@ -186,7 +213,11 @@ mod tests { } fn funds(balance: u64, nonce: u32) -> AddressFunds { - AddressFunds { balance, nonce } + AddressFunds { + balance, + nonce, + as_of_height: 0, + } } /// An address that previously carried a cached balance and is proven @@ -202,7 +233,7 @@ mod tests { let mut absent = BTreeSet::new(); absent.insert((tag(0), p2pkh(1))); - let entries = compute_address_balance_diff(&before, &found, &absent); + let entries = compute_address_balance_diff(&before, &found, &absent, 0); assert_eq!(entries.len(), 1); let entry = &entries[0]; @@ -224,7 +255,7 @@ mod tests { let mut absent = BTreeSet::new(); absent.insert((tag(0), p2pkh(1))); - let entries = compute_address_balance_diff(&before, &found, &absent); + let entries = compute_address_balance_diff(&before, &found, &absent, 0); assert!(entries.is_empty()); } @@ -241,7 +272,7 @@ mod tests { let mut absent = BTreeSet::new(); absent.insert((tag(0), p2pkh(1))); - let entries = compute_address_balance_diff(&before, &found, &absent); + let entries = compute_address_balance_diff(&before, &found, &absent, 0); assert_eq!(entries.len(), 1); assert_eq!(entries[0].funds.balance, 0); assert_eq!(entries[0].funds.nonce, 0); @@ -258,7 +289,7 @@ mod tests { let mut absent = BTreeSet::new(); absent.insert((tag(0), p2pkh(1))); - let entries = compute_address_balance_diff(&before, &found, &absent); + let entries = compute_address_balance_diff(&before, &found, &absent, 0); assert!(entries.is_empty()); } @@ -276,7 +307,7 @@ mod tests { let absent = BTreeSet::new(); - let entries = compute_address_balance_diff(&before, &found, &absent); + let entries = compute_address_balance_diff(&before, &found, &absent, 0); assert_eq!(entries.len(), 1); assert_eq!(entries[0].address_index, 0); assert_eq!(entries[0].address, p2pkh(10)); @@ -300,7 +331,7 @@ mod tests { let mut absent = BTreeSet::new(); absent.insert((tag(0), p2pkh(1))); - let entries = compute_address_balance_diff(&before, &found, &absent); + let entries = compute_address_balance_diff(&before, &found, &absent, 0); assert_eq!(entries.len(), 1); assert_eq!(entries[0].address, p2pkh(1)); @@ -322,7 +353,7 @@ mod tests { let mut absent = BTreeSet::new(); absent.insert((tag(1), p2pkh(20))); - let mut entries = compute_address_balance_diff(&before, &found, &absent); + let mut entries = compute_address_balance_diff(&before, &found, &absent, 0); entries.sort_by_key(|e| e.address_index); assert_eq!(entries.len(), 2); @@ -336,4 +367,82 @@ mod tests { assert_eq!(entries[1].address, p2pkh(20)); assert_eq!(entries[1].funds, funds(0, 0)); } + + /// A stale-but-valid scan result (older pin) must not be emitted over + /// a row a fresher reconcile committed: on a lagging node the scanned + /// absolute predates the ST-attested balance, and persisting it would + /// regress the durable row until the next pass self-heals. + #[test] + fn found_with_older_pin_is_not_emitted() { + let mut before = BTreeMap::new(); + // Committed by the reconcile seam at the funding proof height. + before.insert( + tag(0), + AddressFunds { + balance: 9_985_071_720, + nonce: 0, + as_of_height: 379_731, + }, + ); + let mut found = BTreeMap::new(); + // Scan against a lagging node: pre-funding value at an older pin. + found.insert( + (tag(0), p2pkh(0)), + AddressFunds { + balance: 0, + nonce: 0, + as_of_height: 379_728, + }, + ); + let entries = compute_address_balance_diff(&before, &found, &BTreeSet::new(), 379_728); + assert!( + entries.is_empty(), + "a stale-pinned scan absolute must not clobber a fresher row" + ); + + // A genuinely newer scan (same or later pin) still emits. + found.insert( + (tag(0), p2pkh(0)), + AddressFunds { + balance: 5, + nonce: 1, + as_of_height: 379_740, + }, + ); + let entries = compute_address_balance_diff(&before, &found, &BTreeSet::new(), 379_740); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].funds.balance, 5); + } + + /// A stale absence proof (scan checkpoint below the committed pin) + /// must not zero a row a fresher reconcile committed — on a lagging + /// node an address funded seconds ago legitimately does not exist at + /// the scan checkpoint. + #[test] + fn absent_with_older_checkpoint_is_not_emitted() { + let mut before = BTreeMap::new(); + before.insert( + tag(0), + AddressFunds { + balance: 9_985_071_720, + nonce: 0, + as_of_height: 379_731, + }, + ); + let mut absent = BTreeSet::new(); + absent.insert((tag(0), p2pkh(0))); + + // Lagging checkpoint: no zeroing entry. + let entries = compute_address_balance_diff(&before, &BTreeMap::new(), &absent, 379_728); + assert!( + entries.is_empty(), + "a stale absence proof must not zero a fresher-pinned row" + ); + + // A checkpoint at/after the pin still zeroes (a real drain). + let entries = compute_address_balance_diff(&before, &BTreeMap::new(), &absent, 379_731); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].funds.balance, 0); + assert_eq!(entries[0].funds.as_of_height, 379_731); + } } diff --git a/packages/rs-platform-wallet/src/wallet/platform_addresses/transfer.rs b/packages/rs-platform-wallet/src/wallet/platform_addresses/transfer.rs index f04d17e3f11..a62fa1ca045 100644 --- a/packages/rs-platform-wallet/src/wallet/platform_addresses/transfer.rs +++ b/packages/rs-platform-wallet/src/wallet/platform_addresses/transfer.rs @@ -212,15 +212,13 @@ impl PlatformAddressWallet { // gate and the spend path diverge on a non-latest-pinned SDK. let version = platform_version.unwrap_or_else(|| self.sdk.version()); - // Capture the credited output addresses BEFORE `outputs` is moved - // into the SDK call below. Transfer outputs are recorded on-chain - // as `AddBalanceToAddress` DELTAS (a same-wallet output — e.g. a - // change address — is the ADDR-09 double-count shape); the seam's - // watermark gate needs to know them. See - // `reconcile_address_infos` for the full mechanism. - let credited_outputs = super::credited_outputs_set(outputs.keys()); - - let address_infos = match input_selection { + // `proof_height` is the broadcast proof's committed block — the + // height pin for the reconciled absolutes below. Transfer outputs + // are recorded on-chain as `AddBalanceToAddress` DELTAS at that + // height (a same-wallet output — e.g. a change address — is the + // ADDR-09 double-count shape); the pin is what stops the sync from + // re-applying them. See `reconcile_address_infos`. + let (address_infos, proof_height) = match input_selection { InputSelection::Explicit(inputs) => { if inputs.is_empty() { return Err(PlatformWalletError::AddressOperation( @@ -273,13 +271,12 @@ impl PlatformAddressWallet { // `transfer_address_funds` returns address info for the full // `inputs ∪ outputs` set, including external recipients the wallet // does not own — the shared seam filters those out, applies the - // proof-attested balances, updates the sync seed, persists, and - // (when a wallet-owned output in `credited_outputs` was committed) - // invalidates the incremental sync watermark so the next BLAST - // pass full-scan-reconciles instead of re-applying the on-chain - // credit delta on top of the absolute seed (ADDR-09). + // proof-attested balances pinned at `proof_height`, updates the + // sync seed, and persists. The pin lets the next BLAST pass drop + // the on-chain credit delta instead of re-applying it on top of + // the absolute seed (ADDR-09). Ok(self - .reconcile_address_infos(&address_infos, &credited_outputs, "address transfer") + .reconcile_address_infos(&address_infos, proof_height, "address transfer") .await) } diff --git a/packages/rs-platform-wallet/src/wallet/platform_addresses/wallet.rs b/packages/rs-platform-wallet/src/wallet/platform_addresses/wallet.rs index f2d0fae4bf8..efb88fcee7c 100644 --- a/packages/rs-platform-wallet/src/wallet/platform_addresses/wallet.rs +++ b/packages/rs-platform-wallet/src/wallet/platform_addresses/wallet.rs @@ -1,6 +1,6 @@ //! Platform address wallet for DIP-17 platform payment addresses. -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::BTreeMap; use std::sync::Arc; use dpp::address_funds::PlatformAddress; @@ -189,9 +189,10 @@ impl PlatformAddressWallet { /// balances". /// /// A freshness guard protects against racing the 15s background sync: - /// entries whose nonce is below the committed seed's are dropped (a - /// fresher state was already committed), see - /// [`PlatformPaymentAddressProvider::commit_reconciliation`]. The + /// height-pin authority (see `AddressFunds::as_of_height`) — entries + /// whose pin is below the committed seed's are dropped (a fresher + /// absolute was already committed; the nonce breaks same-block ties), + /// see [`PlatformPaymentAddressProvider::commit_reconciliation`]. The /// provider write lock is held across the provider commit, the /// account-balance write, AND the persist — mirroring /// [`sync_balances`](Self::sync_balances), so the two writers' stores @@ -199,50 +200,33 @@ impl PlatformAddressWallet { /// interleave between (or persist across) the seam's steps; the lock /// order (provider → wallet manager) matches the sync callbacks. /// - /// # `credited_outputs` — the ADDR-09 watermark gate + /// # `as_of_height` — the height pin /// - /// `credited_outputs` names the addresses the transition credited via - /// an on-chain `AddBalanceToAddress` op — a DELTA (`AddToCredits`) in - /// Drive's recent-address-balance-changes tree — as opposed to inputs, - /// which are recorded as absolute `SetBalanceToAddress` ops. Build it - /// with [`super::credited_outputs_set`]; pass an empty set when the - /// transition credits nothing the wallet could own (e.g. withdrawal, - /// which never has a change output). - /// - /// When the seam commits an entry for one of these addresses, it has - /// just written an optimistic ABSOLUTE balance into the sync seed - /// while the on-chain record of the same credit is a delta. An - /// incremental next sync would seed from `current_balances()` (already - /// the absolute `X`) and re-apply the recent delta on top → `X + delta` - /// — the ADDR-09 double-count. So the seam invalidates the provider's - /// incremental sync watermark ([`PlatformPaymentAddressProvider::invalidate_sync_watermark`]) - /// INSIDE this same critical section, forcing the next pass to - /// full-scan-reconcile (absolute re-seed from the tree). Doing it - /// under the lock — rather than caller-side after the seam returns — - /// means no sync pass can slip in between the seed commit and the - /// invalidation with the stale watermark. Input-only reconciliations - /// (empty or non-matching `credited_outputs`) keep the fast - /// incremental cadence: re-applying an absolute op is idempotent. - /// - /// The gate keys on entries actually COMMITTED, not merely requested: - /// an entry skipped as stale/unchanged means a sync already applied - /// this credit (and advanced the watermark past it), so no forced - /// full scan is needed. + /// `as_of_height` is the block height of the proof that attested + /// `address_infos` (the broadcast result's quorum-authenticated + /// `metadata.height`). Every committed entry carries it as its + /// balance pin, which is what makes the optimistic absolute write + /// safe against delta replay: the transition's on-chain + /// `AddBalanceToAddress` credit is recorded as a DELTA + /// (`AddToCredits`) at this same height in Drive's + /// recent-address-balance-changes tree, and the sync's apply loops + /// drop any delta at or below an entry's pin. This replaces the + /// former watermark-invalidation gate (`credited_outputs`), which + /// forced a full rescan but could not stop the rescan itself from + /// replaying the same delta on top of the fresh absolute. /// /// Persistence errors are logged rather than propagated — Platform /// already accepted the transition, and a later sync reconciles. /// /// [`PlatformPaymentAddressProvider::commit_reconciliation`]: /// super::provider::PlatformPaymentAddressProvider::commit_reconciliation - /// [`PlatformPaymentAddressProvider::invalidate_sync_watermark`]: - /// super::provider::PlatformPaymentAddressProvider::invalidate_sync_watermark pub async fn reconcile_address_infos( &self, address_infos: &AddressInfos, - credited_outputs: &BTreeSet, + as_of_height: u64, context: &'static str, ) -> crate::PlatformAddressChangeSet { - self.reconcile_address_infos_with_persistence(address_infos, credited_outputs, context) + self.reconcile_address_infos_with_persistence(address_infos, as_of_height, context) .await .0 } @@ -264,7 +248,7 @@ impl PlatformAddressWallet { pub(super) async fn reconcile_address_infos_with_persistence( &self, address_infos: &AddressInfos, - credited_outputs: &BTreeSet, + as_of_height: u64, context: &'static str, ) -> (crate::PlatformAddressChangeSet, bool) { if address_infos.is_empty() { @@ -319,7 +303,12 @@ impl PlatformAddressWallet { out }; - let outcome = provider.commit_reconciliation(&self.wallet_id, address_infos, &pool_indexes); + let outcome = provider.commit_reconciliation( + &self.wallet_id, + address_infos, + &pool_indexes, + as_of_height, + ); if outcome.resolved == 0 { tracing::warn!( @@ -395,20 +384,6 @@ impl PlatformAddressWallet { ..Default::default() }; - // ADDR-09 watermark gate (see the method docs): a committed entry - // for a delta-credited output means the absolute seed write above - // is inconsistent with incremental delta re-application — force - // the next pass to full-scan. Still under the provider write lock, - // so no sync pass can interleave between the seed commit and this - // invalidation with the stale watermark. - if cs - .addresses - .iter() - .any(|entry| credited_outputs.contains(&entry.address)) - { - provider.invalidate_sync_watermark(); - } - let persisted = match self.persister.store(cs.clone().into()) { Ok(()) => true, Err(e) => { diff --git a/packages/rs-platform-wallet/src/wallet/platform_addresses/withdrawal.rs b/packages/rs-platform-wallet/src/wallet/platform_addresses/withdrawal.rs index 34ef776bf81..f9780a7f79e 100644 --- a/packages/rs-platform-wallet/src/wallet/platform_addresses/withdrawal.rs +++ b/packages/rs-platform-wallet/src/wallet/platform_addresses/withdrawal.rs @@ -113,7 +113,9 @@ impl PlatformAddressWallet { // path diverge on a non-latest-pinned SDK. let version = platform_version.unwrap_or_else(|| self.sdk.version()); - let address_infos = match input_selection { + // `proof_height` is the broadcast proof's committed block — the + // height pin for the reconciled absolutes below. + let (address_infos, proof_height) = match input_selection { InputSelection::Explicit(inputs) => { if inputs.is_empty() { return Err(PlatformWalletError::AddressOperation( @@ -186,24 +188,18 @@ impl PlatformAddressWallet { }; // Apply + persist the proof-attested post-withdrawal balances via - // the shared seam. Input addresses resolve through the provider's - // persisted index bijection (with live-pool fallback), so a - // restored address that is no longer in a live derived pool keeps - // its real derivation index instead of corrupting index 0. + // the shared seam, pinned at `proof_height`. Input addresses + // resolve through the provider's persisted index bijection (with + // live-pool fallback), so a restored address that is no longer in + // a live derived pool keeps its real derivation index instead of + // corrupting index 0. // - // `credited_outputs` is empty (ADDR-09 watermark gate stays off): - // every owned address a withdrawal touches is an INPUT, recorded - // on-chain as `SetBalanceToAddress` (an absolute `SetCredits` op in - // the recent tree); re-applying an absolute op on the next - // incremental pass is idempotent, so there is no delta to - // double-count. The only op that *would* be a delta - // (`AddBalanceToAddress`) is the change output, and every - // `withdraw_address_funds` call above passes `None` for it — the - // account is drained in full with no change. If a future change - // output is ever wired up here, pass it via - // `super::credited_outputs_set` so the seam's gate covers it. + // Withdrawal inputs are recorded on-chain as `SetBalanceToAddress` + // (absolute `SetCredits` ops), which were already replay-safe; the + // pin additionally protects a future change output (an + // `AddBalanceToAddress` delta) without any caller-side bookkeeping. Ok(self - .reconcile_address_infos(&address_infos, &BTreeSet::new(), "address withdrawal") + .reconcile_address_infos(&address_infos, proof_height, "address withdrawal") .await) } diff --git a/packages/rs-platform-wallet/src/wallet/platform_wallet.rs b/packages/rs-platform-wallet/src/wallet/platform_wallet.rs index e64cb41d6cb..3b1e1ab2740 100644 --- a/packages/rs-platform-wallet/src/wallet/platform_wallet.rs +++ b/packages/rs-platform-wallet/src/wallet/platform_wallet.rs @@ -1,6 +1,6 @@ //! The main PlatformWallet struct combining core, identity (+DashPay), and platform sub-wallets. -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::BTreeMap; use std::ops::{Deref, DerefMut}; use std::sync::Arc; @@ -254,18 +254,15 @@ impl PlatformWallet { address_signer: &S, settings: Option, ) -> Result { - let (address_infos, new_balance) = self + let (address_infos, new_balance, proof_height) = self .identity .top_up_from_addresses(identity_id, inputs, address_signer, settings) .await?; - // `credited_outputs` is empty: an identity top-up only DRAINS - // platform addresses (absolute `SetBalanceToAddress` ops, - // idempotent under incremental re-apply) — it has no change - // output. If one is ever added, pass it via - // `platform_addresses::credited_outputs_set` so the seam's - // ADDR-09 watermark gate covers it. + // The reconciled absolutes are pinned at the proof's block height + // (`AddressFunds::as_of_height`), so the sync's delta replay can + // never re-apply this transition's on-chain ops on top of them. self.platform - .reconcile_address_infos(&address_infos, &BTreeSet::new(), "identity top-up") + .reconcile_address_infos(&address_infos, proof_height, "identity top-up") .await; Ok(new_balance) } @@ -298,14 +295,11 @@ impl PlatformWallet { AS: Signer + Send + Sync, { // The optional refund-style `output` is credited on-chain via an - // `AddBalanceToAddress` DELTA; when it lands on a wallet-owned - // address, the seam's ADDR-09 watermark gate must know about it or - // the next incremental BLAST pass would re-apply the delta on top - // of the reconciled absolute balance (double-count). Capture it - // before `output` moves into the identity call. - let credited_outputs = - super::platform_addresses::credited_outputs_set(output.iter().map(|(addr, _)| addr)); - let (registered_identity, address_infos) = self + // `AddBalanceToAddress` DELTA at the proof's block height. The + // reconciled absolutes carry that height as their pin + // (`AddressFunds::as_of_height`), so the sync's delta replay drops + // the credit instead of re-applying it on top (ADDR-09). + let (registered_identity, address_infos, proof_height) = self .identity .register_from_addresses( identity, @@ -318,7 +312,7 @@ impl PlatformWallet { ) .await?; self.platform - .reconcile_address_infos(&address_infos, &credited_outputs, "identity registration") + .reconcile_address_infos(&address_infos, proof_height, "identity registration") .await; Ok(registered_identity) } @@ -345,15 +339,13 @@ impl PlatformWallet { S: Signer + Send + Sync, { // Every recipient is credited on-chain via an `AddBalanceToAddress` - // DELTA — and the primary use of this flow is consolidating identity - // credits into the wallet's OWN platform addresses, so the seam's - // ADDR-09 watermark gate must know the recipient set or the next - // incremental BLAST pass would re-apply the delta on top of the - // reconciled absolute balance (double-count). Capture it before - // `recipient_addresses` moves into the identity call. - let credited_outputs = - super::platform_addresses::credited_outputs_set(recipient_addresses.keys()); - let (address_infos, new_balance) = self + // DELTA at the proof's block height — and the primary use of this + // flow is consolidating identity credits into the wallet's OWN + // platform addresses. The reconciled absolutes carry that height + // as their pin (`AddressFunds::as_of_height`), so the sync's delta + // replay drops the credit instead of re-applying it on top + // (ADDR-09). + let (address_infos, new_balance, proof_height) = self .identity .transfer_credits_to_addresses_with_external_signer( identity_id, @@ -363,11 +355,7 @@ impl PlatformWallet { ) .await?; self.platform - .reconcile_address_infos( - &address_infos, - &credited_outputs, - "credit transfer to addresses", - ) + .reconcile_address_infos(&address_infos, proof_height, "credit transfer to addresses") .await; Ok(new_balance) } diff --git a/packages/rs-sdk-ffi/src/address/transitions/top_up_from_asset_lock.rs b/packages/rs-sdk-ffi/src/address/transitions/top_up_from_asset_lock.rs index 5121206b77e..19d6dd1e6f7 100644 --- a/packages/rs-sdk-ffi/src/address/transitions/top_up_from_asset_lock.rs +++ b/packages/rs-sdk-ffi/src/address/transitions/top_up_from_asset_lock.rs @@ -271,6 +271,10 @@ unsafe fn dash_sdk_address_top_up_from_asset_lock_inner( .map_err(FFIError::from)?; // Convert to FFI type (same as transfer) + let (address_infos, _proof_height) = address_infos; + // The `_proof_height` pin matters to callers that PERSIST these + // absolutes (the platform-wallet reconcile seam); this raw debug + // path only renders the proof entries, so it drops the height. let entries: Vec = address_infos .iter() .map(|(address, info_opt)| { diff --git a/packages/rs-sdk-ffi/src/address/transitions/transfer.rs b/packages/rs-sdk-ffi/src/address/transitions/transfer.rs index 3922e5a067e..7e8523710eb 100644 --- a/packages/rs-sdk-ffi/src/address/transitions/transfer.rs +++ b/packages/rs-sdk-ffi/src/address/transitions/transfer.rs @@ -292,6 +292,10 @@ unsafe fn dash_sdk_address_transfer_funds_inner( .map_err(FFIError::from)?; // Convert to FFI type + let (address_infos, _proof_height) = address_infos; + // The `_proof_height` pin matters to callers that PERSIST these + // absolutes (the platform-wallet reconcile seam); this raw debug + // path only renders the proof entries, so it drops the height. let entries: Vec = address_infos .iter() .map(|(address, info_opt)| { diff --git a/packages/rs-sdk-ffi/src/address/transitions/withdraw.rs b/packages/rs-sdk-ffi/src/address/transitions/withdraw.rs index 418eddd5de7..54f7d590fb3 100644 --- a/packages/rs-sdk-ffi/src/address/transitions/withdraw.rs +++ b/packages/rs-sdk-ffi/src/address/transitions/withdraw.rs @@ -278,6 +278,10 @@ unsafe fn dash_sdk_address_withdraw_funds_inner( .map_err(FFIError::from)?; // Convert to FFI type (same as transfer) + let (address_infos, _proof_height) = address_infos; + // The `_proof_height` pin matters to callers that PERSIST these + // absolutes (the platform-wallet reconcile seam); this raw debug + // path only renders the proof entries, so it drops the height. let entries: Vec = address_infos .iter() .map(|(address, info_opt)| { diff --git a/packages/rs-sdk-ffi/src/address_sync/mod.rs b/packages/rs-sdk-ffi/src/address_sync/mod.rs index daf3af359b0..26aa3c3041f 100644 --- a/packages/rs-sdk-ffi/src/address_sync/mod.rs +++ b/packages/rs-sdk-ffi/src/address_sync/mod.rs @@ -463,6 +463,13 @@ pub unsafe extern "C" fn dash_sdk_sync_addresses_batch_with_result( AddressFunds { nonce: kb_nonces[i], balance: kb_amounts[i], + // This raw batch API predates the height pin + // and its C signature carries no per-address + // height; 0 = "unknown provenance", the + // pre-pin delta-replay semantics. The + // platform-wallet BLAST path (the production + // sync) round-trips real pins. + as_of_height: 0, }, )); } diff --git a/packages/rs-sdk-ffi/src/identity/create_from_addresses.rs b/packages/rs-sdk-ffi/src/identity/create_from_addresses.rs index f786bbb7143..c90e2e3dae3 100644 --- a/packages/rs-sdk-ffi/src/identity/create_from_addresses.rs +++ b/packages/rs-sdk-ffi/src/identity/create_from_addresses.rs @@ -233,7 +233,7 @@ unsafe fn dash_sdk_identity_create_from_addresses_inner( // Execute the creation let result: Result = wrapper.runtime.block_on(async { - let (created_identity, address_infos) = identity + let (created_identity, address_infos, _proof_height) = identity .put_with_address_funding( &wrapper.sdk, input_map, diff --git a/packages/rs-sdk-ffi/src/identity/top_up_from_addresses.rs b/packages/rs-sdk-ffi/src/identity/top_up_from_addresses.rs index fa606238511..c40346e102b 100644 --- a/packages/rs-sdk-ffi/src/identity/top_up_from_addresses.rs +++ b/packages/rs-sdk-ffi/src/identity/top_up_from_addresses.rs @@ -175,7 +175,7 @@ unsafe fn dash_sdk_identity_top_up_from_addresses_inner( // Execute the top-up let result: Result = wrapper.runtime.block_on(async { - let (address_infos, identity_balance) = identity + let (address_infos, identity_balance, _proof_height) = identity .top_up_from_addresses(&wrapper.sdk, input_map, &signer, settings) .await .map_err(FFIError::from)?; diff --git a/packages/rs-sdk-ffi/src/identity/transfer_to_addresses.rs b/packages/rs-sdk-ffi/src/identity/transfer_to_addresses.rs index 72feb19a450..4c9c193ac80 100644 --- a/packages/rs-sdk-ffi/src/identity/transfer_to_addresses.rs +++ b/packages/rs-sdk-ffi/src/identity/transfer_to_addresses.rs @@ -183,7 +183,7 @@ unsafe fn dash_sdk_identity_transfer_credits_to_addresses_inner( // Execute the transfer let result: Result = wrapper.runtime.block_on(async { - let (address_infos, identity_balance) = identity + let (address_infos, identity_balance, _proof_height) = identity .transfer_credits_to_addresses( &wrapper.sdk, recipient_map, diff --git a/packages/rs-sdk/src/platform/address_sync/mod.rs b/packages/rs-sdk/src/platform/address_sync/mod.rs index d2c6c24d8e2..faac4648773 100644 --- a/packages/rs-sdk/src/platform/address_sync/mod.rs +++ b/packages/rs-sdk/src/platform/address_sync/mod.rs @@ -61,7 +61,7 @@ use dapi_grpc::platform::v0::{ GetRecentAddressBalanceChangesRequest, GetRecentCompactedAddressBalanceChangesRequest, Proof, }; use dpp::address_funds::PlatformAddress; -use dpp::balances::credits::{BlockAwareCreditOperation, CreditOperation, Credits}; +use dpp::balances::credits::{BlockAwareCreditOperation, CreditOperation}; use dpp::prelude::AddressNonce; use dpp::version::PlatformVersion; use drive::drive::{Drive, RootTree}; @@ -162,7 +162,11 @@ impl TrunkBranchSyncOps for AddressOps

{ for (tag, address) in pending { let key_bytes = address.to_bytes(); if let Some(element) = trunk_result.elements.get(&key_bytes) { - let funds = AddressFunds::try_from(element)?; + // Pin the scan absolute at the snapshot height: the trunk + // proof attests this balance as of `checkpoint_height`, so + // any delta recorded at or below it is already included. + let mut funds = AddressFunds::try_from(element)?; + funds.as_of_height = context.result.checkpoint_height; context.result.found.insert((tag, address), funds); context .provider @@ -212,7 +216,10 @@ impl TrunkBranchSyncOps for AddressOps

{ }; if let Some(element) = branch_result.elements.get(&target_key) { - let funds = AddressFunds::try_from(element)?; + // Branch queries are checkpointed to the trunk query's + // height, so branch absolutes carry the same pin. + let mut funds = AddressFunds::try_from(element)?; + funds.as_of_height = context.result.checkpoint_height; context.result.found.insert((tag, address), funds); context .provider @@ -661,10 +668,14 @@ async fn incremental_catch_up( Err(e) => return Err(e), }; - let entries = match changes { + let mut entries = match changes { Some(c) => c.into_inner(), None => break, }; + // Apply in ascending range order so per-address pins advance + // monotonically — an out-of-order apply could gate off a + // genuinely newer delta. + entries.sort_by_key(|e| e.end_block_height); result.new_sync_timestamp = metadata.time_ms / 1000; result.metrics.compacted_queries += 1; @@ -687,7 +698,9 @@ async fn incremental_catch_up( .changes .iter() .map(|(a, op)| (a, BalanceOp::Compacted(op))), - current_height, + // The pin gate needs the height this change is recorded + // AS OF — the range end, not the pagination cursor. + entry.end_block_height, provider, result, &mut pending_unknown, @@ -714,9 +727,14 @@ async fn incremental_catch_up( let mut highest_recent_block: u64 = 0; if let Some(changes) = recent_changes { - let entries = changes.into_inner(); + let mut entries = changes.into_inner(); result.metrics.recent_entries_returned += entries.len(); + // Apply in ascending block order so per-address pins advance + // monotonically — an out-of-order apply could gate off a genuinely + // newer delta. + entries.sort_by_key(|e| e.block_height); + for entry in &entries { // Track the highest block height in recent entries if entry.block_height > highest_recent_block { @@ -729,7 +747,9 @@ async fn incremental_catch_up( .changes .iter() .map(|(a, op)| (a, BalanceOp::Recent(op))), - current_height, + // The pin gate needs the block this change is recorded AT, + // not the pagination cursor. + entry.block_height, provider, result, &mut pending_unknown, @@ -783,29 +803,68 @@ pub(crate) enum OwnedBalanceOp { Compacted(BlockAwareCreditOperation), } -/// A buffered miss: raw GroveDB key, owned change, and the catch-up cursor -/// height at the original block (feeds the compacted height filter on replay; -/// ignored by `Recent`). +/// A buffered miss: raw GroveDB key, owned change, and the height the change +/// is recorded as of (recent: the entry's block height; compacted: the range +/// end). Feeds the height-pin gate on replay exactly as on the forward pass. type PendingMiss = (Vec, OwnedBalanceOp, u64); -/// Resolve the post-change balance from the current balance and the catch-up -/// cursor height. Compacted sums only operations at or after `current_height`; -/// recent applies a flat set/add. -fn apply_op(op: BalanceOp<'_>, current_balance: Credits, current_height: u64) -> Credits { +/// Resolve the post-change funds from the current funds and the height the +/// change is recorded as of (`op_height` — recent: the entry's block height; +/// compacted: the range end). +/// +/// This is where the height pin (`AddressFunds::as_of_height`) gates delta +/// replay: the pinned balance already includes every block up to and +/// including the pin, so a change at or below it is already inside the +/// absolute — re-applying it is the ADDR-09 double-count (a fresh trunk/ST +/// absolute plus the same block's `AddToCredits` replayed on top). Returns +/// `current` unchanged when the change is fully gated off; applying advances +/// the pin so later passes gate correctly too. +fn apply_op(op: BalanceOp<'_>, current: AddressFunds, op_height: u64) -> AddressFunds { match op { - BalanceOp::Recent(op) => match op { - CreditOperation::SetCredits(credits) => *credits, - CreditOperation::AddToCredits(credits) => current_balance.saturating_add(*credits), - }, + BalanceOp::Recent(op) => { + if op_height <= current.as_of_height { + return current; + } + let balance = match op { + CreditOperation::SetCredits(credits) => *credits, + CreditOperation::AddToCredits(credits) => current.balance.saturating_add(*credits), + }; + AddressFunds { + nonce: current.nonce, + balance, + as_of_height: op_height, + } + } BalanceOp::Compacted(op) => match op { - BlockAwareCreditOperation::SetCredits(credits) => *credits, + BlockAwareCreditOperation::SetCredits(credits) => { + // Absolute as of the end of this compacted range — + // authoritative only if it postdates the pin. + if op_height <= current.as_of_height { + return current; + } + AddressFunds { + nonce: current.nonce, + balance: *credits, + as_of_height: op_height, + } + } BlockAwareCreditOperation::AddToCreditsOperations(operations) => { + // Each operation carries its own block height, so a pin that + // falls inside the compacted range drops exactly the ops the + // pinned absolute already includes and applies the rest. let total_to_add: u64 = operations .iter() - .filter(|(height, _)| **height >= current_height) + .filter(|(height, _)| **height > current.as_of_height) .map(|(_, credits)| *credits) .fold(0u64, |acc, c| acc.saturating_add(c)); - current_balance.saturating_add(total_to_add) + AddressFunds { + nonce: current.nonce, + balance: current.balance.saturating_add(total_to_add), + // The entry aggregates every change for this address + // through its range end, so the balance is now current + // through there. + as_of_height: current.as_of_height.max(op_height), + } } }, } @@ -835,30 +894,34 @@ fn apply_change( tag: P::Tag, address: P::Address, op: BalanceOp<'_>, - current_height: u64, + op_height: u64, ) -> Option { let result_key = (tag, address); - let current_balance = result - .found - .get(&result_key) - .map(|f| f.balance) - .unwrap_or(0); - - let new_balance = apply_op(op, current_balance, current_height); - if new_balance == current_balance { - return None; - } - // INTENTIONAL — accepted risk: incremental RPCs carry no nonce, so a // catch-up-discovered address synthesizes nonce=0. It is published and // persisted but NON-AUTHORITATIVE — every spend re-fetches the on-chain // nonce. Callers MUST NOT treat this as the authoritative nonce. // (Option rework deliberately skipped.) - let nonce = result.found.get(&result_key).map(|f| f.nonce).unwrap_or(0); - let funds = AddressFunds { - nonce, - balance: new_balance, - }; + // + // `as_of_height: 0` = "unknown provenance" for an address with no + // recorded funds: every change applies, and the first one pins it. + let current = result + .found + .get(&result_key) + .copied() + .unwrap_or(AddressFunds { + nonce: 0, + balance: 0, + as_of_height: 0, + }); + + let funds = apply_op(op, current, op_height); + // Commit on ANY funds change — a balance-neutral change that advances + // the pin still persists, hardening future replay gating. + if funds == current { + return None; + } + // Keep `found` and `absent` disjoint: a post-checkpoint funding can land // on an address the branch scan proved absent, so clear any stale // `absent` entry before recording it as found. @@ -876,7 +939,7 @@ fn apply_change( async fn apply_block_changes<'a, P, I>( address_lookup: &HashMap, (P::Tag, P::Address)>, changes: I, - current_height: u64, + op_height: u64, provider: &mut P, result: &mut AddressSyncResult, pending_unknown: &mut Vec, @@ -889,7 +952,7 @@ async fn apply_block_changes<'a, P, I>( for (platform_addr, change) in changes { let addr_bytes = platform_addr.to_bytes(); if let Some(&(tag, address)) = address_lookup.get(&addr_bytes) { - if let Some(funds) = apply_change::

(result, tag, address, change, current_height) { + if let Some(funds) = apply_change::

(result, tag, address, change, op_height) { local_applied.push((tag, address, funds)); } } else { @@ -897,7 +960,7 @@ async fn apply_block_changes<'a, P, I>( BalanceOp::Recent(op) => OwnedBalanceOp::Recent(*op), BalanceOp::Compacted(op) => OwnedBalanceOp::Compacted(op.clone()), }; - pending_unknown.push((addr_bytes, owned, current_height)); + pending_unknown.push((addr_bytes, owned, op_height)); // NOTE: this buffer is intentionally unbounded — premature // optimization here would couple the catch-up loop to ad-hoc // memory heuristics. We log a one-shot warning above a generous @@ -1237,6 +1300,10 @@ impl TryFrom<&Element> for AddressFunds { /// Convert a GroveDB element into address funds (nonce and balance). /// /// The address funds tree stores the nonce as the item value and the balance as the sum item. + /// + /// The element itself carries no block height, so the produced funds + /// have `as_of_height == 0`; the caller must pin them at the proof + /// height of the query that returned the element. fn try_from(element: &Element) -> Result { if let Element::ItemWithSumItem(nonce_bytes, balance, _) = element { let nonce_bytes: [u8; 4] = nonce_bytes.as_slice().try_into().map_err(|_| { @@ -1248,7 +1315,11 @@ impl TryFrom<&Element> for AddressFunds { let balance: u64 = (*balance).try_into().map_err(|_| { Error::InvalidProvedResponse("address funds balance must fit into u64".to_string()) })?; - return Ok(AddressFunds { nonce, balance }); + return Ok(AddressFunds { + nonce, + balance, + as_of_height: 0, + }); } Err(Error::InvalidProvedResponse( @@ -1718,6 +1789,7 @@ mod tests { AddressFunds { nonce: 0, balance: 500, + as_of_height: 0, }, ); result.found.insert( @@ -1725,6 +1797,7 @@ mod tests { AddressFunds { nonce: 1, balance: 0, + as_of_height: 0, }, ); result.found.insert( @@ -1732,6 +1805,7 @@ mod tests { AddressFunds { nonce: 2, balance: 1500, + as_of_height: 0, }, ); @@ -1829,7 +1903,9 @@ mod tests { apply_block_changes( &lookup, changes.iter().map(|(a, c)| (*a, *c)), - 0, + // A real block height — heights are never 0 in production, and + // the pin gate drops changes at or below the current pin. + 100, &mut provider, &mut result, &mut pending_unknown, @@ -1915,7 +1991,9 @@ mod tests { apply_block_changes( &lookup, changes.iter().map(|(a, c)| (*a, *c)), - 0, + // A real block height — heights are never 0 in production, and + // the pin gate drops changes at or below the current pin. + 100, &mut NoopProvider, &mut result, &mut pending_unknown, @@ -1991,13 +2069,14 @@ mod tests { AddressFunds { nonce: 0, balance: 1_000, + as_of_height: 0, }, ); let mut provider = GrowingProvider { late }; let known_op = BlockAwareCreditOperation::AddToCreditsOperations( - std::iter::once((0u64, 500u64)).collect(), + std::iter::once((50u64, 500u64)).collect(), ); let late_op = BlockAwareCreditOperation::SetCredits(7_000); let changes = [ @@ -2009,7 +2088,9 @@ mod tests { apply_block_changes( &lookup, changes.iter().map(|(a, c)| (*a, *c)), - 0, + // A real block height — heights are never 0 in production, and + // the pin gate drops changes at or below the current pin. + 100, &mut provider, &mut result, &mut pending_unknown, @@ -2112,7 +2193,9 @@ mod tests { apply_block_changes( &lookup, changes.iter().map(|(a, c)| (*a, *c)), - 0, + // A real block height — heights are never 0 in production, and + // the pin gate drops changes at or below the current pin. + 100, &mut provider, &mut result, &mut pending_unknown, @@ -2260,8 +2343,8 @@ mod tests { let op_a = BlockAwareCreditOperation::SetCredits(1_111); let op_b = BlockAwareCreditOperation::SetCredits(2_222); let pending_unknown: Vec = vec![ - (a.to_bytes(), OwnedBalanceOp::Compacted(op_a), 0), - (b.to_bytes(), OwnedBalanceOp::Compacted(op_b), 0), + (a.to_bytes(), OwnedBalanceOp::Compacted(op_a), 100), + (b.to_bytes(), OwnedBalanceOp::Compacted(op_b), 100), ]; refresh_and_replay_unknown(&lookup, pending_unknown, &mut provider, &mut result).await; @@ -2373,7 +2456,7 @@ mod tests { ( addrs[i].to_bytes(), OwnedBalanceOp::Compacted(BlockAwareCreditOperation::SetCredits(*credits)), - 0, + 100, ) }) .collect(); @@ -2499,14 +2582,14 @@ mod tests { OwnedBalanceOp::Compacted(BlockAwareCreditOperation::SetCredits( 1_000 + i as u64, )), - 0, + 100, ) }) .collect(); pending_unknown.push(( foreign.to_bytes(), OwnedBalanceOp::Compacted(BlockAwareCreditOperation::SetCredits(9_999)), - 0, + 100, )); let lookup: HashMap, (u32, PlatformAddress)> = HashMap::new(); @@ -2566,12 +2649,12 @@ mod tests { ( owned.to_bytes(), OwnedBalanceOp::Compacted(BlockAwareCreditOperation::SetCredits(5_000)), - 0, + 100, ), ( p2pkh(0x12).to_bytes(), OwnedBalanceOp::Compacted(BlockAwareCreditOperation::SetCredits(6_000)), - 0, + 100, ), ]; @@ -2657,6 +2740,7 @@ mod tests { AddressFunds { nonce: 0, balance: 1_000, + as_of_height: 0, }, ); @@ -2678,7 +2762,9 @@ mod tests { apply_block_changes( &lookup, changes.iter().map(|(a, c)| (*a, *c)), - 0, + // A real block height — heights are never 0 in production, and + // the pin gate drops changes at or below the current pin. + 100, &mut provider, &mut result, &mut pending_unknown, @@ -2765,19 +2851,21 @@ mod tests { lookup.insert(addr.to_bytes(), (1u32, addr)); let mut result: AddressSyncResult = AddressSyncResult::new(); + // The base balance is pinned at height 99: the scan absolute that + // produced it already includes every block through 99. result.found.insert( (1u32, addr), AddressFunds { nonce: 0, balance: 10_000, + as_of_height: 99, }, ); - // Cursor at height 100: ops at 98 and 99 are below (already - // accounted for by the tree scan, must be filtered out as a - // double-count guard); ops at 100 and 101 are at/above (must - // apply). Below sum = 700, at/above sum = 30. - let current_height = 100u64; + // Pin at height 99: ops at 98 and 99 are at/below the pin (already + // inside the pinned absolute, must be dropped as a double-count + // guard); ops at 100 and 101 postdate it (must apply). Dropped + // sum = 700, applied sum = 30. let op = BlockAwareCreditOperation::AddToCreditsOperations( [ (98u64, 300u64), @@ -2794,19 +2882,25 @@ mod tests { apply_block_changes( &lookup, changes.iter().map(|(a, c)| (*a, *c)), - current_height, + // The compacted range's end height — becomes the new pin. + 101, &mut NoopProvider, &mut result, &mut pending_unknown, ) .await; - // 10_000 base + only the at/above deltas (10 + 20) = 10_030. The - // below-cursor 300 + 400 must NOT be counted. + // 10_000 base + only the post-pin deltas (10 + 20) = 10_030. The + // at/below-pin 300 + 400 must NOT be counted, and the pin advances + // to the range end so later replays gate correctly too. assert_eq!( - result.found.get(&(1u32, addr)).map(|f| f.balance), - Some(10_030), - "only deltas at heights >= current_height may apply (anti-double-count)" + result.found.get(&(1u32, addr)).copied(), + Some(AddressFunds { + nonce: 0, + balance: 10_030, + as_of_height: 101, + }), + "only deltas at heights above the pin may apply (anti-double-count)" ); assert!( pending_unknown.is_empty(), @@ -2814,6 +2908,112 @@ mod tests { ); } + /// The exact ADDR-09 double-count shape, on the pure seam: a fresh + /// proof-attested absolute (e.g. an asset-lock top-up reconcile) is + /// pinned at its proof height, and the recent replay then delivers the + /// same block's `AddToCredits` delta. The pin gate must drop it — the + /// delta is already inside the absolute — instead of producing + /// `X + X = 2X`. A genuinely newer delta still applies and advances + /// the pin. + #[tokio::test] + async fn apply_block_changes_drops_recent_delta_already_inside_pinned_absolute() { + use async_trait::async_trait; + + struct NoopProvider; + + #[async_trait] + impl AddressProvider for NoopProvider { + type Tag = u32; + type Address = PlatformAddress; + + fn gap_limit(&self) -> AddressIndex { + 0 + } + + fn pending_addresses(&self) -> impl Iterator + '_ { + std::iter::empty() + } + + async fn on_address_found( + &mut self, + _tag: Self::Tag, + _address: &Self::Address, + _funds: AddressFunds, + ) { + } + + async fn on_address_absent(&mut self, _tag: Self::Tag, _address: &Self::Address) {} + + fn current_balances( + &self, + ) -> impl Iterator + '_ { + std::iter::empty() + } + } + + let addr = p2pkh(0x42); + let mut lookup: HashMap, (u32, PlatformAddress)> = HashMap::new(); + lookup.insert(addr.to_bytes(), (1u32, addr)); + + // The reconcile seam committed the proof-attested absolute X, + // pinned at the funding proof's block height. + let mut result: AddressSyncResult = AddressSyncResult::new(); + result.found.insert( + (1u32, addr), + AddressFunds { + nonce: 0, + balance: 9_985_071_720, + as_of_height: 379_731, + }, + ); + + // The recent tree replays the funding credit recorded at the SAME + // block — the ADDR-09 double-count if applied. + let same_block = CreditOperation::AddToCredits(9_985_071_720); + let changes = [(&addr, BalanceOp::Recent(&same_block))]; + let mut pending_unknown: Vec = Vec::new(); + apply_block_changes( + &lookup, + changes.iter().map(|(a, c)| (*a, *c)), + 379_731, + &mut NoopProvider, + &mut result, + &mut pending_unknown, + ) + .await; + assert_eq!( + result.found.get(&(1u32, addr)).copied(), + Some(AddressFunds { + nonce: 0, + balance: 9_985_071_720, + as_of_height: 379_731, + }), + "a delta at the pin height is already inside the absolute (ADDR-09)" + ); + + // A genuinely newer delta still applies and advances the pin. + let newer = CreditOperation::AddToCredits(1_000); + let changes = [(&addr, BalanceOp::Recent(&newer))]; + apply_block_changes( + &lookup, + changes.iter().map(|(a, c)| (*a, *c)), + 379_740, + &mut NoopProvider, + &mut result, + &mut pending_unknown, + ) + .await; + assert_eq!( + result.found.get(&(1u32, addr)).copied(), + Some(AddressFunds { + nonce: 0, + balance: 9_985_072_720, + as_of_height: 379_740, + }), + "a post-pin delta applies once and advances the pin" + ); + } + /// A provider that lists an address in `current_balances` but NOT in /// `pending_addresses` (the FFI two-array shape that violates the /// trait invariant). On the full-scan path the tree scan never sees it, @@ -2865,6 +3065,7 @@ mod tests { AddressFunds { nonce: 0, balance: 5_000, + as_of_height: 0, }, )) } @@ -2903,14 +3104,16 @@ mod tests { let mut pending_unknown: Vec = Vec::new(); let op = BlockAwareCreditOperation::AddToCreditsOperations( - std::iter::once((0u64, 1_500u64)).collect(), + std::iter::once((50u64, 1_500u64)).collect(), ); let changes = [(&seeded, BalanceOp::Compacted(&op))]; apply_block_changes( &key_to_tag, changes.iter().map(|(a, c)| (*a, *c)), - 0, + // A real block height — heights are never 0 in production, and + // the pin gate drops changes at or below the current pin. + 100, &mut provider, &mut result, &mut pending_unknown, @@ -2976,6 +3179,7 @@ mod tests { AddressFunds { nonce: 0, balance: 5_000, + as_of_height: 0, }, )) } @@ -3058,6 +3262,7 @@ mod tests { AddressFunds { nonce: 0, balance: 5_000, + as_of_height: 0, }, )) } @@ -3099,14 +3304,16 @@ mod tests { // A later AddToCredits delta resolves via key_to_tag to (10, addr) and // accumulates on the seeded base: 5000 + 1500 = 6500 under ONE key. let op = BlockAwareCreditOperation::AddToCreditsOperations( - std::iter::once((0u64, 1_500u64)).collect(), + std::iter::once((50u64, 1_500u64)).collect(), ); let changes = [(&addr, BalanceOp::Compacted(&op))]; let mut pending_unknown: Vec = Vec::new(); apply_block_changes( &key_to_tag, changes.iter().map(|(a, c)| (*a, *c)), - 0, + // A real block height — heights are never 0 in production, and + // the pin gate drops changes at or below the current pin. + 100, &mut provider, &mut result, &mut pending_unknown, diff --git a/packages/rs-sdk/src/platform/address_sync/types.rs b/packages/rs-sdk/src/platform/address_sync/types.rs index 74daea4609a..04ad8d6fa7a 100644 --- a/packages/rs-sdk/src/platform/address_sync/types.rs +++ b/packages/rs-sdk/src/platform/address_sync/types.rs @@ -55,6 +55,29 @@ pub struct AddressFunds { pub nonce: AddressNonce, /// Credits balance held by the address. pub balance: Credits, + /// Platform block height this balance is current **as of** — the + /// height pin. + /// + /// The pin means: `balance` includes the effect of every block up to + /// and including `as_of_height`. It is the reconciliation rule between + /// the two sources of truth for an address balance: + /// + /// - **Direct truth** — a proof-attested absolute (state-transition + /// result, trunk/branch scan element). It arrives pinned at its + /// proof's block height. + /// - **Aggregate truth** — the recent/compacted balance-change delta + /// stream. A delta recorded at block `B` may only be applied when + /// `B > as_of_height` (otherwise it is already included in the + /// absolute); applying it advances the pin to `B`. + /// + /// Freshness between two absolutes is decided by comparing pins — a + /// later pin is authoritative *even when it revises the balance + /// downward* (nonces only advance on outgoing ops, so they cannot + /// order receive-only state). + /// + /// `0` means "unknown provenance" (legacy rows persisted before the + /// pin existed): every delta applies, matching pre-pin behavior. + pub as_of_height: u64, } /// Configuration for address synchronization. #[derive(Debug, Clone)] @@ -165,7 +188,8 @@ pub struct AddressSyncResult { /// whether the height has been compacted away. /// /// Store this value and return it from - /// [`AddressProvider::last_known_recent_block_height`] on the next call. + /// [`AddressProvider::last_known_recent_block_height`](super::provider::AddressProvider::last_known_recent_block_height) + /// on the next call. /// A value of `0` means no recent block has been observed yet. pub last_known_recent_block: u64, diff --git a/packages/rs-sdk/src/platform/transition/address_credit_withdrawal.rs b/packages/rs-sdk/src/platform/transition/address_credit_withdrawal.rs index 9e75fba3293..4d91e972fe2 100644 --- a/packages/rs-sdk/src/platform/transition/address_credit_withdrawal.rs +++ b/packages/rs-sdk/src/platform/transition/address_credit_withdrawal.rs @@ -32,7 +32,7 @@ pub trait WithdrawAddressFunds> { output_script: CoreScript, signer: &S, settings: Option, - ) -> Result; + ) -> Result<(AddressInfos, u64), Error>; /// Withdraws address balances with explicitly provided nonces. /// @@ -48,7 +48,7 @@ pub trait WithdrawAddressFunds> { output_script: CoreScript, signer: &S, settings: Option, - ) -> Result; + ) -> Result<(AddressInfos, u64), Error>; } #[async_trait::async_trait] @@ -63,7 +63,7 @@ impl> WithdrawAddressFunds for Sdk { output_script: CoreScript, signer: &S, settings: Option, - ) -> Result { + ) -> Result<(AddressInfos, u64), Error> { let inputs_with_nonce = nonce_inc(fetch_inputs_with_nonce(self, &inputs).await?); self.withdraw_address_funds_with_nonce( inputs_with_nonce, @@ -88,7 +88,7 @@ impl> WithdrawAddressFunds for Sdk { output_script: CoreScript, signer: &S, settings: Option, - ) -> Result { + ) -> Result<(AddressInfos, u64), Error> { let user_fee_increase = settings .as_ref() .and_then(|settings| settings.user_fee_increase) @@ -108,10 +108,12 @@ impl> WithdrawAddressFunds for Sdk { .await?; ensure_valid_state_transition_structure(&state_transition, self.version())?; - match state_transition - .broadcast_and_wait::(self, settings) - .await? - { + // `metadata.height` is the proof's committed block — the height + // pin for these absolutes (`AddressFunds::as_of_height`). + let (st_result, metadata) = state_transition + .broadcast_and_wait_with_metadata::(self, settings) + .await?; + match st_result { StateTransitionProofResult::VerifiedAddressInfos(address_infos_map) => { let mut expected_addresses: BTreeSet = inputs.keys().copied().collect(); @@ -120,6 +122,7 @@ impl> WithdrawAddressFunds for Sdk { } collect_address_infos_from_proof(address_infos_map, &expected_addresses) + .map(|infos| (infos, metadata.height)) } other => Err(Error::InvalidProvedResponse(format!( "unexpected proof result for address withdrawal: {:?}", diff --git a/packages/rs-sdk/src/platform/transition/broadcast.rs b/packages/rs-sdk/src/platform/transition/broadcast.rs index e7217e4deb5..67e846f8e33 100644 --- a/packages/rs-sdk/src/platform/transition/broadcast.rs +++ b/packages/rs-sdk/src/platform/transition/broadcast.rs @@ -5,7 +5,7 @@ use crate::sync::retry; use crate::{Error, Sdk}; use dapi_grpc::platform::v0::wait_for_state_transition_result_response::wait_for_state_transition_result_response_v0; use dapi_grpc::platform::v0::{ - wait_for_state_transition_result_response, BroadcastStateTransitionRequest, + wait_for_state_transition_result_response, BroadcastStateTransitionRequest, ResponseMetadata, WaitForStateTransitionResultResponse, }; use dash_context_provider::ContextProviderError; @@ -24,11 +24,30 @@ pub trait BroadcastStateTransition { sdk: &Sdk, settings: Option, ) -> Result; + /// Like [`wait_for_response`](Self::wait_for_response), but also + /// returns the quorum-authenticated response metadata. + /// `metadata.height` is the committed block the proof attests — + /// callers that persist proof-attested absolute balances need it as + /// the balance's height pin + /// (`dash_sdk::platform::address_sync::AddressFunds::as_of_height`). + async fn wait_for_response_with_metadata + Send>( + &self, + sdk: &Sdk, + settings: Option, + ) -> Result<(T, ResponseMetadata), Error>; async fn broadcast_and_wait + Send>( &self, sdk: &Sdk, settings: Option, ) -> Result; + /// Like [`broadcast_and_wait`](Self::broadcast_and_wait), but also + /// returns the quorum-authenticated response metadata (see + /// [`wait_for_response_with_metadata`](Self::wait_for_response_with_metadata)). + async fn broadcast_and_wait_with_metadata + Send>( + &self, + sdk: &Sdk, + settings: Option, + ) -> Result<(T, ResponseMetadata), Error>; } #[async_trait::async_trait] @@ -94,6 +113,16 @@ impl BroadcastStateTransition for StateTransition { sdk: &Sdk, settings: Option, ) -> Result { + self.wait_for_response_with_metadata::(sdk, settings) + .await + .map(|(result, _metadata)| result) + } + + async fn wait_for_response_with_metadata + Send>( + &self, + sdk: &Sdk, + settings: Option, + ) -> Result<(T, ResponseMetadata), Error> { trace!( transaction_id = %self .transaction_id() @@ -202,6 +231,7 @@ impl BroadcastStateTransition for StateTransition { let variant_name = result.to_string(); let conversion_result = T::try_from(result) + .map(|converted| (converted, metadata)) .map_err(|_| { Error::InvalidProvedResponse(format!( "invalid proved response: cannot convert from {} to {}", @@ -254,11 +284,23 @@ impl BroadcastStateTransition for StateTransition { sdk: &Sdk, settings: Option, ) -> Result { + self.broadcast_and_wait_with_metadata::(sdk, settings) + .await + .map(|(result, _metadata)| result) + } + + async fn broadcast_and_wait_with_metadata + Send>( + &self, + sdk: &Sdk, + settings: Option, + ) -> Result<(T, ResponseMetadata), Error> { trace!(state_transition = %self.name(), "broadcast_and_wait: start"); trace!("broadcast_and_wait: step 1 - broadcasting"); self.broadcast(sdk, settings).await?; trace!("broadcast_and_wait: step 2 - waiting for response"); - let result = self.wait_for_response::(sdk, settings).await; + let result = self + .wait_for_response_with_metadata::(sdk, settings) + .await; match &result { Ok(_) => trace!("broadcast_and_wait: complete success"), Err(e) => warn!(error = ?e, "broadcast_and_wait: failed"), diff --git a/packages/rs-sdk/src/platform/transition/put_identity.rs b/packages/rs-sdk/src/platform/transition/put_identity.rs index 24592500b2b..e41648d0f82 100644 --- a/packages/rs-sdk/src/platform/transition/put_identity.rs +++ b/packages/rs-sdk/src/platform/transition/put_identity.rs @@ -114,7 +114,7 @@ pub trait PutIdentity>: Waitable { identity_signer: &IS, input_address_signer: &AS, settings: Option, - ) -> Result<(Identity, AddressInfos), Error>; + ) -> Result<(Identity, AddressInfos, u64), Error>; /// Creates an identity funded by Platform addresses, fetching the /// current address nonces from Platform automatically. @@ -138,7 +138,7 @@ pub trait PutIdentity>: Waitable { identity_signer: &IS, input_address_signer: &AS, settings: Option, - ) -> Result<(Identity, AddressInfos), Error>; + ) -> Result<(Identity, AddressInfos, u64), Error>; } #[async_trait::async_trait] @@ -243,7 +243,7 @@ impl> PutIdentity for Identity { identity_signer: &IS, input_address_signer: &AS, settings: Option, - ) -> Result<(Identity, AddressInfos), Error> { + ) -> Result<(Identity, AddressInfos, u64), Error> { put_identity_with_address_funding::( self, sdk, @@ -264,7 +264,7 @@ impl> PutIdentity for Identity { identity_signer: &IS, input_address_signer: &AS, settings: Option, - ) -> Result<(Identity, AddressInfos), Error> { + ) -> Result<(Identity, AddressInfos, u64), Error> { // Platform's convention: transitions submit `last_used + 1`. // `fetch_inputs_with_nonce` reads the on-chain "last used", // `nonce_inc` bumps by 1 — same helpers used by @@ -356,7 +356,7 @@ async fn put_identity_with_address_funding< identity_signer: &IS, input_signer: &AS, settings: Option, -) -> Result<(Identity, AddressInfos), Error> { +) -> Result<(Identity, AddressInfos, u64), Error> { let expected_addresses: BTreeSet = inputs.keys().copied().collect::>(); @@ -388,10 +388,12 @@ async fn put_identity_with_address_funding< ensure_valid_state_transition_structure(&state_transition, sdk.version())?; - match state_transition - .broadcast_and_wait::(sdk, settings) - .await? - { + // `metadata.height` is the proof's committed block — the height + // pin for these absolutes (`AddressFunds::as_of_height`). + let (st_result, metadata) = state_transition + .broadcast_and_wait_with_metadata::(sdk, settings) + .await?; + match st_result { StateTransitionProofResult::VerifiedIdentityFullWithAddressInfos( proved_identity, address_infos_map, @@ -407,7 +409,7 @@ async fn put_identity_with_address_funding< let address_infos = collect_address_infos_from_proof(address_infos_map, &expected_addresses)?; - Ok((proved_identity, address_infos)) + Ok((proved_identity, address_infos, metadata.height)) } other => Err(Error::InvalidProvedResponse(format!( "identity proof was expected but not returned: {:?}", diff --git a/packages/rs-sdk/src/platform/transition/top_up_address.rs b/packages/rs-sdk/src/platform/transition/top_up_address.rs index 126bb5857c8..05377713c04 100644 --- a/packages/rs-sdk/src/platform/transition/top_up_address.rs +++ b/packages/rs-sdk/src/platform/transition/top_up_address.rs @@ -23,7 +23,13 @@ use drive_proof_verifier::types::AddressInfos; pub trait TopUpAddress> { /// Tops up addresses using a raw private key for the asset-lock proof. /// - /// Returns proof-backed [`AddressInfos`] for the funded addresses. + /// Returns proof-backed [`AddressInfos`] for the funded addresses, + /// paired with the proof's committed block height — the balance + /// height pin ([`AddressFunds::as_of_height`]) callers that persist + /// the absolutes must record. + /// + /// [`AddressFunds::as_of_height`]: + /// crate::platform::address_sync::AddressFunds::as_of_height /// /// Prefer [`Self::top_up_with_signers`] when the asset-lock private /// key lives outside Rust (Swift / hardware wallet / HSM): the @@ -38,7 +44,7 @@ pub trait TopUpAddress> { fee_strategy: AddressFundsFeeStrategy, signer: &S, settings: Option, - ) -> Result; + ) -> Result<(AddressInfos, u64), Error>; /// Top up addresses with an external asset-lock signer. /// @@ -68,7 +74,7 @@ pub trait TopUpAddress> { signer: &S, asset_lock_signer: &AS, settings: Option, - ) -> Result + ) -> Result<(AddressInfos, u64), Error> where AS: dpp::key_wallet::signer::Signer + Send + Sync; } @@ -89,7 +95,7 @@ where fee_strategy: AddressFundsFeeStrategy, signer: &S, settings: Option, - ) -> Result { + ) -> Result<(AddressInfos, u64), Error> { BTreeMap::from([(self.0, self.1)]) .top_up( sdk, @@ -113,7 +119,7 @@ where signer: &S, asset_lock_signer: &AS, settings: Option, - ) -> Result + ) -> Result<(AddressInfos, u64), Error> where AS: dpp::key_wallet::signer::Signer + Send + Sync, { @@ -141,7 +147,7 @@ impl> TopUpAddress for AddressesWithBalances { fee_strategy: AddressFundsFeeStrategy, signer: &S, settings: Option, - ) -> Result { + ) -> Result<(AddressInfos, u64), Error> { if self.is_empty() { return Err(Error::from(TransitionNoOutputsError::new())); } @@ -177,7 +183,7 @@ impl> TopUpAddress for AddressesWithBalances { signer: &S, asset_lock_signer: &AS, settings: Option, - ) -> Result + ) -> Result<(AddressInfos, u64), Error> where AS: dpp::key_wallet::signer::Signer + Send + Sync, { @@ -215,18 +221,24 @@ impl> TopUpAddress for AddressesWithBalances { } /// Broadcast the address-funding ST and convert the proof into the -/// `AddressInfos` map. Shared between the legacy private-key path and -/// the new signer-pair path — both flows want the same proof-shape -/// guarantee and the same expected-addresses cross-check. +/// `AddressInfos` map, paired with the proof's committed block height. +/// Shared between the legacy private-key path and the new signer-pair +/// path — both flows want the same proof-shape guarantee and the same +/// expected-addresses cross-check. +/// +/// The returned height is the balances' height pin (see +/// `crate::platform::address_sync::AddressFunds::as_of_height`): callers +/// that persist these absolutes must record it so later balance-change +/// deltas at or below it are not re-applied on top. async fn broadcast_and_collect_address_infos( expected: &AddressesWithBalances, state_transition: StateTransition, sdk: &Sdk, settings: Option, -) -> Result { +) -> Result<(AddressInfos, u64), Error> { ensure_valid_state_transition_structure(&state_transition, sdk.version())?; - let st_result = state_transition - .broadcast_and_wait::(sdk, settings) + let (st_result, metadata) = state_transition + .broadcast_and_wait_with_metadata::(sdk, settings) .await?; match st_result { StateTransitionProofResult::VerifiedAddressInfos(address_infos) => { @@ -235,6 +247,7 @@ async fn broadcast_and_collect_address_infos( .copied() .collect::>(); collect_address_infos_from_proof(address_infos, &expected_addresses) + .map(|infos| (infos, metadata.height)) } other => Err(Error::InvalidProvedResponse(format!( "address info proof was expected for {:?}, but received {:?}", diff --git a/packages/rs-sdk/src/platform/transition/top_up_identity_from_addresses.rs b/packages/rs-sdk/src/platform/transition/top_up_identity_from_addresses.rs index 0eca0ea7ffa..68cc2908e02 100644 --- a/packages/rs-sdk/src/platform/transition/top_up_identity_from_addresses.rs +++ b/packages/rs-sdk/src/platform/transition/top_up_identity_from_addresses.rs @@ -28,7 +28,7 @@ pub trait TopUpIdentityFromAddresses>: Waitable { inputs: BTreeMap, signer: &S, settings: Option, - ) -> Result<(AddressInfos, Credits), Error>; + ) -> Result<(AddressInfos, Credits, u64), Error>; /// Top up identity providing explicit address nonces. /// @@ -39,7 +39,7 @@ pub trait TopUpIdentityFromAddresses>: Waitable { inputs: BTreeMap, signer: &S, settings: Option, - ) -> Result<(AddressInfos, Credits), Error>; + ) -> Result<(AddressInfos, Credits, u64), Error>; } #[async_trait::async_trait] @@ -50,7 +50,7 @@ impl> TopUpIdentityFromAddresses for Identity { inputs: BTreeMap, signer: &S, settings: Option, - ) -> Result<(AddressInfos, Credits), Error> { + ) -> Result<(AddressInfos, Credits, u64), Error> { let inputs_with_nonce = nonce_inc(fetch_inputs_with_nonce(sdk, &inputs).await?); self.top_up_from_addresses_with_nonce(sdk, inputs_with_nonce, signer, settings) .await @@ -62,7 +62,7 @@ impl> TopUpIdentityFromAddresses for Identity { inputs: BTreeMap, signer: &S, settings: Option, - ) -> Result<(AddressInfos, Credits), Error> { + ) -> Result<(AddressInfos, Credits, u64), Error> { let user_fee_increase = settings .as_ref() .and_then(|settings| settings.user_fee_increase) @@ -82,10 +82,12 @@ impl> TopUpIdentityFromAddresses for Identity { .await?; ensure_valid_state_transition_structure(&state_transition, sdk.version())?; - match state_transition - .broadcast_and_wait::(sdk, settings) - .await? - { + // `metadata.height` is the proof's committed block — the height + // pin for these absolutes (`AddressFunds::as_of_height`). + let (st_result, metadata) = state_transition + .broadcast_and_wait_with_metadata::(sdk, settings) + .await?; + match st_result { StateTransitionProofResult::VerifiedIdentityWithAddressInfos( identity, address_infos_map, @@ -107,7 +109,7 @@ impl> TopUpIdentityFromAddresses for Identity { ) })?; - Ok((address_infos, balance)) + Ok((address_infos, balance, metadata.height)) } other => Err(Error::InvalidProvedResponse(format!( "identity proof was expected for {:?}, but received {:?}", diff --git a/packages/rs-sdk/src/platform/transition/transfer_address_funds.rs b/packages/rs-sdk/src/platform/transition/transfer_address_funds.rs index 4890f0e2cd1..90be291c901 100644 --- a/packages/rs-sdk/src/platform/transition/transfer_address_funds.rs +++ b/packages/rs-sdk/src/platform/transition/transfer_address_funds.rs @@ -27,7 +27,7 @@ pub trait TransferAddressFunds> { fee_strategy: AddressFundsFeeStrategy, signer: &S, settings: Option, - ) -> Result; + ) -> Result<(AddressInfos, u64), Error>; /// Broadcast address funds transfer with explicitly provided address nonces. /// @@ -39,7 +39,7 @@ pub trait TransferAddressFunds> { fee_strategy: AddressFundsFeeStrategy, signer: &S, settings: Option, - ) -> Result; + ) -> Result<(AddressInfos, u64), Error>; } #[async_trait::async_trait] @@ -51,7 +51,7 @@ impl> TransferAddressFunds for Sdk { fee_strategy: AddressFundsFeeStrategy, signer: &S, settings: Option, - ) -> Result { + ) -> Result<(AddressInfos, u64), Error> { let inputs_with_nonce = nonce_inc(fetch_inputs_with_nonce(self, &inputs).await?); self.transfer_address_funds_with_nonce( inputs_with_nonce, @@ -70,7 +70,7 @@ impl> TransferAddressFunds for Sdk { fee_strategy: AddressFundsFeeStrategy, signer: &S, settings: Option, - ) -> Result { + ) -> Result<(AddressInfos, u64), Error> { if outputs.is_empty() { return Err(Error::from(TransitionNoOutputsError::new())); } @@ -94,12 +94,15 @@ impl> TransferAddressFunds for Sdk { let expected_addresses: BTreeSet = inputs.keys().chain(outputs.keys()).copied().collect(); - match state_transition - .broadcast_and_wait::(self, settings) - .await? - { + // `metadata.height` is the proof's committed block — the height + // pin for these absolutes (`AddressFunds::as_of_height`). + let (st_result, metadata) = state_transition + .broadcast_and_wait_with_metadata::(self, settings) + .await?; + match st_result { StateTransitionProofResult::VerifiedAddressInfos(address_infos_map) => { collect_address_infos_from_proof(address_infos_map, &expected_addresses) + .map(|infos| (infos, metadata.height)) } other => Err(Error::InvalidProvedResponse(format!( "address info proof was expected for {:?}, but received {:?}", diff --git a/packages/rs-sdk/src/platform/transition/transfer_to_addresses.rs b/packages/rs-sdk/src/platform/transition/transfer_to_addresses.rs index ee95e309997..5026f201978 100644 --- a/packages/rs-sdk/src/platform/transition/transfer_to_addresses.rs +++ b/packages/rs-sdk/src/platform/transition/transfer_to_addresses.rs @@ -23,7 +23,14 @@ pub trait TransferToAddresses: Waitable { /// Returns tuple of: /// * Proof-backed address infos for provided recipients /// * Updated identity balance - /// * Proof-backed address infos for provided recipients + /// * The proof's committed block height (`metadata.height`) — the + /// height the returned absolutes are current **as of**. Callers + /// that persist them must record it as the balance height pin + /// ([`AddressFunds::as_of_height`]) so balance-change deltas at or + /// below it are not re-applied on top. + /// + /// [`AddressFunds::as_of_height`]: + /// crate::platform::address_sync::AddressFunds::as_of_height #[allow(clippy::too_many_arguments)] async fn transfer_credits_to_addresses + Send>( &self, @@ -32,7 +39,7 @@ pub trait TransferToAddresses: Waitable { signing_transfer_key_to_use: Option<&IdentityPublicKey>, signer: &S, settings: Option, - ) -> Result<(AddressInfos, Credits), Error>; + ) -> Result<(AddressInfos, Credits, u64), Error>; } #[async_trait::async_trait] @@ -44,7 +51,7 @@ impl TransferToAddresses for Identity { signing_transfer_key_to_use: Option<&IdentityPublicKey>, signer: &S, settings: Option, - ) -> Result<(AddressInfos, Credits), Error> { + ) -> Result<(AddressInfos, Credits, u64), Error> { if recipient_addresses.is_empty() { return Err(Error::Generic( "recipient_addresses must contain at least one address".to_string(), @@ -73,10 +80,12 @@ impl TransferToAddresses for Identity { let expected_addresses: BTreeSet = recipient_addresses.keys().copied().collect(); - match state_transition - .broadcast_and_wait::(sdk, settings) - .await? - { + // `metadata.height` is the proof's committed block — the height + // pin for these absolutes (`AddressFunds::as_of_height`). + let (st_result, metadata) = state_transition + .broadcast_and_wait_with_metadata::(sdk, settings) + .await?; + match st_result { StateTransitionProofResult::VerifiedIdentityWithAddressInfos( identity, address_infos_map, @@ -98,7 +107,7 @@ impl TransferToAddresses for Identity { ) })?; - Ok((address_infos, balance)) + Ok((address_infos, balance, metadata.height)) } other => Err(Error::InvalidProvedResponse(format!( "identity proof was expected for {:?}, but received {:?}", diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentPlatformAddress.swift b/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentPlatformAddress.swift index a231c0264d7..3a46aef846d 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentPlatformAddress.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentPlatformAddress.swift @@ -59,9 +59,12 @@ public final class PersistentPlatformAddress { /// Platform block height where this address first appeared in a /// balance changeset. Zero until the address is seen on-chain. public var firstSeenHeight: UInt32 - /// Platform block height of the most recent balance changeset - /// touching this address. - public var lastSeenHeight: UInt32 + /// Platform block height this row's `balance` is current **as of** + /// — the balance height pin (`AddressFunds::as_of_height` in Rust). + /// Round-tripped verbatim through the persistence callbacks so the + /// sync's delta-replay gate survives restarts. Zero means "unknown + /// provenance" (rows persisted before the pin existed). + public var lastSeenHeight: UInt64 /// 32-byte wallet ID that owns this address. Denormalized from /// `account.wallet.walletId` so per-wallet `@Query` filters don't /// have to traverse two optional relationships. diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/ManagedPlatformAddressWallet.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/ManagedPlatformAddressWallet.swift index 2c5c918cd4f..9db6e9bf961 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/ManagedPlatformAddressWallet.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/ManagedPlatformAddressWallet.swift @@ -217,7 +217,10 @@ public final class ManagedPlatformAddressWallet: @unchecked Sendable { balance: out.credits, nonce: 0, account_index: 0, - address_index: 0 + address_index: 0, + // Request path: names an output amount, carries no + // persisted balance — the height pin is meaningless here. + as_of_height: 0 ) } diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift index 6f3646730a0..c7f67912379 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift @@ -115,12 +115,12 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable { /// only; derivation metadata stays as the address-emit path set it. func persistAddressBalances( walletId: Data, - entries: [(UInt8, Data, UInt64, UInt32, UInt32, UInt32)] + entries: [(UInt8, Data, UInt64, UInt32, UInt32, UInt32, UInt64)] ) { onQueue { // `accountIndex` / `addressIndex` (tuple slots 5 and 6) are // intentionally ignored — see the note above. - for (_, addressHash, balance, nonce, _, _) in entries { + for (_, addressHash, balance, nonce, _, _, asOfHeight) in entries { // Scope by walletId + hash: a hash-only predicate can match // another wallet's row in a multi-wallet store (same seed // imported on coin-type-sharing networks, watch-only @@ -135,6 +135,9 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable { } existing.balance = balance existing.nonce = nonce + // Balance height pin — persisted verbatim so the load + // path can hand it back to Rust (delta-replay gating). + existing.lastSeenHeight = asOfHeight if balance > 0 || nonce > 0 { existing.isUsed = true } @@ -264,7 +267,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable { /// shape matches the Rust-side `AddressBalanceEntryFFI` layout so /// the load-wallet-list path can re-seed the provider on startup /// without a full rescan. - public func loadCachedBalances(walletId: Data) -> [(UInt8, [UInt8], UInt64, UInt32, UInt32, UInt32)] { + public func loadCachedBalances(walletId: Data) -> [(UInt8, [UInt8], UInt64, UInt32, UInt32, UInt32, UInt64)] { onQueue { loadCachedBalancesOnQueue(walletId: walletId) } } @@ -272,7 +275,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable { /// already running on `serialQueue`. Lets internal on-queue /// callers (`loadWalletList`) reuse the body without recursing /// through `onQueue`, which would deadlock. - private func loadCachedBalancesOnQueue(walletId: Data) -> [(UInt8, [UInt8], UInt64, UInt32, UInt32, UInt32)] { + private func loadCachedBalancesOnQueue(walletId: Data) -> [(UInt8, [UInt8], UInt64, UInt32, UInt32, UInt32, UInt64)] { let descriptor = FetchDescriptor( predicate: PersistentPlatformAddress.predicate(walletId: walletId) ) @@ -288,7 +291,8 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable { record.balance, record.nonce, record.accountIndex, - record.addressIndex + record.addressIndex, + record.lastSeenHeight ) } } @@ -4051,7 +4055,8 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable { ) var written = 0 for cached in cachedBalances { - let (addressType, hash, balance, nonce, accountIndex, addressIndex) = cached + let (addressType, hash, balance, nonce, accountIndex, addressIndex, asOfHeight) = + cached guard hash.count == 20 else { continue } var hashTuple: @@ -4068,7 +4073,8 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable { balance: balance, nonce: nonce, account_index: accountIndex, - address_index: addressIndex + address_index: addressIndex, + as_of_height: asOfHeight ) written += 1 } @@ -5557,7 +5563,7 @@ private func persistAddressBalancesCallback( let walletId = Data(bytes: walletIdPtr, count: 32) - var entries: [(UInt8, Data, UInt64, UInt32, UInt32, UInt32)] = [] + var entries: [(UInt8, Data, UInt64, UInt32, UInt32, UInt32, UInt64)] = [] entries.reserveCapacity(Int(count)) for i in 0.. (balance: UInt64, nonce: UInt32, accountIndex: UInt32, addressIndex: UInt32)? { + ) -> (balance: UInt64, nonce: UInt32, accountIndex: UInt32, addressIndex: UInt32, asOfHeight: UInt64)? { let rows = handler.loadCachedBalances(walletId: walletId) - for (_, hash, balance, nonce, accountIndex, addressIndex) in rows + for (_, hash, balance, nonce, accountIndex, addressIndex, asOfHeight) in rows where hash.allSatisfy({ $0 == hashByte }) && hash.count == 20 { - return (balance, nonce, accountIndex, addressIndex) + return (balance, nonce, accountIndex, addressIndex, asOfHeight) } return nil } @@ -111,8 +111,8 @@ final class AddressBalancePersistTests: XCTestCase { // Reconcile removal for `removed`: fully consumed, zero funds, // and — the hazard — carrying `funded`'s index 5, not its own 2. - let removal: [(UInt8, Data, UInt64, UInt32, UInt32, UInt32)] = [ - (0, removedHash, 0, 0, accountIndex, conflictingIndex) + let removal: [(UInt8, Data, UInt64, UInt32, UInt32, UInt32, UInt64)] = [ + (0, removedHash, 0, 0, accountIndex, conflictingIndex, 379_790) ] handler.persistAddressBalances(walletId: walletId, entries: removal) @@ -156,9 +156,10 @@ final class AddressBalancePersistTests: XCTestCase { nonce: 0 ) - // BLAST reports a fresh balance; the entry echoes the true index. - let update: [(UInt8, Data, UInt64, UInt32, UInt32, UInt32)] = [ - (0, fundedHash, 1_000, 7, accountIndex, fundedIndex) + // BLAST reports a fresh balance pinned at the pass's proof + // height; the entry echoes the true index. + let update: [(UInt8, Data, UInt64, UInt32, UInt32, UInt32, UInt64)] = [ + (0, fundedHash, 1_000, 7, accountIndex, fundedIndex, 379_784) ] handler.persistAddressBalances(walletId: walletId, entries: update) @@ -166,6 +167,10 @@ final class AddressBalancePersistTests: XCTestCase { XCTAssertEqual(funded.balance, 1_000) XCTAssertEqual(funded.nonce, 7) XCTAssertEqual(funded.addressIndex, fundedIndex) + XCTAssertEqual( + funded.asOfHeight, 379_784, + "the balance height pin must round-trip through lastSeenHeight" + ) } /// A balance update for an address that was never address-emitted @@ -173,8 +178,8 @@ final class AddressBalancePersistTests: XCTestCase { func testBalanceUpdateForUnknownAddressIsSkipped() throws { let (handler, _) = try makeHandler() - let update: [(UInt8, Data, UInt64, UInt32, UInt32, UInt32)] = [ - (0, removedHash, 42, 1, accountIndex, conflictingIndex) + let update: [(UInt8, Data, UInt64, UInt32, UInt32, UInt32, UInt64)] = [ + (0, removedHash, 42, 1, accountIndex, conflictingIndex, 100) ] handler.persistAddressBalances(walletId: walletId, entries: update) diff --git a/packages/wasm-sdk/src/state_transitions/addresses.rs b/packages/wasm-sdk/src/state_transitions/addresses.rs index be9c2e9aa71..8a21952a3b6 100644 --- a/packages/wasm-sdk/src/state_transitions/addresses.rs +++ b/packages/wasm-sdk/src/state_transitions/addresses.rs @@ -171,7 +171,9 @@ impl WasmSdk { let fee_strategy = fee_strategy_from_steps_or_default(parsed.fee_strategy); // Use the SDK's transfer_address_funds method which handles nonces, building, and broadcasting - let address_infos = self + // The returned proof height pins persisted absolutes; the wasm + // path only renders the proof entries, so it is unused here. + let (address_infos, _proof_height) = self .inner_sdk() .transfer_address_funds(inputs_map, outputs_map, fee_strategy, &signer, settings) .await @@ -295,7 +297,7 @@ impl WasmSdk { let inputs_map = outputs_to_btree_map(parsed.inputs)?; // Use the SDK's top_up_from_addresses method - let (address_infos, new_balance) = identity + let (address_infos, new_balance, _proof_height) = identity .top_up_from_addresses(self.inner_sdk(), inputs_map, &signer, settings) .await .map_err(|e| WasmSdkError::generic(format!("Failed to top up identity: {}", e)))?; @@ -451,7 +453,7 @@ impl WasmSdk { let pooling = parsed.pooling.into(); // Use the SDK's withdraw_address_funds method which handles nonces, building, and broadcasting - let address_infos = self + let (address_infos, _proof_height) = self .inner_sdk() .withdraw_address_funds( inputs_map, @@ -507,7 +509,7 @@ impl WasmSdk { .transpose()?; // Use the SDK's transfer_credits_to_addresses method - let (address_infos, new_balance) = identity + let (address_infos, new_balance, _proof_height) = identity .transfer_credits_to_addresses( self.inner_sdk(), outputs_map, @@ -746,7 +748,7 @@ impl WasmSdk { let fee_strategy = fee_strategy_from_steps_or_default(parsed.fee_strategy); // Use the SDK's top_up method for addresses - let address_infos = outputs_map + let (address_infos, _proof_height) = outputs_map .top_up( self.inner_sdk(), asset_lock_proof, @@ -915,7 +917,7 @@ impl WasmSdk { .transpose()?; // Use the SDK's put_with_address_funding method - let (created_identity, address_infos) = identity + let (created_identity, address_infos, _proof_height) = identity .put_with_address_funding( self.inner_sdk(), inputs,