diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c25f006d2..2601f653e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -61,8 +61,8 @@ jobs: name: Repo guards # GitHub-hosted runner so it runs on pull requests. Runs the workbench # repo guards: the rustc-pin check (Dockerfile.deterministic's pallet-rust - # tag == rust-toolchain.toml channel) and the no-OpenSSL-apt check, plus the - # crate's own fmt/clippy/tests. + # tag == rust-toolchain.toml channel), the no-OpenSSL-apt check, and the + # duplicate-Rust-logic check, plus the crate's own fmt/clippy/tests. runs-on: ubuntu-latest steps: - name: Checkout repository @@ -80,6 +80,9 @@ jobs: - name: Check no OpenSSL apt packages run: cargo run -q --manifest-path tools/workbench/Cargo.toml --bin check-no-openssl-apt + - name: Check for duplicate Rust logic + run: cargo run -q --manifest-path tools/workbench/Cargo.toml --bin check-code-duplication + cargo-deny: name: cargo-deny # Enforces deny.toml — including the openssl / openssl-sys / boring* crate diff --git a/live-tests/clientless/Cargo.toml b/live-tests/clientless/Cargo.toml index c8945a81a..4eee54633 100644 --- a/live-tests/clientless/Cargo.toml +++ b/live-tests/clientless/Cargo.toml @@ -36,6 +36,8 @@ zaino-fetch = { workspace = true } zaino-testutils = { workspace = true } # The lib's z_validate helper calls `ZcashIndexer::z_validate_address`. zaino-state = { workspace = true, features = ["test_dependencies"] } +# The lib's get_block_header oracle helper matches on `zebra_rpc::methods::GetBlock`. +zebra-rpc = { workspace = true } [dev-dependencies] anyhow = { workspace = true } diff --git a/live-tests/clientless/src/lib.rs b/live-tests/clientless/src/lib.rs index 12d93f9be..c40348ed1 100644 --- a/live-tests/clientless/src/lib.rs +++ b/live-tests/clientless/src/lib.rs @@ -2,6 +2,41 @@ //! //! This crate also exposes test-vectors. +/// Assert that `oracle` and `subject` return the same `getblockheader` response for +/// the block at `height`: look the block up on the oracle (verbosity 1) to learn its +/// hash, then compare the two servers' non-verbose header responses for that hash. +/// Shared body of the per-backend `get_block_header` oracle tests. +#[allow(deprecated)] +pub async fn assert_get_block_header_matches( + oracle: &Oracle, + subject: &Subject, + height: u32, +) where + Oracle: zaino_state::ZcashIndexer, + Subject: zaino_state::ZcashIndexer, +{ + let block = oracle + .z_get_block(height.to_string(), Some(1)) + .await + .unwrap(); + + let block_hash = match block { + zebra_rpc::methods::GetBlock::Object(block) => block.hash(), + zebra_rpc::methods::GetBlock::Raw(_) => panic!("Expected block object"), + }; + + let oracle_header = oracle + .get_block_header(block_hash.to_string(), false) + .await + .unwrap(); + + let subject_header = subject + .get_block_header(block_hash.to_string(), false) + .await + .unwrap(); + assert_eq!(oracle_header, subject_header); +} + pub mod rpc { pub mod json_rpc { pub const VALID_P2PKH_ADDRESS: &str = "tmVqEASZxBNKFTbmASZikGa5fPLkd68iJyx"; diff --git a/live-tests/clientless/tests/compact_block_consistency.rs b/live-tests/clientless/tests/compact_block_consistency.rs index 18bf0116a..575e0a503 100644 --- a/live-tests/clientless/tests/compact_block_consistency.rs +++ b/live-tests/clientless/tests/compact_block_consistency.rs @@ -157,20 +157,42 @@ async fn unfiltered_compact_blocks_match_chain_metadata_zebrad() { test_manager.close().await; } -/// Class-1 (consensus) predicate: in the NU5-through-NU6.2 era, a shielded -/// (orchard-receiver) miner's coinbase carries the reward as Orchard actions. +/// The shielded pool a miner's coinbase routes the reward to: Orchard in the +/// NU5-through-NU6.2 era (transaction v5), Ironwood from NU6.3 (v6). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum CoinbaseRewardPool { + Orchard, + Ironwood, +} + +/// Shared body of the class-1 (consensus) coinbase predicates: the first +/// transaction is a coinbase of the era's version, carries the reward as at least +/// one action in `pool`, and has an empty sapling component and an empty +/// other-shielded-pool component. /// /// The action count uses `>= 1` rather than the padded exact count so the predicate /// does not couple to the Orchard bundle-padding rule. -fn is_valid_orchard_coinbase(block: &zebra_chain::block::Block) -> bool { +fn is_valid_shielded_coinbase(block: &zebra_chain::block::Block, pool: CoinbaseRewardPool) -> bool { let Some(coinbase) = block.transactions.first() else { return false; }; + let orchard_actions = coinbase.orchard_actions().count(); + let ironwood_actions = coinbase.ironwood_actions().count(); + let (version, rewarded_pool_actions, other_pool_actions) = match pool { + CoinbaseRewardPool::Orchard => (5, orchard_actions, ironwood_actions), + CoinbaseRewardPool::Ironwood => (6, ironwood_actions, orchard_actions), + }; coinbase.is_coinbase() - && coinbase.version() == 5 + && coinbase.version() == version && coinbase.sapling_outputs().count() == 0 - && coinbase.orchard_actions().count() >= 1 - && coinbase.ironwood_actions().count() == 0 + && rewarded_pool_actions >= 1 + && other_pool_actions == 0 +} + +/// Class-1 (consensus) predicate: in the NU5-through-NU6.2 era, a shielded +/// (orchard-receiver) miner's coinbase carries the reward as Orchard actions. +fn is_valid_orchard_coinbase(block: &zebra_chain::block::Block) -> bool { + is_valid_shielded_coinbase(block, CoinbaseRewardPool::Orchard) } /// Class-1 (consensus) predicate: from NU6.3 the same miner's coinbase must have an @@ -182,14 +204,7 @@ fn is_valid_orchard_coinbase(block: &zebra_chain::block::Block) -> bool { /// builder off-by-one vs missing routing vs a zaino pool-swap). This predicate over /// raw validator blocks is that issue's disambiguator. fn is_valid_ironwood_coinbase(block: &zebra_chain::block::Block) -> bool { - let Some(coinbase) = block.transactions.first() else { - return false; - }; - coinbase.is_coinbase() - && coinbase.version() == 6 - && coinbase.sapling_outputs().count() == 0 - && coinbase.orchard_actions().count() == 0 - && coinbase.ironwood_actions().count() >= 1 + is_valid_shielded_coinbase(block, CoinbaseRewardPool::Ironwood) } /// One-line coinbase summary for assertion messages, so a predicate failure names the diff --git a/live-tests/clientless/tests/json_server.rs b/live-tests/clientless/tests/json_server.rs index b8c17443c..bc62e4322 100644 --- a/live-tests/clientless/tests/json_server.rs +++ b/live-tests/clientless/tests/json_server.rs @@ -286,7 +286,6 @@ mod zcashd { pub(crate) mod zcash_indexer { use zaino_state::LightWalletIndexer; - use zebra_rpc::methods::GetBlock; use super::*; @@ -453,29 +452,12 @@ mod zcashd { &services.zaino_subscriber, &services.zcashd_subscriber, async |i| { - let block = services - .zcashd_subscriber - .z_get_block(i.to_string(), Some(1)) - .await - .unwrap(); - - let block_hash = match block { - GetBlock::Object(block) => block.hash(), - GetBlock::Raw(_) => panic!("Expected block object"), - }; - - let zcashd_get_block_header = services - .zcashd_subscriber - .get_block_header(block_hash.to_string(), false) - .await - .unwrap(); - - let zainod_block_header_response = services - .zaino_subscriber - .get_block_header(block_hash.to_string(), false) - .await - .unwrap(); - assert_eq!(zcashd_get_block_header, zainod_block_header_response); + clientless::assert_get_block_header_matches( + &services.zcashd_subscriber, + &services.zaino_subscriber, + i, + ) + .await; }, ) .await; diff --git a/live-tests/clientless/tests/state_service.rs b/live-tests/clientless/tests/state_service.rs index 59b607d40..c48e07929 100644 --- a/live-tests/clientless/tests/state_service.rs +++ b/live-tests/clientless/tests/state_service.rs @@ -650,7 +650,6 @@ mod zebra { pub(crate) mod lightwallet_indexer { use futures::StreamExt as _; use zaino_proto::proto::service::{BlockId, BlockRange, GetSubtreeRootsArg}; - use zebra_rpc::methods::GetBlock; use super::*; @@ -712,32 +711,12 @@ mod zebra { &services.fetch_subscriber, &services.state_subscriber, async |i| { - let block = services - .fetch_subscriber - .z_get_block(i.to_string(), Some(1)) - .await - .unwrap(); - - let block_hash = match block { - GetBlock::Object(block) => block.hash(), - GetBlock::Raw(_) => panic!("Expected block object"), - }; - - let fetch_service_get_block_header = services - .fetch_subscriber - .get_block_header(block_hash.to_string(), false) - .await - .unwrap(); - - let state_service_block_header_response = services - .state_subscriber - .get_block_header(block_hash.to_string(), false) - .await - .unwrap(); - assert_eq!( - fetch_service_get_block_header, - state_service_block_header_response - ); + clientless::assert_get_block_header_matches( + &services.fetch_subscriber, + &services.state_subscriber, + i, + ) + .await; }, ) .await; diff --git a/live-tests/e2e/src/devtool.rs b/live-tests/e2e/src/devtool.rs index 13ed1679d..ce3c3193e 100644 --- a/live-tests/e2e/src/devtool.rs +++ b/live-tests/e2e/src/devtool.rs @@ -250,3 +250,239 @@ impl Pool { } } } + +/// Launch a shielded-mining validator of `validator` kind (at `activation_heights`, +/// or the kind's defaults on `None`) with Zaino serving gRPC, and build the devtool +/// faucet/recipient wallets against it, without mining or syncing. The shared body +/// of the per-validator `launch_*_and_build_clients` preambles in the devtool test +/// binaries. +pub async fn launch_and_build_devtool_clients( + validator: &zaino_testutils::ValidatorKind, + activation_heights: Option, +) -> (zaino_testutils::TestManager, DevtoolClients) +where + V: zaino_testutils::ValidatorExt, + Conn: zaino_testutils::ValidatorConnectionMarker, +{ + let test_manager = zaino_testutils::TestManager::::launch_mining_to( + zaino_testutils::SHIELDED_FUNDING_POOL, + validator, + None, // network -> Regtest + activation_heights, + None, // no chain cache: build fresh + true, // enable zaino + false, // no json-rpc server + false, // no clients (the devtool wallet is built separately) + ) + .await + .expect("launch TestManager"); + + let clients = build_clients( + test_manager + .zaino_grpc_listen_address + .expect("zaino enabled") + .port(), + &test_manager.local_net, + ) + .await; + + (test_manager, clients) +} + +/// The faucet sends 250_000 zatoshis to the recipient's `pool` address, the send is +/// mined in, and the recipient's synced wallet shows the receipt in that pool. +/// Shared body of the per-validator `send_to_pool` tests; the caller launches and +/// funds the faucet first. +pub async fn assert_send_to_pool( + mut test_manager: zaino_testutils::TestManager, + mut clients: DevtoolClients, + pool: Pool, +) where + V: zaino_testutils::ValidatorExt, + Conn: zaino_testutils::ValidatorConnectionMarker, +{ + let recipient = clients.get_recipient_address(pool.address_kind()).await; + let txid = clients.send_from_faucet(&recipient, 250_000).await; + dbg!(txid); + + test_manager + .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) + .await; + clients.sync_recipient().await; + + assert_eq!( + pool.spendable_balance(&clients.recipient_balance().await), + 250_000 + ); + + test_manager.close().await; +} + +/// The recipient receives a transparent send, confirms it, shields it, and confirms +/// the shielded balance net of the ZIP-317 fee (250_000 − 15_000 = 235_000) in +/// `shielded_pool` — [`Pool::Ironwood`] under NU6.3-era heights (devtool routes +/// `shield` there), [`Pool::Orchard`] under pre-NU6.3 heights. Shared body of the +/// per-validator `shield` tests; the caller launches and funds the faucet first. +pub async fn assert_shield_for_validator( + mut test_manager: zaino_testutils::TestManager, + mut clients: DevtoolClients, + shielded_pool: Pool, +) where + V: zaino_testutils::ValidatorExt, + Conn: zaino_testutils::ValidatorConnectionMarker, +{ + let recipient_taddr = clients.get_recipient_address("transparent").await; + clients.send_from_faucet(&recipient_taddr, 250_000).await; + test_manager + .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) + .await; + clients.sync_recipient().await; + + assert_eq!( + Pool::Transparent.spendable_balance(&clients.recipient_balance().await), + 250_000 + ); + + clients.shield_recipient().await; + test_manager + .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) + .await; + clients.sync_recipient().await; + + assert_eq!( + shielded_pool.spendable_balance(&clients.recipient_balance().await), + 235_000 + ); + + test_manager.close().await; +} + +/// A transparent send returns the same address txids from the non-finalized chain +/// and again after a seam-deep advance lands it in the finalized DB. Shared body of +/// the per-validator gated `send_to_transparent_finalization` tests; the caller +/// launches and funds the faucet first. The advance mines shielded coinbase, so the +/// callers stay `#[ignore]`d until per-call cheap filler mining lands. +pub async fn assert_send_to_transparent_finalization( + mut test_manager: zaino_testutils::TestManager, + mut clients: DevtoolClients, +) where + V: zaino_testutils::ValidatorExt, + Conn: zaino_testutils::ValidatorConnectionMarker, +{ + let recipient_taddr = clients.get_recipient_address("transparent").await; + clients.send_from_faucet(&recipient_taddr, 250_000).await; + test_manager + .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) + .await; + + let fetch_service = test_manager.full_node_jsonrpc_connector().await; + let height = fetch_service.get_blockchain_info().await.unwrap().blocks.0; + let unfinalised_transactions = fetch_service + .get_address_txids(vec![recipient_taddr.clone()], height, height) + .await + .unwrap(); + + // The load-bearing advance: these blocks push the send below the seam + // (`FAST_TEST_MAX_NONFINALISED_DEPTH`) into the finalized DB. + test_manager + .generate_blocks_bulk_and_wait_for_tips( + // Advance past the seam so the send crosses the finalised floor + // (`tip - seam`); a small margin above it keeps the boundary unambiguous. + zaino_common::consensus::FAST_TEST_MAX_NONFINALISED_DEPTH + 5, + test_manager.subscriber(), + test_manager.subscriber(), + ) + .await; + + let finalised_transactions = fetch_service + .get_address_txids(vec![recipient_taddr], height, height) + .await + .unwrap(); + + clients.sync_recipient().await; + assert_eq!( + Pool::Transparent.spendable_balance(&clients.recipient_balance().await), + 250_000 + ); + assert_eq!(unfinalised_transactions, finalised_transactions); + + test_manager.close().await; +} + +/// Broadcast two unmined shielded sends, observe them in the validator mempool, +/// then mine them in. Shared body of the per-validator gated +/// `monitor_unverified_mempool` tests; the caller launches and funds the faucet +/// (two notes — one per unmined send) first. +/// +/// The unconfirmed/confirmed balance assertions of the zingolib original stay +/// commented out: devtool's `WalletBalance` surfaces only `*_spendable` (its sync +/// is block-based and never scans the mempool). Restore them and un-ignore the +/// callers when devtool surfaces unconfirmed balances. +pub async fn assert_monitor_unverified_mempool( + mut test_manager: zaino_testutils::TestManager, + mut clients: DevtoolClients, +) where + V: zaino_testutils::ValidatorExt, + Conn: zaino_testutils::ValidatorConnectionMarker, +{ + let recipient_ua = clients.get_recipient_address("unified").await; + let txid_1 = clients.send_from_faucet(&recipient_ua, 250_000).await; + let recipient_zaddr = clients.get_recipient_address("sapling").await; + let txid_2 = clients.send_from_faucet(&recipient_zaddr, 250_000).await; + + clients.rescan_recipient().await; + + let fetch_service = test_manager.full_node_jsonrpc_connector().await; + let mempool_txids = fetch_service.get_raw_mempool().await.unwrap(); + dbg!(txid_1); + dbg!(txid_2); + dbg!(mempool_txids.clone()); + + let _transaction_1 = dbg!( + fetch_service + .get_raw_transaction(mempool_txids.transactions[0].clone(), Some(1)) + .await + ); + let _transaction_2 = dbg!( + fetch_service + .get_raw_transaction(mempool_txids.transactions[1].clone(), Some(1)) + .await + ); + + // Unconfirmed (mempool) balances — devtool's WalletBalance has no + // unconfirmed_* fields (block-based sync, no mempool scan): + // assert_eq!( + // clients.recipient_balance().await.unconfirmed_orchard_balance.unwrap().into_u64(), + // 250_000 + // ); + // assert_eq!( + // clients.recipient_balance().await.unconfirmed_sapling_balance.unwrap().into_u64(), + // 250_000 + // ); + + test_manager + .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) + .await; + + let _transaction_1 = dbg!( + fetch_service + .get_raw_transaction(mempool_txids.transactions[0].clone(), Some(1)) + .await + ); + let _transaction_2 = dbg!( + fetch_service + .get_raw_transaction(mempool_txids.transactions[1].clone(), Some(1)) + .await + ); + + clients.sync_recipient().await; + + // Confirmed balances — original asserts WalletBalance::confirmed_orchard_balance, + // also absent on devtool. Restore as e.g.: + // assert_eq!( + // Pool::Orchard.spendable_balance(&clients.recipient_balance().await), + // 250_000 + // ); + + test_manager.close().await; +} diff --git a/live-tests/e2e/tests/devtool.rs b/live-tests/e2e/tests/devtool.rs index 94012c554..cc2b45a4a 100644 --- a/live-tests/e2e/tests/devtool.rs +++ b/live-tests/e2e/tests/devtool.rs @@ -100,29 +100,7 @@ async fn launch_and_build_clients() -> (TestManager, Devtool where Conn: zaino_testutils::ValidatorConnectionMarker, { - let test_manager = TestManager::::launch_mining_to( - zaino_testutils::SHIELDED_FUNDING_POOL, - &ValidatorKind::Zebrad, - None, - None, - None, - true, - false, - false, - ) - .await - .expect("launch TestManager"); - - let clients = e2e::devtool::build_clients( - test_manager - .zaino_grpc_listen_address - .expect("zaino enabled") - .port(), - &test_manager.local_net, - ) - .await; - - (test_manager, clients) + e2e::devtool::launch_and_build_devtool_clients(&ValidatorKind::Zebrad, None).await } /// Port of `connect_to_node_get_info` (wallet_to_validator, zebrad): the faucet @@ -173,23 +151,8 @@ async fn send_to_pool(pool: e2e::Pool) where Conn: zaino_testutils::ValidatorConnectionMarker, { - let (mut test_manager, mut clients) = launch_and_fund_faucet::(1).await; - - let recipient = clients.get_recipient_address(pool.address_kind()).await; - let txid = clients.send_from_faucet(&recipient, 250_000).await; - dbg!(txid); - - test_manager - .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) - .await; - clients.sync_recipient().await; - - assert_eq!( - pool.spendable_balance(&clients.recipient_balance().await), - 250_000 - ); - - test_manager.close().await; + let (test_manager, clients) = launch_and_fund_faucet::(1).await; + e2e::devtool::assert_send_to_pool(test_manager, clients, pool).await; } /// Port of `send_to_all` (wallet_to_validator, zebrad): one faucet funds a send @@ -243,32 +206,8 @@ async fn shield_for_validator() where Conn: zaino_testutils::ValidatorConnectionMarker, { - let (mut test_manager, mut clients) = launch_and_fund_faucet::(1).await; - - let recipient_taddr = clients.get_recipient_address("transparent").await; - clients.send_from_faucet(&recipient_taddr, 250_000).await; - test_manager - .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) - .await; - clients.sync_recipient().await; - - assert_eq!( - e2e::Pool::Transparent.spendable_balance(&clients.recipient_balance().await), - 250_000 - ); - - clients.shield_recipient().await; - test_manager - .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) - .await; - clients.sync_recipient().await; - - assert_eq!( - e2e::Pool::Ironwood.spendable_balance(&clients.recipient_balance().await), - 235_000 - ); - - test_manager.close().await; + let (test_manager, clients) = launch_and_fund_faucet::(1).await; + e2e::devtool::assert_shield_for_validator(test_manager, clients, e2e::Pool::Ironwood).await; } /// Launch, fund the faucet with two orchard notes, and broadcast (without @@ -1550,47 +1489,8 @@ async fn send_to_transparent_finalization() where Conn: zaino_testutils::ValidatorConnectionMarker, { - let (mut test_manager, mut clients) = launch_and_fund_faucet::(1).await; - - let recipient_taddr = clients.get_recipient_address("transparent").await; - clients.send_from_faucet(&recipient_taddr, 250_000).await; - test_manager - .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) - .await; - - let fetch_service = test_manager.full_node_jsonrpc_connector().await; - let height = fetch_service.get_blockchain_info().await.unwrap().blocks.0; - let unfinalised_transactions = fetch_service - .get_address_txids(vec![recipient_taddr.clone()], height, height) - .await - .unwrap(); - - // The load-bearing advance: these blocks push the send below the seam - // (`FAST_TEST_MAX_NONFINALISED_DEPTH`) into the finalized DB. Orchard coinbase - // here (see #[ignore] rationale). - test_manager - .generate_blocks_bulk_and_wait_for_tips( - // Advance past the seam so the send crosses the finalised floor - // (`tip - seam`); a small margin above it keeps the boundary unambiguous. - zaino_common::consensus::FAST_TEST_MAX_NONFINALISED_DEPTH + 5, - test_manager.subscriber(), - test_manager.subscriber(), - ) - .await; - - let finalised_transactions = fetch_service - .get_address_txids(vec![recipient_taddr], height, height) - .await - .unwrap(); - - clients.sync_recipient().await; - assert_eq!( - e2e::Pool::Transparent.spendable_balance(&clients.recipient_balance().await), - 250_000 - ); - assert_eq!(unfinalised_transactions, finalised_transactions); - - test_manager.close().await; + let (test_manager, clients) = launch_and_fund_faucet::(1).await; + e2e::devtool::assert_send_to_transparent_finalization(test_manager, clients).await; } /// Port of `zebra::get::address_deltas` (zebrad): `getaddressdeltas` over a @@ -2115,68 +2015,9 @@ async fn monitor_unverified_mempool() where Conn: zaino_testutils::ValidatorConnectionMarker, { - let (mut test_manager, mut clients) = launch_and_fund_faucet::(2).await; - - let recipient_ua = clients.get_recipient_address("unified").await; - let txid_1 = clients.send_from_faucet(&recipient_ua, 250_000).await; - let recipient_zaddr = clients.get_recipient_address("sapling").await; - let txid_2 = clients.send_from_faucet(&recipient_zaddr, 250_000).await; - - clients.rescan_recipient().await; - - let fetch_service = test_manager.full_node_jsonrpc_connector().await; - let mempool_txids = fetch_service.get_raw_mempool().await.unwrap(); - dbg!(txid_1); - dbg!(txid_2); - dbg!(mempool_txids.clone()); - - let _transaction_1 = dbg!( - fetch_service - .get_raw_transaction(mempool_txids.transactions[0].clone(), Some(1)) - .await - ); - let _transaction_2 = dbg!( - fetch_service - .get_raw_transaction(mempool_txids.transactions[1].clone(), Some(1)) - .await - ); - - // Unconfirmed (mempool) balances — devtool's WalletBalance has no - // unconfirmed_* fields (block-based sync, no mempool scan): - // assert_eq!( - // clients.recipient_balance().await.unconfirmed_orchard_balance.unwrap().into_u64(), - // 250_000 - // ); - // assert_eq!( - // clients.recipient_balance().await.unconfirmed_sapling_balance.unwrap().into_u64(), - // 250_000 - // ); - - test_manager - .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) - .await; - - let _transaction_1 = dbg!( - fetch_service - .get_raw_transaction(mempool_txids.transactions[0].clone(), Some(1)) - .await - ); - let _transaction_2 = dbg!( - fetch_service - .get_raw_transaction(mempool_txids.transactions[1].clone(), Some(1)) - .await - ); - - clients.sync_recipient().await; - - // Confirmed balances — original asserts WalletBalance::confirmed_orchard_balance, - // also absent on devtool. Restore as e.g.: - // assert_eq!( - // e2e::Pool::Orchard.spendable_balance(&clients.recipient_balance().await), - // 250_000 - // ); - - test_manager.close().await; + // Two orchard notes — one per unmined send. + let (test_manager, clients) = launch_and_fund_faucet::(2).await; + e2e::devtool::assert_monitor_unverified_mempool(test_manager, clients).await; } /// Port of `test_get_mempool_info` (fetch_service, zebrad): `get_mempool_info` diff --git a/live-tests/e2e/tests/devtool_zcashd.rs b/live-tests/e2e/tests/devtool_zcashd.rs index 5950d0806..dc64e2e33 100644 --- a/live-tests/e2e/tests/devtool_zcashd.rs +++ b/live-tests/e2e/tests/devtool_zcashd.rs @@ -37,30 +37,12 @@ use zebra_rpc::methods::GetAddressTxIdsRequest; /// syncing. The zcashd analogue of devtool.rs's `launch_and_build_clients`, /// concrete on zcashd (which has no StateService backend). async fn launch_zcashd_and_build_clients() -> (TestManager, DevtoolClients) { - let test_manager = TestManager::::launch_mining_to( - zaino_testutils::SHIELDED_FUNDING_POOL, // ORCHARD + e2e::devtool::launch_and_build_devtool_clients( &ValidatorKind::Zcashd, - None, // network -> Regtest // The heights the devtool wallet accepts (same as the zebrad path). Some(zaino_testutils::ZEBRAD_DEFAULT_ACTIVATION_HEIGHTS), - None, // no chain cache: build fresh at these heights - true, // enable zaino - false, // no json-rpc server - false, // no clients (the devtool wallet is built separately) ) .await - .expect("launch zcashd TestManager"); - - let clients = e2e::devtool::build_clients( - test_manager - .zaino_grpc_listen_address - .expect("zaino enabled") - .port(), - &test_manager.local_net, - ) - .await; - - (test_manager, clients) } /// [`launch_zcashd_and_build_clients`] plus `orchard_notes` orchard coinbase @@ -429,55 +411,17 @@ mod json_server { /// zcashd analogue of devtool.rs's `send_to_pool`: the faucet sends 250_000 to /// the recipient's `pool` address and the recipient sees it. async fn send_to_pool(pool: e2e::Pool) { - let (mut test_manager, mut clients) = launch_and_fund_zcashd_faucet(1).await; - - let recipient = clients.get_recipient_address(pool.address_kind()).await; - let txid = clients.send_from_faucet(&recipient, 250_000).await; - dbg!(txid); - - test_manager - .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) - .await; - clients.sync_recipient().await; - - assert_eq!( - pool.spendable_balance(&clients.recipient_balance().await), - 250_000 - ); - - test_manager.close().await; + let (test_manager, clients) = launch_and_fund_zcashd_faucet(1).await; + e2e::devtool::assert_send_to_pool(test_manager, clients, pool).await; } /// zcashd analogue of devtool.rs's `shield_for_validator`: the recipient /// receives a transparent send, then shields it into orchard (235_000 after the /// ZIP-317 shielding fee). async fn shield_for_validator() { - let (mut test_manager, mut clients) = launch_and_fund_zcashd_faucet(1).await; - - let recipient_taddr = clients.get_recipient_address("transparent").await; - clients.send_from_faucet(&recipient_taddr, 250_000).await; - test_manager - .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) - .await; - clients.sync_recipient().await; - - assert_eq!( - e2e::Pool::Transparent.spendable_balance(&clients.recipient_balance().await), - 250_000 - ); - - clients.shield_recipient().await; - test_manager - .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) - .await; - clients.sync_recipient().await; - - assert_eq!( - e2e::Pool::Orchard.spendable_balance(&clients.recipient_balance().await), - 235_000 - ); - - test_manager.close().await; + let (test_manager, clients) = launch_and_fund_zcashd_faucet(1).await; + // Pre-NU6.3 heights on zcashd: `shield` lands in orchard, not ironwood. + e2e::devtool::assert_shield_for_validator(test_manager, clients, e2e::Pool::Orchard).await; } /// Devtool ports of `wallet_to_validator`'s `mod zcashd` send/shield/get-info @@ -528,44 +472,8 @@ mod wallet_to_validator { ignore = "heavy: seam-deep orchard advance (~100 halo2 proofs); un-ignore + transparent filler when cheap filler mining lands" )] async fn send_to_transparent_finalization() { - let (mut test_manager, mut clients) = launch_and_fund_zcashd_faucet(1).await; - - let recipient_taddr = clients.get_recipient_address("transparent").await; - clients.send_from_faucet(&recipient_taddr, 250_000).await; - test_manager - .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) - .await; - - let fetch_service = test_manager.full_node_jsonrpc_connector().await; - let height = fetch_service.get_blockchain_info().await.unwrap().blocks.0; - let unfinalised_transactions = fetch_service - .get_address_txids(vec![recipient_taddr.clone()], height, height) - .await - .unwrap(); - - test_manager - .generate_blocks_bulk_and_wait_for_tips( - // Advance past the seam so the send crosses the finalised floor - // (`tip - seam`); a small margin above it keeps the boundary unambiguous. - zaino_common::consensus::FAST_TEST_MAX_NONFINALISED_DEPTH + 5, - test_manager.subscriber(), - test_manager.subscriber(), - ) - .await; - - let finalised_transactions = fetch_service - .get_address_txids(vec![recipient_taddr], height, height) - .await - .unwrap(); - - clients.sync_recipient().await; - assert_eq!( - e2e::Pool::Transparent.spendable_balance(&clients.recipient_balance().await), - 250_000 - ); - assert_eq!(unfinalised_transactions, finalised_transactions); - - test_manager.close().await; + let (test_manager, clients) = launch_and_fund_zcashd_faucet(1).await; + e2e::devtool::assert_send_to_transparent_finalization(test_manager, clients).await; } /// zcashd port of `sent_to::all` (heavy): one faucet funds a send to all @@ -617,56 +525,7 @@ mod wallet_to_validator { ignore = "devtool WalletBalance has no unconfirmed_*/confirmed_* fields; balance asserts commented out — restore + un-ignore when devtool surfaces unconfirmed balances" )] async fn monitor_unverified_mempool() { - let (mut test_manager, mut clients) = launch_and_fund_zcashd_faucet(2).await; - - let recipient_ua = clients.get_recipient_address("unified").await; - let txid_1 = clients.send_from_faucet(&recipient_ua, 250_000).await; - let recipient_zaddr = clients.get_recipient_address("sapling").await; - let txid_2 = clients.send_from_faucet(&recipient_zaddr, 250_000).await; - - clients.rescan_recipient().await; - - let fetch_service = test_manager.full_node_jsonrpc_connector().await; - let mempool_txids = fetch_service.get_raw_mempool().await.unwrap(); - dbg!(txid_1); - dbg!(txid_2); - dbg!(mempool_txids.clone()); - - let _transaction_1 = dbg!( - fetch_service - .get_raw_transaction(mempool_txids.transactions[0].clone(), Some(1)) - .await - ); - let _transaction_2 = dbg!( - fetch_service - .get_raw_transaction(mempool_txids.transactions[1].clone(), Some(1)) - .await - ); - - // Unconfirmed (mempool) balances — devtool's WalletBalance has no - // unconfirmed_* fields: - // assert_eq!(clients.recipient_balance().await.unconfirmed_orchard_balance.unwrap().into_u64(), 250_000); - // assert_eq!(clients.recipient_balance().await.unconfirmed_sapling_balance.unwrap().into_u64(), 250_000); - - test_manager - .generate_blocks_and_wait_for_tip(1, test_manager.subscriber()) - .await; - - let _transaction_1 = dbg!( - fetch_service - .get_raw_transaction(mempool_txids.transactions[0].clone(), Some(1)) - .await - ); - let _transaction_2 = dbg!( - fetch_service - .get_raw_transaction(mempool_txids.transactions[1].clone(), Some(1)) - .await - ); - - clients.sync_recipient().await; - - // Confirmed balances — restore as Pool::Orchard.spendable_balance(...) when un-ignoring. - - test_manager.close().await; + let (test_manager, clients) = launch_and_fund_zcashd_faucet(2).await; + e2e::devtool::assert_monitor_unverified_mempool(test_manager, clients).await; } } diff --git a/packages/zaino-state/src/broadcast.rs b/packages/zaino-state/src/broadcast.rs index 521a4c067..1f1cca875 100644 --- a/packages/zaino-state/src/broadcast.rs +++ b/packages/zaino-state/src/broadcast.rs @@ -153,19 +153,29 @@ impl Default for Broadcast { } } +/// Formats either side of the broadcast pair: the shared dashmap contents plus a +/// placeholder naming the watch-channel endpoint (the channel itself is not `Debug`-useful). +fn fmt_broadcast_state( + struct_name: &str, + state: &DashMap, + notifier_placeholder: &str, + f: &mut std::fmt::Formatter<'_>, +) -> std::fmt::Result { + let state_contents: Vec<_> = state + .iter() + .map(|entry| (entry.key().clone(), entry.value().clone())) + .collect(); + f.debug_struct(struct_name) + .field("state", &state_contents) + .field("notifier", ¬ifier_placeholder) + .finish() +} + impl std::fmt::Debug for Broadcast { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - let state_contents: Vec<_> = self - .state - .iter() - .map(|entry| (entry.key().clone(), entry.value().clone())) - .collect(); - f.debug_struct("Broadcast") - .field("state", &state_contents) - .field("notifier", &"watch::Sender") - .finish() + fmt_broadcast_state("Broadcast", &self.state, "watch::Sender", f) } } @@ -236,14 +246,11 @@ impl std::fm for BroadcastSubscriber { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - let state_contents: Vec<_> = self - .state - .iter() - .map(|entry| (entry.key().clone(), entry.value().clone())) - .collect(); - f.debug_struct("Broadcast") - .field("state", &state_contents) - .field("notifier", &"watch::Sender") - .finish() + fmt_broadcast_state( + "BroadcastSubscriber", + &self.state, + "watch::Receiver", + f, + ) } } diff --git a/packages/zaino-state/src/chain_index.rs b/packages/zaino-state/src/chain_index.rs index cf6ccc87b..a6b222674 100644 --- a/packages/zaino-state/src/chain_index.rs +++ b/packages/zaino-state/src/chain_index.rs @@ -1282,22 +1282,22 @@ async fn compact_block_from_source( let (sapling_root, sapling_size, orchard_root, orchard_size, ironwood) = TreeRootData::new(tree_roots.0, tree_roots.1, tree_roots.2).extract_with_defaults(); - let metadata = BlockMetadata::new( + let metadata = BlockMetadata { sapling_root, - sapling_size.try_into().map_err(|_| { + sapling_size: sapling_size.try_into().map_err(|_| { ChainIndexError::backing_validator(std::io::Error::new( std::io::ErrorKind::InvalidData, "sapling commitment tree size overflow", )) })?, orchard_root, - orchard_size.try_into().map_err(|_| { + orchard_size: orchard_size.try_into().map_err(|_| { ChainIndexError::backing_validator(std::io::Error::new( std::io::ErrorKind::InvalidData, "orchard commitment tree size overflow", )) })?, - ironwood + ironwood: ironwood .map(|(root, size)| { Ok::<_, ChainIndexError>(( root, @@ -1310,9 +1310,10 @@ async fn compact_block_from_source( )) }) .transpose()?, - None, // parent chainwork unknown — single-block construction + // parent chainwork unknown — single-block construction + parent_chainwork: None, network, - ); + }; let indexed_block = IndexedBlock::try_from(BlockWithMetadata::new(&block, metadata)).map_err(|error| { ChainIndexError::backing_validator(std::io::Error::new( diff --git a/packages/zaino-state/src/chain_index/finalised_state.rs b/packages/zaino-state/src/chain_index/finalised_state.rs index 1a1915e82..e7ba12cae 100644 --- a/packages/zaino-state/src/chain_index/finalised_state.rs +++ b/packages/zaino-state/src/chain_index/finalised_state.rs @@ -332,7 +332,7 @@ pub(crate) async fn build_indexed_block_from_source( || format!("block {block_hash}"), )?; - let metadata = BlockMetadata::new( + let metadata = BlockMetadata { sapling_root, sapling_size, orchard_root, @@ -340,7 +340,7 @@ pub(crate) async fn build_indexed_block_from_source( ironwood, parent_chainwork, network, - ); + }; let block_with_metadata = BlockWithMetadata::new(block.as_ref(), metadata); IndexedBlock::try_from(block_with_metadata).map_err(|_| { @@ -1185,15 +1185,15 @@ impl FinalisedState { || format!("block {block_hash}"), )?; - let metadata = BlockMetadata::new( + let metadata = BlockMetadata { sapling_root, sapling_size, orchard_root, orchard_size, ironwood, parent_chainwork, - cfg.network.clone(), - ); + network: cfg.network.clone(), + }; let block_with_metadata = BlockWithMetadata::new(block.as_ref(), metadata); let chain_block = IndexedBlock::try_from(block_with_metadata).map_err(|_| { FinalisedStateError::BlockchainSourceError(BlockchainSourceError::Unrecoverable( diff --git a/packages/zaino-state/src/chain_index/finalised_state/finalised_source/ephemeral.rs b/packages/zaino-state/src/chain_index/finalised_state/finalised_source/ephemeral.rs index b53aaaaac..8cbbf19b1 100644 --- a/packages/zaino-state/src/chain_index/finalised_state/finalised_source/ephemeral.rs +++ b/packages/zaino-state/src/chain_index/finalised_state/finalised_source/ephemeral.rs @@ -301,15 +301,16 @@ impl EphemeralFinalisedState { format!("block at height {height}") })?; - let block_metadata = BlockMetadata::new( + let block_metadata = BlockMetadata { sapling_root, sapling_size, orchard_root, orchard_size, ironwood, - None, // ephemeral store does not track chainwork - self.network.clone(), - ); + // ephemeral store does not track chainwork + parent_chainwork: None, + network: self.network.clone(), + }; let block_with_metadata = BlockWithMetadata::new(block.as_ref(), block_metadata); let indexed_block = IndexedBlock::try_from(block_with_metadata).map_err(|error| { FinalisedStateError::BlockchainSourceError(BlockchainSourceError::Unrecoverable( diff --git a/packages/zaino-state/src/chain_index/non_finalised_state.rs b/packages/zaino-state/src/chain_index/non_finalised_state.rs index aee18b01a..80b1a52a7 100644 --- a/packages/zaino-state/src/chain_index/non_finalised_state.rs +++ b/packages/zaino-state/src/chain_index/non_finalised_state.rs @@ -838,15 +838,15 @@ impl NonFinalizedState { let (sapling_root, sapling_size, orchard_root, orchard_size, ironwood) = tree_roots.clone().extract_with_defaults(); - let metadata = BlockMetadata::new( + let metadata = BlockMetadata { sapling_root, - sapling_size as u32, + sapling_size: sapling_size as u32, orchard_root, - orchard_size as u32, - ironwood.map(|(root, size)| (root, size as u32)), + orchard_size: orchard_size as u32, + ironwood: ironwood.map(|(root, size)| (root, size as u32)), parent_chainwork, network, - ); + }; let block_with_metadata = BlockWithMetadata::new(block, metadata); IndexedBlock::try_from(block_with_metadata) diff --git a/packages/zaino-state/src/chain_index/source/mockchain_source.rs b/packages/zaino-state/src/chain_index/source/mockchain_source.rs index a6bc290e8..76b6476d3 100644 --- a/packages/zaino-state/src/chain_index/source/mockchain_source.rs +++ b/packages/zaino-state/src/chain_index/source/mockchain_source.rs @@ -2,7 +2,7 @@ use super::validator_connector::{ assemble_block_deltas, build_block_header_object, build_verbose_block, - confirmations_from_depth, final_orchard_root, final_sapling_root, median_of_block_times, + confirmations_from_depth, final_orchard_root, final_sapling_root, median_time_past_via, zebra_block_header_to_wire, }; use super::*; @@ -415,34 +415,10 @@ impl MockchainSource { /// Median time past over the 11-block window ending at `start`, walking backwards via /// verbosity-1 `getblock` lookups against the mock vectors. async fn median_time_past(&self, start: &BlockObject) -> BlockchainSourceResult { - const MEDIAN_TIME_PAST_WINDOW: usize = 11; - let mut times = Vec::with_capacity(MEDIAN_TIME_PAST_WINDOW); - let start_time = start.time().ok_or_else(|| { - BlockchainSourceError::Unrecoverable("getblockdeltas: start block missing time".into()) - })?; - times.push(start_time); - - let mut prev = start.previous_block_hash(); - for _ in 0..(MEDIAN_TIME_PAST_WINDOW - 1) { - let Some(hash) = prev else { - break; // genesis - }; - match self - .get_block_verbose(HashOrHeight::Hash(hash), Some(1)) - .await - { - Ok(GetBlock::Object(object)) => { - if let Some(time) = object.time() { - times.push(time); - } - prev = object.previous_block_hash(); - } - Ok(GetBlock::Raw(_)) => break, - Err(_) => break, - } - } - - median_of_block_times(times) + median_time_past_via(start, |hash| { + self.get_block_verbose(HashOrHeight::Hash(hash), Some(1)) + }) + .await } fn block_height_at_index(&self, block_index: usize) -> Height { diff --git a/packages/zaino-state/src/chain_index/source/validator_connector.rs b/packages/zaino-state/src/chain_index/source/validator_connector.rs index e2130c218..b6bdf8d7e 100644 --- a/packages/zaino-state/src/chain_index/source/validator_connector.rs +++ b/packages/zaino-state/src/chain_index/source/validator_connector.rs @@ -2044,37 +2044,11 @@ impl State { /// Median time past over the 11-block window ending at `start`, walking backwards via /// verbosity-1 `getblock` lookups against the `ReadStateService`. - // TODO(DRY): MockchainSource duplicates this walk; the only difference is the - // per-block fetch. A shared helper generic over an async block fetcher would unify them. async fn median_time_past(&self, start: &BlockObject) -> BlockchainSourceResult { - const MEDIAN_TIME_PAST_WINDOW: usize = 11; - let mut times = Vec::with_capacity(MEDIAN_TIME_PAST_WINDOW); - let start_time = start.time().ok_or_else(|| { - BlockchainSourceError::Unrecoverable("getblockdeltas: start block missing time".into()) - })?; - times.push(start_time); - - let mut prev = start.previous_block_hash(); - for _ in 0..(MEDIAN_TIME_PAST_WINDOW - 1) { - let Some(hash) = prev else { - break; // genesis - }; - match self - .get_block_inner(HashOrHeight::Hash(hash), Some(1)) - .await - { - Ok(GetBlock::Object(object)) => { - if let Some(time) = object.time() { - times.push(time); - } - prev = object.previous_block_hash(); - } - Ok(GetBlock::Raw(_)) => break, - Err(_) => break, // use values collected so far - } - } - - median_of_block_times(times) + median_time_past_via(start, |hash| { + self.get_block_inner(HashOrHeight::Hash(hash), Some(1)) + }) + .await } /// Builds the `getblockchaininfo` response from the `ReadStateService`. @@ -2388,6 +2362,46 @@ pub(crate) fn zebra_block_header_to_wire( serde_json::from_value(value).map_err(BlockchainSourceError::unrecoverable) } +/// Median time past over the 11-block window ending at `start`, walking parent hashes +/// backwards via `fetch_prev` (a verbosity-1 `getblock` lookup against the caller's source). +/// +/// The walk tolerates short histories: it stops at genesis, at a raw (non-verbose) +/// response, or on a fetch error, and takes the median of the times collected so far. +pub(crate) async fn median_time_past_via( + start: &BlockObject, + fetch_prev: Fetch, +) -> BlockchainSourceResult +where + Fetch: Fn(zebra_chain::block::Hash) -> Fut, + Fut: std::future::Future>, +{ + const MEDIAN_TIME_PAST_WINDOW: usize = 11; + let mut times = Vec::with_capacity(MEDIAN_TIME_PAST_WINDOW); + let start_time = start.time().ok_or_else(|| { + BlockchainSourceError::Unrecoverable("getblockdeltas: start block missing time".into()) + })?; + times.push(start_time); + + let mut prev = start.previous_block_hash(); + for _ in 0..(MEDIAN_TIME_PAST_WINDOW - 1) { + let Some(hash) = prev else { + break; // genesis + }; + match fetch_prev(hash).await { + Ok(GetBlock::Object(object)) => { + if let Some(time) = object.time() { + times.push(time); + } + prev = object.previous_block_hash(); + } + Ok(GetBlock::Raw(_)) => break, + Err(_) => break, // use values collected so far + } + } + + median_of_block_times(times) +} + /// Returns the median of a non-empty set of block times. pub(crate) fn median_of_block_times(mut times: Vec) -> BlockchainSourceResult { if times.is_empty() { diff --git a/packages/zaino-state/src/chain_index/tests/finalised_state/v1.rs b/packages/zaino-state/src/chain_index/tests/finalised_state/v1.rs index f59349635..c4b275d4c 100644 --- a/packages/zaino-state/src/chain_index/tests/finalised_state/v1.rs +++ b/packages/zaino-state/src/chain_index/tests/finalised_state/v1.rs @@ -331,15 +331,16 @@ async fn try_write_invalid_block() { .. } = blocks.last().unwrap().clone(); - let metadata = BlockMetadata::new( + let metadata = BlockMetadata { sapling_root, - sapling_tree_size as u32, + sapling_size: sapling_tree_size as u32, orchard_root, - orchard_tree_size as u32, - None, - None, // no parent chainwork for this test - ActivationHeights::default().to_regtest_network(), - ); + orchard_size: orchard_tree_size as u32, + ironwood: None, + // no parent chainwork for this test + parent_chainwork: None, + network: ActivationHeights::default().to_regtest_network(), + }; let mut chain_block = IndexedBlock::try_from(BlockWithMetadata::new(&zebra_block, metadata)).unwrap(); diff --git a/packages/zaino-state/src/chain_index/tests/types.rs b/packages/zaino-state/src/chain_index/tests/types.rs index 2da3fab70..b2afbaad3 100644 --- a/packages/zaino-state/src/chain_index/tests/types.rs +++ b/packages/zaino-state/src/chain_index/tests/types.rs @@ -27,7 +27,15 @@ pub(crate) fn canonical_blockheaderdata() -> BlockHeaderData { let bits = CompactDifficulty::try_from_bits(TEST_VALID_NBITS).expect("valid nBits"); let bctx = BlockContext::new(hash, parent_hash, chainwork, height); - let bdata = BlockData::new(1, 2, [3u8; 32], [4u8; 32], bits, [5u8; 32], solution); + let bdata = BlockData { + version: 1, + time: 2, + merkle_root: [3u8; 32], + block_commitments: [4u8; 32], + bits, + nonce: [5u8; 32], + solution, + }; BlockHeaderData::new(bctx, bdata) } diff --git a/packages/zaino-state/src/chain_index/tests/vectors.rs b/packages/zaino-state/src/chain_index/tests/vectors.rs index 0e8a3ceae..814bc533c 100644 --- a/packages/zaino-state/src/chain_index/tests/vectors.rs +++ b/packages/zaino-state/src/chain_index/tests/vectors.rs @@ -92,14 +92,14 @@ pub(crate) fn indexed_block_chain( ) -> impl Iterator + '_ { let mut parent_chain_work: Option = None; blocks.iter().map(move |vector| { - let metadata = BlockMetadata::new( - vector.sapling_root, - vector.sapling_tree_size as u32, - vector.orchard_root, - vector.orchard_tree_size as u32, - None, - parent_chain_work, - zebra_chain::parameters::Network::new_regtest( + let metadata = BlockMetadata { + sapling_root: vector.sapling_root, + sapling_size: vector.sapling_tree_size as u32, + orchard_root: vector.orchard_root, + orchard_size: vector.orchard_tree_size as u32, + ironwood: None, + parent_chainwork: parent_chain_work, + network: zebra_chain::parameters::Network::new_regtest( zebra_chain::parameters::testnet::ConfiguredActivationHeights { before_overwinter: Some(1), overwinter: Some(1), @@ -117,7 +117,7 @@ pub(crate) fn indexed_block_chain( } .into(), ), - ); + }; let chain_block = IndexedBlock::try_from(BlockWithMetadata::new(&vector.zebra_block, metadata)).unwrap(); parent_chain_work = Some(chain_block.context.chainwork); diff --git a/packages/zaino-state/src/chain_index/types/db/legacy.rs b/packages/zaino-state/src/chain_index/types/db/legacy.rs index ea663352c..6f02f2087 100644 --- a/packages/zaino-state/src/chain_index/types/db/legacy.rs +++ b/packages/zaino-state/src/chain_index/types/db/legacy.rs @@ -793,28 +793,6 @@ pub struct BlockData { } impl BlockData { - /// Creates a new BlockData instance. - #[allow(clippy::too_many_arguments)] - pub fn new( - version: u32, - time: i64, - merkle_root: [u8; 32], - block_commitments: [u8; 32], - bits: CompactDifficulty, - nonse: [u8; 32], - solution: EquihashSolution, - ) -> Self { - Self { - version, - time, - merkle_root, - block_commitments, - bits, - nonce: nonse, - solution, - } - } - /// Convert zebra block commitment to 32-byte array pub fn commitment_to_bytes(commitment: zebra_chain::block::Commitment) -> [u8; 32] { match commitment { @@ -908,15 +886,15 @@ impl ZainoVersionedSerde for BlockData { let solution = EquihashSolution::deserialize(&mut r)?; - Ok(BlockData::new( + Ok(BlockData { version, time, merkle_root, block_commitments, bits, - nonse, + nonce: nonse, solution, - )) + }) } } @@ -1264,15 +1242,15 @@ impl ); // --- Compute chainwork --- - let block_data = BlockData::new( - header.version() as u32, - header.time() as i64, + let block_data = BlockData { + version: header.version() as u32, + time: header.time() as i64, merkle_root, block_commitments, bits, - nonse, + nonce: nonse, solution, - ); + }; let block_work = block_data.bits.to_work(); let chainwork = match parent_chainwork { diff --git a/packages/zaino-state/src/chain_index/types/db/metadata.rs b/packages/zaino-state/src/chain_index/types/db/metadata.rs index efbb3880f..500417adc 100644 --- a/packages/zaino-state/src/chain_index/types/db/metadata.rs +++ b/packages/zaino-state/src/chain_index/types/db/metadata.rs @@ -206,23 +206,34 @@ impl FinalisedTxOutSetInfoAccumulator { outpoint: &Outpoint, out: &TxOutCompact, ) -> Result<(), AccumulatorDeltaError> { - let digest = tx_out_set_entry_digest(outpoint, out); + self.apply_output_delta(outpoint, out, OutputDelta::Added) + } + + /// Applies one output entering or leaving the set: XOR its digest into the multiset + /// commitment and step every per-output counter in the delta's direction. + fn apply_output_delta( + &mut self, + outpoint: &Outpoint, + out: &TxOutCompact, + delta: OutputDelta, + ) -> Result<(), AccumulatorDeltaError> { + self.xor_into_hash_serialized(&tx_out_set_entry_digest(outpoint, out)); + self.transaction_outputs = + delta.step(self.transaction_outputs, 1, "transaction_outputs")?; + self.bytes_serialized = delta.step( + self.bytes_serialized, + ZAINO_TXOUTSET_ENTRY_LEN, + "bytes_serialized", + )?; + self.total_zatoshis = delta.step(self.total_zatoshis, out.value(), "total_zatoshis")?; + Ok(()) + } + + /// XORs `digest` into the multiset commitment (self-inverse and order-independent). + fn xor_into_hash_serialized(&mut self, digest: &[u8; 32]) { for (dst, src) in self.hash_serialized.iter_mut().zip(digest.iter()) { *dst ^= *src; } - self.transaction_outputs = self - .transaction_outputs - .checked_add(1) - .ok_or(AccumulatorDeltaError::Overflow("transaction_outputs"))?; - self.bytes_serialized = self - .bytes_serialized - .checked_add(ZAINO_TXOUTSET_ENTRY_LEN) - .ok_or(AccumulatorDeltaError::Overflow("bytes_serialized"))?; - self.total_zatoshis = self - .total_zatoshis - .checked_add(out.value()) - .ok_or(AccumulatorDeltaError::Overflow("total_zatoshis"))?; - Ok(()) } /// Folds another accumulator into this one: XOR the multiset commitments (self-inverse and @@ -232,13 +243,7 @@ impl FinalisedTxOutSetInfoAccumulator { /// both commutative and associative, the recombined result is independent of the shard count and /// of the order shards are folded in. pub fn combine(&mut self, other: &Self) -> Result<(), AccumulatorDeltaError> { - for (dst, src) in self - .hash_serialized - .iter_mut() - .zip(other.hash_serialized.iter()) - { - *dst ^= *src; - } + self.xor_into_hash_serialized(&other.hash_serialized); self.transactions = self .transactions .checked_add(other.transactions) @@ -265,23 +270,35 @@ impl FinalisedTxOutSetInfoAccumulator { outpoint: &Outpoint, out: &TxOutCompact, ) -> Result<(), AccumulatorDeltaError> { - let digest = tx_out_set_entry_digest(outpoint, out); - for (dst, src) in self.hash_serialized.iter_mut().zip(digest.iter()) { - *dst ^= *src; + self.apply_output_delta(outpoint, out, OutputDelta::Removed) + } +} + +/// Direction of a single-output change applied to the accumulator's counters. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum OutputDelta { + /// The output enters the UTXO set: counters are checked-added. + Added, + /// The output leaves the UTXO set: counters are checked-subtracted. + Removed, +} + +impl OutputDelta { + /// Steps `field` by `amount` in this direction, naming `field_name` in the error. + fn step( + self, + field: u64, + amount: u64, + field_name: &'static str, + ) -> Result { + match self { + OutputDelta::Added => field + .checked_add(amount) + .ok_or(AccumulatorDeltaError::Overflow(field_name)), + OutputDelta::Removed => field + .checked_sub(amount) + .ok_or(AccumulatorDeltaError::Underflow(field_name)), } - self.transaction_outputs = self - .transaction_outputs - .checked_sub(1) - .ok_or(AccumulatorDeltaError::Underflow("transaction_outputs"))?; - self.bytes_serialized = self - .bytes_serialized - .checked_sub(ZAINO_TXOUTSET_ENTRY_LEN) - .ok_or(AccumulatorDeltaError::Underflow("bytes_serialized"))?; - self.total_zatoshis = self - .total_zatoshis - .checked_sub(out.value()) - .ok_or(AccumulatorDeltaError::Underflow("total_zatoshis"))?; - Ok(()) } } diff --git a/packages/zaino-state/src/chain_index/types/helpers.rs b/packages/zaino-state/src/chain_index/types/helpers.rs index dbfb5297a..d6bbc3021 100644 --- a/packages/zaino-state/src/chain_index/types/helpers.rs +++ b/packages/zaino-state/src/chain_index/types/helpers.rs @@ -124,29 +124,6 @@ pub struct BlockMetadata { pub network: zebra_chain::parameters::Network, } -impl BlockMetadata { - /// Create new block metadata - pub fn new( - sapling_root: zebra_chain::sapling::tree::Root, - sapling_size: u32, - orchard_root: zebra_chain::orchard::tree::Root, - orchard_size: u32, - ironwood: Option<(zebra_chain::orchard::tree::Root, u32)>, - parent_chainwork: Option, - network: zebra_chain::parameters::Network, - ) -> Self { - Self { - sapling_root, - sapling_size, - orchard_root, - orchard_size, - ironwood, - parent_chainwork, - network, - } - } -} - /// Intermediate type combining a block with its metadata #[derive(Debug, Clone)] pub struct BlockWithMetadata<'a> { @@ -402,15 +379,15 @@ mod create_commitment_tree_data { fn absent_ironwood_root_is_stored_as_none() { let (sapling_root, sapling_size, orchard_root, orchard_size, ironwood) = TreeRootData::new(None, None, None).extract_with_defaults(); - let metadata = BlockMetadata::new( + let metadata = BlockMetadata { sapling_root, - sapling_size as u32, + sapling_size: sapling_size as u32, orchard_root, - orchard_size as u32, - ironwood.map(|(root, size)| (root, size as u32)), - None, - zebra_chain::parameters::Network::Mainnet, - ); + orchard_size: orchard_size as u32, + ironwood: ironwood.map(|(root, size)| (root, size as u32)), + parent_chainwork: None, + network: zebra_chain::parameters::Network::Mainnet, + }; let commitment_tree_data = metadata.create_commitment_tree_data(); diff --git a/packages/zaino-state/src/error.rs b/packages/zaino-state/src/error.rs index b0eb87b31..d64ec78f8 100644 --- a/packages/zaino-state/src/error.rs +++ b/packages/zaino-state/src/error.rs @@ -192,29 +192,11 @@ impl From> for NodeBackedIndexerServiceError { /// in favor of a new type with the new chain cache is complete impl From> for MempoolError { fn from(value: RpcRequestError) -> Self { - match value { - RpcRequestError::Transport(transport_error) => { + match RpcRequestFallback::classify(value) { + RpcRequestFallback::Transport(transport_error) => { MempoolError::JsonRpcConnectorError(transport_error) } - RpcRequestError::JsonRpc(error) => { - MempoolError::Critical(format!("argument failed to serialze: {error}")) - } - RpcRequestError::InternalUnrecoverable(e) => { - MempoolError::Critical(format!("Internal unrecoverable error: {e}")) - } - RpcRequestError::ServerWorkQueueFull => MempoolError::Critical( - "Server queue full. Handling for this not yet implemented".to_string(), - ), - RpcRequestError::Method(e) => MempoolError::Critical(format!( - "unhandled rpc-specific {} error: {}", - type_name::(), - e.to_string() - )), - RpcRequestError::UnexpectedErrorResponse(error) => MempoolError::Critical(format!( - "unhandled rpc-specific {} error: {}", - type_name::(), - error - )), + RpcRequestFallback::Message(message) => MempoolError::Critical(message), } } } @@ -320,30 +302,40 @@ pub enum NonFinalisedStateError { StatusError(StatusError), } -/// These aren't the best conversions, but the NonFinalizedStateError should go away -/// in favor of a new type with the new chain cache is complete -impl From> for NonFinalisedStateError { - fn from(value: RpcRequestError) -> Self { +/// Either a transport error passed through intact, or the formatted message for the +/// target error type's degraded catch-all variant. +/// +/// Shared classification behind the `From>` conversions for +/// [`MempoolError`], [`NonFinalisedStateError`], and [`FinalisedStateError`]: all +/// three map transport errors to their `JsonRpcConnectorError` variant and degrade +/// every other case to a message-carrying variant (`Critical` or `Custom`). +enum RpcRequestFallback { + Transport(zaino_fetch::jsonrpsee::error::TransportError), + Message(String), +} + +impl RpcRequestFallback { + fn classify(value: RpcRequestError) -> Self { match value { RpcRequestError::Transport(transport_error) => { - NonFinalisedStateError::JsonRpcConnectorError(transport_error) + RpcRequestFallback::Transport(transport_error) } RpcRequestError::JsonRpc(error) => { - NonFinalisedStateError::Custom(format!("argument failed to serialze: {error}")) + RpcRequestFallback::Message(format!("argument failed to serialze: {error}")) } RpcRequestError::InternalUnrecoverable(e) => { - NonFinalisedStateError::Custom(format!("Internal unrecoverable error: {e}")) + RpcRequestFallback::Message(format!("Internal unrecoverable error: {e}")) } - RpcRequestError::ServerWorkQueueFull => NonFinalisedStateError::Custom( + RpcRequestError::ServerWorkQueueFull => RpcRequestFallback::Message( "Server queue full. Handling for this not yet implemented".to_string(), ), - RpcRequestError::Method(e) => NonFinalisedStateError::Custom(format!( + RpcRequestError::Method(e) => RpcRequestFallback::Message(format!( "unhandled rpc-specific {} error: {}", type_name::(), e.to_string() )), RpcRequestError::UnexpectedErrorResponse(error) => { - NonFinalisedStateError::Custom(format!( + RpcRequestFallback::Message(format!( "unhandled rpc-specific {} error: {}", type_name::(), error @@ -353,6 +345,19 @@ impl From> for NonFinalisedStateError { } } +/// These aren't the best conversions, but the NonFinalizedStateError should go away +/// in favor of a new type with the new chain cache is complete +impl From> for NonFinalisedStateError { + fn from(value: RpcRequestError) -> Self { + match RpcRequestFallback::classify(value) { + RpcRequestFallback::Transport(transport_error) => { + NonFinalisedStateError::JsonRpcConnectorError(transport_error) + } + RpcRequestFallback::Message(message) => NonFinalisedStateError::Custom(message), + } + } +} + /// Errors related to the `FinalisedState`. // TODO: Update name to DbError when FinalisedState replaces legacy finalised state. #[derive(Debug, thiserror::Error)] @@ -423,31 +428,11 @@ pub enum FinalisedStateError { /// in favor of a new type with the new chain cache is complete impl From> for FinalisedStateError { fn from(value: RpcRequestError) -> Self { - match value { - RpcRequestError::Transport(transport_error) => { + match RpcRequestFallback::classify(value) { + RpcRequestFallback::Transport(transport_error) => { FinalisedStateError::JsonRpcConnectorError(transport_error) } - RpcRequestError::JsonRpc(error) => { - FinalisedStateError::Custom(format!("argument failed to serialze: {error}")) - } - RpcRequestError::InternalUnrecoverable(e) => { - FinalisedStateError::Custom(format!("Internal unrecoverable error: {e}")) - } - RpcRequestError::ServerWorkQueueFull => FinalisedStateError::Custom( - "Server queue full. Handling for this not yet implemented".to_string(), - ), - RpcRequestError::Method(e) => FinalisedStateError::Custom(format!( - "unhandled rpc-specific {} error: {}", - type_name::(), - e.to_string() - )), - RpcRequestError::UnexpectedErrorResponse(error) => { - FinalisedStateError::Custom(format!( - "unhandled rpc-specific {} error: {}", - type_name::(), - error - )) - } + RpcRequestFallback::Message(message) => FinalisedStateError::Custom(message), } } } diff --git a/packages/zaino-state/src/indexer/node_backed_indexer.rs b/packages/zaino-state/src/indexer/node_backed_indexer.rs index 8c9078cc9..91f109925 100644 --- a/packages/zaino-state/src/indexer/node_backed_indexer.rs +++ b/packages/zaino-state/src/indexer/node_backed_indexer.rs @@ -373,6 +373,121 @@ impl NodeBackedIndexerServiceSubscriber { monitor: self.indexer.source().chain_tip_change()?, }) } + + /// Shared body of `get_block_range` and `get_block_range_nullifiers`: streams the + /// requested compact-block range through a channel, applying `map_block` to every + /// block before it is sent. `rpc_name` labels log lines and nothing else. + #[allow(deprecated)] + async fn spawn_block_range_stream( + &self, + request: &BlockRange, + rpc_name: &'static str, + map_block: impl Fn(CompactBlock) -> CompactBlock + Send + Sync + 'static, + ) -> Result { + let validated_request = ValidatedBlockRangeRequest::new_from_block_range(request) + .map_err(NodeBackedIndexerServiceError::from)?; + + let pool_type_filter = validated_request.pool_type_filter().clone(); + + let start = validated_request.start(); + let end = validated_request.end(); + + let service_clone = self.clone(); + let service_timeout = self.config.service.timeout; + let (channel_tx, channel_rx) = mpsc::channel(self.config.service.channel_size as usize); + let snapshot = service_clone.indexer.snapshot_nonfinalized_state().await?; + + tokio::spawn(async move { + let timeout_result = timeout( + time::Duration::from_secs((service_timeout * 4) as u64), + async { + let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { + // TODO: This probably shouldn't be an error. + // this is an improvement over previous behaviour of + // acting as if we are only synced to the genesis block + if let Err(e) = channel_tx + .send(Err(tonic::Status::failed_precondition( + "zaino not yet synced".to_string(), + ))) + .await + { + warn!(%e, "{rpc_name} channel closed unexpectedly"); + }; + return; + }; + // Use the snapshot tip directly, as this function doesn't support passthrough + let chain_height = non_finalized_snapshot.best_tip.height.0; + + let height_out_of_range_status = move || { + let offending_height = if start > chain_height { start } else { end }; + tonic::Status::out_of_range(format!( + "Error: Height out of range [{offending_height}]. \ + Height requested is greater than the best \ + chain tip [{chain_height}].", + )) + }; + + match service_clone + .indexer + .get_compact_block_stream( + &snapshot, + types::Height(start), + types::Height(end), + pool_type_filter.clone(), + ) + .await + { + Ok(Some(mut compact_block_stream)) => { + while let Some(stream_item) = compact_block_stream.next().await { + if channel_tx.send(stream_item.map(&map_block)).await.is_err() { + break; + } + } + } + Ok(None) => { + // Per `get_compact_block_stream` semantics: `None` means at least one bound is above the tip. + if let Err(e) = channel_tx.send(Err(height_out_of_range_status())).await + { + warn!(%e, "{rpc_name} channel closed unexpectedly"); + } + } + Err(e) => { + // Preserve previous behaviour: if the request is above tip, surface OutOfRange; + // otherwise return the error (currently exposed for dev). + if start > chain_height || end > chain_height { + if let Err(e) = + channel_tx.send(Err(height_out_of_range_status())).await + { + warn!(%e, "{rpc_name} channel closed unexpectedly"); + } + } else { + // TODO: Hide server error from clients before release. Currently useful for dev purposes. + if channel_tx + .send(Err(tonic::Status::unknown(e.to_string()))) + .await + .is_err() + { + warn!(%e, "{rpc_name} stream closed unexpectedly"); + } + } + } + } + }, + ) + .await; + + if timeout_result.is_err() { + channel_tx + .send(Err(tonic::Status::deadline_exceeded( + "Error: get_block_range gRPC request timed out.", + ))) + .await + .ok(); + } + }); + + Ok(CompactBlockStream::new(channel_rx)) + } } /// Methods available only on the production (`ValidatorConnector`-backed) subscriber: @@ -1251,122 +1366,8 @@ impl LightWalletIndexer for NodeBackedIndexerServiceSu &self, request: BlockRange, ) -> Result { - let validated_request = ValidatedBlockRangeRequest::new_from_block_range(&request) - .map_err(NodeBackedIndexerServiceError::from)?; - - let pool_type_filter = validated_request.pool_type_filter().clone(); - - let start = validated_request.start(); - let end = validated_request.end(); - - let service_clone = self.clone(); - let service_timeout = self.config.service.timeout; - let (channel_tx, channel_rx) = mpsc::channel(self.config.service.channel_size as usize); - let snapshot = service_clone.indexer.snapshot_nonfinalized_state().await?; - - tokio::spawn(async move { - let timeout_result = timeout( - time::Duration::from_secs((service_timeout * 4) as u64), - async { - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - if let Err(e) = channel_tx - .send(Err(tonic::Status::failed_precondition( - "zaino not yet synced".to_string(), - ))) - .await - { - warn!(%e, "GetBlockRange channel closed unexpectedly"); - }; - return; - }; - // Use the snapshot tip directly, as this function doesn't support passthrough - let chain_height = non_finalized_snapshot.best_tip.height.0; - - match service_clone - .indexer - .get_compact_block_stream( - &snapshot, - types::Height(start), - types::Height(end), - pool_type_filter.clone(), - ) - .await - { - Ok(Some(mut compact_block_stream)) => { - while let Some(stream_item) = compact_block_stream.next().await { - if channel_tx.send(stream_item).await.is_err() { - break; - } - } - } - Ok(None) => { - // Per `get_compact_block_stream` semantics: `None` means at least one bound is above the tip. - let offending_height = if start > chain_height { start } else { end }; - - match channel_tx - .send(Err(tonic::Status::out_of_range(format!( - "Error: Height out of range [{offending_height}]. \ - Height requested is greater than the best \ - chain tip [{chain_height}].", - )))) - .await - { - Ok(_) => {} - Err(e) => { - warn!(%e, "GetBlockRange channel closed unexpectedly"); - } - } - } - Err(e) => { - // Preserve previous behaviour: if the request is above tip, surface OutOfRange; - // otherwise return the error (currently exposed for dev). - if start > chain_height || end > chain_height { - let offending_height = - if start > chain_height { start } else { end }; - - match channel_tx - .send(Err(tonic::Status::out_of_range(format!( - "Error: Height out of range [{offending_height}]. \ - Height requested is greater than the best \ - chain tip [{chain_height}].", - )))) - .await - { - Ok(_) => {} - Err(e) => { - warn!(%e, "GetBlockRange channel closed unexpectedly"); - } - } - } else { - // TODO: Hide server error from clients before release. Currently useful for dev purposes. - if channel_tx - .send(Err(tonic::Status::unknown(e.to_string()))) - .await - .is_err() - { - warn!(%e, "GetBlockRangeStream closed unexpectedly"); - } - } - } - } - }, - ) - .await; - - if timeout_result.is_err() { - channel_tx - .send(Err(tonic::Status::deadline_exceeded( - "Error: get_block_range gRPC request timed out.", - ))) - .await - .ok(); - } - }); - - Ok(CompactBlockStream::new(channel_rx)) + self.spawn_block_range_stream(&request, "GetBlockRange", |block| block) + .await } /// Same as GetBlockRange except actions contain only nullifiers @@ -1377,138 +1378,12 @@ impl LightWalletIndexer for NodeBackedIndexerServiceSu &self, request: BlockRange, ) -> Result { - let validated_request = ValidatedBlockRangeRequest::new_from_block_range(&request) - .map_err(NodeBackedIndexerServiceError::from)?; - - let pool_type_filter = validated_request.pool_type_filter().clone(); - - let start = validated_request.start(); - let end = validated_request.end(); - - let service_clone = self.clone(); - let service_timeout = self.config.service.timeout; - let (channel_tx, channel_rx) = mpsc::channel(self.config.service.channel_size as usize); - let snapshot = service_clone.indexer.snapshot_nonfinalized_state().await?; - - tokio::spawn(async move { - let timeout_result = timeout( - time::Duration::from_secs((service_timeout * 4) as u64), - async { - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - if let Err(e) = channel_tx - .send(Err(tonic::Status::failed_precondition( - "zaino not yet synced".to_string(), - ))) - .await - { - warn!(%e, "GetBlockRangeNullifiers channel closed unexpectedly"); - }; - return; - }; - - // Use the snapshot tip directly, as this function doesn't support passthrough - let chain_height = non_finalized_snapshot.best_tip.height.0; - - match service_clone - .indexer - .get_compact_block_stream( - &snapshot, - types::Height(start), - types::Height(end), - pool_type_filter.clone(), - ) - .await - { - Ok(Some(mut compact_block_stream)) => { - while let Some(stream_item) = compact_block_stream.next().await { - match stream_item { - Ok(block) => { - if channel_tx - .send(Ok(compact_block_to_nullifiers(block))) - .await - .is_err() - { - break; - } - } - Err(status) => { - if channel_tx.send(Err(status)).await.is_err() { - break; - } - } - } - } - } - Ok(None) => { - // Per `get_compact_block_stream` semantics: `None` means at least one bound is above the tip. - let offending_height = if start > chain_height { start } else { end }; - - match channel_tx - .send(Err(tonic::Status::out_of_range(format!( - "Error: Height out of range [{offending_height}]. \ - Height requested is greater than the best \ - chain tip [{chain_height}].", - )))) - .await - { - Ok(_) => {} - Err(e) => { - warn!(%e, "GetBlockRange channel closed unexpectedly"); - } - } - } - Err(e) => { - // Preserve previous behaviour: if the request - // is above tip, surface OutOfRange; - // otherwise return the error (currently exposed for dev). - if start > chain_height || end > chain_height { - let offending_height = - if start > chain_height { start } else { end }; - - match channel_tx - .send(Err(tonic::Status::out_of_range(format!( - "Error: Height out of range [{offending_height}]. \ - Height requested is greater than the best chain tip \ - [{chain_height}].", - )))) - .await - { - Ok(_) => {} - Err(e) => { - warn!(%e, "GetBlockRange channel closed unexpectedly"); - } - } - } else { - // TODO: Hide server error from clients before release. - // Currently useful for dev purposes. - if channel_tx - .send(Err(tonic::Status::unknown(e.to_string()))) - .await - .is_err() - { - warn!(%e, "GetBlockRangeStream closed unexpectedly"); - } - } - } - } - }, - ) - .await; - - if timeout_result.is_err() { - channel_tx - .send(Err(tonic::Status::deadline_exceeded( - "Error: get_block_range gRPC request timed out.", - ))) - .await - .ok(); - } - }); - - Ok(CompactBlockStream::new(channel_rx)) + self.spawn_block_range_stream( + &request, + "GetBlockRangeNullifiers", + compact_block_to_nullifiers, + ) + .await } /// Return the requested full (not compact) transaction (as from zcashd) diff --git a/packages/zaino-state/src/stream.rs b/packages/zaino-state/src/stream.rs index 7816f47fd..50e46abc8 100644 --- a/packages/zaino-state/src/stream.rs +++ b/packages/zaino-state/src/stream.rs @@ -6,191 +6,46 @@ use zaino_proto::proto::{ service::{Address, GetAddressUtxosReply, RawTransaction, SubtreeRoot}, }; -/// Stream of RawTransactions, output type of get_taddress_txids. +/// A stream of `Result` items read from a tokio mpsc receiver. #[derive(Debug)] -pub struct RawTransactionStream { - inner: ReceiverStream>, +pub struct ChannelStream { + inner: ReceiverStream>, } -impl RawTransactionStream { - /// Returns new instance of RawTransactionStream. - pub fn new(rx: tokio::sync::mpsc::Receiver>) -> Self { - RawTransactionStream { +impl ChannelStream { + /// Wraps the receiving half of an mpsc channel as a stream. + pub fn new(rx: tokio::sync::mpsc::Receiver>) -> Self { + ChannelStream { inner: ReceiverStream::new(rx), } } } -impl futures::Stream for RawTransactionStream { - type Item = Result; +impl futures::Stream for ChannelStream { + type Item = Result; fn poll_next( mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> std::task::Poll> { - let poll = std::pin::Pin::new(&mut self.inner).poll_next(cx); - match poll { - std::task::Poll::Ready(Some(Ok(raw_tx))) => std::task::Poll::Ready(Some(Ok(raw_tx))), - std::task::Poll::Ready(Some(Err(e))) => std::task::Poll::Ready(Some(Err(e))), - std::task::Poll::Ready(None) => std::task::Poll::Ready(None), - std::task::Poll::Pending => std::task::Poll::Pending, - } - } -} - -/// Stream of RawTransactions, output type of get_taddress_txids. -pub struct CompactTransactionStream { - inner: ReceiverStream>, -} - -impl CompactTransactionStream { - /// Returns new instance of RawTransactionStream. - pub fn new(rx: tokio::sync::mpsc::Receiver>) -> Self { - CompactTransactionStream { - inner: ReceiverStream::new(rx), - } - } -} - -impl futures::Stream for CompactTransactionStream { - type Item = Result; - - fn poll_next( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - let poll = std::pin::Pin::new(&mut self.inner).poll_next(cx); - match poll { - std::task::Poll::Ready(Some(Ok(raw_tx))) => std::task::Poll::Ready(Some(Ok(raw_tx))), - std::task::Poll::Ready(Some(Err(e))) => std::task::Poll::Ready(Some(Err(e))), - std::task::Poll::Ready(None) => std::task::Poll::Ready(None), - std::task::Poll::Pending => std::task::Poll::Pending, - } - } -} - -/// Stream of CompactBlocks, output type of get_block_range. -pub struct CompactBlockStream { - inner: ReceiverStream>, -} - -impl CompactBlockStream { - /// Returns new instance of CompactBlockStream. - pub fn new(rx: tokio::sync::mpsc::Receiver>) -> Self { - CompactBlockStream { - inner: ReceiverStream::new(rx), - } + std::pin::Pin::new(&mut self.inner).poll_next(cx) } } -impl futures::Stream for CompactBlockStream { - type Item = Result; +/// Stream of `RawTransaction` items, output type of get_taddress_txids. +pub type RawTransactionStream = ChannelStream; - fn poll_next( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - let poll = std::pin::Pin::new(&mut self.inner).poll_next(cx); - match poll { - std::task::Poll::Ready(Some(Ok(raw_tx))) => std::task::Poll::Ready(Some(Ok(raw_tx))), - std::task::Poll::Ready(Some(Err(e))) => std::task::Poll::Ready(Some(Err(e))), - std::task::Poll::Ready(None) => std::task::Poll::Ready(None), - std::task::Poll::Pending => std::task::Poll::Pending, - } - } -} +/// Stream of `CompactTx` items, output type of get_mempool_tx. +pub type CompactTransactionStream = ChannelStream; -/// Stream of CompactBlocks, output type of get_block_range. -pub struct UtxoReplyStream { - inner: ReceiverStream>, -} +/// Stream of `CompactBlock` items, output type of get_block_range. +pub type CompactBlockStream = ChannelStream; -impl UtxoReplyStream { - /// Returns new instance of CompactBlockStream. - pub fn new( - rx: tokio::sync::mpsc::Receiver>, - ) -> Self { - UtxoReplyStream { - inner: ReceiverStream::new(rx), - } - } -} +/// Stream of `GetAddressUtxosReply` items, output type of get_address_utxos_stream. +pub type UtxoReplyStream = ChannelStream; -impl futures::Stream for UtxoReplyStream { - type Item = Result; +/// Stream of `SubtreeRoot` items, output type of get_subtree_roots. +pub type SubtreeRootReplyStream = ChannelStream; - fn poll_next( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - let poll = std::pin::Pin::new(&mut self.inner).poll_next(cx); - match poll { - std::task::Poll::Ready(Some(Ok(raw_tx))) => std::task::Poll::Ready(Some(Ok(raw_tx))), - std::task::Poll::Ready(Some(Err(e))) => std::task::Poll::Ready(Some(Err(e))), - std::task::Poll::Ready(None) => std::task::Poll::Ready(None), - std::task::Poll::Pending => std::task::Poll::Pending, - } - } -} - -/// Stream of CompactBlocks, output type of get_block_range. -pub struct SubtreeRootReplyStream { - inner: ReceiverStream>, -} - -impl SubtreeRootReplyStream { - /// Returns new instance of CompactBlockStream. - pub fn new(rx: tokio::sync::mpsc::Receiver>) -> Self { - SubtreeRootReplyStream { - inner: ReceiverStream::new(rx), - } - } -} - -impl futures::Stream for SubtreeRootReplyStream { - type Item = Result; - - fn poll_next( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - let poll = std::pin::Pin::new(&mut self.inner).poll_next(cx); - match poll { - std::task::Poll::Ready(Some(Ok(raw_tx))) => std::task::Poll::Ready(Some(Ok(raw_tx))), - std::task::Poll::Ready(Some(Err(e))) => std::task::Poll::Ready(Some(Err(e))), - std::task::Poll::Ready(None) => std::task::Poll::Ready(None), - std::task::Poll::Pending => std::task::Poll::Pending, - } - } -} - -/// Stream of `Address`, input type for `get_taddress_balance_stream`. -pub struct AddressStream { - inner: ReceiverStream>, -} - -impl AddressStream { - /// Creates a new `AddressStream` instance. - pub fn new(rx: tokio::sync::mpsc::Receiver>) -> Self { - AddressStream { - inner: ReceiverStream::new(rx), - } - } -} - -impl futures::Stream for AddressStream { - type Item = Result; - - fn poll_next( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - let poll = std::pin::Pin::new(&mut self.inner).poll_next(cx); - match poll { - std::task::Poll::Ready(Some(Ok(address))) => std::task::Poll::Ready(Some(Ok(address))), - std::task::Poll::Ready(Some(Err(e))) => std::task::Poll::Ready(Some(Err(e))), - std::task::Poll::Ready(None) => std::task::Poll::Ready(None), - std::task::Poll::Pending => std::task::Poll::Pending, - } - } -} +/// Stream of `Address` items, input type for get_taddress_balance_stream. +pub type AddressStream = ChannelStream
; diff --git a/tools/makefiles/lints.toml b/tools/makefiles/lints.toml index e6824f01f..df79bc096 100644 --- a/tools/makefiles/lints.toml +++ b/tools/makefiles/lints.toml @@ -92,6 +92,11 @@ description = "Fail if any Dockerfile installs a libssl*/openssl* apt package. T command = "cargo" args = ["run", "-q", "--manifest-path", "tools/workbench/Cargo.toml", "--bin", "check-no-openssl-apt"] +[tasks.lint-code-duplication] +description = "Fail on any exact or near duplicate Rust logic (AST-normalized, min 50 nodes) in packages/ (minus generated zaino-proto and in-crate tests/), live-tests/, and tools/. Escape hatch: annotated entries in .dupes-ignore.toml." +command = "cargo" +args = ["run", "-q", "--manifest-path", "tools/workbench/Cargo.toml", "--bin", "check-code-duplication"] + [tasks.deny] description = "Enforce deny.toml's crate bans (openssl/openssl-sys/boring*). Scoped to `bans`: advisories/licenses have pre-existing failures (e.g. RUSTSEC-2022-0001 lmdb) tracked separately." command = "cargo" @@ -113,6 +118,7 @@ dependencies = [ "lint-toolchain-pin", "lint-crypto-provider", "lint-no-openssl-apt", + "lint-code-duplication", "deny", "lint-shell", ] diff --git a/tools/workbench/Cargo.lock b/tools/workbench/Cargo.lock index 853b2d198..fc020034c 100644 --- a/tools/workbench/Cargo.lock +++ b/tools/workbench/Cargo.lock @@ -2,6 +2,274 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "dupes-core" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "708adfad616d4800a92d6997700d39a82b73bbb85e6ed77701417caea94bad7b" +dependencies = [ + "serde", + "serde_json", + "thiserror", + "toml", + "walkdir", +] + +[[package]] +name = "dupes-rust" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2335d2c56a41c539c58e7203bfb3f94edc8ee0ecc5d6ee8c88df0bcb774de70" +dependencies = [ + "dupes-core", + "proc-macro2", + "syn", +] + +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + +[[package]] +name = "indexmap" +version = "2.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +dependencies = [ + "equivalent", + "hashbrown", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "serde_spanned" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3" +dependencies = [ + "serde", +] + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thiserror" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "toml" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362" +dependencies = [ + "serde", + "serde_spanned", + "toml_datetime", + "toml_edit", +] + +[[package]] +name = "toml_datetime" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" +dependencies = [ + "serde", +] + +[[package]] +name = "toml_edit" +version = "0.22.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" +dependencies = [ + "indexmap", + "serde", + "serde_spanned", + "toml_datetime", + "toml_write", + "winnow", +] + +[[package]] +name = "toml_write" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "winnow" +version = "0.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945" +dependencies = [ + "memchr", +] + [[package]] name = "workbench" version = "0.0.0" +dependencies = [ + "dupes-core", + "dupes-rust", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/tools/workbench/Cargo.toml b/tools/workbench/Cargo.toml index 8f33524ed..660faba5c 100644 --- a/tools/workbench/Cargo.toml +++ b/tools/workbench/Cargo.toml @@ -13,3 +13,7 @@ publish = false # Crate-wide: forbid unsafe across the lib and every binary in one place. [lints.rust] unsafe_code = "forbid" + +[dependencies] +dupes-core = "=0.2.1" +dupes-rust = "=0.2.1" diff --git a/tools/workbench/src/bin/check-code-duplication.rs b/tools/workbench/src/bin/check-code-duplication.rs new file mode 100644 index 000000000..b231b4008 --- /dev/null +++ b/tools/workbench/src/bin/check-code-duplication.rs @@ -0,0 +1,145 @@ +//! Guard: no duplicate Rust logic may enter the tree. +//! +//! Runs cargo-dupes' AST-normalizing analyzer (identifiers become positional +//! placeholders and literals are erased, so renamed copies still match) over the +//! first-party Rust scopes and fails if any exact or near duplicate group remains: +//! +//! - `packages/` — the production crates, excluding `zaino-proto` (tonic/prost +//! generated code) and in-crate `tests/` module directories. +//! - `live-tests/` — the e2e and clientless harness crates, including their +//! integration-test helper code. +//! - `tools/` — the developer tooling. +//! +//! `#[test]` functions and `#[cfg(test)]` modules visible per-file are excluded in +//! every scope. The tree was deduplicated to zero groups when this gate landed, so +//! both thresholds are zero: any new group is a regression. +//! +//! `MIN_NODES` was calibrated empirically against this tree on 2026-07-15: 50 is +//! the floor at which every reported group was a genuine dedup-worthy twin. Below +//! it the analyzer reports structurally-rhyming but semantically unrelated small +//! functions (field-copy constructors, tiny delegating getters), whose "dedup" +//! would couple unrelated types. +//! +//! Escape hatch: a genuinely irreducible group may be granted an annotated entry +//! in `.dupes-ignore.toml` at the repo root — run +//! `code-dupes ignore --reason "..."` (crates.io package +//! `code-dupes`, the CLI over the same libraries). Prefer deduplicating; entries +//! are reviewed like code. + +use std::path::{Path, PathBuf}; + +use dupes_core::config::Config; +use dupes_core::scanner::{scan_files, ScanConfig}; +use dupes_rust::RustAnalyzer; +use workbench::{repo_root, run}; + +/// One scanned scope: a repo-relative directory and its excluded path substrings +/// (matched against repo-relative paths, so the checkout location cannot affect +/// the result). +struct Scope { + path: &'static str, + excludes: &'static [&'static str], +} + +const SCOPES: &[Scope] = &[ + Scope { + path: "packages", + excludes: &["zaino-proto", "/tests/"], + }, + Scope { + path: "live-tests", + excludes: &[], + }, + Scope { + path: "tools", + excludes: &[], + }, +]; + +/// Minimum AST-node count for a code unit to participate in duplicate analysis. +/// See the module docs for the calibration rationale. +const MIN_NODES: usize = 50; + +fn main() { + run("check-code-duplication", check, |n| { + println!("check-code-duplication: ok — no duplicate groups across {n} Rust file(s)"); + }) +} + +fn check() -> Result> { + let root = repo_root()?; + let analyzer = RustAnalyzer; + let mut scanned_files = 0usize; + let mut offenders: Vec = Vec::new(); + + for scope in SCOPES { + let files: Vec = scan_files(&ScanConfig::new(root.join(scope.path))) + .into_iter() + .filter(|path| !is_excluded(path, &root, scope.excludes)) + .collect(); + scanned_files += files.len(); + + let config = Config { + min_nodes: MIN_NODES, + exclude_tests: true, + // The ignore file (`.dupes-ignore.toml`) is discovered at this root. + root: root.clone(), + ..Config::default() + }; + + let result = dupes_core::analyze(&analyzer, &files, &config) + .map_err(|e| vec![format!("analysis of {} failed: {e}", scope.path)])?; + for warning in &result.warnings { + eprintln!( + "check-code-duplication: warning ({}): {warning}", + scope.path + ); + } + + for (label, groups) in [ + ("exact", &result.exact_groups), + ("near", &result.near_groups), + ] { + for group in groups { + offenders.push(format!( + "{label} duplicate group in {} (fingerprint {}, similarity {:.0}%):", + scope.path, + group.fingerprint, + group.similarity * 100.0 + )); + for member in &group.members { + let file = member.file.strip_prefix(&root).unwrap_or(&member.file); + offenders.push(format!( + " - {} ({}) at {}:{}-{}", + member.name, + member.kind, + file.display(), + member.line_start, + member.line_end + )); + } + } + } + } + + if offenders.is_empty() { + return Ok(scanned_files); + } + let mut msg = vec![ + "duplicate Rust logic found — deduplicate it (prefer a plain fn; see CLAUDE.md §DRY):" + .to_string(), + ]; + msg.extend(offenders); + msg.push( + "if a group is genuinely irreducible, add an annotated entry to .dupes-ignore.toml \ + (`code-dupes ignore --reason \"...\"`)." + .to_string(), + ); + Err(msg) +} + +/// Whether `path`, taken relative to the repo root, matches any exclude substring. +fn is_excluded(path: &Path, root: &Path, excludes: &[&str]) -> bool { + let relative = path.strip_prefix(root).unwrap_or(path).to_string_lossy(); + excludes.iter().any(|pattern| relative.contains(pattern)) +}