diff --git a/packages/rs-platform-wallet-storage/src/sqlite/schema/identity_keys.rs b/packages/rs-platform-wallet-storage/src/sqlite/schema/identity_keys.rs index e985717c40b..b8046091d04 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/schema/identity_keys.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/schema/identity_keys.rs @@ -6,7 +6,7 @@ //! the key with bincode's native `Encode`/`Decode` and rides the surrounding //! fields on the serde encoder, keeping one blob per row. -use rusqlite::{params, Connection, Transaction}; +use rusqlite::{params, Connection, OptionalExtension, Transaction}; use serde::{Deserialize, Serialize}; use dpp::identity::identity_public_key::accessors::v0::IdentityPublicKeyGettersV0; @@ -126,6 +126,16 @@ fn map_fk_violation( /// cross-checked against it so the typed columns and the blob stay aligned. /// A cross-wallet write is refused with /// [`WalletStorageError::IdentityKeyWalletMismatch`]. +/// +/// An upsert without a derivation breadcrumb keeps the one the same row +/// already holds, as the Swift and Kotlin persisters do (they keep it on any +/// upsert of the row; this writer also checks the row first): a write that only +/// knows the key's new state (a refresh, a raised limit, a disable) must not +/// turn a wallet-derived key watch-only. Only a row that holds the same +/// public key (type, data and hash) and passes the reader's checks +/// (`contradicts_columns`) lends its breadcrumb; any other stored row (one +/// that does not decode, is over the blob size limit, contradicts its +/// columns or holds another key) holds nothing to keep and is replaced. pub fn apply( tx: &Transaction<'_>, wallet_id: &WalletId, @@ -161,6 +171,12 @@ pub fn apply( // 32 zero bytes instead would produce a row that matches no // wallet, no NULL-scoped reader, and no guard. let wallet_id_param = super::wallet_id_to_param(wallet_id); + let mut stored = tx.prepare_cached( + "SELECT public_key_blob, public_key_hash FROM identity_keys \ + WHERE identity_id = ?1 AND key_id = ?2 AND wallet_id IS ?3 \ + AND length(public_key_blob) <= ?4", + )?; + let size_limit = i64::try_from(blob::BLOB_SIZE_LIMIT_BYTES).unwrap_or(i64::MAX); for ((identity_id, key_id), entry) in &cs.upserts { // Typed columns and blob fields must agree so a row can never // diverge on disk. @@ -172,7 +188,40 @@ pub fn apply( return Err(WalletStorageError::IdentityKeyEntryMismatch); } } - let wire = IdentityKeyWire::from_entry(entry)?; + let mut wire = IdentityKeyWire::from_entry(entry)?; + if wire.derivation_indices.is_none() { + let held: Option<(Vec, Vec)> = stored + .query_row( + params![ + identity_id.as_slice(), + i64::from(*key_id), + wallet_id_param, + size_limit + ], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .optional()?; + let held = held.and_then(|(payload, typed_public_key_hash)| { + decode_entry(&payload).ok().filter(|held| { + !contradicts_columns( + held, + identity_id, + *key_id, + &typed_public_key_hash, + wallet_id, + ) + }) + }); + if let Some(held) = held { + let same_key = held.public_key.key_type() == entry.public_key.key_type() + && held.public_key.data() == entry.public_key.data() + && held.public_key_hash == entry.public_key_hash; + if same_key && held.derivation_indices.is_some() { + wire.wallet_id = held.wallet_id; + wire.derivation_indices = held.derivation_indices; + } + } + } let entry_blob = blob::encode(&wire)?; stmt.execute(params![ wallet_id_param, @@ -217,6 +266,25 @@ pub fn apply( Ok(()) } +/// Whether a decoded row names a different identity, key, public-key hash +/// or wallet than the typed columns it was selected by: corruption, which +/// the reader skips and the writer never borrows a breadcrumb from. +fn contradicts_columns( + entry: &IdentityKeyEntry, + identity_id: &Identifier, + key_id: KeyID, + typed_public_key_hash: &[u8], + wallet_id: &WalletId, +) -> bool { + entry.identity_id != *identity_id + || entry.key_id != key_id + || entry.public_key.id() != key_id + || entry.public_key_hash[..] != typed_public_key_hash[..] + || entry + .wallet_id + .is_some_and(|entry_wallet_id| entry_wallet_id != *wallet_id) +} + /// Decode an `identity_keys.public_key_blob` cell back to the entry. pub fn decode_entry(payload: &[u8]) -> Result { let wire: IdentityKeyWire = blob::decode(payload)?; @@ -285,14 +353,13 @@ pub fn load_state( // `public_key_hash` is the indexed lookup column while the blob is // what callers receive, so a divergence makes a key findable under a // hash it does not carry. - let contradicts_columns = entry.identity_id != identity_id - || entry.key_id != key_id - || entry.public_key.id() != key_id - || entry.public_key_hash[..] != typed_public_key_hash[..] - || entry - .wallet_id - .is_some_and(|entry_wallet_id| entry_wallet_id != *wallet_id); - if contradicts_columns { + if contradicts_columns( + &entry, + &identity_id, + key_id, + &typed_public_key_hash, + wallet_id, + ) { ctx.tolerate( LoadSite::IdentityKeyRow, WalletStorageError::IdentityKeyEntryMismatch, diff --git a/packages/rs-platform-wallet-storage/tests/integration/main.rs b/packages/rs-platform-wallet-storage/tests/integration/main.rs index 918d0c6bac2..0f0a6c6f051 100644 --- a/packages/rs-platform-wallet-storage/tests/integration/main.rs +++ b/packages/rs-platform-wallet-storage/tests/integration/main.rs @@ -47,6 +47,7 @@ mod sqlite_foreign_keys; mod sqlite_identity_hard_delete; mod sqlite_identity_index_concurrency; mod sqlite_identity_index_uniqueness; +mod sqlite_identity_key_restore; mod sqlite_identity_keys_reader; mod sqlite_identity_scan_state; mod sqlite_legacy_state_migration; diff --git a/packages/rs-platform-wallet-storage/tests/integration/sqlite_identity_key_restore.rs b/packages/rs-platform-wallet-storage/tests/integration/sqlite_identity_key_restore.rs new file mode 100644 index 00000000000..cdf8a10cbfb --- /dev/null +++ b/packages/rs-platform-wallet-storage/tests/integration/sqlite_identity_key_restore.rs @@ -0,0 +1,659 @@ +//! Identity keys across a restart: a restore whose key write the disk +//! refuses, and keys a refresh finds. +//! +//! Every test drives the real `PlatformWalletManager` + `IdentityWallet` +//! over a mock SDK and the real SQLite persister (Immediate mode, the +//! default and what dash-evo-tool runs). A disk-full failure is the genuine +//! rusqlite `SQLITE_FULL` error primed into the persister's own flush seam +//! (`force_next_flush_to_fail`), so classification, buffer restore and the +//! `FlushRetryable` surface are the production code paths; only the moment +//! the disk fills is chosen by the test. A "restart" closes the SQLite file, +//! reopens it and rebuilds a fresh manager through `load_from_persistor`. +//! + +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; + +use dash_sdk::platform::types::identity::PublicKeyHash; +use dpp::identity::accessors::{IdentityGettersV0, IdentitySettersV0}; +use dpp::identity::identity_public_key::accessors::v0::{ + IdentityPublicKeyGettersV0, IdentityPublicKeySettersV0, +}; +use dpp::identity::identity_public_key::v0::IdentityPublicKeyV0; +use dpp::identity::v0::IdentityV0; +use dpp::identity::{Identity, IdentityPublicKey, KeyID, KeyType, Purpose, SecurityLevel}; +use dpp::platform_value::BinaryData; +use dpp::prelude::Identifier; +use key_wallet::bip32::ExtendedPrivKey; +use key_wallet::mnemonic::Mnemonic; +use key_wallet::wallet::initialization::WalletAccountCreationOptions; +use key_wallet::Network; +use platform_wallet::changeset::{ + ClientStartState, IdentityKeyEntry, IdentityScanStateEntry, PersistenceCapabilities, + PersistenceError, PersistenceErrorKind, PlatformWalletChangeSet, PlatformWalletPersistence, +}; +use platform_wallet::wallet::identity::network::{ + derive_ecdsa_identity_auth_keypair_from_master, derive_identity_auth_key_hash_from_master, + IdentityDiscoveryOptions, +}; +use platform_wallet::wallet::platform_wallet::WalletId; +use platform_wallet::{PlatformEventHandler, PlatformWalletManager}; +use platform_wallet_storage::sqlite::schema::{identity_keys, identity_scan_states}; +use platform_wallet_storage::{ + LoadCtx, SqlitePersister, SqlitePersisterConfig, WalletStorageError, +}; + +const MNEMONIC_A: &str = "abandon abandon abandon abandon abandon abandon \ + abandon abandon abandon abandon abandon about"; +const NET: Network = Network::Testnet; + +// --------------------------------------------------------------------------- +// Persister: real SQLite, with a disk that can fill +// --------------------------------------------------------------------------- + +#[derive(Clone, Copy, PartialEq, Eq)] +enum DiskFull { + /// The disk never fills. + Never, + /// The disk fills on the first write that carries a key breadcrumb and + /// stays full until the process ends. + FromBreadcrumbWriteOn, +} + +struct Disk { + inner: Mutex>, + mode: DiskFull, + tripped: AtomicBool, + full: AtomicBool, + /// Every error `store`/`flush` returned, with its kind. + failures: Mutex>>, +} + +fn sqlite_full() -> WalletStorageError { + WalletStorageError::Sqlite(rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_FULL), + Some("database or disk is full".into()), + )) +} + +impl Disk { + fn open(path: &std::path::Path, mode: DiskFull) -> Arc { + let p = SqlitePersister::open(SqlitePersisterConfig::new(path)).expect("open persister"); + Arc::new(Self { + inner: Mutex::new(Some(p)), + mode, + tripped: AtomicBool::new(false), + full: AtomicBool::new(false), + failures: Mutex::new(Vec::new()), + }) + } + + /// Process death: the SQLite handle closes; anything only buffered is gone. + fn close(&self) { + self.inner.lock().unwrap().take(); + } + + fn with(&self, f: impl FnOnce(&SqlitePersister) -> R) -> R { + let guard = self.inner.lock().unwrap(); + f(guard.as_ref().expect("persister closed")) + } + + fn before_write(&self, p: &SqlitePersister) { + if self.full.load(Ordering::SeqCst) { + p.force_next_flush_to_fail(sqlite_full()); + } + } + + fn record(&self, r: Result) -> Result { + if let Err(e) = &r { + self.failures.lock().unwrap().push(e.kind()); + } + r + } + + fn key_rows(&self, w: &WalletId, id: &Identifier) -> BTreeMap { + self.with(|p| { + let conn = p.lock_conn_for_test(); + identity_keys::load_state(&conn, w, &LoadCtx::strict()) + .expect("read identity_keys") + .upserts + .into_iter() + .filter(|((owner, _), _)| owner == id) + .map(|((_, key_id), e)| (key_id, e)) + .collect() + }) + } + + fn scan_verdict(&self, w: &WalletId) -> Option { + self.with(|p| { + let conn = p.lock_conn_for_test(); + identity_scan_states::load_for_wallet(&conn, w, &LoadCtx::strict()) + .expect("read identity_scan_states") + }) + } +} + +fn closed() -> PersistenceError { + PersistenceError::backend("persister closed (process ended)") +} + +impl PlatformWalletPersistence for Disk { + fn store_commits_inline(&self) -> bool { + self.with(|p| p.store_commits_inline()) + } + fn persistence_capabilities(&self) -> PersistenceCapabilities { + self.with(|p| p.persistence_capabilities()) + } + fn store_transient_is_reissuable(&self) -> bool { + self.with(|p| p.store_transient_is_reissuable()) + } + fn stores_dashpay_backfill(&self) -> bool { + self.with(|p| p.stores_dashpay_backfill()) + } + fn store(&self, w: WalletId, cs: PlatformWalletChangeSet) -> Result<(), PersistenceError> { + let carries_breadcrumb = cs + .identity_keys + .as_ref() + .is_some_and(|k| k.upserts.values().any(|e| e.derivation_indices.is_some())); + if self.mode != DiskFull::Never + && carries_breadcrumb + && !self.tripped.swap(true, Ordering::SeqCst) + { + self.full.store(true, Ordering::SeqCst); + } + let guard = self.inner.lock().unwrap(); + let Some(p) = guard.as_ref() else { + return Err(closed()); + }; + self.before_write(p); + let r = p.store(w, cs); + drop(guard); + self.record(r) + } + fn flush(&self, w: WalletId) -> Result<(), PersistenceError> { + let guard = self.inner.lock().unwrap(); + let Some(p) = guard.as_ref() else { + return Err(closed()); + }; + self.before_write(p); + let r = p.flush(w); + drop(guard); + self.record(r) + } + fn load(&self) -> Result { + self.with(|p| p.load()) + } + fn persist_tracked_masternodes( + &self, + network: dashcore::Network, + records: &[platform_wallet::masternode::TrackedMasternode], + ) -> Result<(), PersistenceError> { + self.with(|p| p.persist_tracked_masternodes(network, records)) + } + fn load_tracked_masternodes( + &self, + network: dashcore::Network, + ) -> Result, PersistenceError> { + self.with(|p| p.load_tracked_masternodes(network)) + } +} + +// --------------------------------------------------------------------------- +// Wallet, identity and Platform fixtures +// --------------------------------------------------------------------------- + +struct Noop; +impl platform_wallet::events::EventHandler for Noop {} +impl PlatformEventHandler for Noop {} + +fn seed(phrase: &str) -> [u8; 64] { + Mnemonic::from_phrase(phrase).unwrap().to_seed("") +} + +fn master(phrase: &str) -> ExtendedPrivKey { + ExtendedPrivKey::new_master(NET, &seed(phrase)).unwrap() +} + +/// The DIP-9 ECDSA auth key a wallet derives at `(identity_index, key_index)`. +fn derived_key( + phrase: &str, + identity_index: u32, + key_index: u32, + id: KeyID, + level: SecurityLevel, +) -> IdentityPublicKey { + let kp = derive_ecdsa_identity_auth_keypair_from_master( + &master(phrase), + NET, + identity_index, + key_index, + ) + .unwrap(); + IdentityPublicKey::V0(IdentityPublicKeyV0 { + id, + purpose: Purpose::AUTHENTICATION, + security_level: level, + contract_bounds: None, + key_type: KeyType::ECDSA_SECP256K1, + read_only: false, + data: BinaryData::new(kp.public_key.to_vec()), + disabled_at: None, + }) +} + +/// Identity registered by wallet A at index 0 the way registration does it: +/// MASTER, HIGH and CRITICAL auth keys at key ids 0..=2, all from the seed. +fn registered_identity() -> Identity { + let levels = [ + SecurityLevel::MASTER, + SecurityLevel::HIGH, + SecurityLevel::CRITICAL, + ]; + let public_keys = levels + .iter() + .enumerate() + .map(|(i, level)| { + let i = i as u32; + (i, derived_key(MNEMONIC_A, 0, i, i, *level)) + }) + .collect(); + Identity::V0(IdentityV0 { + id: Identifier::from([0x5A; 32]), + public_keys, + balance: 1_000_000, + revision: 1, + }) +} + +fn probe(phrase: &str, index: u32) -> PublicKeyHash { + PublicKeyHash( + derive_identity_auth_key_hash_from_master(&master(phrase), NET, index, 0).unwrap(), + ) +} + +/// Platform: `phrase`'s index 0 answers `at_0`, index 1 holds nothing, and a +/// fetch by id answers `by_id`. +async fn platform( + phrase: &str, + at_0: Option, + by_id: Option, +) -> Arc { + let mut sdk = dash_sdk::SdkBuilder::new_mock() + .with_network(NET) + .build() + .unwrap(); + sdk.mock() + .expect_fetch(probe(phrase, 0), at_0) + .await + .unwrap(); + sdk.mock() + .expect_fetch(probe(phrase, 1), None::) + .await + .unwrap(); + if let Some(identity) = by_id { + sdk.mock() + .expect_fetch(identity.id(), Some(identity)) + .await + .unwrap(); + } + Arc::new(sdk) +} + +type Manager = Arc>; + +/// First launch: a fresh install restores the wallet from its mnemonic. +async fn first_launch( + sdk: Arc, + disk: Arc, + phrase: &str, +) -> (Manager, WalletId) { + let manager = Arc::new(PlatformWalletManager::new(sdk, disk, Arc::new(Noop))); + let wallet = manager + .create_wallet_from_seed_bytes( + NET, + &seed(phrase), + WalletAccountCreationOptions::Default, + Some(0), + ) + .await + .expect("create wallet"); + let id = wallet.wallet_id(); + (manager, id) +} + +/// A later launch: rebuild from what SQLite holds (watch-only wallet, the +/// iOS shape — signing keys come from the Keychain seed via the breadcrumb). +async fn relaunch(sdk: Arc, disk: Arc) -> Manager { + let manager = Arc::new(PlatformWalletManager::new(sdk, disk, Arc::new(Noop))); + manager + .load_from_persistor() + .await + .expect("load from SQLite"); + manager +} + +async fn end_process(manager: Manager, disk: &Disk) { + // No `shutdown()`: it needs a multi-thread runtime. Dropping the manager + // and closing the SQLite handle under it is the process dying. + drop(manager); + disk.close(); +} + +/// Startup's identity step (`start_wallet_subsystems`, startup.rs ~543): a +/// local identity on file and no recorded incomplete scan takes the warm +/// shortcut — no discovery runs this launch. +async fn startup_takes_warm_shortcut(manager: &Manager, w: &WalletId) -> bool { + let wm = manager.wallet_manager_arc(); + let wm = wm.read().await; + let info = wm.get_wallet_info(w).expect("wallet info"); + let known = !info.identity_manager.wallet_identity_ids(w).is_empty(); + known && !info.identity_manager.identity_scan_is_incomplete(w) +} + +async fn held_key_ids(manager: &Manager, w: &WalletId, id: &Identifier) -> Vec { + let wm = manager.wallet_manager_arc(); + let wm = wm.read().await; + wm.get_wallet_info(w) + .expect("wallet info") + .identity_manager + .identity(id) + .expect("identity in memory") + .identity + .public_keys() + .keys() + .copied() + .collect() +} + +fn breadcrumbed(rows: &BTreeMap) -> Vec { + rows.iter() + .filter(|(_, e)| e.derivation_indices.is_some()) + .map(|(k, _)| *k) + .collect() +} + +fn scan_from_zero() -> IdentityDiscoveryOptions { + // The options startup's discovery uses (startup.rs ~835). + IdentityDiscoveryOptions { + start_index: Some(0), + gap_limit: 1, + } +} + +// --------------------------------------------------------------------------- +// A key-row store fails during a restore +// --------------------------------------------------------------------------- + +/// A restore runs discovery, and the disk fills on the write that carries +/// the key breadcrumbs and stays full for the rest of the session (the scan +/// verdict cannot land either). The new identity and its keys go to the store +/// as one changeset, so nothing of it lands: after a restart no key row is +/// left without its breadcrumb, startup does not take the warm shortcut (no +/// identity is on file), and its discovery restores every breadcrumb. +#[tokio::test] +async fn should_not_leave_watch_only_key_rows_when_the_disk_fills_during_a_restore() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("wallet.db"); + let identity = registered_identity(); + let id = identity.id(); + + // Launch 1. + let disk = Disk::open(&path, DiskFull::FromBreadcrumbWriteOn); + let (manager, w) = first_launch( + platform(MNEMONIC_A, Some(identity.clone()), None).await, + disk.clone(), + MNEMONIC_A, + ) + .await; + let err = manager + .get_wallet(&w) + .await + .unwrap() + .identity() + .discover_from_master(scan_from_zero(), &master(MNEMONIC_A)) + .await + .expect_err("the key write fails"); + assert!( + err.to_string() + .contains("identity keys not persisted during discovery"), + "{err}" + ); + assert!( + disk.failures + .lock() + .unwrap() + .iter() + .all(|k| *k == Some(PersistenceErrorKind::Transient)), + "disk full reaches the wallet as Transient" + ); + // In memory the session is fine: all three keys are held. + assert_eq!(held_key_ids(&manager, &w, &id).await, vec![0, 1, 2]); + end_process(manager, &disk).await; + + // Launch 2, disk has space again. + let disk = Disk::open(&path, DiskFull::Never); + let rows = disk.key_rows(&w, &id); + assert!( + rows.values().all(|row| row.derivation_indices.is_some()), + "no key row is left watch-only: {rows:?}" + ); + assert!(disk.scan_verdict(&w).is_none(), "the verdict never landed"); + + let manager = relaunch( + platform(MNEMONIC_A, Some(identity.clone()), None).await, + disk.clone(), + ) + .await; + assert!( + !startup_takes_warm_shortcut(&manager, &w).await, + "no identity is on file, so startup runs discovery" + ); + manager + .get_wallet(&w) + .await + .unwrap() + .identity() + .discover_from_master(scan_from_zero(), &master(MNEMONIC_A)) + .await + .expect("startup's discovery"); + assert_eq!( + breadcrumbed(&disk.key_rows(&w, &id)), + vec![0, 1, 2], + "the startup discovery restores every breadcrumb" + ); + end_process(manager, &disk).await; +} + +// --------------------------------------------------------------------------- +// Keys a refresh finds (added from another device) +// --------------------------------------------------------------------------- + +/// The identity after the user's other device (same seed) added a MEDIUM key +/// at key id 3 — derived, as that device's add-key flow does, at (0, 3). +fn with_key_from_other_device(identity: &Identity) -> Identity { + let mut updated = identity.clone(); + updated.add_public_key(derived_key(MNEMONIC_A, 0, 3, 3, SecurityLevel::MEDIUM)); + updated.set_revision(2); + updated +} + +/// A refresh that finds a key the user's other device added stores its row +/// (without a breadcrumb, which leaves the stored ones as they are), so after +/// a restart the identity holds the key set its stored revision claims. +#[tokio::test] +async fn should_store_a_key_a_refresh_finds_so_it_survives_a_restart() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("wallet.db"); + let identity = registered_identity(); + let id = identity.id(); + let updated = with_key_from_other_device(&identity); + + let disk = Disk::open(&path, DiskFull::Never); + let (manager, w) = first_launch( + platform(MNEMONIC_A, Some(identity.clone()), Some(updated)).await, + disk.clone(), + MNEMONIC_A, + ) + .await; + let iw = manager.get_wallet(&w).await.unwrap(); + iw.identity() + .discover_from_master(scan_from_zero(), &master(MNEMONIC_A)) + .await + .expect("restore"); + iw.identity().refresh_identity(&id).await.expect("refresh"); + assert_eq!(held_key_ids(&manager, &w, &id).await, vec![0, 1, 2, 3]); + let rows = disk.key_rows(&w, &id); + assert!(rows.contains_key(&3), "the refreshed key has a row"); + assert_eq!( + breadcrumbed(&rows), + vec![0, 1, 2], + "the restore's breadcrumbs stay" + ); + drop(iw); + end_process(manager, &disk).await; + + let disk = Disk::open(&path, DiskFull::Never); + let manager = relaunch( + platform(MNEMONIC_A, Some(identity), None).await, + disk.clone(), + ) + .await; + assert_eq!( + held_key_ids(&manager, &w, &id).await, + vec![0, 1, 2, 3], + "key 3 is there after restart" + ); + let wm = manager.wallet_manager_arc(); + let revision = wm + .read() + .await + .get_wallet_info(&w) + .unwrap() + .identity_manager + .identity(&id) + .unwrap() + .identity + .revision(); + assert_eq!(revision, 2); + end_process(manager, &disk).await; +} + +/// A refresh at the cache's revision is taken whole, key states included: +/// Core disables and re-enables a masternode identity's keys at revision 0 +/// (`ReEnableIdentityKeys` in the masternode identity updates), so a cached +/// disable must not stick. The identity here has that shape: revision 0, key +/// 2 disabled when restored, active again in the refresh. The re-enabled row +/// is stored with its breadcrumb kept and survives a restart. +#[tokio::test] +async fn should_store_a_key_core_re_enabled_so_it_survives_a_restart() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("wallet.db"); + let mut active = registered_identity(); + active.set_revision(0); + let id = active.id(); + let mut disabled = active.clone(); + let mut key_2 = disabled.public_keys()[&2].clone(); + key_2.set_disabled_at(1_700_000_000_000); + disabled.add_public_key(key_2); + + let disk = Disk::open(&path, DiskFull::Never); + let (manager, w) = first_launch( + platform(MNEMONIC_A, Some(disabled), Some(active.clone())).await, + disk.clone(), + MNEMONIC_A, + ) + .await; + let iw = manager.get_wallet(&w).await.unwrap(); + iw.identity() + .discover_from_master(scan_from_zero(), &master(MNEMONIC_A)) + .await + .expect("restore"); + assert!(disk.key_rows(&w, &id)[&2].public_key.is_disabled()); + iw.identity().refresh_identity(&id).await.expect("refresh"); + let rows = disk.key_rows(&w, &id); + assert!( + !rows[&2].public_key.is_disabled(), + "the re-enable is stored" + ); + assert_eq!(breadcrumbed(&rows), vec![0, 1, 2], "breadcrumbs stay"); + drop(iw); + end_process(manager, &disk).await; + + let disk = Disk::open(&path, DiskFull::Never); + let manager = relaunch(platform(MNEMONIC_A, None, None).await, disk.clone()).await; + let wm = manager.wallet_manager_arc(); + let held = wm + .read() + .await + .get_wallet_info(&w) + .unwrap() + .identity_manager + .identity(&id) + .unwrap() + .identity + .clone(); + assert!( + !held.public_keys()[&2].is_disabled(), + "key 2 is active after restart" + ); + drop(wm); + end_process(manager, &disk).await; +} + +/// A load at the identity's index (the host's "load identity at index") +/// re-fetches the identity and stores every on-chain key with a breadcrumb +/// when it derives from the seed, a key another device added included. +#[tokio::test] +async fn should_store_a_key_a_load_at_index_finds_with_its_breadcrumb() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("wallet.db"); + let identity = registered_identity(); + let id = identity.id(); + let updated = with_key_from_other_device(&identity); + + let disk = Disk::open(&path, DiskFull::Never); + let (manager, w) = first_launch( + platform(MNEMONIC_A, Some(identity.clone()), None).await, + disk.clone(), + MNEMONIC_A, + ) + .await; + manager + .get_wallet(&w) + .await + .unwrap() + .identity() + .discover_from_master(scan_from_zero(), &master(MNEMONIC_A)) + .await + .expect("restore"); + end_process(manager, &disk).await; + + // Next launch: Platform now holds the other device's key 3. + let disk = Disk::open(&path, DiskFull::Never); + let manager = relaunch( + platform(MNEMONIC_A, Some(updated.clone()), None).await, + disk.clone(), + ) + .await; + manager + .get_wallet(&w) + .await + .unwrap() + .identity() + .load_identity_by_index_from_master(0, &master(MNEMONIC_A)) + .await + .expect("load at index 0") + .expect("found"); + assert_eq!(breadcrumbed(&disk.key_rows(&w, &id)), vec![0, 1, 2, 3]); + end_process(manager, &disk).await; + + let disk = Disk::open(&path, DiskFull::Never); + let manager = relaunch( + platform(MNEMONIC_A, Some(updated), None).await, + disk.clone(), + ) + .await; + assert_eq!(held_key_ids(&manager, &w, &id).await, vec![0, 1, 2, 3]); + end_process(manager, &disk).await; +} diff --git a/packages/rs-platform-wallet-storage/tests/integration/sqlite_identity_keys_reader.rs b/packages/rs-platform-wallet-storage/tests/integration/sqlite_identity_keys_reader.rs index ceae21c6219..6736ec108eb 100644 --- a/packages/rs-platform-wallet-storage/tests/integration/sqlite_identity_keys_reader.rs +++ b/packages/rs-platform-wallet-storage/tests/integration/sqlite_identity_keys_reader.rs @@ -48,6 +48,123 @@ fn key_entry(identity: Identifier, key_id: u32, byte: u8) -> IdentityKeyEntry { } } +/// An upsert without a derivation breadcrumb keeps the one the row holds +/// and takes the rest of the entry: a key's derivation never changes, and a +/// write that only knows the key's new state (a raised limit, a disable) +/// must not turn a wallet-derived key watch-only (a refresh stores the rows +/// of the keys it adds or changes without one). A row without one stays +/// without one. +#[test] +fn should_keep_a_stored_breadcrumb_when_an_upsert_carries_none() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0xD2); + ensure_wallet_meta(&persister, &w); + let id = Identifier::from([0x0C; 32]); + platform_wallet_storage::sqlite::schema::identities::ensure_exists( + &persister.lock_conn_for_test(), + &w, + id.as_slice().try_into().unwrap(), + ) + .unwrap(); + let store = |entries: Vec| { + let mut keys = IdentityKeysChangeSet::default(); + for entry in entries { + keys.upserts + .insert((entry.identity_id, entry.key_id), entry); + } + persister + .store( + w, + PlatformWalletChangeSet { + identity_keys: Some(keys), + ..Default::default() + }, + ) + .unwrap(); + }; + + let derived = key_entry(id, 0, 0x11); + let mut watch_only = key_entry(id, 1, 0x22); + watch_only.derivation_indices = None; + store(vec![derived.clone(), watch_only.clone()]); + + let mut disabled = derived.clone(); + disabled.derivation_indices = None; + if let IdentityPublicKey::V0(key) = &mut disabled.public_key { + key.disabled_at = Some(7); + } + let mut watch_only_disabled = watch_only.clone(); + if let IdentityPublicKey::V0(key) = &mut watch_only_disabled.public_key { + key.disabled_at = Some(8); + } + store(vec![disabled.clone(), watch_only_disabled.clone()]); + drop(persister); + + let p2 = reopen(&path); + let conn = p2.lock_conn_for_test(); + let cs = identity_keys::load_state(&conn, &w, &platform_wallet_storage::LoadCtx::strict()) + .expect("load_state"); + drop(conn); + + let mut expected = disabled; + expected.derivation_indices = derived.derivation_indices; + assert_eq!(cs.upserts.get(&(id, 0)), Some(&expected)); + assert_eq!(cs.upserts.get(&(id, 1)), Some(&watch_only_disabled)); +} + +/// A stored row over the blob size limit is not read into memory to look for +/// a breadcrumb; the upsert replaces it. +#[test] +fn should_replace_an_oversized_stored_row_without_reading_it() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0xD4); + ensure_wallet_meta(&persister, &w); + let id = Identifier::from([0x0E; 32]); + platform_wallet_storage::sqlite::schema::identities::ensure_exists( + &persister.lock_conn_for_test(), + &w, + id.as_slice().try_into().unwrap(), + ) + .unwrap(); + let entry = key_entry(id, 0, 0x11); + persister + .lock_conn_for_test() + .execute( + "INSERT INTO identity_keys \ + (wallet_id, identity_id, key_id, public_key_blob, public_key_hash, derivation_blob) \ + VALUES (?1, ?2, 0, zeroblob(?3), ?4, NULL)", + rusqlite::params![ + w.as_slice(), + id.as_slice(), + (platform_wallet_storage::SIZE_LIMIT_BYTES + 1) as i64, + &entry.public_key_hash[..], + ], + ) + .unwrap(); + + let mut upsert = entry.clone(); + upsert.derivation_indices = None; + let mut keys = IdentityKeysChangeSet::default(); + keys.upserts.insert((id, 0), upsert.clone()); + persister + .store( + w, + PlatformWalletChangeSet { + identity_keys: Some(keys), + ..Default::default() + }, + ) + .unwrap(); + drop(persister); + + let p2 = reopen(&path); + let conn = p2.lock_conn_for_test(); + let cs = identity_keys::load_state(&conn, &w, &platform_wallet_storage::LoadCtx::strict()) + .expect("load_state"); + drop(conn); + assert_eq!(cs.upserts.get(&(id, 0)), Some(&upsert)); +} + /// Identity-key rows round-trip bit-exact into the keyless /// `IdentityKeysChangeSet`. #[test] @@ -150,3 +267,151 @@ fn gk3_corrupt_blob_is_hard_error() { "corrupt public_key_blob must be a typed BincodeDecode; got {result:?}" ); } + +/// Store `entries` under `w` as one identity-keys changeset. +fn store_entries( + persister: &platform_wallet_storage::SqlitePersister, + w: [u8; 32], + entries: Vec, +) { + let mut keys = IdentityKeysChangeSet::default(); + for entry in entries { + keys.upserts + .insert((entry.identity_id, entry.key_id), entry); + } + persister + .store( + w, + PlatformWalletChangeSet { + identity_keys: Some(keys), + ..Default::default() + }, + ) + .expect("store identity keys"); +} + +/// Reopen the database at `path` and read `w`'s identity keys back. +fn reload(path: &std::path::Path, w: &[u8; 32]) -> IdentityKeysChangeSet { + let persister = reopen(path); + let conn = persister.lock_conn_for_test(); + identity_keys::load_state(&conn, w, &platform_wallet_storage::LoadCtx::strict()) + .expect("load_state") +} + +/// `entry` as an upsert that carries no derivation breadcrumb. +fn without_breadcrumb(mut entry: IdentityKeyEntry) -> IdentityKeyEntry { + entry.derivation_indices = None; + entry +} + +/// A kept breadcrumb keeps the wallet scope the row's blob was stored with: +/// an upsert that carries neither a breadcrumb nor a `wallet_id` takes both +/// from the row, so the reloaded entry still names its wallet. +#[test] +fn should_keep_a_stored_wallet_id_with_the_breadcrumb_when_an_upsert_carries_neither() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0xE0); + ensure_wallet_meta(&persister, &w); + let id = Identifier::from([0x20; 32]); + platform_wallet_storage::sqlite::schema::identities::ensure_exists( + &persister.lock_conn_for_test(), + &w, + id.as_slice().try_into().unwrap(), + ) + .unwrap(); + let mut stored = key_entry(id, 0, 0x11); + stored.wallet_id = Some(w); + store_entries(&persister, w, vec![stored.clone()]); + + let mut upsert = without_breadcrumb(stored.clone()); + upsert.wallet_id = None; + if let IdentityPublicKey::V0(key) = &mut upsert.public_key { + key.disabled_at = Some(9); + } + store_entries(&persister, w, vec![upsert.clone()]); + drop(persister); + + let row = reload(&path, &w).upserts.remove(&(id, 0)).expect("row"); + assert_eq!(row.wallet_id, Some(w)); + assert_eq!(row.derivation_indices, stored.derivation_indices); + assert_eq!(row.public_key, upsert.public_key); +} + +/// A stored breadcrumb is lent only to the same public key from a row that +/// passes the reader's checks: an upsert of another key at the same id, or +/// over a row whose blob contradicts its typed hash column, replaces the row +/// without its coordinates. +#[test] +fn should_lend_a_stored_breadcrumb_only_to_the_same_key_from_a_valid_row() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0xE2); + ensure_wallet_meta(&persister, &w); + let id = Identifier::from([0x21; 32]); + platform_wallet_storage::sqlite::schema::identities::ensure_exists( + &persister.lock_conn_for_test(), + &w, + id.as_slice().try_into().unwrap(), + ) + .unwrap(); + store_entries( + &persister, + w, + vec![key_entry(id, 0, 0x11), key_entry(id, 1, 0x22)], + ); + // Row 1's typed hash no longer matches its blob. + persister + .lock_conn_for_test() + .execute( + "UPDATE identity_keys SET public_key_hash = ?1 WHERE key_id = 1", + rusqlite::params![&[0x44u8; 20][..]], + ) + .unwrap(); + + let other_key = without_breadcrumb(key_entry(id, 0, 0x33)); + let same_key = without_breadcrumb(key_entry(id, 1, 0x22)); + store_entries(&persister, w, vec![other_key.clone(), same_key.clone()]); + drop(persister); + + let rows = reload(&path, &w).upserts; + assert_eq!(rows.get(&(id, 0)), Some(&other_key)); + assert_eq!(rows.get(&(id, 1)), Some(&same_key)); +} + +/// A stored row within the size limit that does not decode holds no +/// breadcrumb to keep: the upsert replaces it and the store succeeds. +#[test] +fn should_replace_a_stored_row_that_does_not_decode() { + let (persister, _tmp, path) = fresh_persister(); + let w = wid(0xE1); + ensure_wallet_meta(&persister, &w); + // The identity and key id the crafted blob itself names. + let id = Identifier::from([0xAA; 32]); + platform_wallet_storage::sqlite::schema::identities::ensure_exists( + &persister.lock_conn_for_test(), + &w, + id.as_slice().try_into().unwrap(), + ) + .unwrap(); + let upsert = without_breadcrumb(key_entry(id, 0, 0x11)); + let crafted = identity_keys::crafted_entry_blob_with_bad_pk_bincode_for_test(); + assert!(identity_keys::decode_entry(&crafted).is_err()); + persister + .lock_conn_for_test() + .execute( + "INSERT INTO identity_keys \ + (wallet_id, identity_id, key_id, public_key_blob, public_key_hash, derivation_blob) \ + VALUES (?1, ?2, 0, ?3, ?4, NULL)", + rusqlite::params![ + w.as_slice(), + id.as_slice(), + crafted, + &upsert.public_key_hash[..] + ], + ) + .unwrap(); + + store_entries(&persister, w, vec![upsert.clone()]); + drop(persister); + + assert_eq!(reload(&path, &w).upserts.get(&(id, 0)), Some(&upsert)); +} diff --git a/packages/rs-platform-wallet/src/changeset/changeset.rs b/packages/rs-platform-wallet/src/changeset/changeset.rs index 8b8f8418fe2..e8beae1b385 100644 --- a/packages/rs-platform-wallet/src/changeset/changeset.rs +++ b/packages/rs-platform-wallet/src/changeset/changeset.rs @@ -40,6 +40,7 @@ use crate::wallet::platform_wallet::WalletId; use dpp::balances::credits::Credits; use dpp::identity::accessors::IdentityGettersV0; +use dpp::identity::identity_public_key::accessors::v0::IdentityPublicKeyGettersV0; use dpp::identity::{IdentityPublicKey, KeyID}; use dpp::prelude::{Identifier, Revision}; @@ -1006,8 +1007,10 @@ pub struct IdentityKeyDerivationIndices { /// A derivation breadcrumb as the raw `(wallet_id, identity_index, /// key_index)` triple passed to `ManagedIdentity::add_key` / `add_keys`. -/// `Some` lets the client re-derive the private key from the wallet seed; -/// `None` marks a watch-only key. +/// `Some` lets the client re-derive the private key from the wallet seed. +/// `None` gives no coordinates: a key the persister does not hold yet is +/// watch-only, and one it holds keeps its breadcrumb (see +/// [`IdentityKeyEntry::derivation_indices`]). pub type KeyDerivationBreadcrumb = ([u8; 32], u32, u32); /// One public key paired with its derivation breadcrumb — the unit @@ -1017,12 +1020,13 @@ pub type KeyDerivationBreadcrumb = ([u8; 32], u32, u32); /// Discovery derives a candidate scalar, validates it against the on-chain /// key, and emits a breadcrumb (the DIP-9 coordinates) only when it matches. /// The scalar itself is never carried out — the client derives the key on -/// demand from the Keychain seed at the breadcrumb path. `breadcrumb` is -/// `None` for a watch-only key. +/// demand from the Keychain seed at the breadcrumb path. A `None` +/// `breadcrumb` gives no coordinates (see +/// [`IdentityKeyEntry::derivation_indices`]). pub struct KeyWithBreadcrumb { /// The DPP public-key record. pub key: dpp::identity::IdentityPublicKey, - /// Derivation coordinates for re-derivable keys; `None` for watch-only. + /// Derivation coordinates for re-derivable keys; `None` gives none. pub breadcrumb: Option, } @@ -1032,8 +1036,9 @@ pub struct KeyWithBreadcrumb { /// derivation that produced it. No private material crosses here: the client /// derives the key on demand from the Keychain seed at the breadcrumb path /// (`m/9'/coin'/5'/0'/ECDSA'/identity_index'/key_index'`). When -/// `derivation_indices` is `None` the key is watch-only from this wallet's -/// point of view. +/// `derivation_indices` is `None`, the entry supplies no new coordinates; +/// an update preserves any breadcrumb the persister already holds (see the +/// field). #[derive(Clone, Debug, PartialEq)] #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] pub struct IdentityKeyEntry { @@ -1047,12 +1052,22 @@ pub struct IdentityKeyEntry { /// Precomputed on the Rust side so clients don't need /// RIPEMD-160 implementations of their own. pub public_key_hash: [u8; 20], - /// The wallet whose mnemonic derives this key. `None` for - /// watched-only identities with no wallet association. + /// The wallet whose mnemonic derives this key. It travels with + /// [`Self::derivation_indices`]: `None` alongside them gives no wallet + /// either, and an update to a held row keeps the `wallet_id` the row + /// already holds, as it keeps the breadcrumb. pub wallet_id: Option<[u8; 32]>, /// DIP-9 `(identity_index, key_index)` pair the client needs to - /// re-derive the private key. `None` means "view-only, no - /// private key recoverable". + /// re-derive the private key. + /// + /// `None` gives no coordinates: for a row the persister does not hold + /// yet it means "view-only, no private key recoverable", but for an + /// update to a held row it means "no new coordinates", and the + /// persister keeps the breadcrumb (and its `wallet_id`) the row already + /// holds. A refresh emits such entries for keys a wallet derived, so a + /// persister that replaced the held breadcrumb with `None` would lose + /// the key's signing recovery. The SQLite writer, the Swift and the + /// Kotlin persisters all keep it, and changeset merges do the same. pub derivation_indices: Option, } @@ -1077,8 +1092,29 @@ impl Merge for IdentityKeysChangeSet { // Last-write-wins per composite key. Matches the discipline // used by other scalar-entry changesets in this module; the // "don't emit both upsert + remove for the same key in one - // mutation" invariant is enforced on the emitter side. - self.upserts.extend(other.upserts); + // mutation" invariant is enforced on the emitter side. A later + // entry without a derivation breadcrumb (a refresh, a limits + // update) keeps the earlier one's, as every persister keeps a + // stored one: a buffered write that only knows the key's new state + // must not turn a wallet-derived key watch-only. Only for the same + // public key in the same wallet scope (or none): the coordinates + // derive that key, not another one under the same row id. + for (id, mut entry) in other.upserts { + if entry.derivation_indices.is_none() { + if let Some(earlier) = self.upserts.get(&id) { + let same_key = earlier.public_key_hash == entry.public_key_hash + && earlier.public_key.key_type() == entry.public_key.key_type() + && earlier.public_key.data() == entry.public_key.data(); + let same_scope = + entry.wallet_id.is_none() || entry.wallet_id == earlier.wallet_id; + if earlier.derivation_indices.is_some() && same_key && same_scope { + entry.wallet_id = earlier.wallet_id; + entry.derivation_indices = earlier.derivation_indices; + } + } + } + self.upserts.insert(id, entry); + } self.removed.extend(other.removed); } @@ -2527,6 +2563,100 @@ mod tests { } } + /// A later key entry without a derivation breadcrumb keeps the earlier + /// one's and takes the rest; one with a breadcrumb replaces it whole. + #[test] + fn should_keep_an_earlier_breadcrumb_when_a_later_key_entry_has_none() { + use dpp::identity::identity_public_key::v0::IdentityPublicKeyV0; + let id = Identifier::from([0x54; 32]); + let entry = |disabled_at: Option, key_index: Option| IdentityKeyEntry { + identity_id: id, + key_id: 3, + public_key: IdentityPublicKeyV0 { + id: 3, + disabled_at, + ..Default::default() + } + .into(), + public_key_hash: [3; 20], + wallet_id: key_index.map(|_| [9; 32]), + derivation_indices: key_index.map(|key_index| IdentityKeyDerivationIndices { + identity_index: 0, + key_index, + }), + }; + let changeset = |entry: IdentityKeyEntry| { + let mut cs = IdentityKeysChangeSet::default(); + cs.upserts.insert((id, 3), entry); + cs + }; + + let mut merged = changeset(entry(None, Some(3))); + merged.merge(changeset(entry(Some(7), None))); + assert_eq!(merged.upserts[&(id, 3)], entry(Some(7), Some(3))); + + merged.merge(changeset(entry(Some(8), Some(4)))); + assert_eq!(merged.upserts[&(id, 3)], entry(Some(8), Some(4))); + + // Nothing to keep: the later entry stands as it is. + let mut merged = changeset(entry(None, None)); + merged.merge(changeset(entry(Some(9), None))); + assert_eq!(merged.upserts[&(id, 3)], entry(Some(9), None)); + } + + /// A later entry without a breadcrumb takes the earlier one's only for + /// the same public key (data and hash) in the same wallet scope or none: + /// a replacement key under the same row id, or an entry scoped to + /// another wallet, stands as it is. + #[test] + fn should_lend_a_breadcrumb_only_to_the_same_public_key_in_the_same_scope() { + use dpp::identity::identity_public_key::v0::IdentityPublicKeyV0; + let id = Identifier::from([0x55; 32]); + let entry = |byte: u8, hash: u8, wallet_id: Option<[u8; 32]>, key_index: Option| { + IdentityKeyEntry { + identity_id: id, + key_id: 3, + public_key: IdentityPublicKeyV0 { + id: 3, + data: vec![byte; 33].into(), + ..Default::default() + } + .into(), + public_key_hash: [hash; 20], + wallet_id, + derivation_indices: key_index.map(|key_index| IdentityKeyDerivationIndices { + identity_index: 0, + key_index, + }), + } + }; + let merged = |later: IdentityKeyEntry| { + let mut cs = IdentityKeysChangeSet::default(); + cs.upserts + .insert((id, 3), entry(1, 1, Some([9; 32]), Some(3))); + let mut next = IdentityKeysChangeSet::default(); + next.upserts.insert((id, 3), later); + cs.merge(next); + cs.upserts[&(id, 3)].clone() + }; + + for later in [entry(1, 1, None, None), entry(1, 1, Some([9; 32]), None)] { + assert_eq!(merged(later), entry(1, 1, Some([9; 32]), Some(3))); + } + for later in [ + // A replacement key: other data, other hash. + entry(2, 2, None, None), + // The same hash over other data. + entry(2, 1, None, None), + // The same data with another hash. + entry(1, 2, None, None), + // Another wallet's scope. + entry(1, 1, Some([8; 32]), None), + ] { + assert_eq!(merged(later.clone()), later); + } + } + #[test] fn should_merge_balance_and_watermark_under_the_same_revision_gate() { let id = Identifier::from([0x53; 32]); diff --git a/packages/rs-platform-wallet/src/error.rs b/packages/rs-platform-wallet/src/error.rs index 5696d681d2b..a1e0103c7ab 100644 --- a/packages/rs-platform-wallet/src/error.rs +++ b/packages/rs-platform-wallet/src/error.rs @@ -3,6 +3,7 @@ use dash_sdk::platform::dashpay::InvitationError; use dashcore::sml::llmq_entry_verification::LLMQEntryVerificationStatus; use dpp::address_funds::PlatformAddress; use dpp::consensus::state::address_funds::AddressInvalidNonceError; +use dpp::consensus::state::identity::invalid_identity_revision_error::InvalidIdentityRevisionError; use dpp::consensus::ConsensusError; use dpp::fee::Credits; use dpp::identifier::Identifier; @@ -1219,6 +1220,21 @@ pub fn as_address_invalid_nonce(error: &dash_sdk::Error) -> Option<&AddressInval } } +/// Extract the `InvalidIdentityRevisionError` (DPP consensus code 40203) +/// when Platform rejected an identity update whose revision is not one above +/// the identity's, exposing `identity_id()` and Drive's `current_revision()`; +/// `None` otherwise. +pub fn as_identity_revision_invalid( + error: &dash_sdk::Error, +) -> Option<&InvalidIdentityRevisionError> { + use dpp::consensus::state::state_error::StateError; + + match error.consensus_error() { + Some(ConsensusError::StateError(StateError::InvalidIdentityRevisionError(e))) => Some(e), + _ => None, + } +} + /// Promote a nonce-rejection SDK error to the typed /// [`PlatformWalletError::AddressNonceMismatch`] so callers can recover /// `expected_nonce` and retry, instead of receiving the rejection flattened diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs b/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs index 6407953c2a6..3bf6ed652cf 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs @@ -6,6 +6,7 @@ use key_wallet::bip32::ExtendedPrivKey; use crate::error::PlatformWalletError; use crate::wallet::identity::network::contact_requests::budget_spent; +use crate::BlockTime; use super::*; @@ -412,10 +413,13 @@ impl IdentityWallet { // Query Platform for an identity registered with this key // hash. No locks are held during this network call. - let fetch_result = Identity::fetch(&self.sdk, PublicKeyHash(key_hash_array)).await; + let fetch_result = + Identity::fetch_with_metadata(&self.sdk, PublicKeyHash(key_hash_array), None) + .await; match fetch_result { - Ok(Some(identity)) => { + Ok((Some(identity), metadata)) => { + let block_time = BlockTime::from(metadata); let identity_id = identity.id(); // Derive + verify a candidate for every on-chain key @@ -447,11 +451,12 @@ impl IdentityWallet { })?; let is_new = info_guard.identity_manager.identity(&identity_id).is_none(); if is_new { - info_guard.identity_manager.add_identity( + // Stored below, with its verified keys, as one + // changeset. + info_guard.identity_manager.add_identity_unstored( identity.clone(), identity_index, wallet_id, - &self.persister, )?; } @@ -459,7 +464,7 @@ impl IdentityWallet { .identity_manager .managed_identity_mut(&identity_id) { - managed.set_status(IdentityStatus::Active, &self.persister); + managed.status = IdentityStatus::Active; managed.wallet_id = Some(wallet_id); // Breadcrumbs for every re-derivable key (not just the // MASTER key) so the client (iOS Keychain) can @@ -468,9 +473,17 @@ impl IdentityWallet { // imported identity cannot sign with its HIGH / // CRITICAL authentication keys. A failed persist here // would silently leave the identity watch-only after - // restart, so surface it (matching `add_identity` above). + // restart, so surface it. A known identity is folded + // as loading and refresh fold it (a fetch older than + // the cache keeps the cache's revision, keys and + // disables; the balance follows the proof's block). managed - .add_keys(key_decisions, &self.persister) + .store_fetched_identity_keys( + &identity, + key_decisions, + block_time, + &self.persister, + ) .map_err(|e| { PlatformWalletError::Persistence(format!( "identity keys not persisted during discovery: {e}" @@ -484,7 +497,7 @@ impl IdentityWallet { } tally.record_sighting(); } - Ok(None) => { + Ok((None, _)) => { tally.record_miss(); } Err(e) => { diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/lagging_fetch_tests.rs b/packages/rs-platform-wallet/src/wallet/identity/network/lagging_fetch_tests.rs new file mode 100644 index 00000000000..91e447983e3 --- /dev/null +++ b/packages/rs-platform-wallet/src/wallet/identity/network/lagging_fetch_tests.rs @@ -0,0 +1,413 @@ +//! Network ordering and the fetch fold, end to end: a load by index or a +//! refresh answered by a node behind the block an update landed in, and a +//! fetch that lands between an update's snapshot and its post-broadcast +//! apply. +//! +//! What is real and what is modelled: +//! - The fetches are real: `load_identity_by_index_from_master` (the FFI +//! `platform_wallet_load_identity_at_index` path DashConnect calls before +//! every update) and `refresh_identity`, answered by mock DAPI expectations. +//! A lagging node is modelled by answering with the identity as it was one +//! block before the update landed. The mock reports every proof at height +//! 0, which stands for "below the height the cache's balance came from". +//! - The update's broadcast is not run: the mock SDK cannot answer +//! `broadcast_and_wait`. What runs instead is the update's post-broadcast +//! local apply, `apply_landed_update`. +//! - The key-limits update's broadcast is not run either; its post-broadcast +//! apply (`ManagedIdentity::replace_key`) is called as is. +//! - The balance a client read after the update (`refresh_identity_balance`) +//! is recorded with `persist_refreshed_balance`, the call that path makes, +//! at height 100 instead of the mock's flat 0. + +use std::sync::Arc; + +use dash_sdk::platform::transition::update_identity_key_limits::raised_key_limits; +use dash_sdk::platform::types::identity::PublicKeyHash; +use dpp::identity::accessors::{IdentityGettersV0, IdentitySettersV0}; +use dpp::identity::identity_public_key::accessors::v0::{ + IdentityPublicKeyGettersV0, IdentityPublicKeySettersV0, +}; +use dpp::identity::identity_public_key::accessors::v1::IdentityPublicKeyGettersV1; +use dpp::identity::identity_public_key::v0::IdentityPublicKeyV0; +use dpp::identity::{Identity, IdentityPublicKey, KeyType, Purpose, SecurityLevel}; +use dpp::prelude::Identifier; +use dpp::version::PlatformVersion; +use key_wallet::bip32::ExtendedPrivKey; +use key_wallet::wallet::initialization::WalletAccountCreationOptions; +use key_wallet::Network; + +use super::super::identity_handle::{derive_identity_auth_key_hash_from_master, MASTER_KEY_INDEX}; +use super::IdentityWallet; +use crate::events::{EventHandler, PlatformEventHandler}; +use crate::wallet::persister::NoPlatformPersistence; +use crate::{BlockTime, PlatformWallet, PlatformWalletManager}; + +const SEED: [u8; 64] = [7; 64]; +/// Balance before the update, and after it paid its fee. +const BALANCE_BEFORE: u64 = 10_000_000; +const BALANCE_AFTER: u64 = 9_000_000; +/// The block the update landed in; a lagging node answers from the one below. +const LANDED_AT: u64 = 100; + +fn key(id: u32, level: SecurityLevel) -> IdentityPublicKey { + IdentityPublicKeyV0 { + id, + purpose: Purpose::AUTHENTICATION, + security_level: level, + key_type: KeyType::ECDSA_SECP256K1, + data: vec![id as u8 + 1; KeyType::ECDSA_SECP256K1.default_size()].into(), + ..Default::default() + } + .into() +} + +/// The identity on Platform at revision 1: MASTER 0, HIGH 1, CRITICAL 2. +fn identity_at_revision_1() -> Identity { + let mut identity = Identity::default_versioned(PlatformVersion::latest()).unwrap(); + identity.add_public_key(key(0, SecurityLevel::MASTER)); + identity.add_public_key(key(1, SecurityLevel::HIGH)); + identity.add_public_key(key(2, SecurityLevel::CRITICAL)); + identity.set_revision(1); + identity.set_balance(BALANCE_BEFORE); + identity +} + +/// The update every scenario lands: add HIGH key 3, disable key 1. +fn update_payload() -> (Vec, Vec) { + (vec![key(3, SecurityLevel::HIGH)], vec![1]) +} + +/// The identity on Platform once the update landed at revision 2. +fn identity_at_revision_2() -> Identity { + let mut identity = identity_at_revision_1(); + let mut key_1 = identity.public_keys()[&1].clone(); + key_1.set_disabled_at(1_700_000_000_000); + identity.add_public_key(key_1); + identity.add_public_key(key(3, SecurityLevel::HIGH)); + identity.set_revision(2); + identity.set_balance(BALANCE_AFTER); + identity +} + +fn master() -> ExtendedPrivKey { + ExtendedPrivKey::new_master(Network::Testnet, &SEED).unwrap() +} + +/// The unique-hash probe a load by index 0 sends. +fn probe() -> PublicKeyHash { + PublicKeyHash( + derive_identity_auth_key_hash_from_master(&master(), Network::Testnet, 0, MASTER_KEY_INDEX) + .unwrap(), + ) +} + +/// A wallet from `SEED` holding the identity at revision 1 at index 0, as a +/// registration or an earlier load leaves it. +async fn wallet_on(sdk: dash_sdk::Sdk) -> Arc { + struct Events; + impl EventHandler for Events {} + impl PlatformEventHandler for Events {} + let manager = PlatformWalletManager::new( + Arc::new(sdk), + Arc::new(NoPlatformPersistence), + Arc::new(Events), + ); + let wallet = manager + .create_wallet_from_seed_bytes( + Network::Testnet, + &SEED, + WalletAccountCreationOptions::Default, + Some(0), + ) + .await + .unwrap(); + let mut wm = manager.wallet_manager.write().await; + wm.get_wallet_info_mut(&wallet.wallet_id()) + .unwrap() + .identity_manager + .add_identity( + identity_at_revision_1(), + 0, + wallet.wallet_id(), + &wallet.identity().persister, + ) + .unwrap(); + drop(wm); + wallet +} + +async fn cached(wallet: &IdentityWallet) -> Identity { + let wm = wallet.wallet_manager.read().await; + wm.get_wallet_info(&wallet.wallet_id) + .unwrap() + .identity_manager + .identity(&Identifier::default()) + .unwrap() + .identity + .clone() +} + +/// What an identity update reads from the cache when it starts: the +/// revision its transition carries (cache + 1, the same in both trees). +async fn update_snapshot(wallet: &IdentityWallet) -> u64 { + cached(wallet).await.revision() + 1 +} + +/// The post-broadcast local apply of an update Drive accepted at +/// `transition_revision`. +async fn land_update( + wallet: &IdentityWallet, + transition_revision: u64, + added: Vec, + disabled: Vec, +) { + wallet + .apply_landed_update( + &Identifier::default(), + Some(0), + transition_revision, + added, + disabled, + ) + .await + .unwrap(); +} + +/// A client read the balance after the update from an up-to-date node. +async fn balance_read_after_update(wallet: &IdentityWallet) { + let mut wm = wallet.wallet_manager.write().await; + let managed = wm + .get_wallet_info_mut(&wallet.wallet_id) + .unwrap() + .identity_manager + .managed_identity_mut(&Identifier::default()) + .unwrap(); + managed + .persist_refreshed_balance( + BALANCE_AFTER, + BlockTime::new(LANDED_AT, 0, 1_700_000_000_000), + &wallet.persister, + ) + .unwrap(); +} + +#[derive(Debug)] +struct Seen { + revision: u64, + key_1_disabled: bool, + has_key_3: bool, + balance: u64, +} + +fn seen(identity: &Identity) -> Seen { + Seen { + revision: identity.revision(), + key_1_disabled: identity.public_keys()[&1].is_disabled(), + has_key_3: identity.public_keys().contains_key(&3), + balance: identity.balance(), + } +} + +#[derive(Clone, Copy, Debug)] +enum Fetch { + /// `load_identity_by_index_from_master(0)`: DashConnect before an + /// update, pull-to-refresh in Storage Explorer. + Load, + /// `refresh_identity`: the post-trade refresh of the DPNS marketplace. + Refresh, +} + +/// A mock SDK whose one node answers `fetch` with `answer`. +async fn sdk_answering(fetch: Fetch, answer: Identity) -> dash_sdk::Sdk { + let mut sdk = dash_sdk::SdkBuilder::new_mock().build().unwrap(); + match fetch { + Fetch::Load => { + sdk.mock() + .expect_fetch(probe(), Some(answer)) + .await + .unwrap(); + } + Fetch::Refresh => { + sdk.mock() + .expect_fetch(Identifier::default(), Some(answer)) + .await + .unwrap(); + } + } + sdk +} + +async fn run_fetch(wallet: &IdentityWallet, fetch: Fetch) { + match fetch { + Fetch::Load => { + wallet + .load_identity_by_index_from_master(0, &master()) + .await + .unwrap() + .unwrap(); + } + Fetch::Refresh => { + wallet + .refresh_identity(&Identifier::default()) + .await + .unwrap(); + } + } +} + +/// A load or refresh answered by a node one block behind an update (the +/// identity at revision 1, after the update landed at 2) keeps the cache at +/// the update: its revision, the disable, the added key and the balance the +/// update left, so the next update carries the revision Drive takes. +#[tokio::test] +async fn should_keep_a_landed_update_when_a_lagging_fetch_answers_older() { + for fetch in [Fetch::Load, Fetch::Refresh] { + let wallet = wallet_on(sdk_answering(fetch, identity_at_revision_1()).await).await; + let wallet = wallet.identity(); + + let transition_revision = update_snapshot(wallet).await; + let (added, disabled) = update_payload(); + land_update(wallet, transition_revision, added, disabled).await; + balance_read_after_update(wallet).await; + let after_update = seen(&cached(wallet).await); + assert_eq!(after_update.revision, 2, "{fetch:?}"); + assert!(after_update.key_1_disabled && after_update.has_key_3); + assert_eq!(after_update.balance, BALANCE_AFTER); + + run_fetch(wallet, fetch).await; + let after_fetch = seen(&cached(wallet).await); + assert_eq!(after_fetch.revision, 2, "{fetch:?}: cache revision kept"); + assert!(after_fetch.key_1_disabled, "{fetch:?}: disabled key stays"); + assert!(after_fetch.has_key_3, "{fetch:?}: added key stays"); + assert_eq!( + after_fetch.balance, BALANCE_AFTER, + "{fetch:?}: balance from the newer proof stays" + ); + // The next update carries cache + 1, which Drive takes: chain + 1. + assert_eq!(after_fetch.revision + 1, 3, "{fetch:?}"); + } +} + +/// The identity is at revision 2 (key 1 disabled), and an update at 3 +/// disables key 1 again, which only moves its `disabled_at`. Two loads or +/// refreshes in a row answered by a node still at revision 2 keep the cache +/// at the update: an older answer never rewrites a disable the cache holds, +/// so the second one finds nothing the first one changed. +#[tokio::test] +async fn should_keep_a_second_disable_when_two_lagging_fetches_answer_older() { + for fetch in [Fetch::Load, Fetch::Refresh] { + let wallet = wallet_on(sdk_answering(fetch, identity_at_revision_2()).await).await; + let wallet = wallet.identity(); + { + // The cache as a fetch at revision 2 left it. + let mut wm = wallet.wallet_manager.write().await; + wm.get_wallet_info_mut(&wallet.wallet_id) + .unwrap() + .identity_manager + .managed_identity_mut(&Identifier::default()) + .unwrap() + .identity = identity_at_revision_2(); + } + let transition_revision = update_snapshot(wallet).await; + assert_eq!(transition_revision, 3); + land_update(wallet, transition_revision, vec![], vec![1]).await; + let after_update = cached(wallet).await; + assert_eq!(after_update.revision(), 3, "{fetch:?}"); + + for answer in 1..=2 { + run_fetch(wallet, fetch).await; + let held = cached(wallet).await; + assert_eq!(held.revision(), 3, "{fetch:?}: answer {answer}"); + assert_eq!( + held.public_keys()[&1].disabled_at(), + after_update.public_keys()[&1].disabled_at(), + "{fetch:?}: answer {answer} keeps the second disable" + ); + } + } +} + +/// A key-limits update raises key 3's budget 1_000 -> 5_000 and expiry +/// T1 -> T2 (no revision is claimed); a load or refresh answered by a node +/// one block behind, with the old limits at the same revision, keeps the +/// raised ones, so the next top-up is computed from what the chain holds. +#[tokio::test] +async fn should_keep_raised_key_limits_when_a_lagging_fetch_answers_older() { + const T1: u64 = 1_800_000_000_000; + const T2: u64 = 1_900_000_000_000; + let limited = + |budget, expires| key(3, SecurityLevel::HIGH).with_limits(Some(budget), Some(expires)); + let mut before = identity_at_revision_1(); + before.add_public_key(limited(1_000, T1)); + + for fetch in [Fetch::Load, Fetch::Refresh] { + let wallet = wallet_on(sdk_answering(fetch, before.clone()).await).await; + let wallet = wallet.identity(); + { + // The key as registered: the cache took it with the update that added it. + let mut wm = wallet.wallet_manager.write().await; + wm.get_wallet_info_mut(&wallet.wallet_id) + .unwrap() + .identity_manager + .managed_identity_mut(&Identifier::default()) + .unwrap() + .identity = before.clone(); + } + { + // Post-broadcast apply of `update_identity_key_limits_with_external_signer`. + let mut wm = wallet.wallet_manager.write().await; + wm.get_wallet_info_mut(&wallet.wallet_id) + .unwrap() + .identity_manager + .managed_identity_mut(&Identifier::default()) + .unwrap() + .replace_key(limited(5_000, T2), &wallet.persister) + .unwrap(); + } + + run_fetch(wallet, fetch).await; + let held = cached(wallet).await; + let key_3 = held.public_keys()[&3].clone(); + assert_eq!(key_3.total_budget(), Some(5_000), "{fetch:?}: budget kept"); + assert_eq!(key_3.expires_at(), Some(T2), "{fetch:?}: expiry kept"); + // The next top-up of 500 is computed from the cached key: above the + // 5_000 the chain holds, so consensus takes it as a raise. + let next_top_up = raised_key_limits(&held, 3, Some(500), None).unwrap(); + assert_eq!(next_top_up.0, Some(5_500), "{fetch:?}"); + } +} + +/// The update read revision 1 from the cache (transition at 2); while it +/// waits for its result, a load or refresh is answered by an up-to-date +/// node, which already holds the update (revision 2); then the +/// post-broadcast apply runs. The cache stays at the chain's revision, so +/// the next update is not refused, and keeps the on-chain `disabled_at` +/// rather than the local clock's. +#[tokio::test] +async fn should_not_put_the_cache_past_the_chain_when_a_fetch_lands_before_the_apply() { + for fetch in [Fetch::Load, Fetch::Refresh] { + let wallet = wallet_on(sdk_answering(fetch, identity_at_revision_2()).await).await; + let wallet = wallet.identity(); + + let transition_revision = update_snapshot(wallet).await; + assert_eq!(transition_revision, 2); + // ... nonce fetch, broadcast, the block commits at revision 2 ... + run_fetch(wallet, fetch).await; + assert_eq!( + cached(wallet).await.revision(), + 2, + "{fetch:?}: fetch took the landed update" + ); + let (added, disabled) = update_payload(); + land_update(wallet, transition_revision, added, disabled).await; + + let held = cached(wallet).await; + assert_eq!(held.revision(), 2, "{fetch:?}: cache at the chain"); + assert_eq!( + held.public_keys()[&1].disabled_at(), + identity_at_revision_2().public_keys()[&1].disabled_at(), + "{fetch:?}: on-chain disable time kept" + ); + assert!(held.public_keys().contains_key(&3), "{fetch:?}"); + } +} diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/loading.rs b/packages/rs-platform-wallet/src/wallet/identity/network/loading.rs index 3133c1e2bea..1e032ed2690 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/loading.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/loading.rs @@ -7,6 +7,7 @@ use dpp::prelude::Identifier; use key_wallet::bip32::ExtendedPrivKey; use crate::error::PlatformWalletError; +use crate::BlockTime; use super::*; @@ -206,16 +207,19 @@ impl IdentityWallet { }; // Query Platform for an identity registered with this key hash. - let identity = match Identity::fetch(&self.sdk, PublicKeyHash(key_hash_array)).await { - Ok(Some(identity)) => identity, - Ok(None) => return Ok(None), - Err(e) => { - return Err(PlatformWalletError::InvalidIdentityData(format!( - "Failed to fetch identity at index {}: {}", - identity_index, e - ))); - } - }; + let (mut identity, block_time) = + match Identity::fetch_with_metadata(&self.sdk, PublicKeyHash(key_hash_array), None) + .await + { + Ok((Some(identity), metadata)) => (identity, BlockTime::from(metadata)), + Ok((None, _)) => return Ok(None), + Err(e) => { + return Err(PlatformWalletError::InvalidIdentityData(format!( + "Failed to fetch identity at index {}: {}", + identity_index, e + ))); + } + }; let identity_id = identity.id(); @@ -243,11 +247,11 @@ impl IdentityWallet { ) })?; if info.identity_manager.identity(&identity_id).is_none() { - info.identity_manager.add_identity( + // Stored below, with its verified keys, as one changeset. + info.identity_manager.add_identity_unstored( identity.clone(), identity_index, wallet_id, - &self.persister, )?; } @@ -257,22 +261,31 @@ impl IdentityWallet { // new ones, and without this a re-load kept serving the // stale key set (keys added from another device never // appeared, and contact requests referencing them failed - // key-index validation). Same replace `refresh_identity` - // does; the `set_status` below persists the full snapshot - // changeset, so the updated keys reach the store. - managed.identity = identity.clone(); - managed.set_status(IdentityStatus::Active, &self.persister); + // key-index validation). Same fold `refresh_identity` + // does: a fetch older than the cache (a node behind the + // block an update landed in) keeps the cache's revision, key + // set and every disable, and the balance follows the proof's + // block, not the revision. The keys and the folded scalar + // snapshot reach the store together. + managed.status = IdentityStatus::Active; managed.wallet_id = Some(wallet_id); // Breadcrumbs for every re-derivable key (was MASTER-only). A // failed persist would silently leave the identity watch-only // after restart, so surface it rather than swallow. managed - .add_keys(key_decisions, &self.persister) + .store_fetched_identity_keys( + &identity, + key_decisions, + block_time, + &self.persister, + ) .map_err(|e| { PlatformWalletError::Persistence(format!( "identity keys not persisted during load: {e}" )) })?; + // The cache as folded. + identity = managed.identity.clone(); } } @@ -305,10 +318,22 @@ impl IdentityWallet { /// re-fetching it from Platform. /// /// The identity must already exist in the [`IdentityManager`]. Its - /// on-chain state (keys, balance, revision) is replaced with the latest - /// version from Platform and the status is set to `Active`. + /// on-chain state (keys, revision) is replaced with the latest version + /// from Platform and the status is set to `Active`, key states included + /// (Core re-enables a masternode identity's keys at revision 0); a key + /// limit the cache holds higher stays. The + /// balance, at any revision, is taken only when the fetch's proof is at a + /// block above the cache's balance watermark. A fetch at a lower revision + /// than the cache's was answered before an update the cache already + /// holds (a node behind the block it landed in): the cache keeps its + /// revision, key set and every disable, and takes only a disable the + /// fetch shows on a key it holds enabled, or a raised limit + /// (`ManagedIdentity::take_fetched_identity`). The row of every key the + /// fold adds or changes is stored with the scalar snapshot, without a + /// derivation breadcrumb (the persisters keep the one they hold); a + /// failed store is logged, as the status write it replaces was. /// - /// Returns the refreshed identity. + /// Returns the identity the cache holds afterwards. /// /// # Errors /// @@ -337,20 +362,21 @@ impl IdentityWallet { } // Fetch the latest state from Platform. - let identity = Identity::fetch(&self.sdk, *identity_id) + let (identity, metadata) = Identity::fetch_with_metadata(&self.sdk, *identity_id, None) .await .map_err(|e| { PlatformWalletError::InvalidIdentityData(format!( "Failed to fetch identity {} from Platform: {}", identity_id, e )) - })? - .ok_or_else(|| { - PlatformWalletError::InvalidIdentityData(format!( - "Identity {} not found on Platform", - identity_id - )) })?; + let mut identity = identity.ok_or_else(|| { + PlatformWalletError::InvalidIdentityData(format!( + "Identity {} not found on Platform", + identity_id + )) + })?; + let block_time = BlockTime::from(metadata); // Update the managed identity. { @@ -361,8 +387,14 @@ impl IdentityWallet { ) })?; if let Some(managed) = info.identity_manager.managed_identity_mut(identity_id) { - managed.identity = identity.clone(); - managed.set_status(IdentityStatus::Active, &self.persister); + managed.status = IdentityStatus::Active; + if let Err(e) = + managed.store_fetched_identity(&identity, block_time, &self.persister) + { + tracing::error!("Failed to persist changeset: {}", e); + } + // The cache as folded. + identity = managed.identity.clone(); } } @@ -502,14 +534,26 @@ impl IdentityWallet { #[cfg(test)] mod tests { use super::super::identity_handle::{ - derive_identity_auth_key_hash, derive_identity_auth_key_hash_from_master, MASTER_KEY_INDEX, + derive_ecdsa_identity_auth_keypair_from_master, derive_identity_auth_key_hash, + derive_identity_auth_key_hash_from_master, MASTER_KEY_INDEX, }; + use super::super::signing_key::tests::CapturingStore; use super::{derive_load_probe_hash, ResolvedLoadKeyHashSource}; + use crate::changeset::IdentityKeyDerivationIndices; + use crate::events::{EventHandler, PlatformEventHandler}; + use crate::PlatformWalletManager; + use dash_sdk::platform::types::identity::PublicKeyHash; + use dpp::identity::identity_public_key::v0::IdentityPublicKeyV0; + use dpp::identity::v0::IdentityV0; + use dpp::identity::{Identity, IdentityPublicKey, KeyType, Purpose, SecurityLevel}; + use dpp::platform_value::BinaryData; + use dpp::prelude::Identifier; use key_wallet::bip32::ExtendedPrivKey; use key_wallet::mnemonic::Mnemonic; use key_wallet::wallet::initialization::WalletAccountCreationOptions; use key_wallet::wallet::Wallet; use key_wallet::Network; + use std::sync::Arc; /// English BIP-39 test vector (all-zero entropy). Same fixture the /// FFI-side derive tests and the `identity_handle` tests use, so the @@ -668,4 +712,119 @@ mod tests { } } } + + /// A load that finds an identity the wallet does not hold yet stores it + /// and its keys, each with the breadcrumb verified against the seed, as + /// one changeset. No snapshot of the identity or its keys reaches the + /// store ahead of the breadcrumbs, so a store that fails between two + /// writes cannot leave watch-only key rows behind. + #[tokio::test] + async fn should_store_a_newly_loaded_identity_with_its_keys_in_one_changeset() { + struct Events; + impl EventHandler for Events {} + impl PlatformEventHandler for Events {} + + const SEED: [u8; 64] = [7; 64]; + let network = Network::Testnet; + let master = ExtendedPrivKey::new_master(network, &SEED).unwrap(); + let identity_id = Identifier::from([0x5A; 32]); + // Registered at index 0 the way registration does it: MASTER, HIGH + // and CRITICAL auth keys at key ids 0..=2, all from the seed. + let levels = [ + SecurityLevel::MASTER, + SecurityLevel::HIGH, + SecurityLevel::CRITICAL, + ]; + let public_keys = (0u32..) + .zip(levels) + .map(|(key_id, security_level)| { + let derived = + derive_ecdsa_identity_auth_keypair_from_master(&master, network, 0, key_id) + .unwrap(); + let key: IdentityPublicKey = IdentityPublicKeyV0 { + id: key_id, + purpose: Purpose::AUTHENTICATION, + security_level, + contract_bounds: None, + key_type: KeyType::ECDSA_SECP256K1, + read_only: false, + data: BinaryData::new(derived.public_key.to_vec()), + disabled_at: None, + } + .into(); + (key_id, key) + }) + .collect(); + let on_chain = Identity::V0(IdentityV0 { + id: identity_id, + public_keys, + balance: 1_000_000, + revision: 1, + }); + + let mut sdk = dash_sdk::SdkBuilder::new_mock().build().unwrap(); + let probe = + derive_load_probe_hash(ResolvedLoadKeyHashSource::Master(&master), network, 0).unwrap(); + sdk.mock() + .expect_fetch(PublicKeyHash(probe), Some(on_chain)) + .await + .unwrap(); + let store = Arc::new(CapturingStore::default()); + let manager = PlatformWalletManager::new(Arc::new(sdk), store.clone(), Arc::new(Events)); + let wallet = manager + .create_wallet_from_seed_bytes( + network, + &SEED, + WalletAccountCreationOptions::Default, + Some(0), + ) + .await + .unwrap(); + let wallet_id = wallet.wallet_id(); + store.take(); + + wallet + .identity() + .load_identity_by_index_from_master(0, &master) + .await + .unwrap() + .expect("the identity at index 0 is found"); + + let stored = store.take(); + let identity_changesets: Vec<_> = stored + .iter() + .filter(|changeset| changeset.identities.is_some() || changeset.identity_keys.is_some()) + .collect(); + assert_eq!( + identity_changesets.len(), + 1, + "the identity and its keys go out together: {identity_changesets:?}" + ); + let changeset = identity_changesets[0]; + assert!( + changeset + .identities + .as_ref() + .is_some_and(|identities| identities.identities.contains_key(&identity_id)), + "the changeset carries the identity" + ); + let upserts = &changeset + .identity_keys + .as_ref() + .expect("the changeset carries the keys") + .upserts; + assert_eq!(upserts.len(), 3); + for key_id in 0..3 { + let row = &upserts[&(identity_id, key_id)]; + assert_eq!(row.wallet_id, Some(wallet_id)); + assert_eq!( + row.derivation_indices, + Some(IdentityKeyDerivationIndices { + identity_index: 0, + key_index: key_id, + }), + "key {key_id} carries its breadcrumb" + ); + } + } } diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/signing_key.rs b/packages/rs-platform-wallet/src/wallet/identity/network/signing_key.rs index 5e09f7af239..4853014ffcf 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/signing_key.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/signing_key.rs @@ -364,9 +364,75 @@ pub(super) mod tests { } } + /// Records every changeset, so a test can read what reached the store. + #[derive(Default)] + pub(crate) struct CapturingStore( + std::sync::Mutex>, + ); + + impl crate::changeset::PlatformWalletPersistence for CapturingStore { + fn store( + &self, + _: crate::wallet::platform_wallet::WalletId, + changeset: crate::changeset::PlatformWalletChangeSet, + ) -> Result<(), crate::changeset::PersistenceError> { + self.0.lock().unwrap().push(changeset); + Ok(()) + } + fn flush( + &self, + _: crate::wallet::platform_wallet::WalletId, + ) -> Result<(), crate::changeset::PersistenceError> { + Ok(()) + } + fn load( + &self, + ) -> Result + { + Ok(Default::default()) + } + } + + impl CapturingStore { + /// Every changeset stored so far, removed from the record. + pub(crate) fn take(&self) -> Vec { + std::mem::take(&mut *self.0.lock().unwrap()) + } + + /// The last key row stored for `key_id`. + pub(crate) fn stored_key( + &self, + identity_id: Identifier, + key_id: dpp::identity::KeyID, + ) -> Option { + self.0 + .lock() + .unwrap() + .iter() + .rev() + .filter_map(|changeset| changeset.identity_keys.as_ref()) + .find_map(|keys| keys.upserts.get(&(identity_id, key_id)).cloned()) + } + } + pub(crate) async fn wallet_with_signing_keys() -> std::sync::Arc { + wallet_with_signing_keys_on( + dash_sdk::SdkBuilder::new_mock().build().unwrap(), + std::sync::Arc::new(crate::wallet::persister::NoPlatformPersistence), + ) + .await + } + + /// [`wallet_with_signing_keys`] on a given SDK and store, for a test + /// that sets mock expectations first or reads what was persisted. + pub(crate) async fn wallet_with_signing_keys_on

