diff --git a/Cargo.lock b/Cargo.lock index a3801c22cf..6165f3018e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2150,6 +2150,7 @@ dependencies = [ "zcash_note_encryption", "zcash_primitives", "zcash_protocol", + "zcash_script", "zcash_transparent", "zingo-memo", "zingo-netutils", diff --git a/Cargo.toml b/Cargo.toml index 3e03fb6361..45c817fe78 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -82,6 +82,7 @@ zcash_primitives = { version = "0.30.0", features = ["non-standard-fees"] } zcash_proofs = "0.30.0" zcash_protocol = "0.10.4" # public API (TODO: remove from public API) zcash_transparent = "0.10.0" +zcash_script = "0.4.5" ### Blockchain Protocol bip0039 = { version = "0.14", features = ["rand"] } diff --git a/libtonode-tests/tests/concrete.rs b/libtonode-tests/tests/concrete.rs index 0bf52398c7..89f9c379bf 100644 --- a/libtonode-tests/tests/concrete.rs +++ b/libtonode-tests/tests/concrete.rs @@ -1690,7 +1690,6 @@ async fn mine_to_transparent_coinbase_maturity() { } mod testnet_test { - use pepper_sync::sync_status; use zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED; use zingolib::{ config::{ChainType, ClientConfig, WalletConfig}, @@ -1730,11 +1729,9 @@ mod testnet_test { let mut interval = tokio::time::interval(std::time::Duration::from_millis(100)); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); interval.tick().await; - while sync_status(&*lightclient.wallet().read().await) - .await - .unwrap() - .percentage_total_outputs_scanned - > 1.0 + while lightclient + .latest_sync_status() + .is_none_or(|status| status.percentage_total_outputs_scanned < 1.0) { interval.tick().await; } diff --git a/libtonode-tests/tests/sync.rs b/libtonode-tests/tests/sync.rs index 9747be413f..db3583de6c 100644 --- a/libtonode-tests/tests/sync.rs +++ b/libtonode-tests/tests/sync.rs @@ -1,6 +1,7 @@ use std::{num::NonZeroU32, time::Duration}; use incrementalmerkletree::Position; +use pepper_sync::error::SyncError; use pepper_sync::sync::ScanPriority; use pepper_sync::test_support::block; use pepper_sync::wallet::ShardTrees; @@ -15,6 +16,8 @@ use zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED; use zingolib::config::{ChainType, ClientConfig, WalletConfig}; use zingolib::data::PollReport; use zingolib::lightclient::DEFAULT_REQUEST_TIMEOUT; +use zingolib::lightclient::error::LightClientError; +use zingolib::sync::SyncModeError; use zingolib::testutils::default_test_wallet_settings; use zingolib::testutils::lightclient::from_inputs::quick_send; use zingolib::testutils::paths::get_cargo_manifest_dir; @@ -32,57 +35,7 @@ use zingolib_testutils::scenarios::{ self, IndexerConvergence, increase_height_and_wait_for_client, }; -#[ignore = "temporary mainnet test for sync development"] -#[tokio::test] -async fn sync_mainnet_test() { - zingolib::ensure_default_crypto_provider(); - tracing_subscriber::fmt().init(); - - let uri = construct_indexer_uri(SYNC_TEST_INDEXER.to_string()).unwrap(); - let temp_dir = TempDir::new().unwrap(); - let temp_path = temp_dir.path().to_path_buf(); - let config = ClientConfig::builder() - .set_indexer_uri(uri.clone()) - .set_chain_type(ChainType::Mainnet) - .set_wallet_dir(temp_path) - .set_wallet_config(WalletConfig::MnemonicPhrase { - mnemonic_phrase: HOSPITAL_MUSEUM_SEED.to_string(), - no_of_accounts: NonZeroU32::try_from(1).expect("hard-coded integer"), - birthday: 1_500_000, - wallet_settings: default_test_wallet_settings(), - }) - .build() - .unwrap(); - let mut lightclient = LightClient::new(config, true).await.unwrap(); - - lightclient.sync().await.unwrap(); - let mut interval = tokio::time::interval(zingo_netutils::time::test::SETTLE_POLL_INTERVAL); - loop { - interval.tick().await; - { - let wallet = lightclient.wallet().read().await; - tracing::info!( - "{}", - json::JsonValue::from(pepper_sync::sync_status(&*wallet).await.unwrap()) - ); - tracing::info!("WALLET DEBUG:"); - tracing::info!("uas: {}", wallet.unified_addresses().len()); - tracing::info!("taddrs: {}", wallet.transparent_addresses().len()); - tracing::info!("blocks: {}", wallet.wallet_blocks.len()); - tracing::info!("txs: {}", wallet.wallet_transactions.len()); - tracing::info!("nullifiers o: {}", wallet.nullifier_map.orchard.len()); - tracing::info!("nullifiers s: {}", wallet.nullifier_map.sapling.len()); - tracing::info!("outpoints: {}", wallet.outpoint_map.len()); - } - lightclient.flush().await.unwrap(); - } - - // let wallet = lightclient.wallet.read().await; - // dbg!(&wallet.wallet_blocks); - // dbg!(&wallet.nullifier_map); - // dbg!(&wallet.sync_state); -} - +// TODO: migrate to mock test as this test is long and connects to mainnet #[tokio::test] async fn add_subtree_roots() { fn assert_subtree_roots_match_server( @@ -230,11 +183,19 @@ async fn add_subtree_roots() { tokio::time::sleep(Duration::from_secs(1)).await; } let _ = lightclient.stop_sync(); - let _ = lightclient.await_sync().await; + match lightclient.await_sync().await { + Ok(_) => {} + Err(LightClientError::SyncError(SyncError::SyncModeError( + SyncModeError::SyncNotRunning, + ))) + | Err(LightClientError::SyncNotRunning) => {} + Err(e) => { + panic!("{e}"); + } + } { let shard_trees = &mut lightclient.wallet().write().await.shard_trees; - assert_subtree_roots_match_server( shard_trees, sapling_subtree_roots_server.clone(), @@ -258,6 +219,9 @@ async fn add_subtree_roots() { assert!(orchard_shard_addrs.len() != orchard_subtree_roots_server.len()); } + // must wait for a new block to be mined to trigger get_subtree_roots in sync + tokio::time::sleep(Duration::from_secs(100)).await; + lightclient.sync().await.unwrap(); while !(lightclient .wallet() @@ -272,7 +236,16 @@ async fn add_subtree_roots() { tokio::time::sleep(Duration::from_secs(1)).await; } let _ = lightclient.stop_sync(); - let _ = lightclient.await_sync().await; + match lightclient.await_sync().await { + Ok(_) => {} + Err(LightClientError::SyncError(SyncError::SyncModeError( + SyncModeError::SyncNotRunning, + ))) + | Err(LightClientError::SyncNotRunning) => {} + Err(e) => { + panic!("{e}"); + } + } { let shard_trees = &mut lightclient.wallet().write().await.shard_trees; diff --git a/pepper-sync/Cargo.toml b/pepper-sync/Cargo.toml index 6d8d9935b2..6e2cb7405f 100644 --- a/pepper-sync/Cargo.toml +++ b/pepper-sync/Cargo.toml @@ -39,6 +39,7 @@ zcash_note_encryption.workspace = true zcash_primitives.workspace = true zcash_protocol.workspace = true zcash_transparent.workspace = true +zcash_script.workspace = true # Protocol bip32.workspace = true diff --git a/pepper-sync/src/client/fetch.rs b/pepper-sync/src/client/fetch.rs index bd6d639448..24a349077b 100644 --- a/pepper-sync/src/client/fetch.rs +++ b/pepper-sync/src/client/fetch.rs @@ -102,27 +102,27 @@ where { match fetch_request { FetchRequest::ChainTip(sender) => { - tracing::debug!("Fetching chain tip."); + tracing::info!("Fetching chain tip."); let block_id = get_latest_block(client).await; let _ignore_error = sender.send(block_id); } FetchRequest::CompactBlock(sender, block_height) => { - tracing::debug!("Fetching compact block. {:?}", &block_height); + tracing::info!("Fetching compact block. {:?}", &block_height); let block = get_block(client, block_height).await; let _ignore_error = sender.send(block); } FetchRequest::CompactBlockRange(sender, block_range) => { - tracing::debug!("Fetching compact blocks. {:?}", &block_range); + tracing::info!("Fetching compact blocks. {:?}", &block_range); let block_stream = get_block_range(client, block_range).await; let _ignore_error = sender.send(block_stream); } FetchRequest::NullifierRange(sender, block_range) => { - tracing::debug!("Fetching nullifiers. {:?}", &block_range); + tracing::info!("Fetching nullifiers. {:?}", &block_range); let block_stream = get_block_range_nullifiers(client, block_range).await; let _ignore_error = sender.send(block_stream); } FetchRequest::SubtreeRoots(sender, start_index, shielded_protocol, max_entries) => { - tracing::debug!( + tracing::info!( "Fetching subtree roots. start index: {}. shielded protocol: {}", start_index, shielded_protocol @@ -132,17 +132,17 @@ where let _ignore_error = sender.send(subtree_roots); } FetchRequest::TreeState(sender, block_height) => { - tracing::debug!("Fetching tree state. {:?}", &block_height); + tracing::info!("Fetching tree state. {:?}", &block_height); let tree_state = get_tree_state(client, block_height).await; let _ignore_error = sender.send(tree_state); } FetchRequest::Transaction(sender, txid) => { - tracing::debug!("Fetching transaction. {:?}", txid); + tracing::info!("Fetching transaction. {:?}", txid); let transaction = get_transaction(client, txid).await; let _ignore_error = sender.send(transaction); } FetchRequest::UtxoMetadata(sender, (addresses, start_height)) => { - tracing::debug!( + tracing::info!( "Fetching unspent transparent output metadata from {:?} for addresses:\n{:?}", &start_height, &addresses @@ -151,7 +151,7 @@ where let _ignore_error = sender.send(utxo_metadata); } FetchRequest::TransparentAddressTxs(sender, (address, block_range)) => { - tracing::debug!( + tracing::info!( "Fetching raw transactions in block range {:?} for address {:?}", &block_range, &address diff --git a/pepper-sync/src/config.rs b/pepper-sync/src/config.rs index a71e0b9311..ce65e2b951 100644 --- a/pepper-sync/src/config.rs +++ b/pepper-sync/src/config.rs @@ -86,12 +86,21 @@ pub struct SyncConfig { pub transparent_address_discovery: TransparentAddressDiscovery, /// Performance level pub performance_level: PerformanceLevel, + /// Shutdown on completion + /// + /// If not set, sync will not shutdown until the consumer sets the `SyncMode` to `Shutdown` variant. + /// The sync engine will regularly check for new blocks mined so the wallet will always be updated to the state + /// of the latest chain. + /// + /// If set, sync will still check for any newly mined blocks during scanning. But when the wallet is completely + /// up-to-date with the latest chain, sync will shutdown. + pub shutdown_on_completion: bool, } #[cfg(feature = "wallet_essentials")] impl SyncConfig { fn serialized_version() -> u8 { - 1 + 2 } /// Deserialize into `reader` @@ -101,10 +110,16 @@ impl SyncConfig { let gap_limit = reader.read_u8()?; let scopes = reader.read_u8()?; let performance_level = if version >= 1 { - PerformanceLevel::read(reader)? + PerformanceLevel::read(&mut reader)? } else { PerformanceLevel::High }; + let shutdown_on_completion = if version >= 2 { + reader.read_u8()? != 0 + } else { + false + }; + Ok(Self { transparent_address_discovery: TransparentAddressDiscovery { gap_limit, @@ -115,6 +130,7 @@ impl SyncConfig { }, }, performance_level, + shutdown_on_completion, }) } @@ -133,7 +149,8 @@ impl SyncConfig { scopes |= 0b100; } writer.write_u8(scopes)?; - self.performance_level.write(writer)?; + self.performance_level.write(&mut writer)?; + writer.write_u8(self.shutdown_on_completion as u8)?; Ok(()) } diff --git a/pepper-sync/src/error.rs b/pepper-sync/src/error.rs index 00835f7af0..800b5ada5f 100644 --- a/pepper-sync/src/error.rs +++ b/pepper-sync/src/error.rs @@ -245,7 +245,7 @@ pub enum ScanError { /// Continuity error. #[error("continuity error")] ContinuityError(#[from] ContinuityError), - /// Zcash client backend scan error + /// Invalid encoding. #[error(transparent)] EncodingError(#[from] EncodingInvalid), /// Invalid sapling nullifier @@ -298,6 +298,17 @@ pub enum ScanError { /// Failed to parse encoded address. #[error("failed to parse encoded address")] AddressParseError(#[from] zcash_address::unified::ParseError), + /// Compact transaction contained transparent output with value outside the valid zatoshi range. + #[error( + "compact transaction contained transparent output with value {0} which is outside the valid zatoshi range" + )] + TransparentOutputInvalidValue(u64), + /// All transparent addresses are already in use. + #[error("all transparent addresses are already in use")] + AllAddressesInUse, + /// Transparent address derivation error. + #[error("transparent address derivation error. {0}")] + TransparentAddressDerivationError(bip32::Error), } /// The encoding of a compact Sapling output or compact Orchard action was invalid. diff --git a/pepper-sync/src/scan.rs b/pepper-sync/src/scan.rs index ad316dafe7..215c616be2 100644 --- a/pepper-sync/src/scan.rs +++ b/pepper-sync/src/scan.rs @@ -17,6 +17,7 @@ use zip32::AccountId; use crate::{ client::FetchRequest, error::{ScanError, ServerError}, + keys::transparent::TransparentAddressId, sync::ScanPriority, utils::{block, transaction}, wallet::{NullifierMap, OutputId, ScanTarget, WalletBlock, WalletTransaction}, @@ -29,6 +30,7 @@ pub(crate) mod compact_blocks; pub(crate) mod task; pub(crate) mod transactions; +#[derive(Debug, Clone)] struct InitialScanData { start_seam_block: Option, end_seam_block: Option, @@ -82,12 +84,16 @@ impl InitialScanData { struct ScanData { nullifiers: NullifierMap, + outpoints: BTreeMap, wallet_blocks: BTreeMap, decrypted_scan_targets: BTreeSet, decrypted_note_data: DecryptedNoteData, witness_data: WitnessData, + new_transparent_inuse_addresses: HashMap, + updated_transparent_gap_addresses: HashMap, } +#[derive(Debug)] pub(crate) struct ScanResults { pub(crate) nullifiers: NullifierMap, pub(crate) outpoints: BTreeMap, @@ -96,6 +102,8 @@ pub(crate) struct ScanResults { pub(crate) sapling_located_trees: Vec>, pub(crate) orchard_located_trees: Vec>, pub(crate) ironwood_located_trees: Vec>, + pub(crate) new_transparent_inuse_addresses: HashMap, + pub(crate) updated_transparent_gap_addresses: HashMap, } pub(crate) struct DecryptedNoteData { @@ -126,6 +134,7 @@ pub(crate) async fn scan

( ufvks: &HashMap, scan_task: ScanTask, max_outputs: usize, + transparent_gap_limit: u32, ) -> Result where P: consensus::Parameters + Sync + Send + 'static, @@ -136,7 +145,8 @@ where start_seam_block, end_seam_block, mut scan_targets, - transparent_addresses, + mut transparent_inuse_addresses, + transparent_gap_addresses, } = scan_task; if compact_blocks @@ -157,7 +167,7 @@ where let mut nullifiers = NullifierMap::new(); for block in &compact_blocks { for transaction in &block.vtx { - collect_nullifiers( + collect_nullifiers_compact( &mut nullifiers, block::get_compact_height(block), transaction, @@ -173,6 +183,8 @@ where sapling_located_trees: Vec::new(), orchard_located_trees: Vec::new(), ironwood_located_trees: Vec::new(), + new_transparent_inuse_addresses: HashMap::new(), + updated_transparent_gap_addresses: transparent_gap_addresses, }); } @@ -189,13 +201,18 @@ where let consensus_parameters_clone = consensus_parameters.clone(); let ufvks_clone = ufvks.clone(); + let initial_scan_data_clone = initial_scan_data.clone(); + let transparent_inuse_addresses_clone = transparent_inuse_addresses.clone(); let scan_data = tokio::task::spawn_blocking(move || { scan_compact_blocks( compact_blocks, &consensus_parameters_clone, &ufvks_clone, - initial_scan_data, + initial_scan_data_clone, max_outputs / 8, + transparent_inuse_addresses_clone, + transparent_gap_addresses, + transparent_gap_limit, ) }) .await @@ -203,15 +220,17 @@ where let ScanData { nullifiers, + mut outpoints, wallet_blocks, mut decrypted_scan_targets, decrypted_note_data, witness_data, + new_transparent_inuse_addresses, + updated_transparent_gap_addresses, } = scan_data; - scan_targets.append(&mut decrypted_scan_targets); + transparent_inuse_addresses.extend(new_transparent_inuse_addresses.clone()); - let mut outpoints = BTreeMap::new(); let wallet_transactions = scan_transactions( fetch_request_sender, consensus_parameters, @@ -220,7 +239,7 @@ where decrypted_note_data, &wallet_blocks, &mut outpoints, - transparent_addresses, + transparent_inuse_addresses, ) .await?; @@ -264,15 +283,19 @@ where sapling_located_trees, orchard_located_trees, ironwood_located_trees, + new_transparent_inuse_addresses, + updated_transparent_gap_addresses, }) } /// Converts the nullifiers from a compact transaction and adds them to the nullifier map -fn collect_nullifiers( +fn collect_nullifiers_compact( nullifier_map: &mut NullifierMap, block_height: BlockHeight, transaction: &CompactTx, ) -> Result<(), ScanError> { + let txid = transaction::get_compact_txid(transaction); + transaction .spends .iter() @@ -284,7 +307,7 @@ fn collect_nullifiers( nullifier, ScanTarget { block_height, - txid: transaction::get_compact_txid(transaction), + txid, narrow_scan_area: false, }, ); @@ -308,7 +331,7 @@ fn collect_nullifiers( nullifier, ScanTarget { block_height, - txid: transaction::get_compact_txid(transaction), + txid, narrow_scan_area: false, }, ); @@ -332,10 +355,33 @@ fn collect_nullifiers( nullifier, ScanTarget { block_height, - txid: transaction::get_compact_txid(transaction), + txid, narrow_scan_area: false, }, ); }); Ok(()) } + +/// Adds the outpoints from a compact transaction to the outpoint map. +fn collect_outpoints_compact( + outpoint_map: &mut BTreeMap, + block_height: BlockHeight, + transaction: &CompactTx, +) { + let txid = transaction::get_compact_txid(transaction); + + transaction.vin.iter().for_each(|outpoint| { + let mut txid_bytes = [0u8; 32]; + txid_bytes.copy_from_slice(&outpoint.prevout_txid); + let prevout_txid = TxId::from_bytes(txid_bytes); + outpoint_map.insert( + OutputId::new(prevout_txid, outpoint.prevout_index), + ScanTarget { + block_height, + txid, + narrow_scan_area: true, + }, + ); + }); +} diff --git a/pepper-sync/src/scan/compact_blocks.rs b/pepper-sync/src/scan/compact_blocks.rs index f723dde3d0..f7bbf9e17f 100644 --- a/pepper-sync/src/scan/compact_blocks.rs +++ b/pepper-sync/src/scan/compact_blocks.rs @@ -10,7 +10,11 @@ use tokio::sync::mpsc; use zcash_keys::keys::UnifiedFullViewingKey; use zcash_note_encryption::Domain; use zcash_primitives::block::BlockHash; -use zcash_protocol::consensus::{self, BlockHeight}; +use zcash_protocol::{ + consensus::{self, BlockHeight}, + value::Zatoshis, +}; +use zcash_transparent::address::Script; use zingo_netutils::lightwallet_protocol::{ CompactBlock, CompactOrchardAction, CompactSaplingOutput, }; @@ -19,9 +23,13 @@ use zip32::AccountId; use crate::{ client::{self, FetchRequest}, error::{ContinuityError, ScanError, ServerError}, - keys::{KeyId, ScanningKeyOps, ScanningKeys}, + keys::{ + self, KeyId, ScanningKeyOps, ScanningKeys, + transparent::{TransparentAddressId, TransparentScope}, + }, + scan::collect_outpoints_compact, utils::{block, get_compact_action, get_compact_output_description, transaction}, - wallet::{NullifierMap, OutputId, ScanTarget, TreeBounds, WalletBlock}, + wallet::{KeyIdInterface as _, NullifierMap, OutputId, ScanTarget, TreeBounds, WalletBlock}, witness::WitnessData, }; @@ -29,16 +37,20 @@ use zcash_protocol::{PoolType, ShieldedPool}; use self::runners::{DecryptedOutput, DecryptionBatchRunners}; -use super::{DecryptedNoteData, InitialScanData, ScanData, collect_nullifiers}; +use super::{DecryptedNoteData, InitialScanData, ScanData, collect_nullifiers_compact}; mod runners; +#[allow(clippy::complexity)] pub(super) fn scan_compact_blocks

( compact_blocks: Vec, consensus_parameters: &P, ufvks: &HashMap, initial_scan_data: InitialScanData, output_decryptions_in_batch: usize, + transparent_inuse_addresses: HashMap, + mut transparent_gap_addresses: HashMap, + transparent_gap_limit: u32, ) -> Result where P: consensus::Parameters + Sync + Send + 'static, @@ -78,21 +90,15 @@ where ironwood_initial_tree_size = ironwood_final_tree_size; let block_height = block::get_compact_height(block); + let block_hash = block::get_compact_hash(block); for transaction in &block.vtx { + let txid = transaction::get_compact_txid(transaction); + // collect trial decryption results by transaction - let incoming_sapling_outputs = runners.sapling.collect_results( - block::get_compact_hash(block), - transaction::get_compact_txid(transaction), - ); - let incoming_orchard_outputs = runners.orchard.collect_results( - block::get_compact_hash(block), - transaction::get_compact_txid(transaction), - ); - let incoming_ironwood_outputs = runners.ironwood.collect_results( - block::get_compact_hash(block), - transaction::get_compact_txid(transaction), - ); + let incoming_sapling_outputs = runners.sapling.collect_results(block_hash, txid); + let incoming_orchard_outputs = runners.orchard.collect_results(block_hash, txid); + let incoming_ironwood_outputs = runners.ironwood.collect_results(block_hash, txid); // gather the txids of all transactions relevant to the wallet // the edge case of transactions that this capability created but did not receive change @@ -119,11 +125,7 @@ where }); } - collect_nullifiers( - &mut nullifiers, - block::get_compact_height(block), - transaction, - )?; + collect_nullifiers_compact(&mut nullifiers, block_height, transaction)?; witness_data.sapling_leaves_and_retentions.extend( calculate_sapling_leaves_and_retentions( @@ -169,52 +171,203 @@ where transaction::shielded_output_count(transaction, ShieldedPool::Orchard); ironwood_final_tree_size += transaction::shielded_output_count(transaction, ShieldedPool::Ironwood); + + set_checkpoint_retentions( + block_height, + &mut witness_data.sapling_leaves_and_retentions, + ); + set_checkpoint_retentions( + block_height, + &mut witness_data.orchard_leaves_and_retentions, + ); + set_checkpoint_retentions( + block_height, + &mut witness_data.ironwood_leaves_and_retentions, + ); + + let wallet_block = WalletBlock { + block_height, + block_hash, + prev_hash: block::get_compact_prev_hash(block), + time: block.time, + txids: block + .vtx + .iter() + .map(transaction::get_compact_txid) + .collect(), + tree_bounds: TreeBounds { + sapling_initial_tree_size, + sapling_final_tree_size, + orchard_initial_tree_size, + orchard_final_tree_size, + ironwood_initial_tree_size, + ironwood_final_tree_size, + }, + }; + + check_tree_size(block, &wallet_block)?; + + wallet_blocks.insert(wallet_block.block_height(), wallet_block); } + } - set_checkpoint_retentions( - block_height, - &mut witness_data.sapling_leaves_and_retentions, - ); - set_checkpoint_retentions( - block_height, - &mut witness_data.orchard_leaves_and_retentions, - ); - set_checkpoint_retentions( - block_height, - &mut witness_data.ironwood_leaves_and_retentions, - ); + // retry transparent compact block scanning until the gap limit has been satisfied + let mut outpoints = BTreeMap::new(); + let mut new_transparent_inuse_addresses = HashMap::new(); + 'gap: loop { + let mut gap_addresses_in_use = BTreeSet::new(); + + for block in &compact_blocks { + let block_height = block::get_compact_height(block); + + for transaction in &block.vtx { + let txid = transaction::get_compact_txid(transaction); + + // check transparent outputs against inuse and gap addresses and add outpoints to map + // TODO: only enable for blocks above the initial chain height when sync session started + for output in transaction.vout.iter() { + let output = zcash_transparent::bundle::TxOut::new( + Zatoshis::from_u64(output.value) + .map_err(|_| ScanError::TransparentOutputInvalidValue(output.value))?, + Script(zcash_script::script::Code(output.script_pub_key.clone())), + ); + if let Some(address) = output.recipient_address() { + let encoded_address = + keys::transparent::encode_address(consensus_parameters, address); + if let Some((_address, _key_id)) = + transparent_inuse_addresses.get_key_value(&encoded_address) + { + decrypted_scan_targets.insert(ScanTarget { + block_height, + txid, + narrow_scan_area: true, + }); + } + if let Some((_address, key_id)) = + transparent_gap_addresses.get_key_value(&encoded_address) + { + // NOTE: the new transparent in-use addresses do not need to be appended to the transparent + // in-use addresses in this loop as the scan target has already been added here + gap_addresses_in_use.insert(*key_id); + decrypted_scan_targets.insert(ScanTarget { + block_height, + txid, + narrow_scan_area: true, + }); + } + } + } + collect_outpoints_compact(&mut outpoints, block_height, transaction); + } + } - let wallet_block = WalletBlock { - block_height: block::get_compact_height(block), - block_hash: block::get_compact_hash(block), - prev_hash: block::get_compact_prev_hash(block), - time: block.time, - txids: block - .vtx - .iter() - .map(transaction::get_compact_txid) - .collect(), - tree_bounds: TreeBounds { - sapling_initial_tree_size, - sapling_final_tree_size, - orchard_initial_tree_size, - orchard_final_tree_size, - ironwood_initial_tree_size, - ironwood_final_tree_size, - }, - }; + if gap_addresses_in_use.is_empty() { + break 'gap; + } - check_tree_size(block, &wallet_block)?; + for (account_id, ufvk) in ufvks.iter() { + let Some(account_pubkey) = ufvk.transparent() else { + continue; + }; + + for scope in [ + TransparentScope::External, + TransparentScope::Internal, + TransparentScope::Refund, + ] { + // TODO: collect as nonempty? + let gap_addresses_in_use_scoped = gap_addresses_in_use + .iter() + .filter(|id| id.account_id() == *account_id && id.scope() == scope) + .collect::>(); + + if gap_addresses_in_use_scoped.is_empty() { + continue; + } - wallet_blocks.insert(wallet_block.block_height(), wallet_block); + // NOTE: the `gap_addresses_in_use` cannot be used to determine the first gap address index as there is no + // guarantee the first gap address is in use + let lowest_gap_address_index = transparent_gap_addresses + .values() + .filter(|id| id.account_id() == *account_id && id.scope() == scope) + .map(TransparentAddressId::address_index) + .min() + .expect( + "gap addresses must exist as some are guaranteed to be in use in this scope", + ); + let highest_gap_address_index_in_use = gap_addresses_in_use_scoped + .last() + .expect("non-empty in this scope") + .address_index(); + let no_of_gap_addresses_in_use = highest_gap_address_index_in_use + .saturating_sub(lowest_gap_address_index.index()) + .index() + + 1; + // NOTE: if we saturating add `gap_limit` to directly find the first index to derive we will not error if + // all addresses are already in use + let mut address_index_for_derivation = lowest_gap_address_index + .saturating_add(transparent_gap_limit - 1) + .next() + .ok_or_else(|| ScanError::AllAddressesInUse)?; + let highest_address_index_for_derivation = address_index_for_derivation + .index() + .saturating_add(no_of_gap_addresses_in_use - 1); + loop { + // derive new gap address for each gap address in use + let new_gap_address_id = + TransparentAddressId::new(*account_id, scope, address_index_for_derivation); + let new_gap_address = keys::transparent::derive_address( + consensus_parameters, + account_pubkey, + new_gap_address_id, + ) + .map_err(ScanError::TransparentAddressDerivationError)?; + transparent_gap_addresses.insert(new_gap_address, new_gap_address_id); + + // move the used gap address into inuse addresses + let new_inuse_address = transparent_gap_addresses + .iter() + .find(|(_address, id)| { + id.account_id() == *account_id + && id.scope() == scope + && id.address_index().index() + == new_gap_address_id + .address_index() + .index() + .checked_sub(transparent_gap_limit) + .expect("new gap address index was derived directly from transparent gap addresses. should never underflow!") + }) + .expect("new gap address index was derived directly from transparent gap addresses. should always exist!") + .0 + .clone(); + let new_inuse_address_entry = transparent_gap_addresses + .remove_entry(&new_inuse_address) + .expect("must exist in this scope!"); + new_transparent_inuse_addresses + .insert(new_inuse_address_entry.0, new_inuse_address_entry.1); + + // increment the address index until we have derived all the new gap addresses + if address_index_for_derivation.index() < highest_address_index_for_derivation { + address_index_for_derivation = address_index_for_derivation + .next() + .ok_or_else(|| ScanError::AllAddressesInUse)?; + } else { + break; + } + } + } + } } Ok(ScanData { nullifiers, + outpoints, wallet_blocks, decrypted_scan_targets, decrypted_note_data, witness_data, + new_transparent_inuse_addresses, + updated_transparent_gap_addresses: transparent_gap_addresses, }) } @@ -669,6 +822,9 @@ mod tests { &HashMap::new(), initial_scan_data(100), 100, + HashMap::new(), + HashMap::new(), + 10, ) .unwrap(); @@ -694,6 +850,9 @@ mod tests { &HashMap::new(), initial_scan_data(0), 100, + HashMap::new(), + HashMap::new(), + 10, ); assert!(matches!( diff --git a/pepper-sync/src/scan/task.rs b/pepper-sync/src/scan/task.rs index be9f34ea56..832de8fbad 100644 --- a/pepper-sync/src/scan/task.rs +++ b/pepper-sync/src/scan/task.rs @@ -30,7 +30,7 @@ use crate::{ utils::block, wallet::{ ScanTarget, WalletBlock, - traits::{SyncBlocks, SyncNullifiers, SyncWallet}, + traits::{SyncBlocks, SyncNullifiers, SyncTransactions, SyncWallet}, }, }; @@ -44,28 +44,34 @@ use zingo_netutils::time::{SCANNER_SHUTDOWN_TIMEOUT, STREAM_MSG_TIMEOUT}; pub(crate) enum ScannerState { Verification, Scan, - Shutdown, + Complete, } impl ScannerState { - fn verified(&mut self) { + pub(crate) fn verified(&mut self) { *self = ScannerState::Scan; } - fn shutdown(&mut self) { - *self = ScannerState::Shutdown; + fn completed(&mut self) { + *self = ScannerState::Complete; + } + + pub(crate) fn reverify(&mut self) { + *self = ScannerState::Verification; } } pub(crate) struct Scanner

{ pub(crate) state: ScannerState, loader: Option>, - workers: Vec>, + pub(crate) workers: Vec>, unique_id: usize, scan_results_sender: mpsc::UnboundedSender<(ScanRange, Result)>, fetch_request_sender: mpsc::UnboundedSender, consensus_parameters: P, ufvks: HashMap, + transparent_gap_limit: u32, + pub(crate) transparent_gap_addresses: HashMap, } impl

Scanner

@@ -77,6 +83,7 @@ where scan_results_sender: mpsc::UnboundedSender<(ScanRange, Result)>, fetch_request_sender: mpsc::UnboundedSender, ufvks: HashMap, + transparent_gap_limit: u32, ) -> Self { let workers: Vec> = Vec::with_capacity(MAX_WORKER_POOLSIZE); @@ -89,6 +96,8 @@ where fetch_request_sender, consensus_parameters, ufvks, + transparent_gap_limit, + transparent_gap_addresses: HashMap::new(), } } @@ -131,7 +140,7 @@ where Ok(()) } - async fn shutdown_loader(&mut self) -> Result<(), ServerError> { + pub(crate) async fn shutdown_loader(&mut self) -> Result<(), ServerError> { let loader = self.loader.take(); if let Some(mut loader) = loader { loader.shutdown().await @@ -151,6 +160,7 @@ where self.scan_results_sender.clone(), self.fetch_request_sender.clone(), self.ufvks.clone(), + self.transparent_gap_limit, ); worker.run(max_outputs); self.workers.push(worker); @@ -166,7 +176,7 @@ where } } - fn idle_worker(&self) -> Option<&ScanWorker

> { + pub(crate) fn idle_worker(&self) -> Option<&ScanWorker

> { if let Some(idle_worker) = self.workers.iter().find(|worker| !worker.is_scanning()) { Some(idle_worker) } else { @@ -177,7 +187,7 @@ where /// Shutdown worker by `worker_id`. /// /// Panics if worker with given `worker_id` is not found. - async fn shutdown_worker(&mut self, worker_id: usize) { + pub(crate) async fn shutdown_worker(&mut self, worker_id: usize) { let worker_index = self .workers .iter() @@ -199,11 +209,10 @@ where pub(crate) async fn update( &mut self, wallet: &mut W, - shutdown_mempool: Arc, nullifier_map_limit_exceeded: bool, ) -> Result<(), SyncError> where - W: SyncWallet + SyncBlocks + SyncNullifiers, + W: SyncWallet + SyncBlocks + SyncNullifiers + SyncTransactions, { self.check_loader_error()?; @@ -247,13 +256,7 @@ where self.update_loader(wallet, nullifier_map_limit_exceeded) .map_err(SyncError::WalletError)?; } - ScannerState::Shutdown => { - shutdown_mempool.store(true, atomic::Ordering::Release); - while let Some(worker) = self.idle_worker() { - self.shutdown_worker(worker.id).await; - } - self.shutdown_loader().await?; - } + ScannerState::Complete => {} } Ok(()) @@ -279,7 +282,7 @@ where nullifier_map_limit_exceeded: bool, ) -> Result<(), W::Error> where - W: SyncWallet + SyncBlocks + SyncNullifiers, + W: SyncWallet + SyncBlocks + SyncNullifiers + SyncTransactions, { let loader = self.loader.as_ref().expect("loader should be running"); if !loader.is_loading() { @@ -287,15 +290,32 @@ where &self.consensus_parameters, wallet, nullifier_map_limit_exceeded, + self.transparent_gap_addresses.clone(), )? { loader.add_scan_task(scan_task); } else if wallet.get_sync_state()?.scan_complete() { - self.state.shutdown(); + // if sync is complete, all nullifiers will have been re-fetched so this note metadata can be discarded. + for transaction in wallet.get_wallet_transactions_mut()?.values_mut() { + for note in transaction.sapling_notes.as_mut_slice() { + note.refetch_nullifier_ranges = Vec::new(); + } + for note in transaction.orchard_notes.as_mut_slice() { + note.refetch_nullifier_ranges = Vec::new(); + } + for note in transaction.ironwood_notes.as_mut_slice() { + note.refetch_nullifier_ranges = Vec::new(); + } + } + self.state.completed(); } } Ok(()) } + + pub(crate) fn is_verified(&self) -> bool { + !matches!(self.state, ScannerState::Verification) + } } struct Loader

{ @@ -612,11 +632,11 @@ where .expect("loader should always have a handle to take!"); match tokio::time::timeout(SCANNER_SHUTDOWN_TIMEOUT, &mut handle).await { - Ok(join_res) => join_res.expect("task panicked")?, + Ok(res) => res.expect("task panicked")?, Err(_) => { + tracing::warn!("Loader shutdown timed out!"); handle.abort(); let _ = handle.await; - return Err(tonic::Status::deadline_exceeded("loader shutdown timeout").into()); } } @@ -633,6 +653,7 @@ pub(crate) struct ScanWorker

{ scan_results_sender: mpsc::UnboundedSender<(ScanRange, Result)>, fetch_request_sender: mpsc::UnboundedSender, ufvks: HashMap, + transparent_gap_limit: u32, } impl

ScanWorker

@@ -645,6 +666,7 @@ where scan_results_sender: mpsc::UnboundedSender<(ScanRange, Result)>, fetch_request_sender: mpsc::UnboundedSender, ufvks: HashMap, + transparent_gap_limit: u32, ) -> Self { Self { id, @@ -655,12 +677,18 @@ where scan_results_sender, fetch_request_sender, ufvks, + transparent_gap_limit, } } + pub(crate) fn id(&self) -> usize { + self.id + } + /// Runs the worker in a new tokio task. /// /// Waits for a scan task and then calls [`crate::scan::scan`] on the given range. + // TODO: max_outputs can be moved to scan worker field fn run(&mut self, max_outputs: usize) { let (scan_task_sender, mut scan_task_receiver) = mpsc::channel::(1); @@ -669,6 +697,7 @@ where let fetch_request_sender = self.fetch_request_sender.clone(); let consensus_parameters = self.consensus_parameters.clone(); let ufvks = self.ufvks.clone(); + let transparent_gap_limit = self.transparent_gap_limit; let handle = tokio::spawn(async move { while let Some(scan_task) = scan_task_receiver.recv().await { @@ -679,6 +708,7 @@ where &ufvks, scan_task, max_outputs, + transparent_gap_limit, ) .await; let _ignore_error = scan_results_sender.send((scan_range, scan_results)); @@ -709,7 +739,8 @@ where /// Shuts down worker by dropping the sender to the worker task and awaiting the handle. /// - /// This should always be called in the context of the scanner as it must be also be removed from the worker pool. + /// This should always be called in the context of the scanner as it must be also be removed from the worker pool + /// (See `Scanner::shutdown_worker`). async fn shutdown(&mut self) -> Result<(), JoinError> { tracing::debug!("Shutting down worker {}", self.id); if let Some(sender) = self.scan_task_sender.take() { @@ -724,6 +755,7 @@ where match tokio::time::timeout(SCANNER_SHUTDOWN_TIMEOUT, &mut handle).await { Ok(res) => res, Err(_) => { + tracing::warn!("Worker shutdown timed out!"); handle.abort(); let _ = handle.await; // ignore join error after abort Ok(()) @@ -739,7 +771,8 @@ pub(crate) struct ScanTask { pub(crate) start_seam_block: Option, pub(crate) end_seam_block: Option, pub(crate) scan_targets: BTreeSet, - pub(crate) transparent_addresses: HashMap, + pub(crate) transparent_inuse_addresses: HashMap, + pub(crate) transparent_gap_addresses: HashMap, } impl ScanTask { @@ -748,7 +781,8 @@ impl ScanTask { start_seam_block: Option, end_seam_block: Option, scan_targets: BTreeSet, - transparent_addresses: HashMap, + transparent_inuse_addresses: HashMap, + transparent_gap_addresses: HashMap, ) -> Self { Self { compact_blocks: Vec::new(), @@ -756,7 +790,8 @@ impl ScanTask { start_seam_block, end_seam_block, scan_targets, - transparent_addresses, + transparent_inuse_addresses, + transparent_gap_addresses, } } @@ -827,7 +862,8 @@ impl ScanTask { start_seam_block: self.start_seam_block, end_seam_block: upper_task_first_block, scan_targets: lower_task_scan_targets, - transparent_addresses: self.transparent_addresses.clone(), + transparent_inuse_addresses: self.transparent_inuse_addresses.clone(), + transparent_gap_addresses: self.transparent_gap_addresses.clone(), }, ScanTask { compact_blocks: upper_compact_blocks, @@ -838,7 +874,8 @@ impl ScanTask { start_seam_block: lower_task_last_block, end_seam_block: self.end_seam_block, scan_targets: upper_task_scan_targets, - transparent_addresses: self.transparent_addresses, + transparent_inuse_addresses: self.transparent_inuse_addresses, + transparent_gap_addresses: self.transparent_gap_addresses, }, )) } diff --git a/pepper-sync/src/scan/transactions.rs b/pepper-sync/src/scan/transactions.rs index 9f3555b1ed..697165ddaf 100644 --- a/pepper-sync/src/scan/transactions.rs +++ b/pepper-sync/src/scan/transactions.rs @@ -34,7 +34,7 @@ use zip32::AccountId; use crate::{ client::{self, FetchRequest}, - error::ScanError, + error::{ScanError, ServerError}, keys::{self, KeyId, transparent::TransparentAddressId}, wallet::{ IronwoodNote, NullifierMap, OrchardNote, OutgoingIronwoodNote, OutgoingNote, @@ -90,21 +90,32 @@ pub(crate) async fn scan_transactions( decrypted_note_data: DecryptedNoteData, wallet_blocks: &BTreeMap, outpoint_map: &mut BTreeMap, - transparent_addresses: HashMap, + transparent_inuse_addresses: HashMap, ) -> Result, ScanError> { let mut wallet_transactions = HashMap::with_capacity(scan_targets.len()); for scan_target in scan_targets { + // TODO: replace wth optional txid in scan targets if scan_target.txid == TxId::from_bytes([0u8; 32]) { continue; } - let (transaction, block_height) = client::get_transaction_and_block_height( + // in case of re-orgs or consumers manually adding incorrect scan target txids, skip if request fails. + let (transaction, block_height) = match client::get_transaction_and_block_height( fetch_request_sender.clone(), consensus_parameters, scan_target.txid, ) - .await?; + .await + { + Ok((tx, height)) => (tx, height), + Err(ServerError::RequestFailed(_)) => { + continue; + } + Err(e) => { + return Err(e.into()); + } + }; if transaction.txid() != scan_target.txid { return Err(ScanError::IncorrectTxid { @@ -134,7 +145,7 @@ pub(crate) async fn scan_transactions( Some(&decrypted_note_data), &mut NullifierMap::new(), outpoint_map, - &transparent_addresses, + &transparent_inuse_addresses, wallet_block.time(), )?; wallet_transactions.insert(scan_target.txid, wallet_transaction); @@ -163,7 +174,7 @@ pub(crate) fn scan_transaction( decrypted_note_data: Option<&DecryptedNoteData>, nullifier_map: &mut NullifierMap, outpoint_map: &mut BTreeMap, - transparent_addresses: &HashMap, + transparent_inuse_addresses: &HashMap, datetime: u32, ) -> Result { let block_height = status.get_height(); @@ -247,7 +258,7 @@ pub(crate) fn scan_transaction( consensus_parameters, &mut transparent_coins, txid, - transparent_addresses, + transparent_inuse_addresses, transparent_outputs, ); @@ -429,13 +440,15 @@ fn scan_incoming_coins( consensus_parameters: &P, transparent_coins: &mut Vec, txid: TxId, - transparent_addresses: &HashMap, + transparent_inuse_addresses: &HashMap, transparent_outputs: &[zcash_transparent::bundle::TxOut], ) { for (output_index, output) in transparent_outputs.iter().enumerate() { if let Some(address) = output.recipient_address() { let encoded_address = keys::transparent::encode_address(consensus_parameters, address); - if let Some((address, key_id)) = transparent_addresses.get_key_value(&encoded_address) { + if let Some((address, key_id)) = + transparent_inuse_addresses.get_key_value(&encoded_address) + { let output_id = OutputId::new( txid, output_index diff --git a/pepper-sync/src/sync.rs b/pepper-sync/src/sync.rs index 7afb39619e..62edf24976 100644 --- a/pepper-sync/src/sync.rs +++ b/pepper-sync/src/sync.rs @@ -3,9 +3,9 @@ use std::collections::{BTreeMap, HashMap}; use std::convert::Infallible; use std::ops::Range; -use std::sync::Arc; use std::sync::atomic::{self, AtomicBool, AtomicU8, AtomicU32}; -use std::time::{Duration, SystemTime}; +use std::sync::{Arc, OnceLock}; +use std::time::{Duration, Instant, SystemTime}; use shardtree::ShardTree; use shardtree::store::memory::MemoryShardStore; @@ -35,7 +35,7 @@ use crate::scan::ScanResults; use crate::scan::task::{Scanner, ScannerState}; use crate::scan::transactions::scan_transaction; use crate::shardtree_ext::{RollbackOutcome, ShardTreeExt}; -use crate::sync::state::truncate_scan_ranges; +use crate::sync::state::VerifyEnd; use crate::wallet::traits::{ SyncBlocks, SyncNullifiers, SyncOutPoints, SyncShardTrees, SyncTransactions, SyncWallet, }; @@ -52,6 +52,12 @@ pub(crate) mod state; pub(crate) mod transparent; pub mod truncate; +// TODO: investigate a potential case where: +// - a wallet syncs, including the latest incomplete shard +// - the wallet is not opened for some time, the incomplete shard has completed since +// - on next sync, the shard roots *after* the incomplete shard are fetched +// - the wallet can't spend because the incomplete shard is not prioritized to be completed + /// The deepest chain reorganization the wallet tolerates, and the /// repository's single source of truth for that depth. It mirrors the /// validator's finalization boundary, zebra's @@ -177,7 +183,7 @@ impl std::fmt::Display for SyncResult { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!( f, - "Sync completed succesfully: + "Sync result: {{ sync start height: {} sync end height: {} @@ -368,6 +374,11 @@ impl ScanRange { } } +enum MempoolMessage { + Transaction(RawTransaction), + NewBlockMined, +} + /// Syncs a wallet to the latest state of the blockchain. /// /// `sync_mode` is intended to be stored in a struct that owns the wallet(s) (i.e. lightclient) and has a non-atomic @@ -378,6 +389,8 @@ impl ScanRange { /// times in quick sucession without the sync engine interrupting. /// Set `sync_mode` back to `Running` to resume scanning. /// Set `sync_mode` to `Shutdown` to stop the sync process. +/// Wallet keys must not change while sync is running. Sync must be stoppped and run again after the key material has +/// been updated. pub async fn sync( client: C, consensus_parameters: &P, @@ -422,7 +435,7 @@ where let unprocessed_mempool_transactions_count = Arc::new(AtomicU32::new(0)); let unprocessed_mempool_transactions_count_clone = unprocessed_mempool_transactions_count.clone(); - let mempool_stream_connected_at = Arc::new(std::sync::OnceLock::new()); + let mempool_stream_connected_at = Arc::new(OnceLock::new()); let mempool_stream_connected_at_clone = mempool_stream_connected_at.clone(); let mempool_handle = tokio::spawn(async move { mempool_monitor( @@ -435,210 +448,328 @@ where .await }); - // pre-scan initialisation - let chain_height = client::get_chain_height(fetch_request_sender.clone()).await?; - if chain_height == 0.into() { - return Err(SyncError::ServerError(ServerError::GenesisBlockOnly)); - } - let last_known_chain_height = checked_wallet_height( - &mut *wallet.write().await, - chain_height, - consensus_parameters, - )?; - + // create channel for receiving scan results and launch scanner let ufvks = wallet .read() .await .get_unified_full_viewing_keys() .map_err(SyncError::WalletError)?; - - transparent::update_addresses_and_scan_targets( - consensus_parameters, - wallet.clone(), - fetch_request_sender.clone(), - &ufvks, - last_known_chain_height, - chain_height, - config.transparent_address_discovery, - ) - .await?; - - update_subtree_roots( - consensus_parameters, - fetch_request_sender.clone(), - &mut *wallet.write().await, - ) - .await?; - - add_initial_frontier( - consensus_parameters, - fetch_request_sender.clone(), - &mut *wallet.write().await, - ) - .await?; - - let initial_reorg_detection_start_height = state::update_scan_ranges( - consensus_parameters, - fetch_request_sender.clone(), - last_known_chain_height, - chain_height, - &mut *wallet.write().await, - ) - .await?; - - state::set_initial_state( - consensus_parameters, - fetch_request_sender.clone(), - &mut *wallet.write().await, - chain_height, - ) - .await?; - - expire_transactions(&mut *wallet.write().await)?; - - publish_sync_status(&*wallet.read().await, &progress).await; - - // create channel for receiving scan results and launch scanner let (scan_results_sender, mut scan_results_receiver) = mpsc::unbounded_channel(); let mut scanner = Scanner::new( consensus_parameters.clone(), scan_results_sender, fetch_request_sender.clone(), ufvks.clone(), + config.transparent_address_discovery.gap_limit as u32, ); scanner.launch(config.performance_level); - // TODO: implement an option for continuous scanning where it doesnt exit when complete + state::reset_scan_ranges( + wallet + .write() + .await + .get_sync_state_mut() + .map_err(SyncError::WalletError)?, + ); + let mut check_for_new_blocks = false; + let mut first_verification_complete = false; + let mut mempool_shutdown_timer = None; let mut nullifier_map_limit_exceeded = false; + let mut continuous_sync_interval = tokio::time::interval(Duration::from_secs(120)); + continuous_sync_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + continuous_sync_interval.tick().await; let mut interval = tokio::time::interval(Duration::from_millis(50)); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - loop { - tokio::select! { - Some((scan_range, scan_results)) = scan_results_receiver.recv() => { - let mut wallet_guard = wallet.write().await; - process_scan_results( - consensus_parameters, - &mut *wallet_guard, - fetch_request_sender.clone(), - &ufvks, - scan_range, - scan_results, + 'continuous_sync: loop { + let mut reorg_occured = false; + scanner.state.reverify(); + let chain_height = client::get_chain_height(fetch_request_sender.clone()).await?; + if chain_height == 0.into() { + return Err(SyncError::ServerError(ServerError::GenesisBlockOnly)); + } + + // hold wallet guard until initial sync state is set to avoid inconsistencies and potential subtraction overflows + // when calculating sync status. + let mut wallet_guard = wallet.write().await; + let last_known_chain_height = + checked_wallet_height(&mut *wallet_guard, chain_height, consensus_parameters)?; + let new_blocks_mined = chain_height > last_known_chain_height; + state::create_scan_range( + last_known_chain_height, + chain_height, + wallet_guard + .get_sync_state_mut() + .map_err(SyncError::WalletError)?, + ); + // only set intiial sync state on first continuous sync loop. + // otherwise, only extend the wallet tree bounds to include new blocks. + if first_verification_complete && new_blocks_mined { + state::update_wallet_tree_bounds( + consensus_parameters, + fetch_request_sender.clone(), + &mut *wallet_guard, + chain_height, + ) + .await?; + } else if !first_verification_complete { + state::set_initial_state( + consensus_parameters, + fetch_request_sender.clone(), + &mut *wallet_guard, + chain_height, + ) + .await?; + } + drop(wallet_guard); + + // on the first verification we verify the previous wallet state. + // afterwards, we only verify the newly mined blocks if they exist. + let mut initial_reorg_detection_start_height_opt = + if first_verification_complete && new_blocks_mined { + Some(last_known_chain_height + 1) + } else if first_verification_complete { + None + } else { + wallet + .read() + .await + .get_sync_state() + .map_err(SyncError::WalletError)? + .highest_scanned_height() + .map(|highest_scanned_height| highest_scanned_height + 1) + }; + + if let Some(initial_reorg_detection_start_height) = initial_reorg_detection_start_height_opt + { + if initial_reorg_detection_start_height <= chain_height { + state::set_verify_scan_range( + wallet + .write() + .await + .get_sync_state_mut() + .map_err(SyncError::WalletError)?, initial_reorg_detection_start_height, - config.performance_level, - &mut nullifier_map_limit_exceeded, - ) - .await?; - wallet_guard.set_save_flag().map_err(SyncError::WalletError)?; - publish_sync_status(&*wallet_guard, &progress).await; - drop(wallet_guard); + VerifyEnd::VerifyLowest, + ); + } else { + // in this case, no blocks have been mined since last sync session but a re-org may have occured + let chain_height_server_block = + client::get_compact_block(fetch_request_sender.clone(), chain_height).await?; + let chain_height_wallet_block = wallet + .read() + .await + .get_wallet_block(chain_height) + .map_err(SyncError::WalletError)?; + if chain_height_wallet_block.block_hash().0.to_vec() + != chain_height_server_block.hash + { + tracing::info!("Re-org detected."); + reorg_occured = true; + // hold wallet guard until initial sync state is set to avoid inconsistencies and potential subtraction overflows + // when calculating sync status + let mut wallet_guard = wallet.write().await; + let scan_range_to_verify = state::set_verify_scan_range( + wallet_guard + .get_sync_state_mut() + .map_err(SyncError::WalletError)?, + chain_height, + VerifyEnd::VerifyHighest, + ); + initial_reorg_detection_start_height_opt = + Some(scan_range_to_verify.block_range().start); + truncate_wallet_data( + &mut *wallet_guard, + initial_reorg_detection_start_height_opt + .expect("value must exist in this scope") + - 1, + )?; + state::set_initial_state( + consensus_parameters, + fetch_request_sender.clone(), + &mut *wallet_guard, + chain_height, + ) + .await?; + } else { + // there are no new blocks to verify and no re-org has occured + scanner.state.verified(); + } } - - Some(raw_transaction) = mempool_transaction_receiver.recv() => { - let mut wallet_guard = wallet.write().await; - process_mempool_transaction( - consensus_parameters, - &ufvks, - &mut *wallet_guard, - raw_transaction, - ) - .await?; - unprocessed_mempool_transactions_count.fetch_sub(1, atomic::Ordering::Release); - wallet_guard.set_save_flag().map_err(SyncError::WalletError)?; - drop(wallet_guard); + } else { + // first verification not complete: first time sync, there are no previously synced blocks to verify. + // first verification complete: no newly mined blocks to verify. + scanner.state.verified(); + } + + if new_blocks_mined || reorg_occured { + // only perform transparent address discovery on the first continuous sync loop. + // transparent data in newly mined blocks during the sync session will be scanned in compact blocks. + // address discovery is still necessary as scanning compact blocks non-linearly may lead to missing funds + // or requiring rescanning multiple times. + if !first_verification_complete { + scanner.transparent_gap_addresses.extend( + transparent::address_discovery( + consensus_parameters, + wallet.clone(), + fetch_request_sender.clone(), + &ufvks, + last_known_chain_height, + chain_height, + config.transparent_address_discovery.clone(), + ) + .await?, + ); } - _update_scanner = interval.tick() => { - sync_mode_enum = SyncMode::from_atomic_u8(sync_mode.clone())?; - match sync_mode_enum { - SyncMode::Paused => { - let mut pause_interval = tokio::time::interval(Duration::from_secs(1)); - pause_interval.tick().await; - while sync_mode_enum == SyncMode::Paused { - pause_interval.tick().await; - sync_mode_enum = SyncMode::from_atomic_u8(sync_mode.clone())?; + update_subtree_roots( + consensus_parameters, + fetch_request_sender.clone(), + &mut *wallet.write().await, + ) + .await?; + + expire_transactions(&mut *wallet.write().await)?; + } + + // now transparent scan targets and subtree roots have been added, set ranges to be prioritized for scanning. + state::prioritize_scan_ranges( + consensus_parameters, + chain_height, + &mut *wallet.write().await, + ) + .map_err(SyncError::WalletError)?; + + // frontier is added after subtree roots to retain subtree roots below birthday + add_initial_frontier( + consensus_parameters, + fetch_request_sender.clone(), + &mut *wallet.write().await, + ) + .await?; + + // publish sync status prior to scanning + publish_sync_status(&*wallet.read().await, &progress).await; + + 'scan: loop { + tokio::select! { + Some((scan_range, scan_results)) = scan_results_receiver.recv() => { + let mut wallet_guard = wallet.write().await; + if let Some(updated_transparent_gap_addresses) = process_scan_results( + consensus_parameters, + &mut *wallet_guard, + fetch_request_sender.clone(), + &ufvks, + scan_range, + scan_results, + initial_reorg_detection_start_height_opt, + config.performance_level, + &mut nullifier_map_limit_exceeded, + ) + .await? { + // NOTE: this is safe in the current architecture as the correct set of gap addressses will be + // determined before scanning begins and this update will only apply to the latest newly mined + // block(s). If the sync engine is modified so there are cases where compact blocks may be scanned + // for transparent data out-of-order, more checks must be applied here to ensure gap addresses are + // not lost and correctly follow on from the wallets current in-use address list. + scanner.transparent_gap_addresses = updated_transparent_gap_addresses; + } + publish_sync_status(&*wallet_guard, &progress).await; + wallet_guard.set_save_flag().map_err(SyncError::WalletError)?; + drop(wallet_guard); + } + + Some(mempool_message) = mempool_transaction_receiver.recv() => { + match mempool_message { + MempoolMessage::Transaction(raw_transaction) => { + let mut wallet_guard = wallet.write().await; + process_mempool_transaction( + consensus_parameters, + &ufvks, + &mut *wallet_guard, + raw_transaction, + ) + .await?; + unprocessed_mempool_transactions_count.fetch_sub(1, atomic::Ordering::Release); + wallet_guard.set_save_flag().map_err(SyncError::WalletError)?; + drop(wallet_guard); + } + MempoolMessage::NewBlockMined => { + check_for_new_blocks = true; } - }, - SyncMode::Shutdown => { - let mut wallet_guard = wallet.write().await; - let sync_status = match sync_status(&*wallet_guard).await { - Ok(status) => status, - Err(SyncStatusError::WalletError(e)) => { - return Err(SyncError::WalletError(e)); - } - Err(SyncStatusError::NoSyncData) => { - panic!("sync data must exist!"); - } - }; - wallet_guard - .set_save_flag() - .map_err(SyncError::WalletError)?; - let _ = progress.send(Some(sync_status.clone())); - drop(wallet_guard); - mempool_handle.abort(); - fetcher_handle.abort(); - tracing::info!("Sync successfully shutdown."); - - return Ok(SyncResult { - sync_start_height: sync_status.sync_start_height, - sync_end_height: (sync_status - .scan_ranges - .last() - .expect("should be non-empty after syncing") - .block_range() - .end - - 1), - blocks_scanned: sync_status.session_blocks_scanned, - sapling_outputs_scanned: sync_status.session_sapling_outputs_scanned, - orchard_outputs_scanned: sync_status.session_orchard_outputs_scanned, - ironwood_outputs_scanned: sync_status.session_ironwood_outputs_scanned, - percentage_total_outputs_scanned: sync_status.percentage_total_outputs_scanned, - }); } - SyncMode::Running => (), - SyncMode::NotRunning => { - panic!("sync mode should not be manually set to NotRunning!"); - }, } - scanner.update(&mut *wallet.write().await, shutdown_mempool.clone(), nullifier_map_limit_exceeded).await?; - - if matches!(scanner.state, ScannerState::Shutdown) { - // Drain check on a 25ms cadence instead of the old - // unconditional one-second sleep. The policy lives in - // [`drain_verdict`]: shutdown requires a drained - // scanner AND a mempool stream that has been connected - // long enough to have served pre-existing content, so - // a first-loop shutdown on a fully synced chain waits - // for the subscription instead of closing the session - // before the monitor ever connects. The old one-second - // ceiling remains the worst case. - let shutdown_poll_started = std::time::Instant::now(); - let mempool_drained = loop { - let verdict = drain_verdict( - scanner.worker_poolsize(), - unprocessed_mempool_transactions_count - .load(atomic::Ordering::Acquire), - mempool_stream_connected_at.get().map(|at| at.elapsed()), - shutdown_poll_started.elapsed(), - ); - match verdict { - DrainVerdict::Shutdown => break true, - DrainVerdict::Reenter => break false, - DrainVerdict::KeepPolling => { - tokio::time::sleep(std::time::Duration::from_millis(25)).await; + _update_scanner = interval.tick() => { + sync_mode_enum = SyncMode::from_atomic_u8(sync_mode.clone())?; + match sync_mode_enum { + SyncMode::Paused => { + let mut pause_interval = tokio::time::interval(Duration::from_secs(1)); + pause_interval.tick().await; + while sync_mode_enum == SyncMode::Paused { + pause_interval.tick().await; + sync_mode_enum = SyncMode::from_atomic_u8(sync_mode.clone())?; + } + }, + SyncMode::Shutdown => { + match mempool_drain_verdict( + shutdown_mempool.clone(), + unprocessed_mempool_transactions_count.clone(), mempool_shutdown_timer.get_or_insert_with(Instant::now).elapsed(), + mempool_stream_connected_at.get().map(|at| at.elapsed()) + ).await { + MempoolDrainVerdict::NotShutdown | MempoolDrainVerdict::ShutdownNotDrained => { + continue 'scan; + } + MempoolDrainVerdict::ShutdownAndDrainComplete => { + break 'continuous_sync; + } } } - }; - if mempool_drained { - tracing::info!("Sync successfully shutdown."); - break; + SyncMode::Running => (), + SyncMode::NotRunning => { + panic!("sync mode should not be manually set to NotRunning!"); + }, + } + + if check_for_new_blocks && scanner.is_verified() { + continuous_sync_interval.reset(); + check_for_new_blocks = false; + first_verification_complete = true; + continue 'continuous_sync; } + + scanner.update(&mut *wallet.write().await, nullifier_map_limit_exceeded).await?; + + if matches!(scanner.state, ScannerState::Complete) && config.shutdown_on_completion { + sync_mode_enum = SyncMode::Shutdown; + sync_mode.store(sync_mode_enum as u8, atomic::Ordering::Release); + + } + } + + _check_new_block_mined = continuous_sync_interval.tick() => { + // if the mempool has not triggered a new block within the time expected, force a new block check. + check_for_new_blocks = true; } } } } + + // shutdown workers and loader + while scanner.worker_poolsize() != 0 { + let worker_id = scanner + .workers + .first() + .expect("non empty in this scope!") + .id(); + scanner.shutdown_worker(worker_id).await; + } + scanner.shutdown_loader().await?; + let mut wallet_guard = wallet.write().await; + wallet_guard + .set_save_flag() + .map_err(SyncError::WalletError)?; let sync_status = match sync_status(&*wallet_guard).await { Ok(status) => status, Err(SyncStatusError::WalletError(e)) => { @@ -648,28 +779,8 @@ where panic!("sync data must exist!"); } }; - // all blocks up to the last known chain height are now scanned, so any transaction still - // pending past its expiry height is genuinely expired. - expire_transactions(&mut *wallet_guard)?; - // once sync is complete, all nullifiers will have been re-fetched so this note metadata can be discarded. - for transaction in wallet_guard - .get_wallet_transactions_mut() - .map_err(SyncError::WalletError)? - .values_mut() - { - for note in transaction.sapling_notes.as_mut_slice() { - note.refetch_nullifier_ranges = Vec::new(); - } - for note in transaction.orchard_notes.as_mut_slice() { - note.refetch_nullifier_ranges = Vec::new(); - } - for note in transaction.ironwood_notes.as_mut_slice() { - note.refetch_nullifier_ranges = Vec::new(); - } - } - wallet_guard - .set_save_flag() - .map_err(SyncError::WalletError)?; + // TODO: return an error if progress is not updated + let _ignore_error = progress.send(Some(sync_status.clone())); drop(wallet_guard); drop(scanner); @@ -681,6 +792,7 @@ where Err(e) => return Err(e.into()), } fetcher_handle.await.expect("task panicked"); + tracing::info!("Sync successfully shutdown."); Ok(SyncResult { sync_start_height: sync_status.sync_start_height, @@ -739,7 +851,7 @@ where // The wallet reported height is above the current proxy height // reset to the proxy height. truncate_wallet_data(wallet, chain_height)?; - truncate_scan_ranges( + state::truncate_scan_ranges( chain_height, wallet .get_sync_state_mut() @@ -1217,60 +1329,51 @@ pub(crate) fn set_transactions_failed_unchecked( reset_spends(wallet_transactions, failed_txids); } -/// Returns true if the scanner and mempool are shutdown. -/// Verdict for one pass of the scanner-shutdown drain poll. Pure over a -/// snapshot: the caller loads the atomics and clocks. This only -/// decides, so the whole policy is table-testable without a runtime. #[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum DrainVerdict { - /// Drained and the mempool stream has settled: end the session. - Shutdown, - /// Work appeared, or the ceiling expired with workers running: - /// re-enter the processing loop. - Reenter, - /// Undecided: sleep one cadence and poll again. - KeepPolling, +enum MempoolDrainVerdict { + /// The mempool stream has been connected for sufficient duration to be shutdown and all recieved mempool + /// transactions are processed. + ShutdownAndDrainComplete, + /// The mempool stream has been connected for sufficient duration to be shutdown but not all recieved mempool + /// transactions have been processed. + ShutdownNotDrained, + /// The mempool stream has not been connected for sufficient duration to be shutdown. + NotShutdown, } -/// The drain policy for scanner shutdown. -/// -/// `connected_for` is the age of the mempool stream subscription -/// (`None` until it is established). Connection, not first delivery, -/// is deliberately the grace trigger: an empty mempool never delivers, -/// and a delivery-based grace would hold every such session for the -/// full ceiling, restoring the fixed second c90f8d309 removed. The -/// settle window covers the gap between subscribing and receiving -/// pre-existing mempool content (served within ~100ms of connect). -fn drain_verdict( - scan_workers: usize, - unprocessed_mempool_transactions: u32, - connected_for: Option, - poll_elapsed: Duration, -) -> DrainVerdict { +async fn mempool_drain_verdict( + shutdown_mempool: Arc, + unprocessed_mempool_transactions_count: Arc, + mempool_shutdown_timer: Duration, + mempool_stream_connection_timer: Option, +) -> MempoolDrainVerdict { use zingo_netutils::time::{MEMPOOL_DRAIN_CEILING, MEMPOOL_DRAIN_SETTLE}; - if unprocessed_mempool_transactions > 0 { - return DrainVerdict::Reenter; - } - if poll_elapsed >= MEMPOOL_DRAIN_CEILING { - return if scan_workers == 0 { - DrainVerdict::Shutdown - } else { - DrainVerdict::Reenter + if mempool_shutdown_timer < MEMPOOL_DRAIN_CEILING { + let Some(mempool_elapsed) = mempool_stream_connection_timer else { + // if mempool stream has not connected yet, continue scanning unless its timed out + return MempoolDrainVerdict::NotShutdown; }; + // wait if the mempool stream has not been connected for sufficient time to receive mempool transactions + if let Some(mempool_startup_remaining) = MEMPOOL_DRAIN_SETTLE.checked_sub(mempool_elapsed) { + tokio::time::sleep(mempool_startup_remaining).await; + } } - match connected_for { - Some(age) if age >= MEMPOOL_DRAIN_SETTLE && scan_workers == 0 => DrainVerdict::Shutdown, - // Settled, but the scanner still holds workers. Polling cannot retire - // them: only the main loop's `scanner.update()` can, and it runs - // outside this loop, so waiting here burns the ceiling on a value that - // cannot move. Re-enter and let the loop make progress instead. - Some(age) if age >= MEMPOOL_DRAIN_SETTLE => DrainVerdict::Reenter, - _ => DrainVerdict::KeepPolling, + + // shutdown mempool monitor + shutdown_mempool.store(true, atomic::Ordering::Release); + + // continue scanning if mempool transaction have been received but haven't finished being preocessed + if unprocessed_mempool_transactions_count.load(atomic::Ordering::Acquire) > 0 { + return MempoolDrainVerdict::ShutdownNotDrained; } + + MempoolDrainVerdict::ShutdownAndDrainComplete } -/// Scan post-processing +/// Scan post-processing. +/// +/// Returns the updated transparent gap addresses or none in the case of a recovered error i.e. re-org. #[allow(clippy::too_many_arguments)] async fn process_scan_results( consensus_parameters: &impl consensus::Parameters, @@ -1279,10 +1382,10 @@ async fn process_scan_results( ufvks: &HashMap, scan_range: ScanRange, scan_results: Result, - initial_reorg_detection_start_height: BlockHeight, + initial_reorg_detection_start_height: Option, performance_level: PerformanceLevel, nullifier_map_limit_exceeded: &mut bool, -) -> Result<(), SyncError> +) -> Result>, SyncError> where W: SyncWallet + SyncBlocks @@ -1302,6 +1405,8 @@ where sapling_located_trees, orchard_located_trees, ironwood_located_trees, + new_transparent_inuse_addresses, + updated_transparent_gap_addresses, } = results; if scan_range.priority() == ScanPriority::ScannedWithoutMapping { @@ -1388,7 +1493,7 @@ where "Nullifiers discarded and will be re-fetched to avoid missing spends." ); - return Ok(()); + return Ok(Some(updated_transparent_gap_addresses)); } spend::update_shielded_spends( @@ -1426,6 +1531,7 @@ where } let mut map_nullifiers = !*nullifier_map_limit_exceeded; + // TODO: do we need to remove this now we scan compact blocks? // all transparent spend locations are known before scanning so there is no need to map outpoints from untargetted ranges. // outpoints of untargetted ranges will still be checked before being discarded. let map_outpoints = scan_range.priority() >= ScanPriority::FoundNote; @@ -1482,6 +1588,7 @@ where sapling_located_trees, orchard_located_trees, ironwood_located_trees, + new_transparent_inuse_addresses, ) .await?; spend::update_transparent_spends( @@ -1532,6 +1639,8 @@ where ); remove_irrelevant_data(wallet).map_err(SyncError::WalletError)?; tracing::debug!("Scan results processed."); + + Ok(Some(updated_transparent_gap_addresses)) } Err(ScanError::ContinuityError(ContinuityError::HashDiscontinuity { height, .. })) => { tracing::warn!("Hash discontinuity detected before block {height}."); @@ -1563,7 +1672,9 @@ where .start; state::merge_scan_ranges(sync_state, ScanPriority::Verify); - if initial_reorg_detection_start_height - current_reorg_detection_start_height + if initial_reorg_detection_start_height + .expect("re-org can only be detected in wallets that have synced previously!") + - current_reorg_detection_start_height > MAX_REORG_ALLOWANCE { clear_wallet_data(wallet)?; @@ -1580,8 +1691,12 @@ where last_known_chain_height, ) .await?; + + Ok(None) } else { - scan_results?; + Err(scan_results + .expect_err("must be error variant in this scope") + .into()) } } Err(ScanError::IncorrectTreeSize { @@ -1596,7 +1711,7 @@ where wallet's {pool:?} records are being cleared back to the pool activation height \ and the next sync rescans from there." ); - return Err(truncate_to_pool_activation_height( + Err(truncate_to_pool_activation_height( consensus_parameters, fetch_request_sender.clone(), wallet, @@ -1605,12 +1720,10 @@ where block_metadata_size, calculated_size, ) - .await?); + .await?) } - Err(e) => return Err(e.into()), + Err(e) => Err(e.into()), } - - Ok(()) } /// Truncates the wallet back to the `target_pool` activation height. @@ -1656,7 +1769,7 @@ where truncate_stores(wallet, rescan_from - 1, false)?; - let frontiers = client::get_frontiers(fetch_request_sender, birthday).await?; + let frontiers = client::get_frontiers(fetch_request_sender.clone(), birthday).await?; let retention = Retention::Checkpoint { id: birthday, marking: Marking::None, @@ -1698,11 +1811,19 @@ where } } - let sync_state = wallet - .get_sync_state_mut() - .map_err(SyncError::WalletError)?; - state::reopen_scan_ranges_from(sync_state, rescan_from); - add_scan_targets(sync_state, &rescan_targets); + state::reopen_scan_ranges_from( + consensus_parameters, + fetch_request_sender, + wallet, + rescan_from, + ) + .await?; + add_scan_targets( + wallet + .get_sync_state_mut() + .map_err(SyncError::WalletError)?, + &rescan_targets, + ); wallet.set_save_flag().map_err(SyncError::WalletError)?; Ok(SyncError::PoolHistoryReopened { @@ -1845,7 +1966,7 @@ where match truncate::plan_truncation(wallet_state, truncate_height) { truncate::TruncationPlan::NoOp => Ok(()), truncate::TruncationPlan::ClearAll => { - truncate_stores(wallet, consensus::H0, true)?; + truncate_stores(wallet, consensus::H0, false)?; wallet.clear_shard_trees() } truncate::TruncationPlan::Truncate { height } => { @@ -1912,7 +2033,7 @@ where }) .collect::>(); truncate_wallet_data(wallet, consensus::H0)?; - truncate_scan_ranges( + state::truncate_scan_ranges( consensus::H0, wallet .get_sync_state_mut() @@ -1945,9 +2066,16 @@ async fn update_wallet_data( sapling_located_trees: Vec>, orchard_located_trees: Vec>, ironwood_located_trees: Vec>, + new_transparent_inuse_addresses: HashMap, ) -> Result<(), SyncError> where - W: SyncBlocks + SyncTransactions + SyncNullifiers + SyncOutPoints + SyncShardTrees + Send, + W: SyncWallet + + SyncBlocks + + SyncTransactions + + SyncNullifiers + + SyncOutPoints + + SyncShardTrees + + Send, { let sync_state = wallet .get_sync_state_mut() @@ -2035,6 +2163,12 @@ where ironwood_located_trees, ) .await?; + let wallet_transparent_addresses = wallet + .get_transparent_addresses_mut() + .map_err(SyncError::WalletError)?; + for (address, id) in new_transparent_inuse_addresses { + wallet_transparent_addresses.insert(id, address); + } Ok(()) } @@ -2381,7 +2515,7 @@ where /// If the mempool stream message is `None` (a block was mined) or the request failed, setup a new mempool stream. async fn mempool_monitor( mut client: C, - mempool_transaction_sender: mpsc::Sender, + mempool_transaction_sender: mpsc::Sender, unprocessed_transactions_count: Arc, stream_connected_at: Arc>, shutdown_mempool: Arc, @@ -2409,10 +2543,10 @@ where loop { tokio::select! { mempool_stream_message = mempool_stream.message() => { - match mempool_stream_message.unwrap_or(None) { - Some(raw_transaction) => { + match mempool_stream_message { + Ok(Some(raw_transaction)) => { match mempool_transaction_sender - .send(raw_transaction) + .send(MempoolMessage::Transaction(raw_transaction)) .await { Ok(_) => { unprocessed_transactions_count.fetch_add(1, atomic::Ordering::Release); @@ -2424,7 +2558,22 @@ where } } } - None => { + Ok(None) => { + match mempool_transaction_sender + .send(MempoolMessage::NewBlockMined) + .await { + Ok(_) => { + continue 'main; + } + Err(_) => { + unprocessed_transactions_count.store(0, atomic::Ordering::Release); + shutdown_mempool.store(true, atomic::Ordering::Release); + break 'main; + } + } + } + Err(e) => { + tracing::warn!("Mempool error: {e}"); continue 'main; } } @@ -2456,9 +2605,6 @@ where /// Transaction status will be set to `Failed` if it's still unconfirmed when the chain reaches it's expiry height. /// /// Transactions with an expiry height of 0 never expire (ZIP-203). -/// -/// Must only be called after all blocks up to the wallet's last known chain height have been scanned, otherwise a -/// transaction mined near its expiry height would be marked `Failed` before the block containing it is scanned. fn expire_transactions(wallet: &mut W) -> Result<(), SyncError> where W: SyncWallet + SyncTransactions, @@ -3082,83 +3228,97 @@ mod test { /// The drain policy for scanner shutdown, exercised as a table: /// pure inputs, no runtime, no clocks. mod drain_verdict { - use std::time::Duration; + use std::{ + sync::{ + Arc, + atomic::{AtomicBool, AtomicU32}, + }, + time::Duration, + }; - use crate::sync::DrainVerdict::{self, KeepPolling, Reenter, Shutdown}; - use crate::sync::drain_verdict; - - fn ms(millis: u64) -> Duration { - Duration::from_millis(millis) - } + use crate::sync::{MempoolDrainVerdict, mempool_drain_verdict}; /// One row of the drain-policy table: /// (workers, unprocessed, connected_for, poll_elapsed, verdict, label). - type DrainCase = (usize, u32, Option, u64, DrainVerdict, &'static str); - - /// HYPOTHESIS: a settled stream whose scanner is still busy, and an - /// unsettled stream whose scanner is still busy, are two different - /// states, because only the second one is advanced by waiting. - /// Falsified if the drain policy answers them alike. - #[test] - fn a_settled_and_an_unsettled_busy_scanner_are_not_one_state() { - let settled = drain_verdict(2, 0, Some(ms(1_400)), ms(500)); - let unsettled = drain_verdict(2, 0, Some(ms(50)), ms(500)); - - assert_ne!( - settled, unsettled, - "a settled stream waits on `scanner.update()`, which this loop \ - never calls, while an unsettled one waits on a clock that ticks \ - by itself; one verdict for both spends the ceiling on the state \ - that waiting cannot advance" - ); - assert_eq!(settled, Reenter, "settled and busy: only the loop helps"); - assert_eq!(unsettled, KeepPolling, "unsettled: the clock still runs"); - } + type DrainCase = ( + Arc, + Arc, + Duration, + Option, + MempoolDrainVerdict, + &'static str, + ); - #[test] - fn table() { + #[tokio::test] + async fn table() { let cases: &[DrainCase] = &[ // The reported bug: first-loop shutdown on a fully // synced chain, stream not yet connected. Hold the // session open instead of closing it instantly. - (0, 0, None, 0, KeepPolling, "no stream yet"), - // Connected but inside the settle window: still hold. - (0, 0, Some(50), 100, KeepPolling, "settling"), + ( + Arc::new(AtomicBool::new(false)), + Arc::new(AtomicU32::new(0)), + Duration::from_millis(0), + None, + MempoolDrainVerdict::NotShutdown, + "no stream yet", + ), + // Connected but inside the settle window: waits before checking whether there is work to process. + ( + Arc::new(AtomicBool::new(false)), + Arc::new(AtomicU32::new(0)), + Duration::from_millis(100), + Some(Duration::from_millis(50)), + MempoolDrainVerdict::ShutdownAndDrainComplete, + "settling", + ), // The typical session: stream connected long ago, // scanner drained. Immediate shutdown, no added cost. - (0, 0, Some(1_400), 0, Shutdown, "settled and drained"), - // Exactly the settle boundary counts as settled. - (0, 0, Some(200), 0, Shutdown, "settle boundary"), + ( + Arc::new(AtomicBool::new(false)), + Arc::new(AtomicU32::new(0)), + Duration::from_millis(0), + Some(Duration::from_millis(1_400)), + MempoolDrainVerdict::ShutdownAndDrainComplete, + "settled and drained", + ), // Unprocessed work always re-enters the processing // loop, whatever the stream state. No deadline caps // the processing itself. - (0, 3, Some(1_400), 0, Reenter, "unprocessed work"), - (5, 1, None, 999, Reenter, "work trumps missing stream"), - // Workers still draining: re-enter at once. Polling here - // cannot retire a worker, because only the main loop's - // `scanner.update()` does, so waiting spends the ceiling on - // a value that cannot move. - (2, 0, Some(1_400), 500, Reenter, "workers draining"), - (2, 0, Some(1_400), 1_000, Reenter, "ceiling with workers"), - // Before the stream settles, polling is still right: the - // settle window is wall-clock and does elapse. - (2, 0, Some(50), 500, KeepPolling, "unsettled stream polls"), + ( + Arc::new(AtomicBool::new(false)), + Arc::new(AtomicU32::new(3)), + Duration::from_millis(3_000), + Some(Duration::from_millis(1_400)), + MempoolDrainVerdict::ShutdownNotDrained, + "unprocessed work", + ), // Ceiling with a stream that never connected: the // pre-c90f8d309 semantics. A dead stream must not // hold the session open. - (0, 0, None, 1_000, Shutdown, "ceiling without stream"), + ( + Arc::new(AtomicBool::new(false)), + Arc::new(AtomicU32::new(0)), + Duration::from_millis(1_000), + None, + MempoolDrainVerdict::ShutdownAndDrainComplete, + "ceiling without stream", + ), ]; - for (workers, unprocessed, connected_ms, elapsed_ms, expected, name) in cases { + for (shutdown_mempool, unprocessed, connected_ms, elapsed_ms, expected, name) in cases { assert_eq!( - drain_verdict( - *workers, - *unprocessed, - connected_ms.map(ms), - ms(*elapsed_ms) - ), + mempool_drain_verdict( + shutdown_mempool.clone(), + unprocessed.clone(), + *connected_ms, + *elapsed_ms, + ) + .await, *expected, "{name}" ); + + // TODO: assert shutdown_mempool is true is cases where mempool should be shutdown } } } diff --git a/pepper-sync/src/sync/state.rs b/pepper-sync/src/sync/state.rs index 2ba24f50d3..203eace244 100644 --- a/pepper-sync/src/sync/state.rs +++ b/pepper-sync/src/sync/state.rs @@ -2,7 +2,7 @@ use std::{ cmp, - collections::{BTreeSet, HashMap}, + collections::{BTreeMap, BTreeSet, HashMap}, ops::Range, }; @@ -21,7 +21,7 @@ use crate::{ scan::task::ScanTask, sync::ScanRange, wallet::{ - InitialSyncState, ScanTarget, SyncState, TreeBounds, WalletTransaction, + InitialSyncState, ScanTarget, SyncState, TreeBounds, WalletBlock, WalletTransaction, traits::{SyncBlocks, SyncNullifiers, SyncWallet}, }, }; @@ -60,23 +60,16 @@ fn find_scan_targets( .collect() } -/// Update scan ranges for scanning. -/// Returns the block height that reorg detection will start from. -pub(super) async fn update_scan_ranges( +/// Prioritize scan ranges for scanning. +pub(super) fn prioritize_scan_ranges( consensus_parameters: &impl consensus::Parameters, - fetch_request_sender: mpsc::UnboundedSender, - last_known_chain_height: BlockHeight, chain_height: BlockHeight, wallet: &mut W, -) -> Result> +) -> Result<(), W::Error> where W: SyncWallet + SyncBlocks, { - let sync_state = wallet - .get_sync_state_mut() - .map_err(SyncError::WalletError)?; - reset_scan_ranges(sync_state); - create_scan_range(last_known_chain_height, chain_height, sync_state); + let sync_state = wallet.get_sync_state_mut()?; let scan_targets = sync_state.scan_targets.clone(); set_found_note_scan_ranges( consensus_parameters, @@ -86,33 +79,9 @@ where ); set_chain_tip_scan_range(consensus_parameters, sync_state, chain_height); merge_scan_ranges(sync_state, ScanPriority::ChainTip); + wallet.set_save_flag()?; - let reorg_detection_start_height = sync_state - .highest_scanned_height() - .expect("scan ranges must be non-empty") - + 1; - if reorg_detection_start_height <= chain_height { - set_verify_scan_range( - sync_state, - reorg_detection_start_height, - VerifyEnd::VerifyLowest, - ); - } else { - let chain_height_server_block = - client::get_compact_block(fetch_request_sender, chain_height).await?; - let chain_height_wallet_block = wallet - .get_wallet_block(chain_height) - .map_err(SyncError::WalletError)?; - let sync_state = wallet - .get_sync_state_mut() - .map_err(SyncError::WalletError)?; - if chain_height_wallet_block.block_hash().0.to_vec() != chain_height_server_block.hash { - set_verify_scan_range(sync_state, chain_height, VerifyEnd::VerifyHighest); - } - } - wallet.set_save_flag().map_err(SyncError::WalletError)?; - - Ok(reorg_detection_start_height) + Ok(()) } /// Merges all adjacent ranges of a given `scan_priority`. @@ -153,12 +122,12 @@ pub(super) fn merge_scan_ranges(sync_state: &mut SyncState, scan_priority: ScanP } /// Create scan range between the wallet height and the chain height from the server. -fn create_scan_range( +pub(super) fn create_scan_range( last_known_chain_height: BlockHeight, chain_height: BlockHeight, sync_state: &mut SyncState, ) { - if last_known_chain_height == chain_height { + if last_known_chain_height >= chain_height { return; } @@ -207,7 +176,7 @@ pub(super) fn truncate_scan_ranges(truncate_height: BlockHeight, sync_state: &mu /// scanning. /// A range that was previously refetching nullifiers when sync was last interrupted is set to `ScannedWithoutMapping` /// so the nullifiers can be fetched again. -fn reset_scan_ranges(sync_state: &mut SyncState) { +pub(super) fn reset_scan_ranges(sync_state: &mut SyncState) { let previously_scanning_scan_ranges = sync_state .scan_ranges .iter() @@ -793,6 +762,7 @@ pub(crate) fn create_scan_task( consensus_parameters: &impl consensus::Parameters, wallet: &mut W, nullifier_map_limit_exceeded: bool, + transparent_gap_addresses: HashMap, ) -> Result, W::Error> where W: SyncWallet + SyncBlocks + SyncNullifiers, @@ -811,8 +781,32 @@ where None, BTreeSet::new(), HashMap::new(), + HashMap::new(), ))) } else { + // in continuous sync there is a case where the range directly below the newly mined block (chain tip) is + // currently being scanned. + // sync must be postponed until this range has completed scanning to then reverify in case of re-org. + if selected_range.priority() == ScanPriority::Verify + && wallet.get_sync_state()?.scan_ranges().iter().any(|range| { + range + .block_range() + .contains(&(selected_range.block_range().start - 1)) + && range.priority() == ScanPriority::Scanning + }) + { + // reset from scanning priority to verify until chain tip is scanned. + let range_to_verify = wallet + .get_sync_state_mut()? + .scan_ranges + .iter_mut() + .find(|range| range.block_range().start == selected_range.block_range().start) + .expect("this range must exist in this scope!"); + range_to_verify.priority = ScanPriority::Verify; + + return Ok(None); + } + let start_seam_block = wallet .get_wallet_block(selected_range.block_range().start - 1) .ok(); @@ -822,7 +816,9 @@ where let scan_targets = find_scan_targets(wallet.get_sync_state()?, selected_range.block_range()); - let transparent_addresses: HashMap = wallet + + // convert transparent addreses to hash map with address as the key for efficient scanning. + let transparent_inuse_addresses: HashMap = wallet .get_transparent_addresses()? .iter() .map(|(id, address)| (address.clone(), *id)) @@ -833,7 +829,8 @@ where start_seam_block, end_seam_block, scan_targets, - transparent_addresses, + transparent_inuse_addresses, + transparent_gap_addresses, ))) } } else { @@ -852,9 +849,6 @@ where W: SyncWallet + SyncBlocks, { let sync_state = wallet.get_sync_state().map_err(SyncError::WalletError)?; - let birthday = sync_state - .wallet_birthday() - .expect("scan ranges must be non-empty"); let fully_scanned_height = sync_state .fully_scanned_height() .expect("scan ranges must be non-empty"); @@ -864,6 +858,54 @@ where previously_scanned_orchard_outputs, previously_scanned_ironwood_outputs, ) = calculate_scanned_outputs(wallet).map_err(SyncError::WalletError)?; + + wallet + .get_sync_state_mut() + .map_err(SyncError::WalletError)? + .initial_sync_state = InitialSyncState { + sync_start_height: if chain_height > fully_scanned_height { + fully_scanned_height + 1 + } else { + chain_height + }, + wallet_tree_bounds: TreeBounds { + sapling_initial_tree_size: 0, + sapling_final_tree_size: 0, + orchard_initial_tree_size: 0, + orchard_final_tree_size: 0, + ironwood_initial_tree_size: 0, + ironwood_final_tree_size: 0, + }, + previously_scanned_blocks, + previously_scanned_sapling_outputs, + previously_scanned_orchard_outputs, + previously_scanned_ironwood_outputs, + }; + + update_wallet_tree_bounds( + consensus_parameters, + fetch_request_sender, + wallet, + chain_height, + ) + .await?; + + Ok(()) +} + +pub(super) async fn update_wallet_tree_bounds( + consensus_parameters: &impl consensus::Parameters, + fetch_request_sender: mpsc::UnboundedSender, + wallet: &mut W, + chain_height: BlockHeight, +) -> Result<(), SyncError> +where + W: SyncWallet + SyncBlocks, +{ + let sync_state = wallet.get_sync_state().map_err(SyncError::WalletError)?; + let birthday = sync_state + .wallet_birthday() + .expect("scan ranges must be non-empty"); let ( birthday_sapling_initial_tree_size, birthday_orchard_initial_tree_size, @@ -898,26 +940,18 @@ where wallet .get_sync_state_mut() .map_err(SyncError::WalletError)? - .initial_sync_state = InitialSyncState { - sync_start_height: if chain_height > fully_scanned_height { - fully_scanned_height + 1 - } else { - chain_height - }, - wallet_tree_bounds: TreeBounds { - sapling_initial_tree_size: birthday_sapling_initial_tree_size, - sapling_final_tree_size: chain_tip_sapling_final_tree_size, - orchard_initial_tree_size: birthday_orchard_initial_tree_size, - orchard_final_tree_size: chain_tip_orchard_final_tree_size, - ironwood_initial_tree_size: birthday_ironwood_initial_tree_size, - ironwood_final_tree_size: chain_tip_ironwood_final_tree_size, - }, - previously_scanned_blocks, - previously_scanned_sapling_outputs, - previously_scanned_orchard_outputs, - previously_scanned_ironwood_outputs, + .initial_sync_state + .wallet_tree_bounds = TreeBounds { + sapling_initial_tree_size: birthday_sapling_initial_tree_size, + sapling_final_tree_size: chain_tip_sapling_final_tree_size, + orchard_initial_tree_size: birthday_orchard_initial_tree_size, + orchard_final_tree_size: chain_tip_orchard_final_tree_size, + ironwood_initial_tree_size: birthday_ironwood_initial_tree_size, + ironwood_final_tree_size: chain_tip_ironwood_final_tree_size, }; + wallet.set_save_flag().map_err(SyncError::WalletError)?; + Ok(()) } @@ -1060,7 +1094,42 @@ pub(super) fn pop_newest_shard_range(sync_state: &mut SyncState, shielded_protoc /// /// Ranges being scanned right now are left alone, since their results are /// already in flight against the bounds they were dispatched with. -pub(super) fn reopen_scan_ranges_from(sync_state: &mut SyncState, from_height: BlockHeight) { +pub(super) async fn reopen_scan_ranges_from( + consensus_parameters: &impl consensus::Parameters, + fetch_request_sender: mpsc::UnboundedSender, + wallet: &mut W, + from_height: BlockHeight, +) -> Result<(), SyncError> +where + W: SyncWallet + SyncBlocks, +{ + let sync_state = wallet + .get_sync_state_mut() + .map_err(SyncError::WalletError)?; + reopen_scan_ranges_inner(sync_state, from_height); + + let upper_block_bound_height = from_height - 1; + if wallet.get_wallet_block(upper_block_bound_height).is_err() { + let mut missing_block_bound = BTreeMap::new(); + missing_block_bound.insert( + upper_block_bound_height, + WalletBlock::from_compact_block( + consensus_parameters, + fetch_request_sender.clone(), + &client::get_compact_block(fetch_request_sender.clone(), upper_block_bound_height) + .await?, + ) + .await?, + ); + wallet + .append_wallet_blocks(missing_block_bound) + .map_err(SyncError::WalletError)?; + } + + Ok(()) +} + +fn reopen_scan_ranges_inner(sync_state: &mut SyncState, from_height: BlockHeight) { if let Some((index, range_to_split)) = sync_state .scan_ranges() .iter() @@ -1118,18 +1187,30 @@ pub(super) fn add_shard_ranges( .fold( highest_subtree_completing_height, |previous_subtree_completing_height, subtree_completing_height| { - shard_ranges.push(Range { - start: previous_subtree_completing_height, - end: subtree_completing_height + 1, - }); - - tracing::debug!( - "{:?} subtree root height: {}", - shielded_protocol, + if subtree_completing_height >= previous_subtree_completing_height { + shard_ranges.push(Range { + start: previous_subtree_completing_height, + end: subtree_completing_height + 1, + }); + + tracing::debug!( + "{:?} subtree root height: {}", + shielded_protocol, + subtree_completing_height + ); + subtree_completing_height - ); + } else { + tracing::error!( + "error: first {:?} subtree root from server has completing block height {} which is lower than the + completing block height of latest shard range in wallet with height {}", + shielded_protocol, + previous_subtree_completing_height, + subtree_completing_height + ); - subtree_completing_height + previous_subtree_completing_height + } }, ); } @@ -1383,7 +1464,7 @@ mod tests { ScanRange::from_parts(300.into()..400.into(), ScanPriority::Scanning), ]; - super::reopen_scan_ranges_from(&mut sync_state, 150.into()); + super::reopen_scan_ranges_inner(&mut sync_state, 150.into()); assert_eq!( sync_state.scan_ranges, @@ -1408,7 +1489,7 @@ mod tests { ScanRange::from_parts(200.into()..300.into(), ScanPriority::Scanned), ]; - super::reopen_scan_ranges_from(&mut sync_state, 1.into()); + super::reopen_scan_ranges_inner(&mut sync_state, 1.into()); assert_eq!( sync_state.scan_ranges, diff --git a/pepper-sync/src/sync/transparent.rs b/pepper-sync/src/sync/transparent.rs index 48a9c616aa..26fed2ea84 100644 --- a/pepper-sync/src/sync/transparent.rs +++ b/pepper-sync/src/sync/transparent.rs @@ -23,7 +23,9 @@ use super::MAX_REORG_ALLOWANCE; /// Discovers all addresses in use by the wallet and returns `scan_targets` for any new relevant transactions to scan transparent /// bundles. /// `last_known_chain_height` should be the value before updating to latest chain height. -pub(crate) async fn update_addresses_and_scan_targets( +/// Returns the gap addresses. +// TODO: improve this to not make all the calls at once +pub(crate) async fn address_discovery( consensus_parameters: &impl consensus::Parameters, wallet: Arc>, fetch_request_sender: mpsc::UnboundedSender, @@ -31,9 +33,9 @@ pub(crate) async fn update_addresses_and_scan_targets( last_known_chain_height: BlockHeight, chain_height: BlockHeight, config: TransparentAddressDiscovery, -) -> Result<(), SyncError> { +) -> Result, SyncError> { if !config.scopes.external && !config.scopes.internal && !config.scopes.refund { - return Ok(()); + return Ok(HashMap::new()); } let wallet_addresses = wallet @@ -104,6 +106,7 @@ pub(crate) async fn update_addresses_and_scan_targets( } // discover new addresses and find scan_targets for relevant transactions + let mut gap_addresses = HashMap::new(); for (account_id, ufvk) in ufvks { if let Some(account_pubkey) = ufvk.transparent() { for scope in &scopes { @@ -158,7 +161,13 @@ pub(crate) async fn update_addresses_and_scan_targets( })?; } - addresses.truncate(addresses.len() - config.gap_limit as usize); + let gap_index = addresses.len().saturating_sub(config.gap_limit as usize); + let scope_gap_addresses: HashMap = addresses + .split_off(gap_index) + .into_iter() + .map(|(id, address)| (address, id)) + .collect(); + gap_addresses.extend(scope_gap_addresses); let mut wallet_guard = wallet.write().await; let wallet_addresses_mut = wallet_guard @@ -181,7 +190,7 @@ pub(crate) async fn update_addresses_and_scan_targets( .set_save_flag() .map_err(SyncError::WalletError)?; - Ok(()) + Ok(gap_addresses) } // TODO: process memo encoded address indexes. diff --git a/pepper-sync/src/wallet/traits.rs b/pepper-sync/src/wallet/traits.rs index 9ab0d866ba..4aa650b98f 100644 --- a/pepper-sync/src/wallet/traits.rs +++ b/pepper-sync/src/wallet/traits.rs @@ -67,11 +67,15 @@ pub trait SyncWallet { ) -> Result<(), Self::Error>; /// Returns a reference to all transparent addresses known to this wallet. + /// + /// These addresses must be in-use, they do not include gap addresses. fn get_transparent_addresses( &self, ) -> Result<&BTreeMap, Self::Error>; /// Returns a mutable reference to all transparent addresses known to this wallet. + /// + /// These addresses must be in-use, they do not include gap addresses. fn get_transparent_addresses_mut( &mut self, ) -> Result<&mut BTreeMap, Self::Error>; diff --git a/zingo-cli/src/commands.rs b/zingo-cli/src/commands.rs index c5a49383e9..ceffa0f529 100644 --- a/zingo-cli/src/commands.rs +++ b/zingo-cli/src/commands.rs @@ -478,11 +478,11 @@ async fn new_address( async fn taddress( lightclient: &mut LightClient, - enforce_gap: bool, + enforce_no_gap: bool, ) -> Result { let chain_type = lightclient.chain_type(); let mut wallet = lightclient.wallet().write().await; - match wallet.generate_transparent_address(zip32::AccountId::ZERO, enforce_gap) { + match wallet.generate_transparent_address(zip32::AccountId::ZERO, enforce_no_gap) { Ok((id, transparent_address)) => Ok(json::object! { "account" => u32::from(id.account_id()), "address_index" => id.address_index().index(), diff --git a/zingo-cli/src/lib.rs b/zingo-cli/src/lib.rs index a380399523..86eae9cb37 100644 --- a/zingo-cli/src/lib.rs +++ b/zingo-cli/src/lib.rs @@ -61,6 +61,7 @@ pub fn build_clap_app() -> clap::Command { .long("nosync") .short('n') .action(clap::ArgAction::SetTrue)) + // TODO: make sure waitsync works with cont sync. may need to change wallet settings to turn shutdown_on_completion off .arg(Arg::new("waitsync") .help("Block execution of the specified command until the background sync completes. Has no effect if --nosync is set.") .long("waitsync") @@ -1267,8 +1268,9 @@ async fn build_zingo_config(filled_template: &CliConfigTemplate) -> std::io::Res let no_of_accounts = NonZeroU32::try_from(1).expect("hard-coded integer"); let wallet_settings = WalletSettings { sync_config: SyncConfig { - transparent_address_discovery: TransparentAddressDiscovery::minimal(), + transparent_address_discovery: TransparentAddressDiscovery::default(), performance_level: PerformanceLevel::High, + shutdown_on_completion: false, }, min_confirmations: NonZeroU32::try_from(3).unwrap(), }; diff --git a/zingo-cli/src/tests.rs b/zingo-cli/src/tests.rs index f11c7fc3c4..3e64f24b57 100644 --- a/zingo-cli/src/tests.rs +++ b/zingo-cli/src/tests.rs @@ -1242,8 +1242,9 @@ mod config_template { birthday: 1, wallet_settings: zingolib::wallet::WalletSettings { sync_config: SyncConfig { - transparent_address_discovery: TransparentAddressDiscovery::minimal(), + transparent_address_discovery: TransparentAddressDiscovery::default(), performance_level: PerformanceLevel::High, + shutdown_on_completion: false, }, min_confirmations: NonZeroU32::try_from(3).unwrap(), }, diff --git a/zingolib/examples/sync_timing.rs b/zingolib/examples/sync_timing.rs index 840e12a103..e3740ff396 100644 --- a/zingolib/examples/sync_timing.rs +++ b/zingolib/examples/sync_timing.rs @@ -67,6 +67,7 @@ fn main() { Some("low") => PerformanceLevel::Low, _ => PerformanceLevel::High, }, + shutdown_on_completion: true, }, min_confirmations: NonZeroU32::new(3).unwrap(), }, diff --git a/zingolib/examples/wallet_from_file.rs b/zingolib/examples/wallet_from_file.rs index 5325e4cd2a..09594ef56f 100644 --- a/zingolib/examples/wallet_from_file.rs +++ b/zingolib/examples/wallet_from_file.rs @@ -85,6 +85,7 @@ async fn main() -> Result<(), Box> { sync_config: SyncConfig { transparent_address_discovery: TransparentAddressDiscovery::default(), performance_level: PerformanceLevel::High, + shutdown_on_completion: true, }, min_confirmations: NonZeroU32::new(3).unwrap(), }, diff --git a/zingolib/src/config.rs b/zingolib/src/config.rs index 89e282edb2..75b653a069 100644 --- a/zingolib/src/config.rs +++ b/zingolib/src/config.rs @@ -469,8 +469,9 @@ impl Default for ClientConfigBuilder { chain_height: 1, wallet_settings: WalletSettings { sync_config: SyncConfig { - transparent_address_discovery: TransparentAddressDiscovery::minimal(), + transparent_address_discovery: TransparentAddressDiscovery::default(), performance_level: pepper_sync::config::PerformanceLevel::High, + shutdown_on_completion: false, }, min_confirmations: NonZeroU32::try_from(3) .expect("hard coded non-zero integer"), diff --git a/zingolib/src/lightclient/darkside.rs b/zingolib/src/lightclient/darkside.rs index b716433282..e4d4403641 100644 --- a/zingolib/src/lightclient/darkside.rs +++ b/zingolib/src/lightclient/darkside.rs @@ -57,7 +57,7 @@ async fn fund_at_height_three(net: &MockNet, address: &str) -> Vec { async fn reorg_removes_receipt() { let mut net = MockNet::launch().await; let mut wallet = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; let address = get_base_address(&wallet, PoolType::Shielded(ShieldedPool::Orchard)).await; fund_at_height_three(&net, &address).await; @@ -83,7 +83,7 @@ async fn reorg_removes_receipt() { async fn reorg_moves_receipt_to_new_height() { let mut net = MockNet::launch().await; let mut wallet = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; let address = get_base_address(&wallet, PoolType::Shielded(ShieldedPool::Orchard)).await; fund_at_height_three(&net, &address).await; @@ -131,12 +131,14 @@ async fn reorg_moves_receipt_to_new_height() { async fn reorg_expires_outgoing_transaction() { let mut net = MockNet::launch().await; let mut sender = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; // Not ABANDON_ART: that is the synthetic funding faucet's own seed, // so its change output would land in this wallet and pollute the // recipient's balance assertions. - let mut recipient = net.client(zingo_test_vectors::seeds::DARKSIDE_SEED).await; + let mut recipient = net + .client(zingo_test_vectors::seeds::DARKSIDE_SEED, None) + .await; let sender_address = get_base_address(&sender, PoolType::Shielded(ShieldedPool::Orchard)).await; let recipient_address = get_base_address(&recipient, PoolType::Shielded(ShieldedPool::Orchard)).await; diff --git a/zingolib/src/lightclient/mock_chain_tests.rs b/zingolib/src/lightclient/mock_chain_tests.rs index c9eba0ad2a..cc86d108f6 100644 --- a/zingolib/src/lightclient/mock_chain_tests.rs +++ b/zingolib/src/lightclient/mock_chain_tests.rs @@ -52,7 +52,7 @@ async fn fund(net: &MockNet, receivers: Vec<(&str, u64, Option<&str>)>, extra_bl async fn funded_send_confirms_on_the_mock_chain() { let mut net = MockNet::launch().await; let mut recipient = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; let recipient_ua = get_base_address(&recipient, PoolType::Shielded(ShieldedPool::Orchard)).await; @@ -89,7 +89,7 @@ async fn funded_send_confirms_on_the_mock_chain() { async fn list_value_transfers_check_fees() { let mut net = MockNet::launch().await; let mut recipient = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; let recipient_ua = get_base_address(&recipient, PoolType::Shielded(ShieldedPool::Orchard)).await; @@ -128,7 +128,7 @@ async fn list_value_transfers_check_fees() { async fn self_send_to_t_displays_as_one_transaction() { let mut net = MockNet::launch().await; let mut recipient = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; let recipient_ua = get_base_address(&recipient, PoolType::Shielded(ShieldedPool::Orchard)).await; @@ -230,7 +230,7 @@ async fn send_to_transparent_and_sapling_maintain_balance() { let recipient_initial_funds = 100_000_000; let mut net = MockNet::launch().await; let mut recipient = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; let recipient_ua = get_base_address(&recipient, PoolType::IRONWOOD).await; // The external destinations: the abandon-art wallet's sapling UA and @@ -604,7 +604,7 @@ async fn from_t_z_o_tz_to_zo_tzo_to_orchard() { let mut net = MockNet::launch().await; let mut client = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; let pmc_unified = get_base_address(&client, PoolType::Shielded(ShieldedPool::Orchard)).await; let pmc_taddr = get_base_address(&client, PoolType::Transparent).await; @@ -797,7 +797,7 @@ async fn send_survives_lost_response_and_duplicate_rejection() { let mut net = MockNet::launch().await; let mut recipient = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; recipient.set_transmit_retry_interval(std::time::Duration::ZERO); let recipient_ua = @@ -860,7 +860,7 @@ async fn send_survives_lost_response_and_queued_duplicate_rejection() { let mut net = MockNet::launch().await; let mut recipient = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; recipient.set_transmit_retry_interval(std::time::Duration::ZERO); let recipient_ua = @@ -948,7 +948,7 @@ async fn failed_split_round_transmit_strands_calculated_transactions() { let mut net = MockNet::launch().await; net.chain.write().await.mine_empty_blocks(TIP); let mut client = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; client.set_transmit_retry_interval(std::time::Duration::ZERO); client @@ -1026,11 +1026,22 @@ async fn failed_split_round_transmit_strands_calculated_transactions() { /// The offline twins whose assertions read the editorial surface. #[cfg(feature = "perspective")] mod perspective { + use std::num::NonZeroU32; + use std::time::{Duration, Instant}; + + use pepper_sync::config::{ + PerformanceLevel, SyncConfig, TransparentAddressDiscovery, + TransparentAddressDiscoveryScopes, + }; + use zcash_keys::encoding::AddressCodec; + use zip32::AccountId; + use crate::lightclient::LightClient; use crate::perspective::value_transfer::{ SelfSendValueTransfer, SentValueTransfer, ValueTransfer, ValueTransferKind, ValueTransfers, }; use crate::testutils::synthetic_wallet::inject_confirmed_orchard_notes; + use crate::wallet::WalletSettings; use super::*; @@ -1045,7 +1056,7 @@ mod perspective { let mut net = MockNet::launch().await; net.chain.write().await.mine_empty_blocks(TIP); let mut client = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; client .sync_and_await() @@ -1077,7 +1088,7 @@ mod perspective { async fn zero_value_receipts() { let mut net = MockNet::launch().await; let mut recipient = net - .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED) + .client(zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, None) .await; let recipient_ua = get_base_address(&recipient, PoolType::IRONWOOD).await; @@ -1194,6 +1205,121 @@ mod perspective { migration.memos, ); } + + #[tokio::test] + async fn gap_address_compact_block_scanning() { + let mut net = MockNet::launch().await; + let mut recipient = net + .client( + zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, + Some(WalletSettings { + sync_config: SyncConfig { + transparent_address_discovery: TransparentAddressDiscovery { + gap_limit: 3, + scopes: TransparentAddressDiscoveryScopes::default(), + }, + performance_level: PerformanceLevel::High, + shutdown_on_completion: false, + }, + min_confirmations: NonZeroU32::try_from(1) + .expect("hard-coded non-zero integer"), + }), + ) + .await; + + // start the recipient syncing continuously and wait for the chain to be fully scanned + net.chain.write().await.mine_empty_blocks(10); + recipient.sync().await.unwrap(); + while recipient + .latest_sync_status() + .is_none_or(|status| !status.is_complete()) + { + tokio::time::sleep(Duration::from_millis(100)).await; + } + + // generate gap_limit+2 new taddrs without adding them directly to the wallet and fund them to trigger gap + // address scanning in the chain tip compact blocks + let recipient_clone = net + .client( + zingo_test_vectors::seeds::HOSPITAL_MUSEUM_SEED, + Some(WalletSettings { + sync_config: SyncConfig { + transparent_address_discovery: TransparentAddressDiscovery { + gap_limit: 3, + scopes: TransparentAddressDiscoveryScopes::default(), + }, + performance_level: PerformanceLevel::High, + shutdown_on_completion: false, + }, + min_confirmations: NonZeroU32::try_from(1) + .expect("hard-coded non-zero integer"), + }), + ) + .await; + let mut gap_taddrs = Vec::new(); + let mut funding_txs = Vec::new(); + for _ in 0..5 { + let gap_taddr = recipient_clone + .wallet() + .write() + .await + .generate_transparent_address(AccountId::ZERO, false) + .unwrap(); + gap_taddrs.push(gap_taddr); + + let funding_tx = faucet_funding_transaction(vec![( + &gap_taddr.1.encode(&recipient.chain_type()), + 100_000, + None, + )]) + .await; + funding_txs.push(funding_tx); + } + net.chain.write().await.mine_block(funding_txs); + + // let timeout = Duration::from_secs(150); + // let assert_start = Instant::now(); + // while !(recipient + // .latest_sync_status() + // .unwrap() + // .scan_ranges + // .last() + // .unwrap() + // .block_range() + // .end + // - 1 + // == 11.into() + // && recipient.latest_sync_status().unwrap().is_complete()) + // { + // if assert_start.elapsed() >= timeout { + // panic!("test time exceeded expected time to scan newly mined blocks"); + // } + // tokio::time::sleep(Duration::from_millis(100)).await; + // } + + recipient.stop_sync().unwrap(); + recipient.await_sync().await.unwrap(); + recipient.sync_and_await().await.unwrap(); + + // check all funds have been received and addresses have been added to the wallet, including the addresses + // beyond the gap limit to prove the new blocks are being rescanend until the gap limit is satisfied + check_client_balances!(recipient, i: 0 o: 0 s: 0 t: 500_000); + for (id, addr) in gap_taddrs { + assert!( + recipient + .wallet() + .read() + .await + .transparent_addresses() + .iter() + .any(|(wallet_addr_id, wallet_addr)| *wallet_addr_id == id + && *wallet_addr == addr.encode(&recipient.chain_type())) + ); + } + + recipient.stop_sync().unwrap(); + recipient.await_sync().await.unwrap(); + } } /// A mock-chain send travels the mixnet route and says so: the receipt diff --git a/zingolib/src/lightclient/sync.rs b/zingolib/src/lightclient/sync.rs index 285bd4449a..4d78ddb804 100644 --- a/zingolib/src/lightclient/sync.rs +++ b/zingolib/src/lightclient/sync.rs @@ -143,6 +143,7 @@ impl LightClient { } /// Returns the sync engine's most recently published status without touching the wallet lock. + // TODO: return result with status error pub fn latest_sync_status(&self) -> Option { self.sync_progress.borrow().clone() } diff --git a/zingolib/src/sync.rs b/zingolib/src/sync.rs index dd9a2e8762..5c42c7a52f 100644 --- a/zingolib/src/sync.rs +++ b/zingolib/src/sync.rs @@ -22,7 +22,6 @@ pub use pepper_sync::error::SyncModeError; pub use pepper_sync::keys::transparent; -pub use pepper_sync::sync_status; pub use pepper_sync::wallet::{IronwoodNote, KeyIdInterface, OrchardNote, SaplingNote, SyncMode}; pub use zingo_status::confirmation_status::ConfirmationStatus; diff --git a/zingolib/src/testutils.rs b/zingolib/src/testutils.rs index c6c082e0d9..faf38ca73c 100644 --- a/zingolib/src/testutils.rs +++ b/zingolib/src/testutils.rs @@ -42,6 +42,7 @@ pub fn default_test_wallet_settings() -> WalletSettings { sync_config: SyncConfig { transparent_address_discovery: TransparentAddressDiscovery::minimal(), performance_level: PerformanceLevel::High, + shutdown_on_completion: true, }, min_confirmations: NonZeroU32::try_from(1).expect("hard-coded non-zero integer"), } diff --git a/zingolib/src/testutils/mock_indexer.rs b/zingolib/src/testutils/mock_indexer.rs index ed83fcbb16..2b8128e7ae 100644 --- a/zingolib/src/testutils/mock_indexer.rs +++ b/zingolib/src/testutils/mock_indexer.rs @@ -64,6 +64,7 @@ use crate::lightclient::LightClient; use crate::testutils::default_test_wallet_settings; use crate::testutils::lightclient::from_inputs; use crate::testutils::synthetic_wallet::SyntheticWalletBuilder; +use crate::wallet::WalletSettings; type SaplingTree = CommitmentTree; @@ -877,7 +878,11 @@ impl MockNet { /// Builds a `LightClient` for `mnemonic` (birthday 1) dialed at the /// mock, with its wallet directory in a tempdir this net keeps /// alive. - pub async fn client(&mut self, mnemonic: &str) -> LightClient { + pub async fn client( + &mut self, + mnemonic: &str, + wallet_settings_opt: Option, + ) -> LightClient { let wallet_dir = tempfile::tempdir().expect("a tempdir is creatable"); let config = ClientConfig::builder() .set_chain_type(ChainType::Regtest(ActivationHeights::default())) @@ -887,7 +892,7 @@ impl MockNet { mnemonic_phrase: mnemonic.to_string(), no_of_accounts: 1.try_into().expect("hard-coded non-zero"), birthday: 1, - wallet_settings: default_test_wallet_settings(), + wallet_settings: wallet_settings_opt.unwrap_or_else(default_test_wallet_settings), }) .build() .unwrap(); diff --git a/zingolib/src/wallet/disk.rs b/zingolib/src/wallet/disk.rs index 32a87dcac4..ef002e5af5 100644 --- a/zingolib/src/wallet/disk.rs +++ b/zingolib/src/wallet/disk.rs @@ -497,8 +497,9 @@ impl LightWallet { save_required: false, wallet_settings: WalletSettings { sync_config: SyncConfig { - transparent_address_discovery: TransparentAddressDiscovery::minimal(), + transparent_address_discovery: TransparentAddressDiscovery::default(), performance_level: PerformanceLevel::High, + shutdown_on_completion: false, }, min_confirmations: NonZeroU32::try_from(3).unwrap(), }, @@ -723,8 +724,9 @@ impl LightWallet { } else { WalletSettings { sync_config: SyncConfig { - transparent_address_discovery: TransparentAddressDiscovery::minimal(), + transparent_address_discovery: TransparentAddressDiscovery::default(), performance_level: PerformanceLevel::High, + shutdown_on_completion: false, }, min_confirmations: NonZeroU32::try_from(3).unwrap(), } diff --git a/zingolib/tests/sync_bench.rs b/zingolib/tests/sync_bench.rs index e1c6213a3a..1f009be85b 100644 --- a/zingolib/tests/sync_bench.rs +++ b/zingolib/tests/sync_bench.rs @@ -61,6 +61,7 @@ async fn sync_20k_mainnet_blocks_within_budget() { sync_config: SyncConfig { transparent_address_discovery: TransparentAddressDiscovery::default(), performance_level: PerformanceLevel::High, + shutdown_on_completion: true, }, min_confirmations: NonZeroU32::new(3).unwrap(), }, diff --git a/zingolib/tests/sync_perf_guard.rs b/zingolib/tests/sync_perf_guard.rs index ee31195293..f745de228a 100644 --- a/zingolib/tests/sync_perf_guard.rs +++ b/zingolib/tests/sync_perf_guard.rs @@ -65,6 +65,7 @@ async fn syncing_the_top_window_holds_this_machines_baseline() { sync_config: SyncConfig { transparent_address_discovery: TransparentAddressDiscovery::default(), performance_level: PerformanceLevel::High, + shutdown_on_completion: true, }, min_confirmations: NonZeroU32::new(3).unwrap(), },