( + sdk: dash_sdk::Sdk, + persistence: std::sync::Arc

, + ) -> std::sync::Arc + where + P: crate::changeset::PlatformWalletPersistence + 'static, + { use crate::events::{EventHandler, PlatformEventHandler}; - use crate::wallet::persister::NoPlatformPersistence; use crate::PlatformWalletManager; use key_wallet::wallet::initialization::WalletAccountCreationOptions; use std::sync::Arc; @@ -374,11 +440,7 @@ pub(super) mod tests { struct Events; impl EventHandler for Events {} impl PlatformEventHandler for Events {} - let manager = PlatformWalletManager::new( - Arc::new(dash_sdk::SdkBuilder::new_mock().build().unwrap()), - Arc::new(NoPlatformPersistence), - Arc::new(Events), - ); + let manager = PlatformWalletManager::new(Arc::new(sdk), persistence, Arc::new(Events)); let wallet = manager .create_wallet_from_seed_bytes( key_wallet::Network::Testnet, diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/update.rs b/packages/rs-platform-wallet/src/wallet/identity/network/update.rs index 9806b257692..e8a3777fb63 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/update.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/update.rs @@ -5,7 +5,9 @@ use dpp::address_funds::AddressWitness; use dpp::identity::accessors::IdentityGettersV0; use dpp::identity::accessors::IdentitySettersV0; -use dpp::identity::identity_public_key::accessors::v0::IdentityPublicKeyGettersV0; +use dpp::identity::identity_public_key::accessors::v0::{ + IdentityPublicKeyGettersV0, IdentityPublicKeySettersV0, +}; use dpp::identity::Identity; use dpp::identity::IdentityPublicKey; use dpp::identity::KeyType; @@ -14,12 +16,14 @@ use dpp::identity::SecurityLevel; use dpp::platform_value::BinaryData; use dpp::prelude::Identifier; use dpp::ProtocolError; +use std::time::{SystemTime, UNIX_EPOCH}; use dpp::identity::signer::Signer; use dash_sdk::platform::transition::put_settings::PutSettings; -use crate::error::PlatformWalletError; +use crate::changeset::{KeyWithBreadcrumb, PersistenceError}; +use crate::error::{as_identity_revision_invalid, PlatformWalletError}; use super::signing_key::AvailableSigningKey; use super::*; @@ -77,16 +81,22 @@ impl IdentityWallet { /// in the typical case). The signer here only signs the update /// transition itself; it does not derive the new keys. /// - /// CACHE INVARIANT: this function does NOT refresh the in-process - /// `IdentityManager` after a successful broadcast. The local - /// cached `Identity` keeps the pre-update revision and key set - /// until the caller invokes [`Self::refresh_identity`] (or the - /// next sync round). A subsequent call to this function for the - /// same identity without an intervening refresh will reuse the - /// stale revision and Platform will reject the duplicate. It is - /// documented here rather than fixed because the refresh requires - /// a wallet-manager write lock that may already be held higher in - /// the call stack. + /// CACHE: after a successful broadcast the in-process + /// `IdentityManager` takes the update locally (see + /// `apply_landed_update`): the cached revision moves to the + /// transition's and the added and disabled keys are applied, so a + /// follow-up update needs no refresh in between. A cache already + /// refreshed to the transition or past it holds the update as Platform + /// has it and keeps it. A refresh that fetched the identity before the + /// update landed and finishes after this step does not take the cache + /// back: a fetched identity at a lower revision than the cache's keeps + /// the cache's revision, key set and every disable (see + /// `ManagedIdentity::take_fetched_identity`). When Drive refuses the + /// update for its revision and holds a lower one than the cache, the + /// cache takes Drive's (see `take_rejected_revision`), so the next + /// update carries the right one. When the call fails after the + /// broadcast (the proof wait timed out, say), the update may have + /// landed without this step: refresh before the next update. pub async fn update_identity_with_external_signer( &self, identity_id: &Identifier, @@ -110,7 +120,7 @@ impl IdentityWallet { let (mut identity, identity_index) = { let wm = self.wallet_manager.read().await; let info = wm.get_wallet_info(&self.wallet_id).ok_or_else(|| { - crate::error::PlatformWalletError::WalletNotFound( + PlatformWalletError::WalletNotFound( "Wallet info not found in wallet manager".to_string(), ) })?; @@ -131,8 +141,8 @@ impl IdentityWallet { }; // Increment revision for the update transition. - let original_revision = identity.revision(); - identity.set_revision(original_revision + 1); + let transition_revision = update_revision(identity_id, identity.revision())?; + identity.set_revision(transition_revision); // Pick the MASTER signing key — DPP requires identity update // transitions to be authorized by MASTER specifically. @@ -185,85 +195,211 @@ impl IdentityWallet { )) })?; - state_transition + if let Err(e) = state_transition .broadcast_and_wait::(&self.sdk, settings) .await - .map_err(|e| { - PlatformWalletError::InvalidIdentityData(format!( - "Failed to broadcast identity update: {}", - e - )) - })?; + { + self.take_rejected_revision(identity_id, transition_revision, &e) + .await; + return Err(PlatformWalletError::InvalidIdentityData(format!( + "Failed to broadcast identity update: {}", + e + ))); + } - // Post-broadcast local apply. - // - // Drive accepted the transition, so the on-chain identity - // now carries the new keys / disabled flags. If we leave - // the local `ManagedIdentity` cache untouched here: - // - The next state transition for this identity reuses - // the stale revision and Platform rejects it as a - // duplicate. - // - The Swift `PersistentPublicKey` rows never receive - // the new keys (rows arrive via the - // `IdentityKeysChangeSet` persister callback that - // `add_key` fires). - // - The Identity Keys list in the iOS app keeps showing - // only the pre-update key set. - // - // `ManagedIdentity::add_key` handles both the in-memory - // mutation AND the `IdentityKeysChangeSet` upsert. Run it - // on each new key under a write lock; the breadcrumb lets - // the iOS Keychain re-derive the matching private bytes - // from the wallet seed + DIP-9 path. - if !added_keys_for_local_apply.is_empty() || !disabled_ids_for_local_apply.is_empty() { - let mut wm = self.wallet_manager.write().await; - let info = wm.get_wallet_info_mut(&self.wallet_id).ok_or_else(|| { - crate::error::PlatformWalletError::WalletNotFound( - "Wallet info not found in wallet manager".to_string(), - ) - })?; - if let Some(managed) = info.identity_manager.managed_identity_mut(identity_id) { - // Bump the cached revision so a subsequent update - // doesn't reuse the pre-broadcast value. - let cached_revision = managed.identity.revision(); - managed.identity.set_revision(cached_revision + 1); - - for key in added_keys_for_local_apply { - let breadcrumb = identity_index.map(|idx| (self.wallet_id, idx, key.id())); - managed - .add_key(key, breadcrumb, &self.persister) - .map_err(|e| { - PlatformWalletError::Persistence(format!( - "identity key not persisted after update: {e}" - )) - })?; - } + self.apply_landed_update( + identity_id, + identity_index, + transition_revision, + added_keys_for_local_apply, + disabled_ids_for_local_apply, + ) + .await + } - if !disabled_ids_for_local_apply.is_empty() { - // Disable-side counterpart to the `add_key` loop - // above: stamp `disabled_at` on the matching cached - // keys and fire the persister so the Swift - // `PersistentPublicKey.disabledAt` rows flip without - // a network re-fetch. The local wall-clock timestamp - // is a placeholder — the next Platform refresh - // reconciles it to the authoritative on-chain block - // time. `disable_keys` reuses the same derivation - // breadcrumb `add_key` carries, so the disabled key - // keeps its private-key linkage. - let disabled_at = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or_default(); - managed.disable_keys( - &disabled_ids_for_local_apply, - disabled_at, - &self.persister, - ); - } - } + /// Post-broadcast local apply of an update Drive accepted at + /// `transition_revision`. + /// + /// Drive accepted the transition, so the on-chain identity + /// now carries the new keys / disabled flags. If we leave + /// the local `ManagedIdentity` cache untouched here: + /// - The next state transition for this identity reuses + /// the stale revision and Platform rejects it as a + /// duplicate. + /// - The Swift `PersistentPublicKey` rows never receive + /// the new keys (rows arrive via the + /// `IdentityKeysChangeSet` persister callback that + /// `add_key` fires). + /// - The Identity Keys list in the iOS app keeps showing + /// only the pre-update key set. + /// + /// A cache behind `transition_revision` moves to it, and the update's + /// keys are layered in memory and stored with the scalar snapshot, + /// revision included, as one fallible write under the write lock, so + /// the landed revision is stored even when no key matched. An added + /// key carries the breadcrumb that lets the iOS Keychain re-derive the + /// matching private bytes from the wallet seed + DIP-9 path. + /// + /// A cache already refreshed to `transition_revision` or past it (a + /// fetch answered between the update's snapshot and this step) holds the + /// update as Platform has it, on-chain `disabled_at` included, and + /// whatever later updates did: the payload is not applied over it. The + /// fetch that put it there may not have been stored (a failed mobile + /// round leaves the fold in memory), so the update's keys are stored + /// here from the cache's own copies, with the scalar snapshot, as one + /// fallible write: an added key with the breadcrumb the update gives it + /// (a refresh stores its rows without one), a disabled key without one + /// (the persisters keep what the row holds). The snapshot is stored even + /// when no key matched. + async fn apply_landed_update( + &self, + identity_id: &Identifier, + identity_index: Option, + transition_revision: u64, + added_keys: Vec, + disabled_ids: Vec, + ) -> Result<(), PlatformWalletError> { + let mut wm = self.wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&self.wallet_id).ok_or_else(|| { + PlatformWalletError::WalletNotFound( + "Wallet info not found in wallet manager".to_string(), + ) + })?; + let Some(managed) = info.identity_manager.managed_identity_mut(identity_id) else { + return Ok(()); + }; + + if managed.identity.revision() >= transition_revision { + let added_ids: Vec = added_keys.iter().map(|key| key.id()).collect(); + let mut keys: Vec = added_keys + .into_iter() + .map(|key| KeyWithBreadcrumb { + breadcrumb: identity_index.map(|idx| (self.wallet_id, idx, key.id())), + key: managed + .identity + .public_keys() + .get(&key.id()) + .cloned() + .unwrap_or(key), + }) + .collect(); + keys.extend( + disabled_ids + .iter() + .filter(|id| !added_ids.contains(id)) + .filter_map(|id| managed.identity.public_keys().get(id).cloned()) + .map(|key| KeyWithBreadcrumb { + key, + breadcrumb: None, + }), + ); + return managed + .store_keys_with_snapshot(keys, &self.persister) + .map_err(landed_but_not_stored); } - Ok(()) + // The identity is at the transition's revision now, so a + // subsequent update must not reuse the pre-broadcast value. + managed.identity.set_revision(transition_revision); + + // The added keys, then the disabled ones stamped with the local + // wall-clock time, a placeholder the next fetch at this revision or + // above reconciles to the on-chain block time. A disabled key keeps + // the breadcrumb of the managed identity's own wallet slot, so it + // keeps its private-key linkage. They are stored with the scalar + // snapshot as one write, so the landed revision is stored even when + // no key matched. + let added_ids: Vec = added_keys.iter().map(|key| key.id()).collect(); + let mut keys: Vec = added_keys + .into_iter() + .map(|key| KeyWithBreadcrumb { + breadcrumb: identity_index.map(|idx| (self.wallet_id, idx, key.id())), + key, + }) + .collect(); + let disabled_at = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or_default(); + let slot = managed.wallet_id.zip(managed.identity_index); + keys.extend( + disabled_ids + .iter() + .filter(|id| !added_ids.contains(id)) + .filter_map(|id| managed.identity.public_keys().get(id).cloned()) + .map(|mut key| { + key.set_disabled_at(disabled_at); + KeyWithBreadcrumb { + breadcrumb: slot.map(|(wallet_id, identity_index)| { + (wallet_id, identity_index, key.id()) + }), + key, + } + }), + ); + managed + .store_keys_with_snapshot(keys, &self.persister) + .map_err(landed_but_not_stored) + } + + /// Take Drive's revision from an update it refused for its revision. + /// + /// Drive takes an identity update only at one above the identity's + /// revision and otherwise refuses it in the block with + /// `InvalidIdentityRevisionError`, which carries the revision it holds. + /// A cache ahead of that value cannot heal from a fetch, which comes back + /// older and keeps the cache's revision, so every update would be + /// refused again. A build before this one could leave a cache one past + /// the chain (a fetch answered between an update's snapshot and its + /// bump, which then counted twice). The cached revision moves down to + /// Drive's and is stored with the scalar snapshot, so the next update + /// built from the cache carries the right one; the key set is left as + /// it is. + /// + /// The cache is lowered only while it is still at the revision the + /// refused transition was built from (`transition_revision - 1`): a + /// fetch or another update that moved it since knows more than this + /// refusal. A cache at Drive's revision or below is left alone: raising + /// it would take no keys with it. The error comes from one node without + /// a proof, so a node that lies low gets later updates refused until a + /// fetch, which is taken at or above the lowered revision, brings the + /// cache back; it can never lock the cache ahead. A snapshot at a higher + /// revision still buffered in the store wins the merge over this one, + /// so a restart before it is written can bring the old revision back, + /// to be lowered again by the next refusal. A failed store is logged; + /// the caller gets the refusal either way. + async fn take_rejected_revision( + &self, + identity_id: &Identifier, + transition_revision: u64, + error: &dash_sdk::Error, + ) { + let Some(rejection) = as_identity_revision_invalid(error) else { + return; + }; + if rejection.identity_id() != identity_id { + return; + } + let current = *rejection.current_revision(); + let mut wm = self.wallet_manager.write().await; + let Some(managed) = wm + .get_wallet_info_mut(&self.wallet_id) + .and_then(|info| info.identity_manager.managed_identity_mut(identity_id)) + else { + return; + }; + let built_from = transition_revision.saturating_sub(1); + if managed.identity.revision() != built_from || built_from <= current { + return; + } + managed.identity.set_revision(current); + if let Err(e) = managed.store_keys_with_snapshot(vec![], &self.persister) { + tracing::error!( + identity = %identity_id, + revision = current, + "identity revision from Drive not persisted: {e}" + ); + } } /// Update an identity using an externally-provided identity and signer. @@ -332,3 +468,454 @@ impl IdentityWallet { Ok(result) } } + +/// The revision an update of an identity cached at `cached` carries: one +/// above it, refused at the last revision rather than wrapped. +fn update_revision(identity_id: &Identifier, cached: u64) -> Result { + cached.checked_add(1).ok_or_else(|| { + PlatformWalletError::InvalidIdentityData(format!( + "identity {identity_id} is at the last revision" + )) + }) +} + +/// The error for a landed update the wallet could not store. The update is +/// on Platform and in the cache either way, so the caller must not resubmit +/// it. +fn landed_but_not_stored(e: PersistenceError) -> PlatformWalletError { + PlatformWalletError::Persistence(format!( + "identity update landed on Platform and is in the wallet's cache, but the wallet could not \ + store it ({e}); do not resubmit it" + )) +} + +#[cfg(test)] +mod update_revision_tests { + use super::*; + + #[test] + fn should_carry_one_above_the_cached_revision() { + let id = Identifier::from([7u8; 32]); + assert_eq!(update_revision(&id, 4).unwrap(), 5); + } + + #[test] + fn should_refuse_an_identity_at_the_last_revision() { + let id = Identifier::from([7u8; 32]); + assert!(matches!( + update_revision(&id, u64::MAX), + Err(PlatformWalletError::InvalidIdentityData(_)) + )); + } +} + +#[cfg(test)] +mod update_identity_tests { + use super::super::signing_key::tests::{wallet_with_signing_keys_on, CapturingStore}; + use super::*; + use crate::changeset::PlatformWalletPersistence; + use crate::wallet::persister::NoPlatformPersistence; + use crate::PlatformWallet; + use dpp::identity::identity_public_key::accessors::v0::IdentityPublicKeySettersV0; + use dpp::identity::identity_public_key::v0::IdentityPublicKeyV0; + use std::sync::atomic::Ordering; + use std::sync::Arc; + + async fn cached_identity(wallet: &IdentityWallet, identity_id: &Identifier) -> Identity { + let wm = wallet.wallet_manager.read().await; + wm.get_wallet_info(&wallet.wallet_id) + .unwrap() + .identity_manager + .identity(identity_id) + .unwrap() + .identity + .clone() + } + + /// A wallet whose cached identity is at `revision` and also holds a + /// signable MASTER key (id 0). + async fn wallet_at_revision(revision: u64) -> Arc { + wallet_at_revision_on(revision, Arc::new(NoPlatformPersistence)).await + } + + /// [`wallet_at_revision`] over a given store. + async fn wallet_at_revision_on( + revision: u64, + persistence: Arc

, + ) -> Arc { + let wallet = wallet_with_signing_keys_on( + dash_sdk::SdkBuilder::new_mock().build().unwrap(), + persistence, + ) + .await; + { + let identity_wallet = wallet.identity(); + let mut wm = identity_wallet.wallet_manager.write().await; + let managed = wm + .get_wallet_info_mut(&identity_wallet.wallet_id) + .unwrap() + .identity_manager + .managed_identity_mut(&Identifier::default()) + .unwrap(); + managed.identity.add_public_key( + IdentityPublicKeyV0 { + id: 0, + purpose: Purpose::AUTHENTICATION, + security_level: SecurityLevel::MASTER, + key_type: KeyType::ECDSA_SECP256K1, + data: vec![0; KeyType::ECDSA_SECP256K1.default_size()].into(), + ..Default::default() + } + .into(), + ); + managed.identity.set_revision(revision); + } + wallet + } + + /// The key the landed-update tests add. + fn key_3() -> IdentityPublicKey { + IdentityPublicKeyV0 { + id: 3, + purpose: Purpose::AUTHENTICATION, + security_level: SecurityLevel::HIGH, + key_type: KeyType::ECDSA_SECP256K1, + data: vec![3; KeyType::ECDSA_SECP256K1.default_size()].into(), + ..Default::default() + } + .into() + } + + #[tokio::test] + async fn should_take_a_landed_update_only_into_a_cache_behind_it() { + let identity_id = Identifier::default(); + let added = key_3(); + // A transition at 5 landed and the cache is still at 4: it takes + // the revision and the payload. + let wallet = wallet_at_revision(4).await; + let wallet = wallet.identity(); + wallet + .apply_landed_update(&identity_id, Some(0), 5, vec![added.clone()], vec![]) + .await + .unwrap(); + let identity = cached_identity(wallet, &identity_id).await; + assert_eq!(identity.revision(), 5); + assert_eq!(identity.public_keys().get(&3), Some(&added)); + + let wallet = wallet_at_revision(4).await; + let wallet = wallet.identity(); + wallet + .apply_landed_update(&identity_id, Some(0), 5, vec![], vec![1]) + .await + .unwrap(); + let identity = cached_identity(wallet, &identity_id).await; + assert_eq!(identity.revision(), 5); + assert!(identity.public_keys()[&1].disabled_at().is_some()); + + // A cache refreshed to 5 or past it already holds the update and + // what came after, and is left as it is: one past the chain would + // get the next update refused. + for cached in [5, 6] { + let wallet = wallet_at_revision(cached).await; + let wallet = wallet.identity(); + { + let mut wm = wallet.wallet_manager.write().await; + wm.get_wallet_info_mut(&wallet.wallet_id) + .unwrap() + .identity_manager + .managed_identity_mut(&identity_id) + .unwrap() + .identity + .add_public_key(added.clone()); + } + let before = cached_identity(wallet, &identity_id).await; + wallet + .apply_landed_update(&identity_id, Some(0), 5, vec![added.clone()], vec![1]) + .await + .unwrap(); + assert_eq!( + cached_identity(wallet, &identity_id).await, + before, + "cached at {cached}" + ); + } + } + + /// An update at 5 added key 3, a later one at 6 disabled it, and the + /// cache was refreshed to 6 before the confirmation of 5 came back: key + /// 3 stays disabled, and is stored that way with the breadcrumb the + /// update gives it (a refresh stores its rows without one). A cache at + /// the transition that lacks the added key takes the update's copy. + #[tokio::test] + async fn should_store_an_added_key_with_its_breadcrumb_when_the_cache_is_ahead() { + let identity_id = Identifier::default(); + let store = Arc::new(CapturingStore::default()); + let wallet = wallet_at_revision_on(6, store.clone()).await; + let wallet = wallet.identity(); + let mut disabled = key_3(); + disabled.set_disabled_at(1); + { + let mut wm = wallet.wallet_manager.write().await; + wm.get_wallet_info_mut(&wallet.wallet_id) + .unwrap() + .identity_manager + .managed_identity_mut(&identity_id) + .unwrap() + .identity + .add_public_key(disabled.clone()); + } + wallet + .apply_landed_update(&identity_id, Some(0), 5, vec![key_3()], vec![]) + .await + .unwrap(); + let identity = cached_identity(wallet, &identity_id).await; + assert_eq!(identity.revision(), 6); + assert_eq!(identity.public_keys()[&3], disabled); + let stored = store.stored_key(identity_id, 3).expect("key 3 stored"); + assert_eq!(stored.public_key, disabled); + assert_eq!(stored.wallet_id, Some(wallet.wallet_id)); + assert_eq!( + stored + .derivation_indices + .map(|indices| (indices.identity_index, indices.key_index)), + Some((0, 3)) + ); + + let store = Arc::new(CapturingStore::default()); + let wallet = wallet_at_revision_on(5, store.clone()).await; + let wallet = wallet.identity(); + wallet + .apply_landed_update(&identity_id, Some(0), 5, vec![key_3()], vec![]) + .await + .unwrap(); + let identity = cached_identity(wallet, &identity_id).await; + assert_eq!(identity.revision(), 5); + assert_eq!(identity.public_keys().get(&3), Some(&key_3())); + let stored = store.stored_key(identity_id, 3).expect("key 3 stored"); + assert!(stored.derivation_indices.is_some()); + } + + /// A disable-only update at 5 whose confirmation finds the cache already + /// at 5 with the disable (a concurrent fetch's fold whose store failed): + /// the disabled key and the scalar snapshot are stored now, so a restart + /// does not bring the key back active. With no key matched, the snapshot + /// is still stored. + #[tokio::test] + async fn should_store_a_disable_the_cache_already_holds_when_the_confirmation_comes_back() { + let identity_id = Identifier::default(); + let store = Arc::new(CapturingStore::default()); + let wallet = wallet_at_revision_on(5, store.clone()).await; + let wallet = wallet.identity(); + { + let mut wm = wallet.wallet_manager.write().await; + let managed = wm + .get_wallet_info_mut(&wallet.wallet_id) + .unwrap() + .identity_manager + .managed_identity_mut(&identity_id) + .unwrap(); + let mut key = managed.identity.public_keys()[&1].clone(); + key.set_disabled_at(7); + managed.identity.add_public_key(key); + } + store.take(); + wallet + .apply_landed_update(&identity_id, Some(0), 5, vec![], vec![1]) + .await + .unwrap(); + let stored = store.stored_key(identity_id, 1).expect("key 1 stored"); + assert_eq!(stored.public_key.disabled_at(), Some(7), "the cache's copy"); + assert_eq!(stored.derivation_indices, None); + let changesets = store.take(); + assert_eq!(changesets.len(), 1); + assert!( + changesets[0].identities.is_some(), + "with the scalar snapshot" + ); + + wallet + .apply_landed_update(&identity_id, Some(0), 5, vec![], vec![9]) + .await + .unwrap(); + let changesets = store.take(); + assert_eq!(changesets.len(), 1, "the snapshot alone"); + assert!(changesets[0].identities.is_some()); + } + + /// A cache at 4 behind an update that landed at 5 stores the landed + /// revision with the update's keys as one write: an added key with a + /// disable, and a disable of a key the cache lacks, which stores the + /// snapshot alone. A failed store is returned. + #[tokio::test] + async fn should_store_a_landed_revision_with_its_keys_in_one_write() { + let identity_id = Identifier::default(); + let stored_revision = |changesets: &[crate::changeset::PlatformWalletChangeSet]| { + changesets[0] + .identities + .as_ref() + .and_then(|identities| identities.identities.get(&identity_id)) + .map(|entry| entry.revision) + }; + for (added, disabled) in [(vec![key_3()], vec![1]), (vec![], vec![9])] { + let store = Arc::new(CapturingStore::default()); + let wallet = wallet_at_revision_on(4, store.clone()).await; + let wallet = wallet.identity(); + store.take(); + wallet + .apply_landed_update(&identity_id, Some(0), 5, added.clone(), disabled.clone()) + .await + .unwrap(); + let changesets = store.take(); + assert_eq!( + changesets.len(), + 1, + "added {added:?}, disabled {disabled:?}" + ); + assert_eq!(stored_revision(&changesets), Some(5)); + let rows = changesets[0] + .identity_keys + .as_ref() + .map_or(0, |keys| keys.upserts.len()); + assert_eq!(rows, if added.is_empty() { 0 } else { 2 }); + assert_eq!(cached_identity(wallet, &identity_id).await.revision(), 5); + } + + let store = Arc::new(FailingStore::default()); + let wallet = wallet_at_revision_on(4, store.clone()).await; + store.0.store(true, Ordering::SeqCst); + let error = wallet + .identity() + .apply_landed_update(&identity_id, Some(0), 5, vec![], vec![1]) + .await + .unwrap_err(); + assert!( + matches!(&error, PlatformWalletError::Persistence(text) if text.contains("do not resubmit")), + "{error:?}" + ); + } + + /// Refuses every store once set. + #[derive(Default)] + struct FailingStore(std::sync::atomic::AtomicBool); + + impl PlatformWalletPersistence for FailingStore { + fn store( + &self, + _: crate::wallet::platform_wallet::WalletId, + _: crate::changeset::PlatformWalletChangeSet, + ) -> Result<(), crate::changeset::PersistenceError> { + if self.0.load(Ordering::SeqCst) { + return Err(crate::changeset::PersistenceError::backend("store refused")); + } + Ok(()) + } + fn flush( + &self, + _: crate::wallet::platform_wallet::WalletId, + ) -> Result<(), crate::changeset::PersistenceError> { + Ok(()) + } + fn load( + &self, + ) -> Result + { + Ok(Default::default()) + } + } + + /// Drive's refusal of an update for an identity it holds at `current`, + /// in the shape `shape` picks: the wait-stream envelope Drive answers + /// with, or a bare consensus error, matched as well. + fn revision_refusal(identity_id: Identifier, current: u64, shape: usize) -> dash_sdk::Error { + use dpp::consensus::state::identity::invalid_identity_revision_error::InvalidIdentityRevisionError; + use dpp::consensus::ConsensusError; + + let cause: ConsensusError = InvalidIdentityRevisionError::new(identity_id, current).into(); + match shape { + 0 => dash_sdk::Error::StateTransitionBroadcastError( + dash_sdk::error::StateTransitionBroadcastError { + code: 40203, + message: cause.to_string(), + cause: Some(cause), + }, + ), + _ => dash_sdk::Error::Protocol(ProtocolError::ConsensusError(Box::new(cause))), + } + } + + /// A cache one past the chain (6 over Drive's 5, which a build before + /// this one could leave) is refused on every update, and no fetch heals + /// it: each comes back older. Drive's refusal carries 5, the cache takes + /// it and stores it with the scalar snapshot, keys untouched, so the + /// next update is built at 6. + #[tokio::test] + async fn should_take_drives_revision_from_an_update_refused_for_its_revision() { + let identity_id = Identifier::default(); + for shape in [0, 1] { + let store = Arc::new(CapturingStore::default()); + let wallet = wallet_at_revision_on(6, store.clone()).await; + let wallet = wallet.identity(); + let before = cached_identity(wallet, &identity_id).await; + assert_eq!(update_revision(&identity_id, before.revision()).unwrap(), 7); + store.take(); + + wallet + .take_rejected_revision(&identity_id, 7, &revision_refusal(identity_id, 5, shape)) + .await; + let identity = cached_identity(wallet, &identity_id).await; + assert_eq!(identity.revision(), 5, "shape {shape}"); + assert_eq!(identity.public_keys(), before.public_keys()); + let changesets = store.take(); + assert_eq!(changesets.len(), 1); + let stored = changesets[0] + .identities + .as_ref() + .and_then(|cs| cs.identities.get(&identity_id)) + .expect("the scalar snapshot"); + assert_eq!(stored.revision, 5); + assert_eq!( + update_revision(&identity_id, identity.revision()).unwrap(), + 6 + ); + } + } + + /// The cache is left alone when the refusal does not show it ahead of + /// Drive: Drive at the cache's revision or above it (raising the cache + /// would take no keys, and a lying node could lock it ahead), a cache + /// that moved after the refused transition was built (a later update + /// landed, or a fetch took a newer identity), a refusal for another + /// identity, or another error. + #[tokio::test] + async fn should_keep_the_cached_revision_unless_drive_holds_a_lower_one() { + let identity_id = Identifier::default(); + let other = Identifier::from([7; 32]); + for (cached, transition, error) in [ + (5, 6, revision_refusal(identity_id, 5, 0)), + (5, 6, revision_refusal(identity_id, 9, 1)), + (7, 6, revision_refusal(identity_id, 5, 0)), + (6, 6, revision_refusal(identity_id, 3, 1)), + (5, 6, revision_refusal(other, 2, 0)), + (5, 6, dash_sdk::Error::Generic("timeout".to_string())), + ] { + let store = Arc::new(CapturingStore::default()); + let wallet = wallet_at_revision_on(cached, store.clone()).await; + let wallet = wallet.identity(); + let before = cached_identity(wallet, &identity_id).await; + store.take(); + wallet + .take_rejected_revision(&identity_id, transition, &error) + .await; + assert_eq!( + cached_identity(wallet, &identity_id).await, + before, + "cached at {cached}: {error:?}" + ); + assert!(store.take().is_empty(), "cached at {cached}: {error:?}"); + } + } +} + +#[cfg(test)] +#[path = "lagging_fetch_tests.rs"] +mod lagging_fetch_tests; diff --git a/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/identity_ops.rs b/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/identity_ops.rs index 0e380b080e6..6dd0e7a0110 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/identity_ops.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/identity_ops.rs @@ -2,11 +2,18 @@ use super::key_storage::{DpnsFetch, DpnsNameInfo, IdentityStatus}; use super::ManagedIdentity; -use crate::changeset::{IdentityChangeSet, IdentityEntry, IdentityKeyEntry, IdentityKeysChangeSet}; +use crate::changeset::{ + IdentityChangeSet, IdentityEntry, IdentityKeyEntry, IdentityKeysChangeSet, KeyWithBreadcrumb, + PersistenceError, +}; use crate::wallet::persister::WalletPersister; -use dpp::identity::accessors::IdentityGettersV0; -use dpp::identity::identity_public_key::accessors::v0::IdentityPublicKeyGettersV0; -use dpp::identity::Identity; +use crate::BlockTime; +use dpp::identity::accessors::{IdentityGettersV0, IdentitySettersV0}; +use dpp::identity::identity_public_key::accessors::v0::{ + IdentityPublicKeyGettersV0, IdentityPublicKeySettersV0, +}; +use dpp::identity::identity_public_key::accessors::v1::IdentityPublicKeyGettersV1; +use dpp::identity::{Identity, IdentityPublicKey}; use dpp::identity::{KeyID, TimestampMillis}; use dpp::prelude::Identifier; use dpp::util::hash::ripemd160_sha256; @@ -381,6 +388,143 @@ impl ManagedIdentity { } } + /// Fold an identity fetched from Platform, at `block_time` (its proof's + /// block), into the cached one. + /// + /// The balance is ordered by proof height, not by revision (it changes + /// without a revision bump), as every verified balance is: the fetch's + /// balance replaces the cache's only when its block is above the cache's + /// balance watermark, and is then queued for the next store (see + /// `store_keys_and_balance`). + /// + /// A key's limits are not ordered by revision either (a limits update + /// claims none), but they only ever rise, so each shared key keeps the + /// higher of the two (see `with_raised_limits`) at any revision. + /// + /// A fetch at the cache's revision or above is otherwise taken whole, + /// key states included: a masternode identity's keys are disabled and + /// re-enabled by Core at revision 0, so such a fetch may re-enable a key + /// the cache holds disabled. Revisions only grow on Platform, so a fetch + /// at a lower revision was answered before an update the cache already + /// holds (one applied locally after it landed, or a newer refresh, from a + /// node that lags behind): the cache keeps its revision and key set; it + /// takes only a disable the fetch shows on a key the cache holds enabled + /// (an identity update's disable is permanent) and a raised limit. + /// Returns whether it was taken whole. + /// + /// Only the in-memory identity changes here; the caller persists. + pub fn take_fetched_identity(&mut self, fetched: &Identity, block_time: BlockTime) -> bool { + let balance = self.identity.balance(); + let cached = self.identity.public_keys().clone(); + let taken = fetched.revision() >= self.identity.revision(); + let mut keys = if taken { + self.identity = fetched.clone(); + self.identity.set_balance(balance); + fetched.public_keys().clone() + } else { + cached.clone() + }; + for (key_id, key) in keys.iter_mut() { + if let Some(fetched_key) = fetched.public_keys().get(key_id) { + if let Some(folded) = fold_fetched_key(cached.get(key_id), fetched_key, taken) { + *key = folded; + } + } + } + self.identity.set_public_keys(keys); + self.take_fetched_balance(fetched.balance(), block_time); + taken + } + + /// Fold `fetched` (see [`Self::take_fetched_identity`]) and store the row + /// of every key the fold adds or changes (a disable, a raised limit, a + /// key added at a newer revision, on this device or another) with the + /// identity's scalar snapshot as one changeset, so the stored key set + /// matches the stored revision after a restart. + /// + /// Used where a fetch comes without derived breadcrumbs (refresh), so the + /// rows carry none. Every persister keeps the breadcrumb a row already + /// holds when an upsert carries none, and a load or discovery, which + /// verify a key against the seed, give a new key its breadcrumb. + /// + /// A failed store is returned; memory holds the fold. + pub fn store_fetched_identity( + &mut self, + fetched: &Identity, + block_time: BlockTime, + persister: &WalletPersister, + ) -> Result<(), PersistenceError> { + let before = self.identity.public_keys().clone(); + self.take_fetched_identity(fetched, block_time); + let identity_id = self.id(); + let mut keys_cs = IdentityKeysChangeSet::default(); + for (key_id, key) in self.identity.public_keys() { + if before.get(key_id) == Some(key) { + continue; + } + keys_cs.upserts.insert( + (identity_id, *key_id), + IdentityKeyEntry { + identity_id, + key_id: *key_id, + public_key: key.clone(), + public_key_hash: pubkey_hash_of(key), + wallet_id: None, + derivation_indices: None, + }, + ); + } + self.store_keys_and_balance(keys_cs, persister) + } + + /// Fold `fetched` (see [`Self::take_fetched_identity`]) with the key + /// `decisions` derived for it, and store the folded keys with their + /// breadcrumbs and the identity's scalar snapshot as one changeset (see + /// `store_keys_and_balance`). Used where a fetch comes with derived + /// breadcrumbs (loading and discovery); an identity added from this + /// fetch just before (see `IdentityManager::add_identity_unstored`) + /// reaches the store in the same changeset as its keys. A failed store + /// is returned; memory holds the fold. + pub fn store_fetched_identity_keys( + &mut self, + fetched: &Identity, + decisions: Vec, + block_time: BlockTime, + persister: &WalletPersister, + ) -> Result<(), PersistenceError> { + let decisions = self.key_decisions_for_fetch(fetched, decisions); + self.take_fetched_identity(fetched, block_time); + let keys_cs = self.layer_keys(decisions); + self.store_keys_and_balance(keys_cs, persister) + } + + /// The key decisions to store for `decisions` built from `fetched`, each + /// key folded as [`Self::take_fetched_identity`] folds it, so storing its + /// breadcrumb (a key's derivation never changes) stores what the fold + /// holds: never a raised limit taken back, nor a disable an older fetch + /// lacks. For a fetch at the cache's + /// revision or above that is the fetched key. For an older fetch it is + /// the cache's copy, disabled when only the fetch's copy is; either way + /// each limit is the higher of the two. + /// A key the cache lacks is left out of an older fetch's decisions, since + /// an older answer cannot show whether a later update disabled it, and a + /// fetch at least as new will bring it. + fn key_decisions_for_fetch( + &self, + fetched: &Identity, + decisions: Vec, + ) -> Vec { + let taken = fetched.revision() >= self.identity.revision(); + decisions + .into_iter() + .filter_map(|mut decision| { + let cached = self.identity.public_keys().get(&decision.key.id()); + decision.key = fold_fetched_key(cached, &decision.key, taken)?; + Some(decision) + }) + .collect() + } + /// Layer an `IdentityPublicKey` onto this identity's /// `public_keys` map and emit a single-key /// [`IdentityKeysChangeSet`] upsert carrying the full derivation @@ -419,18 +563,44 @@ impl ManagedIdentity { /// order-dependent watch-only-then-override). No-op on an empty list. pub fn add_keys( &mut self, - keys: Vec, + keys: Vec, persister: &WalletPersister, - ) -> Result<(), crate::changeset::PersistenceError> { - use dpp::identity::accessors::IdentitySettersV0; - + ) -> Result<(), PersistenceError> { if keys.is_empty() { return Ok(()); } + self.store_keys_with_snapshot(keys, persister) + } + + /// Layer `keys` in memory and store them with the identity's scalar + /// snapshot as one changeset, even when `keys` is empty (the snapshot + /// alone, e.g. a landed revision). The store's failure is returned. + pub fn store_keys_with_snapshot( + &mut self, + keys: Vec, + persister: &WalletPersister, + ) -> Result<(), PersistenceError> { + let keys_cs = self.layer_keys(keys); + let cs = crate::changeset::PlatformWalletChangeSet { + identities: Some(self.snapshot_changeset()), + identity_keys: Some(keys_cs), + ..Default::default() + }; + // Surface the persist failure — these rows carry the per-key + // derivation breadcrumb + verified scalar that make an imported / + // restored identity signable. Swallowing a failed store here would + // leave the keys in memory but absent from the client store, so the + // identity comes back watch-only after restart with no signal. + persister.store(cs)?; + Ok(()) + } + + /// Layer `keys` onto this identity in memory, returning their key rows. + fn layer_keys(&mut self, keys: Vec) -> IdentityKeysChangeSet { let identity_id = self.id(); let mut current = self.identity.public_keys().clone(); let mut keys_cs = IdentityKeysChangeSet::default(); - for crate::changeset::KeyWithBreadcrumb { + for KeyWithBreadcrumb { key: public_key, breadcrumb, } in keys @@ -461,18 +631,7 @@ impl ManagedIdentity { ); } self.identity.set_public_keys(current); - let cs = crate::changeset::PlatformWalletChangeSet { - identities: Some(self.snapshot_changeset()), - identity_keys: Some(keys_cs), - ..Default::default() - }; - // Surface the persist failure — these rows carry the per-key - // derivation breadcrumb + verified scalar that make an imported / - // restored identity signable. Swallowing a failed store here would - // leave the keys in memory but absent from the client store, so the - // identity comes back watch-only after restart with no signal. - persister.store(cs)?; - Ok(()) + keys_cs } /// Replace one public key of the identity with `public_key`, the key as it is stored @@ -492,8 +651,6 @@ impl ManagedIdentity { public_key: dpp::identity::IdentityPublicKey, persister: &WalletPersister, ) -> Result<(), crate::changeset::PersistenceError> { - use dpp::identity::accessors::IdentitySettersV0; - let identity_id = self.id(); let key_id = public_key.id(); let public_key_hash = pubkey_hash_of(&public_key); @@ -543,8 +700,9 @@ impl ManagedIdentity { /// `key_ids` not present on the identity are skipped (logged) and /// no changeset is emitted when nothing matched. `disabled_at` is /// the timestamp stamped on every matching key — callers pass the - /// local broadcast time; the next Platform refresh reconciles it to - /// the authoritative on-chain block time. + /// local broadcast time; the next fetch at the update's revision or + /// above reconciles it to the authoritative on-chain block time (an + /// older one keeps it). /// /// Does **not** touch the identity revision — the caller bumps it /// alongside the update transition. @@ -554,9 +712,6 @@ impl ManagedIdentity { disabled_at: TimestampMillis, persister: &WalletPersister, ) { - use dpp::identity::accessors::IdentitySettersV0; - use dpp::identity::identity_public_key::accessors::v0::IdentityPublicKeySettersV0; - let identity_id = self.id(); // Reconstruct the derivation breadcrumb from the cached @@ -629,6 +784,61 @@ impl ManagedIdentity { } } +/// How a fetched key lands in the cache, the one rule the identity fold and +/// its key decisions share. From a fetch taken whole (`taken`) it is the +/// fetched key. From an older fetch it is the cache's copy, disabled when +/// only the fetch's copy is, or nothing for a key the cache lacks. Either +/// way each limit is the higher of the two copies'. +fn fold_fetched_key( + cached: Option<&IdentityPublicKey>, + fetched: &IdentityPublicKey, + taken: bool, +) -> Option { + if !taken { + return cached + .map(|cached| with_raised_limits(with_disable_from(cached, fetched), fetched)); + } + let key = fetched.clone(); + let Some(cached) = cached else { + return Some(key); + }; + Some(with_raised_limits(key, cached)) +} + +/// `key` with each of its limits raised to `other`'s where that is higher. +/// Consensus only ever raises a key's limits (a budget top-up, a later +/// expiry; anything else is refused as not raising it) and never adds or +/// removes one, so the higher value is the newer one whichever answer +/// carries it, and a delayed answer cannot take a raise back. A limit only +/// one copy has is therefore one the other copy lost, so it is taken from +/// whichever copy has it. +fn with_raised_limits(key: IdentityPublicKey, other: &IdentityPublicKey) -> IdentityPublicKey { + fn higher(own: Option, other: Option) -> Option { + match (own, other) { + (Some(own), Some(other)) => Some(own.max(other)), + (own, other) => own.or(other), + } + } + let total_budget = higher(key.total_budget(), other.total_budget()); + let expires_at = higher(key.expires_at(), other.expires_at()); + if (total_budget, expires_at) == (key.total_budget(), key.expires_at()) { + return key; + } + key.with_limits(total_budget, expires_at) +} + +/// The cache's copy of a key, disabled when an older fetch's copy is: an +/// identity update's disable is permanent on Platform. A disable the cache +/// already holds is kept as it is: it is at least as new as the fetch's (a +/// later update may have disabled the key again at a later block). +fn with_disable_from(cached: &IdentityPublicKey, fetched: &IdentityPublicKey) -> IdentityPublicKey { + let mut key = cached.clone(); + if let Some(disabled_at) = cached.disabled_at().or(fetched.disabled_at()) { + key.set_disabled_at(disabled_at); + } + key +} + #[cfg(test)] mod tests { use super::*; @@ -638,8 +848,10 @@ mod tests { use crate::wallet::identity::PaymentEntry; use crate::wallet::platform_wallet::WalletId; use dpp::identity::identity_public_key::v0::IdentityPublicKeyV0; + use dpp::identity::identity_public_key::v1::IdentityPublicKeyV1; use dpp::identity::v0::IdentityV0; use dpp::identity::{Identity, IdentityPublicKey, KeyID, KeyType, Purpose, SecurityLevel}; + use dpp::platform_value::BinaryData; use std::collections::BTreeMap; use std::sync::Mutex; @@ -698,6 +910,248 @@ mod tests { }) } + /// A fetch at a lower revision than the cache's was answered before an + /// update the cache holds: the cache keeps its revision and key set, and + /// takes only a disable the fetch shows on a key it holds enabled and a + /// raised limit. A fetch at the same revision or higher replaces the + /// cache, key states included, but never takes back a raise. Either way + /// the balance follows the proof's block, not the revision. + #[test] + fn should_take_a_fetched_identity_whole_only_when_it_is_not_older() { + fn limited(id: KeyID, budget: u64, disabled_at: Option) -> IdentityPublicKey { + IdentityPublicKey::V1(IdentityPublicKeyV1 { + id, + purpose: Purpose::AUTHENTICATION, + security_level: SecurityLevel::HIGH, + contract_bounds: None, + key_type: KeyType::ECDSA_SECP256K1, + read_only: false, + data: BinaryData::new(vec![id as u8; 33]), + disabled_at, + total_budget: Some(budget), + expires_at: None, + }) + } + fn block(height: u64) -> BlockTime { + BlockTime::new(height, 1, height) + } + fn at(revision: u64, balance: u64, keys: Vec) -> Identity { + Identity::V0(IdentityV0 { + id: Identifier::from([1u8; 32]), + public_keys: keys.into_iter().map(|key| (key.id(), key)).collect(), + balance, + revision, + }) + } + // The cache took an update at 6 that disabled key 3, and holds a + // balance proved at block 9; the fetch at 5 still shows key 3 live, + // a lower budget and key 4, which the cache never saw. Budgets only + // rise, so the lower one is older, and the fetch's balance is no + // newer than the cache's proof. + let cache = at(6, 10, vec![limited(3, 100, Some(9))]); + let older = at(5, 20, vec![limited(3, 50, None), limited(4, 1, None)]); + let mut managed = ManagedIdentity::new(cache.clone(), 0); + managed.last_updated_balance_block_time = Some(block(9)); + for height in [8, 9] { + assert!(!managed.take_fetched_identity(&older, block(height))); + assert_eq!(managed.identity, cache); + } + // A balance proved at a later block is taken whatever the revision. + assert!(!managed.take_fetched_identity(&older, block(10))); + assert_eq!(managed.identity, at(6, 20, vec![limited(3, 100, Some(9))])); + assert_eq!(managed.last_updated_balance_block_time, Some(block(10))); + + // An identity update's disable is permanent: one the older fetch + // shows sticks even when the cache's copy is live. A higher budget + // it shows is a raise the cache missed (a limits update claims no + // revision). + let mut live = ManagedIdentity::new(at(6, 10, vec![limited(3, 100, None)]), 0); + live.last_updated_balance_block_time = Some(block(9)); + let disabling = at(5, 20, vec![limited(3, 500, Some(4))]); + assert!(!live.take_fetched_identity(&disabling, block(9))); + assert_eq!(live.identity, at(6, 10, vec![limited(3, 500, Some(4))])); + // Both disabled: the cache's timestamp is at least as new (a later + // update may have disabled the key again), so an older answer, or a + // second one like it, never rewrites it. + let mut stamped = ManagedIdentity::new(at(6, 10, vec![limited(3, 100, Some(99))]), 0); + for _ in 0..2 { + assert!(!stamped + .take_fetched_identity(&at(5, 10, vec![limited(3, 100, Some(4))]), block(9))); + assert_eq!(stamped.identity, at(6, 10, vec![limited(3, 100, Some(99))])); + } + + // An answer at the cache's revision or above is taken whole... + let same = at(6, 30, vec![limited(3, 600, Some(7)), limited(4, 1, None)]); + assert!(managed.take_fetched_identity(&same, block(11))); + assert_eq!(managed.identity, same); + assert_eq!(managed.last_updated_balance_block_time, Some(block(11))); + let newer = at(7, 40, vec![limited(3, 600, Some(7))]); + assert!(managed.take_fetched_identity(&newer, block(12))); + assert_eq!(managed.identity, newer); + + // ...key states included (Core re-enables a masternode identity's + // key at revision 0), except for a balance its proof does not order + // after the cache's and a budget the cache holds higher: a delayed + // answer cannot take a raise back. + let delayed = at(7, 35, vec![limited(3, 550, None)]); + for height in [11, 12] { + assert!(managed.take_fetched_identity(&delayed, block(height))); + assert_eq!(managed.identity, at(7, 40, vec![limited(3, 600, None)])); + assert_eq!(managed.last_updated_balance_block_time, Some(block(12))); + } + } + + /// Each limit is the higher of the two copies', since consensus only + /// raises a key's limits; a limit only one copy has is taken from it, + /// since none is ever added or removed, so the other copy lost it. + #[test] + fn should_keep_the_higher_of_each_key_limit() { + let limited = |total_budget, expires_at| key(3).with_limits(total_budget, expires_at); + assert_eq!( + with_raised_limits(limited(Some(100), Some(10)), &limited(Some(50), Some(20))), + limited(Some(100), Some(20)) + ); + assert_eq!( + with_raised_limits(limited(Some(100), None), &limited(Some(50), Some(20))), + limited(Some(100), Some(20)) + ); + assert_eq!( + with_raised_limits(key(3), &limited(Some(50), Some(20))), + limited(Some(50), Some(20)) + ); + assert_eq!(with_raised_limits(key(3), &key(3)), key(3)); + } + + /// An older fetch's key decisions carry the cache's copy of each key + /// (its disabled state, each limit the higher of the two), and a key + /// only the fetch has is left out. + #[test] + fn should_fold_an_older_fetch_s_key_decisions_into_the_cache_s_keys() { + let mut disabled = key(3); + disabled.set_disabled_at(9); + let managed = ManagedIdentity::new( + Identity::V0(IdentityV0 { + id: Identifier::from([1u8; 32]), + public_keys: BTreeMap::from([(3, disabled.clone())]), + balance: 0, + revision: 6, + }), + 0, + ); + let fetched = |revision| { + Identity::V0(IdentityV0 { + id: Identifier::from([1u8; 32]), + public_keys: BTreeMap::from([(3, key(3)), (4, key(4))]), + balance: 0, + revision, + }) + }; + let decisions = || { + vec![ + KeyWithBreadcrumb { + key: key(3), + breadcrumb: None, + }, + KeyWithBreadcrumb { + key: key(4), + breadcrumb: None, + }, + ] + }; + let folded = managed.key_decisions_for_fetch(&fetched(5), decisions()); + assert_eq!(folded.len(), 1); + assert_eq!(folded[0].key, disabled); + assert_eq!( + managed + .key_decisions_for_fetch(&fetched(6), decisions()) + .len(), + 2 + ); + } + + /// A key decision keeps the higher of each limit at any revision, so a + /// load answered before a limits update the cache holds does not store + /// the lower one. + #[test] + fn should_keep_a_raised_limit_in_a_fetch_s_key_decisions() { + let identity = |revision, budget| { + Identity::V0(IdentityV0 { + id: Identifier::from([1u8; 32]), + public_keys: BTreeMap::from([(3, key(3).with_limits(Some(budget), None))]), + balance: 0, + revision, + }) + }; + let managed = ManagedIdentity::new(identity(6, 500), 0); + for revision in [5, 6, 7] { + let decisions = vec![KeyWithBreadcrumb { + key: key(3).with_limits(Some(100), None), + breadcrumb: None, + }]; + let folded = managed.key_decisions_for_fetch(&identity(revision, 100), decisions); + assert_eq!(folded[0].key.total_budget(), Some(500), "at {revision}"); + } + } + + /// A refresh stores, with the scalar snapshot and in one changeset, the + /// row of every key its fold adds or changes (here a key another device + /// added and a disable), without a breadcrumb, and no row for a key it + /// leaves as it was. An older answer stores no key row. + #[test] + fn should_store_the_keys_a_refresh_adds_or_changes() { + let id = Identifier::from([1u8; 32]); + let wallet_id: WalletId = [0xAB; 32]; + let identity = |revision, keys: Vec| { + Identity::V0(IdentityV0 { + id, + public_keys: keys.into_iter().map(|key| (key.id(), key)).collect(), + balance: 0, + revision, + }) + }; + let block = |height| BlockTime::new(height, 1, height); + let persister = std::sync::Arc::new(CapturingPersister::default()); + let p = WalletPersister::new(wallet_id, std::sync::Arc::clone(&persister) as _); + let mut managed = ManagedIdentity::new(identity(6, vec![key(1), key(2)]), 0); + managed.wallet_id = Some(wallet_id); + + let mut disabled = key(2); + disabled.set_disabled_at(5); + managed + .store_fetched_identity( + &identity(7, vec![key(1), disabled.clone(), key(4)]), + block(1), + &p, + ) + .unwrap(); + { + let stores = persister.stores.lock().unwrap(); + assert_eq!(stores.len(), 1, "keys and snapshot in one changeset"); + let snapshot = stores[0].identities.as_ref().unwrap(); + assert_eq!(snapshot.identities[&id].revision, 7); + let rows = &stores[0].identity_keys.as_ref().unwrap().upserts; + assert_eq!( + rows.keys().map(|(_, key_id)| *key_id).collect::>(), + vec![2, 4] + ); + assert_eq!(rows[&(id, 2)].public_key, disabled); + assert!(rows + .values() + .all(|row| row.derivation_indices.is_none() && row.wallet_id.is_none())); + } + + // An answer from before the cache's revision changes no key. + persister.stores.lock().unwrap().clear(); + managed + .store_fetched_identity(&identity(6, vec![key(1), key(2)]), block(2), &p) + .unwrap(); + let stores = persister.stores.lock().unwrap(); + assert_eq!(stores.len(), 1); + assert!(stores[0].identity_keys.as_ref().unwrap().upserts.is_empty()); + assert_eq!(managed.identity.revision(), 7); + assert_eq!(managed.identity.public_keys()[&2], disabled); + } + /// A batch of fetched names lands as one snapshot carrying every name, /// skips labels already held, and persists nothing when all are known. /// One snapshot per name made the first read as the complete owned set. diff --git a/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/sync.rs b/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/sync.rs index 0450f0f690c..8bf238c99f7 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/sync.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/sync.rs @@ -1,6 +1,7 @@ //! Synchronization and block time management for ManagedIdentity use super::ManagedIdentity; +use crate::changeset::{IdentityKeysChangeSet, PersistenceError, PlatformWalletChangeSet}; use crate::error::PlatformWalletError; use crate::wallet::persister::WalletPersister; use crate::BlockTime; @@ -87,6 +88,43 @@ impl ManagedIdentity { Ok(()) } + /// Take a verified balance when its proof's block is above the cache's + /// balance watermark (see `balance_snapshot_is_newer`). Memory only: the + /// balance is queued, so the next store carries it (see + /// `store_keys_and_balance`) and a failed store keeps it queued. + pub(crate) fn take_fetched_balance(&mut self, balance: u64, block_time: BlockTime) { + if self.balance_snapshot_is_newer(block_time) { + self.identity.set_balance(balance); + self.last_updated_balance_block_time = Some(block_time); + self.pending_balance_snapshot = Some((balance, block_time)); + } + } + + /// Store `keys_cs` with the identity's scalar snapshot (status, revision + /// and any queued balance) as one changeset. The store's failure is + /// returned. A store that commits inline has committed the queued + /// balance, which is then cleared; a buffered one keeps it queued, so it + /// rides later snapshots until a balance write flushes it (see + /// `retry_pending_balance`), and no flush runs here. + pub(crate) fn store_keys_and_balance( + &mut self, + keys_cs: IdentityKeysChangeSet, + persister: &WalletPersister, + ) -> Result<(), PersistenceError> { + let (balance, block_time) = self.balance_snapshot_for_persistence(); + persister.store(PlatformWalletChangeSet { + identities: Some(self.snapshot_changeset()), + identity_keys: Some(keys_cs), + ..Default::default() + })?; + if self.pending_balance_snapshot.is_some() && persister.store_commits_inline() { + self.identity.set_balance(balance); + self.last_updated_balance_block_time = block_time; + self.pending_balance_snapshot = None; + } + Ok(()) + } + pub(crate) fn persist_refreshed_balance( &mut self, balance: u64, @@ -110,11 +148,7 @@ impl ManagedIdentity { block_time: BlockTime, persister: &WalletPersister, ) -> u64 { - if self.balance_snapshot_is_newer(block_time) { - self.identity.set_balance(balance); - self.last_updated_balance_block_time = Some(block_time); - self.pending_balance_snapshot = Some((balance, block_time)); - } + self.take_fetched_balance(balance, block_time); // A transaction response can confirm the exact pending query snapshot // whose cache write failed. Publish that verified value now, while // retaining the same write obligation if storage is still unavailable. diff --git a/packages/rs-platform-wallet/src/wallet/identity/state/manager/lifecycle.rs b/packages/rs-platform-wallet/src/wallet/identity/state/manager/lifecycle.rs index f5af15abaf1..2efbf0f4345 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/state/manager/lifecycle.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/state/manager/lifecycle.rs @@ -3,7 +3,8 @@ //! Adds, removes, and looks up the HD index for managed identities. //! Every mutation here persists a corresponding //! [`IdentityChangeSet`](crate::changeset::IdentityChangeSet) via -//! the supplied [`WalletPersister`]. +//! the supplied [`WalletPersister`], except +//! [`IdentityManager::add_identity_unstored`], whose caller stores. use super::{IdentityLocation, IdentityManager}; use crate::changeset::{IdentityChangeSet, IdentityEntry, PlatformWalletChangeSet}; @@ -41,6 +42,50 @@ impl IdentityManager { persister: &WalletPersister, ) -> Result<(), PlatformWalletError> { let identity_id = identity.id(); + self.add_identity_unstored(identity, identity_index, wallet_id)?; + let managed_identity = self + .managed_identity_mut(&identity_id) + .ok_or(PlatformWalletError::IdentityNotFound(identity_id))?; + + let entry = IdentityEntry::from_managed(managed_identity); + // No `key_storage` to snapshot here anymore — the field is + // dropped from `ManagedIdentity`. Public keys still arrive + // via the separate `IdentityKeysChangeSet` path. + let keys_cs = managed_identity.keys_snapshot_changeset(); + + let mut id_cs = IdentityChangeSet::default(); + id_cs.identities.insert(identity_id, entry); + + let cs = PlatformWalletChangeSet { + identities: Some(id_cs), + identity_keys: if keys_cs.upserts.is_empty() { + None + } else { + Some(keys_cs) + }, + ..Default::default() + }; + + if let Err(e) = persister.store(cs) { + tracing::error!("Failed to persist changeset: {}", e); + } + + Ok(()) + } + + /// [`Self::add_identity`] in memory only: the caller stores the identity + /// and its keys itself, as one changeset. Load and discovery use it, so + /// the keys they verified against the seed reach the store with the + /// identity, never after a breadcrumb-less snapshot of them that could + /// commit on its own (a disk that fills between the two writes would + /// otherwise leave every key row without its breadcrumb). + pub(crate) fn add_identity_unstored( + &mut self, + identity: Identity, + identity_index: u32, + wallet_id: WalletId, + ) -> Result<(), PlatformWalletError> { + let identity_id = identity.id(); if self.identity(&identity_id).is_some() { return Err(PlatformWalletError::IdentityAlreadyExists(identity_id)); @@ -52,12 +97,6 @@ impl IdentityManager { // without re-threading the keys. managed_identity.wallet_id = Some(wallet_id); - let entry = IdentityEntry::from_managed(&managed_identity); - // No `key_storage` to snapshot here anymore — the field is - // dropped from `ManagedIdentity`. Public keys still arrive - // via the separate `IdentityKeysChangeSet` path. - let keys_cs = managed_identity.keys_snapshot_changeset(); - self.wallet_identities .entry(wallet_id) .or_default() @@ -72,23 +111,6 @@ impl IdentityManager { }, ); - let mut id_cs = IdentityChangeSet::default(); - id_cs.identities.insert(identity_id, entry); - - let cs = PlatformWalletChangeSet { - identities: Some(id_cs), - identity_keys: if keys_cs.upserts.is_empty() { - None - } else { - Some(keys_cs) - }, - ..Default::default() - }; - - if let Err(e) = persister.store(cs) { - tracing::error!("Failed to persist changeset: {}", e); - } - Ok(()) }