From 2b6d1345f6c845c677786a057fd8125cbdd3df5b Mon Sep 17 00:00:00 2001 From: Enrico Rubboli Date: Tue, 13 Oct 2026 11:02:00 +0200 Subject: [PATCH 1/3] fix(scanner): decrement Staked by the full balance before zeroing it Stake-pool spends zeroed the stored pool record before reading the pledge for the statistic decrement, so the Staked statistic was never decremented when a pool was closed. Capture the pool's full stake balance (pledge plus accumulated staker rewards) before decommissioning and decrement with that, matching what creation and block rewards add. Both spend paths are covered: pool outputs spent as regular transaction inputs and pool outputs spent as block-reward outputs. The same fix is mirrored in the sync simulation's statistics model. Regression test covers decommissioning all three spend shapes with the statistic checked after every block, including a delegation that outlives its pool. --- .../scanner-lib/src/blockchain_state/mod.rs | 210 +++--- api-server/scanner-lib/src/sync/tests/mod.rs | 613 +++++++++++++++++- .../scanner-lib/src/sync/tests/simulation.rs | 79 ++- 3 files changed, 778 insertions(+), 124 deletions(-) diff --git a/api-server/scanner-lib/src/blockchain_state/mod.rs b/api-server/scanner-lib/src/blockchain_state/mod.rs index 24bfabe304..318aefc1df 100644 --- a/api-server/scanner-lib/src/blockchain_state/mod.rs +++ b/api-server/scanner-lib/src/blockchain_state/mod.rs @@ -103,8 +103,7 @@ impl BlockchainState { genesis.utxos(), genesis.get_id().into(), ) - .await - .expect("Unable to update tables from block reward"); + .await?; db_tx.commit().await.expect("Unable to commit transaction"); logging::log::info!("Database commit completed successfully"); @@ -217,8 +216,7 @@ impl LocalBlockchainState for BlockchainState tx, next_order_number + idx as u64, ) - .await - .expect("Unable to update tables from transaction"); + .await?; let tx_info = TransactionInfo { tx: tx.clone(), @@ -277,8 +275,7 @@ impl LocalBlockchainState for BlockchainState &block, block_fees, ) - .await - .expect("Unable to update tables from block"); + .await?; } // Note: sent within the transaction, so the notification is only delivered if the @@ -465,8 +462,7 @@ async fn update_tables_from_block( block.block_reward().outputs(), block.get_id().into(), ) - .await - .expect("Unable to update tables from block reward"); + .await?; update_tables_from_consensus_data( chain_config.clone(), @@ -475,8 +471,7 @@ async fn update_tables_from_block( block, total_tx_fees, ) - .await - .expect("Unable to update tables from consensus data"); + .await?; Ok(()) } @@ -535,7 +530,7 @@ async fn update_tables_from_block_reward( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; increase_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -543,7 +538,7 @@ async fn update_tables_from_block_reward( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; increase_statistic_amount( db_tx, CoinOrTokenStatistic::Staked, @@ -551,7 +546,7 @@ async fn update_tables_from_block_reward( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; } TxOutput::Transfer(output_value, destination) | TxOutput::LockThenTransfer(output_value, destination, _) => { @@ -575,7 +570,7 @@ async fn update_tables_from_block_reward( CoinOrTokenId::TokenId(*token_id), block_height, ) - .await; + .await?; increase_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -583,7 +578,7 @@ async fn update_tables_from_block_reward( CoinOrTokenId::TokenId(*token_id), block_height, ) - .await; + .await?; Some(token_decimals(*token_id, &BTreeMap::new(), db_tx).await?.1) } OutputValue::Coin(amount) => { @@ -602,7 +597,7 @@ async fn update_tables_from_block_reward( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; increase_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -610,7 +605,7 @@ async fn update_tables_from_block_reward( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; None } }; @@ -1055,7 +1050,7 @@ async fn update_tables_from_consensus_data( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; increase_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1063,7 +1058,7 @@ async fn update_tables_from_consensus_data( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; for (delegation_id, rewards) in adapter.rewards_per_delegation() { let delegation = delegation_shares.get(delegation_id).expect("must exist").clone(); @@ -1108,8 +1103,7 @@ async fn update_tables_from_transaction( transaction, tx_global_index, ) - .await - .expect("Unable to update tables from transaction inputs"); + .await?; update_tables_from_transaction_outputs( Arc::clone(&chain_config), @@ -1119,8 +1113,7 @@ async fn update_tables_from_transaction( transaction.transaction(), tx_global_index, ) - .await - .expect("Unable to update tables from transaction outputs"); + .await?; Ok(()) } @@ -1169,7 +1162,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::TokenId(*token_id), block_height, ) - .await; + .await?; let amount = chain_config.token_supply_change_fee(block_height); increase_statistic_amount( db_tx, @@ -1178,7 +1171,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1186,7 +1179,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; transaction_tokens.insert(*token_id); } AccountCommand::UnmintTokens(token_id) => { @@ -1206,7 +1199,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1214,7 +1207,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; transaction_tokens.insert(*token_id); } AccountCommand::FreezeToken(token_id, is_unfreezable) => { @@ -1231,7 +1224,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1239,7 +1232,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; transaction_tokens.insert(*token_id); } AccountCommand::UnfreezeToken(token_id) => { @@ -1256,7 +1249,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1264,7 +1257,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; transaction_tokens.insert(*token_id); } AccountCommand::LockTokenSupply(token_id) => { @@ -1281,7 +1274,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1289,7 +1282,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; transaction_tokens.insert(*token_id); } AccountCommand::ChangeTokenAuthority(token_id, destination) => { @@ -1306,7 +1299,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1314,7 +1307,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; transaction_tokens.insert(*token_id); } AccountCommand::ChangeTokenMetadataUri(token_id, metadata_uri) => { @@ -1331,7 +1324,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1339,7 +1332,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; transaction_tokens.insert(*token_id); } AccountCommand::FillOrder(order_id, fill_amount_in_ask_currency, _) => { @@ -1422,7 +1415,7 @@ async fn update_tables_from_transaction_inputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; } } } @@ -1453,16 +1446,12 @@ async fn update_tables_from_transaction_inputs( | TxOutput::CreateOrder(_) => {} TxOutput::CreateStakePool(pool_id, _) | TxOutput::ProduceBlockFromStake(_, pool_id) => { - let pool_data = db_tx - .get_pool_data(pool_id) - .await? - .expect("pool data should exist") - .decommission_pool(); - - db_tx - .set_pool_data_at_height(pool_id, &pool_data, block_height) - .await - .expect("unable to update pool data"); + let pool_data = decommission_pool_and_decrement_staked( + db_tx, + pool_id, + block_height, + ) + .await?; let address = Address::::new( &chain_config, @@ -1474,14 +1463,6 @@ async fn update_tables_from_transaction_inputs( .entry(address.clone()) .or_default() .insert(tx.get_id()); - decrease_statistic_amount( - db_tx, - CoinOrTokenStatistic::Staked, - &pool_data.pledge_amount(), - CoinOrTokenId::Coin, - block_height, - ) - .await; } } } @@ -1507,24 +1488,8 @@ async fn update_tables_from_transaction_inputs( | TxOutput::CreateOrder(_) => {} TxOutput::CreateStakePool(pool_id, _) | TxOutput::ProduceBlockFromStake(_, pool_id) => { - let pool_data = db_tx - .get_pool_data(pool_id) - .await? - .expect("pool data should exist") - .decommission_pool(); - - db_tx - .set_pool_data_at_height(pool_id, &pool_data, block_height) - .await - .expect("unable to update pool data"); - decrease_statistic_amount( - db_tx, - CoinOrTokenStatistic::Staked, - &pool_data.pledge_amount(), - CoinOrTokenId::Coin, - block_height, - ) - .await; + decommission_pool_and_decrement_staked(db_tx, pool_id, block_height) + .await?; } TxOutput::IssueNft(token_id, _, destination) => { let address = Address::::new(&chain_config, destination) @@ -1682,7 +1647,7 @@ async fn update_tables_from_transaction_outputs( coin_or_token_id, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1690,7 +1655,7 @@ async fn update_tables_from_transaction_outputs( coin_or_token_id, block_height, ) - .await; + .await?; } TxOutput::DataDeposit(_) => { let amount = chain_config.data_deposit_fee(block_height); @@ -1701,7 +1666,7 @@ async fn update_tables_from_transaction_outputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1709,7 +1674,7 @@ async fn update_tables_from_transaction_outputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; } TxOutput::IssueFungibleToken(issuance) => { let token_id = make_token_id(chain_config.as_ref(), block_height, inputs)?; @@ -1735,7 +1700,7 @@ async fn update_tables_from_transaction_outputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1743,7 +1708,7 @@ async fn update_tables_from_transaction_outputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; transaction_tokens.insert(token_id); } TxOutput::IssueNft(token_id, issuance, destination) => { @@ -1771,7 +1736,7 @@ async fn update_tables_from_transaction_outputs( CoinOrTokenId::TokenId(*token_id), block_height, ) - .await; + .await?; let amount = chain_config.nft_issuance_fee(block_height); increase_statistic_amount( db_tx, @@ -1780,7 +1745,7 @@ async fn update_tables_from_transaction_outputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; decrease_statistic_amount( db_tx, CoinOrTokenStatistic::CirculatingSupply, @@ -1788,7 +1753,7 @@ async fn update_tables_from_transaction_outputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; set_utxo( outpoint, @@ -1835,7 +1800,7 @@ async fn update_tables_from_transaction_outputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; set_utxo( outpoint, output, @@ -1880,7 +1845,7 @@ async fn update_tables_from_transaction_outputs( CoinOrTokenId::Coin, block_height, ) - .await; + .await?; let pool_id = *new_delegation.pool_id(); let pool_data = db_tx @@ -2127,19 +2092,54 @@ async fn increase_statistic_amount( amount: &Amount, coin_or_token_id: CoinOrTokenId, block_height: BlockHeight, -) { - let current_balance = db_tx - .get_statistic(statistic, coin_or_token_id) - .await - .expect("Unable to get statistic") - .unwrap_or(Amount::ZERO); +) -> Result<(), ApiServerStorageError> { + let current_balance = + db_tx.get_statistic(statistic, coin_or_token_id).await?.unwrap_or(Amount::ZERO); - let new_amount = current_balance.add(*amount).expect("Balance should not overflow"); + let new_amount = current_balance.add(*amount).ok_or_else(|| { + ApiServerStorageError::LowLevelStorageError(format!( + "statistic {statistic:?} overflow: current balance {current_balance:?}, \ + increment {amount:?}", + )) + })?; db_tx .set_statistic(statistic, coin_or_token_id, block_height, new_amount) - .await - .expect("Unable to update statistic") + .await?; + Ok(()) +} + +/// Decommissions a stake pool and removes its whole balance (pledge plus +/// accumulated staker rewards, matching what the creation and block-reward +/// paths added to the `Staked` statistic) from it. The balance is captured +/// before `decommission_pool` zeroes the record; the zeroed record is stored +/// and returned so callers can read the decommission destination. +async fn decommission_pool_and_decrement_staked( + db_tx: &mut T, + pool_id: PoolId, + block_height: BlockHeight, +) -> Result { + let pool_data = db_tx.get_pool_data(pool_id).await?.ok_or_else(|| { + ApiServerStorageError::LowLevelStorageError(format!("pool data for {pool_id} should exist")) + })?; + let staker_balance = pool_data.staker_balance().map_err(|error| { + ApiServerStorageError::LowLevelStorageError(format!( + "staker balance overflow for pool {pool_id}: {error}", + )) + })?; + let pool_data = pool_data.decommission_pool(); + + db_tx.set_pool_data_at_height(pool_id, &pool_data, block_height).await?; + + decrease_statistic_amount( + db_tx, + CoinOrTokenStatistic::Staked, + &staker_balance, + CoinOrTokenId::Coin, + block_height, + ) + .await?; + Ok(pool_data) } async fn decrease_statistic_amount( @@ -2148,19 +2148,21 @@ async fn decrease_statistic_amount( amount: &Amount, coin_or_token_id: CoinOrTokenId, block_height: BlockHeight, -) { - let current_balance = db_tx - .get_statistic(statistic, coin_or_token_id) - .await - .expect("Unable to get statistic") - .unwrap_or(Amount::ZERO); +) -> Result<(), ApiServerStorageError> { + let current_balance = + db_tx.get_statistic(statistic, coin_or_token_id).await?.unwrap_or(Amount::ZERO); - let new_amount = current_balance.sub(*amount).expect("Balance should not underflow"); + let new_amount = current_balance.sub(*amount).ok_or_else(|| { + ApiServerStorageError::LowLevelStorageError(format!( + "statistic {statistic:?} underflow: current balance {current_balance:?}, \ + decrement {amount:?}", + )) + })?; db_tx .set_statistic(statistic, coin_or_token_id, block_height, new_amount) - .await - .expect("Unable to update statistic") + .await?; + Ok(()) } async fn increase_address_amount( diff --git a/api-server/scanner-lib/src/sync/tests/mod.rs b/api-server/scanner-lib/src/sync/tests/mod.rs index 2ffac40116..22b01e032d 100644 --- a/api-server/scanner-lib/src/sync/tests/mod.rs +++ b/api-server/scanner-lib/src/sync/tests/mod.rs @@ -38,9 +38,9 @@ use chainstate_test_framework::{TestFramework, TransactionBuilder}; use common::{ address::Address, chain::{ - AccountCommand, AccountNonce, CoinUnit, Destination, OrderAccountCommand, OrderData, - OrderId, OutPointSourceId, PoolId, SignedTransaction, Transaction, TxInput, TxOutput, - UtxoOutPoint, + AccountCommand, AccountNonce, AccountOutPoint, AccountSpending, CoinUnit, Destination, + OrderAccountCommand, OrderData, OrderId, OutPointSourceId, PoolId, SignedTransaction, + Transaction, TxInput, TxOutput, UtxoOutPoint, htlc::{HashedTimelockContract, HtlcSecret}, make_delegation_id, make_order_id, make_token_id, output_value::OutputValue, @@ -66,8 +66,8 @@ use common::{ primitives::{Amount, CoinOrTokenId, H256, Idable, per_thousand::PerThousand}, }; use crypto::{ - key::{KeyKind, PrivateKey}, - vrf::{VRFKeyKind, VRFPrivateKey}, + key::{KeyKind, PrivateKey, PublicKey}, + vrf::{VRFKeyKind, VRFPrivateKey, VRFPublicKey}, }; use logging::log; use mempool::FeeRate; @@ -649,6 +649,609 @@ async fn compare_pool_rewards_with_chainstate_real_state(#[case] seed: Seed) { assert_eq!(decommissioned_pool.staker_balance().unwrap(), Amount::ZERO); } +/// Reads the `Staked` statistic from the scanner state. +async fn staked_statistic( + local_state: &BlockchainState, +) -> Amount { + let stats = local_state + .storage() + .transaction_ro() + .await + .unwrap() + .get_all_statistic(CoinOrTokenId::Coin) + .await + .unwrap(); + statistic_value(&stats, CoinOrTokenStatistic::Staked) +} + +fn pool_staker_balance(tf: &TestFramework, pool_id: PoolId) -> Amount { + tf.chainstate + .get_stake_pool_data(&pool_id) + .unwrap() + .expect("pool must exist") + .staker_balance() + .unwrap() +} + +/// The pledge and staker balance of a decommissioned pool's record must be zeroed. +/// (Other fields, notably `delegations_balance`, legitimately survive decommissioning; +/// see `PoolDataWithExtraInfo::decommission_pool`.) +async fn assert_pool_record_zeroed( + local_state: &BlockchainState, + pool_id: PoolId, + expected_delegations_balance: Amount, +) { + let pool_data = local_state + .storage() + .transaction_ro() + .await + .unwrap() + .get_pool_data(pool_id) + .await + .unwrap() + .expect("decommissioned pool record must still exist"); + assert_eq!(pool_data.pledge_amount(), Amount::ZERO); + assert_eq!(pool_data.staker_balance().unwrap(), Amount::ZERO); + // delegations legitimately survive decommissioning; assert they are intact + assert_eq!(pool_data.delegations_balance, expected_delegations_balance); +} + +fn new_pool_data( + rng: &mut impl CryptoRng, + staking_key: &Destination, + vrf_pk: VRFPublicKey, + pledge_amount: Amount, +) -> StakePoolData { + // the test framework produces blocks with a single shared VRF key, so every + // pool must be registered with that same key + StakePoolData::new( + pledge_amount, + staking_key.clone(), + vrf_pk, + staking_key.clone(), + PerThousand::new_from_rng(rng), + Amount::ZERO, + ) +} + +/// Signs the single stake-spend input of a decommissioning transaction. +#[allow(clippy::too_many_arguments)] +// The builder output carries placeholder `NoSignature` witnesses; input 0 is +// signed here. +fn sign_pool_spend_transaction( + tx: SignedTransaction, + spent_output: TxOutput, + pool_id: PoolId, + staker_balance: Amount, + chain_config: &ChainConfig, + next_height: BlockHeight, + staking_sk: &PrivateKey, + staking_pk: &PublicKey, + rng: &mut impl CryptoRng, +) -> SignedTransaction { + assert_eq!( + tx.transaction().inputs().len(), + 1, + "this helper only signs the single stake-spend input of a decommissioning transaction" + ); + let transaction = tx.take_transaction(); + let input = std::slice::from_ref(&transaction.inputs()[0]); + let spent_utxos = [Some(spent_output)]; + let input_commitments = make_sighash_input_commitments_for_transaction_inputs_at_height( + input, + &TrivialUtxoProvider(&spent_utxos), + &BTreeMap::::from([(pool_id, PoolInfo { staker_balance })]), + &BTreeMap::::new(), + chain_config, + next_height, + ) + .unwrap(); + let sighash = + signature_hash(SigHashType::default(), &transaction, &input_commitments, 0).unwrap(); + let signature = sign_public_key_spending(staking_sk, staking_pk, &sighash, rng).unwrap(); + let mut witnesses = vec![InputWitness::NoSignature(None); transaction.inputs().len()]; + witnesses[0] = InputWitness::Standard(StandardInputSignature::new( + SigHashType::default(), + signature.encode(), + )); + SignedTransaction::new(transaction, witnesses).unwrap() +} + +/// The payout of every stake-spend: the whole balance, locked for the +/// stake-spend maturity so the value-accumulator's constraint check passes. +fn stake_spend_payout_output( + chain_config: &ChainConfig, + tf: &TestFramework, + payout: Amount, +) -> TxOutput { + let maturity = chain_config + .staking_pool_spend_maturity_block_count(tf.next_block_height()) + .to_int(); + TxOutput::LockThenTransfer( + OutputValue::Coin(payout), + Destination::AnyoneCanSpend, + OutputTimeLock::ForBlockCount(maturity), + ) +} + +/// Builds and signs a decommissioning transaction that spends a pool output and +/// pays out to `AnyoneCanSpend` under the stake-spend maturity lock. +#[allow(clippy::too_many_arguments)] +fn decommission_spend_transaction( + chain_config: &ChainConfig, + tf: &TestFramework, + input_source: OutPointSourceId, + input_index: u32, + spent_output: TxOutput, + pool_id: PoolId, + payout: Amount, + staking_sk: &PrivateKey, + staking_pk: &PublicKey, + rng: &mut impl CryptoRng, +) -> SignedTransaction { + // `payout` doubles as the sighash `PoolInfo.staker_balance` when + // `spent_output` is a `ProduceBlockFromStake` (a `CreateStakePool` spend + // commits just the UTXO); it must equal the pool's live balance, which + // holds because a decommission spend always pays out the pool's full + // balance. + assert_eq!( + payout, + pool_staker_balance(tf, pool_id), + "payout must equal the pool's live staker balance, otherwise a ProduceBlockFromStake spend would be signed over a stale sighash commitment", + ); + let transaction = TransactionBuilder::new() + .add_input( + TxInput::from_utxo(input_source, input_index), + InputWitness::NoSignature(None), + ) + .add_output(stake_spend_payout_output(chain_config, tf, payout)) + .build(); + sign_pool_spend_transaction( + transaction, + spent_output, + pool_id, + payout, + chain_config, + tf.next_block_height(), + staking_sk, + staking_pk, + rng, + ) +} + +// Decommissioning a pool must decrement the `Staked` statistic by the pool's full +// stake balance (pledge plus accumulated staker rewards) whichever output type is +// spent, and a delegation spend must keep decrementing the statistic after the pool +// it belongs to is gone. +#[rstest] +#[trace] +#[case(test_utils::random::Seed::from_entropy())] +#[tokio::test] +async fn decommission_and_delegation_spends_update_staked_statistic(#[case] seed: Seed) { + let mut rng = make_seedable_rng(seed); + + let pledge = 40_000 * CoinUnit::ATOMS_PER_COIN; + // enough supply to fund three more pools above the 40_000 COIN minimum pledge + let mint = 400_000 * CoinUnit::ATOMS_PER_COIN; + let (staking_sk, staking_pk) = PrivateKey::new_from_rng(&mut rng, KeyKind::Secp256k1Schnorr); + let staking_key = Destination::PublicKey(staking_pk.clone()); + let (vrf_sk, vrf_pk) = VRFPrivateKey::new_from_rng(&mut rng, VRFKeyKind::Schnorrkel); + let pool_data = new_pool_data( + &mut rng, + &staking_key, + vrf_pk.clone(), + Amount::from_atoms(pledge), + ); + let pool_id = PoolId::new(H256::random_using(&mut rng)); + + let chain_config = chainstate_test_framework::create_chain_config_with_staking_pool( + &mut rng, + Amount::from_atoms(mint), + pool_id, + pool_data, + ) + .build(); + let mut tf = TestFramework::builder(&mut rng).with_chain_config(chain_config).build(); + + let chain_config = Arc::clone(tf.chainstate.get_chain_config()); + let storage = { + let mut storage = TransactionalApiServerInMemoryStorage::new(&chain_config); + + let mut db_tx = storage.transaction_rw().await.unwrap(); + db_tx.reinitialize_storage(&chain_config).await.unwrap(); + db_tx.commit().await.unwrap(); + + storage + }; + let mut local_state = BlockchainState::new(chain_config.clone(), storage); + local_state.scan_genesis(chain_config.genesis_block().as_ref()).await.unwrap(); + + let target_block_time = chain_config.target_block_spacing(); + + // Genesis: the pool pledge is the entire staked supply. Each checkpoint below + // reassigns this from the pool balances and the pledges and delegations that + // are alive at that point. + let mut expected_staked = Amount::from_atoms(pledge); + assert_eq!(staked_statistic(&local_state).await, expected_staked); + + // Block 1: the pool produces a block whose transaction pays a fee, so the pool + // earns staker rewards on top of its pledge. + let fee = Amount::from_atoms(100 * CoinUnit::ATOMS_PER_COIN); + let transaction = TransactionBuilder::new() + .add_input( + TxInput::from_utxo( + OutPointSourceId::BlockReward(chain_config.genesis_block_id()), + 0, + ), + InputWitness::NoSignature(None), + ) + .add_output(TxOutput::Transfer( + OutputValue::Coin( + (Amount::from_atoms(mint) - fee).expect("no underflow in test amounts"), + ), + Destination::AnyoneCanSpend, + )) + .build(); + let tx1_id = transaction.transaction().get_id(); + let block = create_block( + &mut rng, + &mut tf, + target_block_time, + chain_config.genesis_block_id(), + staking_sk.clone(), + vrf_sk.clone(), + pool_id, + vec![transaction], + ); + let block1_id = block.get_id(); + sync_and_compare(&mut tf, block, &mut local_state, pool_id).await; + + let pool_id_balance = pool_staker_balance(&tf, pool_id); + assert!( + pool_id_balance > Amount::from_atoms(pledge), + "the pool must have earned staker rewards for this test to be meaningful" + ); + expected_staked = pool_id_balance; + assert_eq!(staked_statistic(&local_state).await, expected_staked); + + // Blocks 2 to 4: three fresh pools are created, one transaction per block (a + // pool id is derived from the transaction input that pays for its creation). + // The pools are created one per block because value constraints are checked per + // block as a whole, so a chain of transactions inside a single block would count + // the intermediate outputs twice. + let pledge2 = 45_000 * CoinUnit::ATOMS_PER_COIN; + let pledge3 = 42_000 * CoinUnit::ATOMS_PER_COIN; + let pledge4 = 43_000 * CoinUnit::ATOMS_PER_COIN; + let amount2 = Amount::from_atoms(pledge2); + let amount3 = Amount::from_atoms(pledge3); + let amount4 = Amount::from_atoms(pledge4); + + let pool2_id = PoolId::from_utxo(&UtxoOutPoint::new(OutPointSourceId::Transaction(tx1_id), 0)); + let rest2 = (Amount::from_atoms(mint) - fee).expect("no underflow in test amounts"); + let rest2 = (rest2 - amount2).expect("no underflow in test amounts"); + let tx2a = TransactionBuilder::new() + .add_input( + TxInput::from_utxo(OutPointSourceId::Transaction(tx1_id), 0), + InputWitness::NoSignature(None), + ) + .add_output(TxOutput::Transfer( + OutputValue::Coin(rest2), + Destination::AnyoneCanSpend, + )) + .add_output(TxOutput::CreateStakePool( + pool2_id, + Box::new(new_pool_data( + &mut rng, + &staking_key, + vrf_pk.clone(), + amount2, + )), + )) + .build(); + let tx2a_id = tx2a.transaction().get_id(); + let pool2_creation_output = tx2a.transaction().outputs()[1].clone(); + + let block = create_block( + &mut rng, + &mut tf, + target_block_time, + block1_id.into(), + staking_sk.clone(), + vrf_sk.clone(), + pool_id, + vec![tx2a], + ); + let block2_id = block.get_id(); + sync_and_compare(&mut tf, block, &mut local_state, pool_id).await; + + let pool_id_balance = pool_staker_balance(&tf, pool_id); + expected_staked = (pool_id_balance + amount2).expect("no overflow in test amounts"); + assert_eq!(staked_statistic(&local_state).await, expected_staked); + + // Block 3: pool 3 is created. + let pool3_id = PoolId::from_utxo(&UtxoOutPoint::new( + OutPointSourceId::Transaction(tx2a_id), + 0, + )); + let rest3 = (rest2 - amount3).expect("no underflow in test amounts"); + let tx2b = TransactionBuilder::new() + .add_input( + TxInput::from_utxo(OutPointSourceId::Transaction(tx2a_id), 0), + InputWitness::NoSignature(None), + ) + .add_output(TxOutput::Transfer( + OutputValue::Coin(rest3), + Destination::AnyoneCanSpend, + )) + .add_output(TxOutput::CreateStakePool( + pool3_id, + Box::new(new_pool_data( + &mut rng, + &staking_key, + vrf_pk.clone(), + amount3, + )), + )) + .build(); + let tx2b_id = tx2b.transaction().get_id(); + let pool3_creation_output = tx2b.transaction().outputs()[1].clone(); + + let block = create_block( + &mut rng, + &mut tf, + target_block_time, + block2_id.into(), + staking_sk.clone(), + vrf_sk.clone(), + pool_id, + vec![tx2b], + ); + let block3_id = block.get_id(); + sync_and_compare(&mut tf, block, &mut local_state, pool_id).await; + + let pool_id_balance = pool_staker_balance(&tf, pool_id); + expected_staked = ((pool_id_balance + amount2).expect("no overflow in test amounts") + amount3) + .expect("no overflow in test amounts"); + assert_eq!(staked_statistic(&local_state).await, expected_staked); + + // Block 4: pool 4 is created. + let pool4_id = PoolId::from_utxo(&UtxoOutPoint::new( + OutPointSourceId::Transaction(tx2b_id), + 0, + )); + let rest4 = (rest3 - amount4).expect("no underflow in test amounts"); + let tx2c = TransactionBuilder::new() + .add_input( + TxInput::from_utxo(OutPointSourceId::Transaction(tx2b_id), 0), + InputWitness::NoSignature(None), + ) + .add_output(TxOutput::Transfer( + OutputValue::Coin(rest4), + Destination::AnyoneCanSpend, + )) + .add_output(TxOutput::CreateStakePool( + pool4_id, + Box::new(new_pool_data( + &mut rng, + &staking_key, + vrf_pk.clone(), + amount4, + )), + )) + .build(); + let tx2c_id = tx2c.transaction().get_id(); + + let block = create_block( + &mut rng, + &mut tf, + target_block_time, + block3_id.into(), + staking_sk.clone(), + vrf_sk.clone(), + pool_id, + vec![tx2c], + ); + let block4_id = block.get_id(); + sync_and_compare(&mut tf, block, &mut local_state, pool_id).await; + + let pool_id_balance = pool_staker_balance(&tf, pool_id); + expected_staked = [pool_id_balance, amount2, amount3, amount4] + .into_iter() + .sum::>() + .expect("no overflow in test amounts"); + assert_eq!(staked_statistic(&local_state).await, expected_staked); + + // Block 5: pool 2 is decommissioned by spending its creation output in a + // transaction (the `OutPointSourceId::Transaction` path). The payout must stay + // locked for the stake-spend maturity or the block violates timelock constraints. + let transaction = decommission_spend_transaction( + &chain_config, + &tf, + OutPointSourceId::Transaction(tx2a_id), + 1, + pool2_creation_output, + pool2_id, + amount2, + &staking_sk, + &staking_pk, + &mut rng, + ); + let block = create_block( + &mut rng, + &mut tf, + target_block_time, + block4_id.into(), + staking_sk.clone(), + vrf_sk.clone(), + pool_id, + vec![transaction], + ); + let block5_id = block.get_id(); + let block5_reward_out0 = block.block_reward().outputs()[0].clone(); + sync_and_compare(&mut tf, block, &mut local_state, pool_id).await; + + assert!( + tf.chainstate.get_stake_pool_data(&pool2_id).unwrap().is_none(), + "pool 2 must be gone from the chainstate" + ); + assert_pool_record_zeroed(&local_state, pool2_id, Amount::ZERO).await; + let pool_id_balance = pool_staker_balance(&tf, pool_id); + expected_staked = ((pool_id_balance + amount3).expect("no overflow in test amounts") + amount4) + .expect("no overflow in test amounts"); + assert_eq!(staked_statistic(&local_state).await, expected_staked); + + // Block 6: the first pool is decommissioned by spending the + // `ProduceBlockFromStake` output of the block it last produced (the + // `OutPointSourceId::BlockReward` path), which is worth its full stake balance. + // Pool 4 produces this block because the spent output is the first pool's latest + // block reward. + let pool_id_balance = pool_staker_balance(&tf, pool_id); + let transaction = TransactionBuilder::new() + .add_input( + TxInput::from_utxo(OutPointSourceId::BlockReward(block5_id.into()), 0), + InputWitness::NoSignature(None), + ) + .add_output(stake_spend_payout_output( + &chain_config, + &tf, + pool_id_balance, + )) + .build(); + let transaction = sign_pool_spend_transaction( + transaction, + block5_reward_out0, + pool_id, + pool_id_balance, + &chain_config, + tf.next_block_height(), + &staking_sk, + &staking_pk, + &mut rng, + ); + // The kernel outpoint is pinned to pool 4's creation output: the default + // resolution would pick the parent block's reward output, which is the exact + // output this block's transaction spends. + let block = tf + .make_pos_block_builder() + .with_parent(block5_id.into()) + .with_stake_spending_key(staking_sk.clone()) + .with_vrf_key(vrf_sk.clone()) + .with_stake_pool_id(pool4_id) + .with_kernel_input(UtxoOutPoint::new(OutPointSourceId::Transaction(tx2c_id), 1)) + .with_transactions(vec![transaction]) + .build(&mut rng); + let block6_id = block.get_id(); + sync_and_compare(&mut tf, block, &mut local_state, pool4_id).await; + + assert!( + tf.chainstate.get_stake_pool_data(&pool_id).unwrap().is_none(), + "the first pool must be gone from the chainstate" + ); + assert_pool_record_zeroed(&local_state, pool_id, Amount::ZERO).await; + // pool 4's pledge is now part of its own staker balance: producing a block + // moves the pledge from a standalone `CreateStakePool` output into the pool + let pool4_balance = pool_staker_balance(&tf, pool4_id); + expected_staked = (pool4_balance + amount3).expect("no overflow in test amounts"); + assert_eq!(staked_statistic(&local_state).await, expected_staked); + + // Block 7: part of the supply is delegated to pool 3. + let delegated = Amount::from_atoms(500 * CoinUnit::ATOMS_PER_COIN); + let tx5_input = TxInput::from_utxo(OutPointSourceId::Transaction(tx2c_id), 0); + let delegation_id = make_delegation_id(std::slice::from_ref(&tx5_input)).unwrap(); + let transaction = TransactionBuilder::new() + .add_input(tx5_input, InputWitness::NoSignature(None)) + .add_output(TxOutput::CreateDelegationId( + Destination::AnyoneCanSpend, + pool3_id, + )) + .add_output(TxOutput::DelegateStaking(delegated, delegation_id)) + .add_output(TxOutput::Transfer( + OutputValue::Coin((rest4 - delegated).expect("no underflow in test amounts")), + Destination::AnyoneCanSpend, + )) + .build(); + let block = create_block( + &mut rng, + &mut tf, + target_block_time, + block6_id.into(), + staking_sk.clone(), + vrf_sk.clone(), + pool4_id, + vec![transaction], + ); + let block7_id = block.get_id(); + sync_and_compare(&mut tf, block, &mut local_state, pool4_id).await; + + let pool4_balance = pool_staker_balance(&tf, pool4_id); + expected_staked = ((pool4_balance + amount3).expect("no overflow in test amounts") + delegated) + .expect("no overflow in test amounts"); + assert_eq!(staked_statistic(&local_state).await, expected_staked); + + // Block 8: pool 3 is decommissioned while the delegation to it is still active + // (the same `OutPointSourceId::Transaction` path as in block 5). + let transaction = decommission_spend_transaction( + &chain_config, + &tf, + OutPointSourceId::Transaction(tx2b_id), + 1, + pool3_creation_output, + pool3_id, + amount3, + &staking_sk, + &staking_pk, + &mut rng, + ); + let block = create_block( + &mut rng, + &mut tf, + target_block_time, + block7_id.into(), + staking_sk.clone(), + vrf_sk.clone(), + pool4_id, + vec![transaction], + ); + let block8_id = block.get_id(); + sync_and_compare(&mut tf, block, &mut local_state, pool4_id).await; + + assert!( + tf.chainstate.get_stake_pool_data(&pool3_id).unwrap().is_none(), + "pool 3 must be gone from the chainstate" + ); + assert_pool_record_zeroed(&local_state, pool3_id, delegated).await; + let pool4_balance = pool_staker_balance(&tf, pool4_id); + expected_staked = (pool4_balance + delegated).expect("no overflow in test amounts"); + assert_eq!(staked_statistic(&local_state).await, expected_staked); + + // Block 9: the delegation is spent even though pool 3 no longer exists, and the + // statistic must be decremented by exactly the delegated amount. + let transaction = TransactionBuilder::new() + .add_input( + TxInput::Account(AccountOutPoint::new( + AccountNonce::new(0), + AccountSpending::DelegationBalance(delegation_id, delegated), + )), + InputWitness::NoSignature(None), + ) + .add_output(stake_spend_payout_output(&chain_config, &tf, delegated)) + .build(); + let block = create_block( + &mut rng, + &mut tf, + target_block_time, + block8_id.into(), + staking_sk.clone(), + vrf_sk.clone(), + pool4_id, + vec![transaction], + ); + sync_and_compare(&mut tf, block, &mut local_state, pool4_id).await; + + expected_staked = pool_staker_balance(&tf, pool4_id); + assert_eq!(staked_statistic(&local_state).await, expected_staked); +} + #[rstest] #[trace] #[case(test_utils::random::Seed::from_entropy())] diff --git a/api-server/scanner-lib/src/sync/tests/simulation.rs b/api-server/scanner-lib/src/sync/tests/simulation.rs index 4e971052e7..989e54976d 100644 --- a/api-server/scanner-lib/src/sync/tests/simulation.rs +++ b/api-server/scanner-lib/src/sync/tests/simulation.rs @@ -385,7 +385,18 @@ async fn simulation( let orders_store = OrderAccountingAdapterToCheckFees::new(tf.chainstate.as_ref()); let mut new_orders_cache = orders_accounting::OrdersAccountingCache::new(&orders_store); + let mut all_inputs_utxos = Vec::new(); for tx in block.transactions() { + let inputs_utxos = tx + .inputs() + .iter() + .map(|inp| match inp.utxo_outpoint() { + Some(outpoint) => { + tf.chainstate.utxo(outpoint).unwrap().map(|utxo| utxo.take_output()) + } + None => None, + }) + .collect::>(); let new_utxos = (0..tx.outputs().len()).map(|output_index| { UtxoOutPoint::new( OutPointSourceId::Transaction(tx.transaction().get_id()), @@ -428,8 +439,11 @@ async fn simulation( &mut statistics, tx.transaction(), &chain_config, + tf.chainstate.as_ref(), + inputs_utxos.as_slice(), block_height, ); + all_inputs_utxos.push(inputs_utxos); } let block_subsidy = chain_config.block_subsidy_at_height(&block_height.next_height()); @@ -450,18 +464,8 @@ async fn simulation( let tokens_store = StubTokensAccounting; let mut total_fees = AccumulatedFee::new(); - for tx in block.transactions().iter() { - let inputs_utxos: Vec<_> = tx - .inputs() - .iter() - .map(|inp| match inp.utxo_outpoint() { - Some(outpoint) => { - tf.chainstate.utxo(outpoint).unwrap().map(|utxo| utxo.take_output()) - } - None => None, - }) - .collect(); - + assert_eq!(all_inputs_utxos.len(), block.transactions().len()); + for (tx, inputs_utxos) in block.transactions().iter().zip(all_inputs_utxos.iter()) { let inputs_accumulator = ConstrainedValueAccumulator::from_inputs( &chain_config, block_height, @@ -469,7 +473,7 @@ async fn simulation( &pos_store, &tokens_store, tx.inputs(), - &inputs_utxos, + inputs_utxos.as_slice(), ) .expect("valid block"); let outputs_accumulator = ConstrainedValueAccumulator::from_outputs( @@ -571,6 +575,8 @@ fn update_statistics( statistics: &mut BTreeMap>, tx: &Transaction, chain_config: &ChainConfig, + chainstate: &dyn ChainstateInterface, + inputs_utxos: &[Option], block_height: BlockHeight, ) { tx.outputs().iter().for_each(|out| match out { @@ -617,8 +623,51 @@ fn update_statistics( | TxOutput::CreateOrder(_) => {} }); - tx.inputs().iter().for_each(|inp| match inp { - TxInput::Utxo(_) => {} + let mut spent_pools = BTreeSet::new(); + assert_eq!( + inputs_utxos.len(), + tx.inputs().len(), + "inputs_utxos must map 1:1 to tx inputs" + ); + tx.inputs().iter().enumerate().for_each(|(input_index, inp)| match inp { + TxInput::Utxo(_) => { + // spending a stake output moves the pool's whole balance (pledge plus + // staker rewards) out of the `Staked` statistic; the utxos were resolved + // against the parent tip before the block was applied so the balance is + // still intact + let spent_output = inputs_utxos[input_index] + .as_ref() + .unwrap_or_else(|| panic!("input utxo spent by the simulation must exist")); + match spent_output { + // subtract each pool's balance only once per transaction, even if + // several spent inputs reference the same pool + TxOutput::CreateStakePool(pool_id, _) + | TxOutput::ProduceBlockFromStake(_, pool_id) + if spent_pools.insert(*pool_id) => + { + let pool_balance = chainstate + .get_stake_pool_data(pool_id) + .unwrap() + .unwrap_or_else(|| { + panic!("pool {pool_id:?} spent by the simulation must exist") + }) + .staker_balance() + .expect("no staker balance overflow in the simulation"); + let staked = statistics + .entry(CoinOrTokenStatistic::Staked) + .or_default() + .entry(CoinOrTokenId::Coin) + .or_insert(Amount::ZERO); + *staked = (*staked - pool_balance).unwrap_or_else(|| { + panic!( + "Staked statistic underflow spending pool {pool_id:?} \ + (balance {pool_balance:?})" + ) + }); + } + _ => {} + } + } TxInput::Account(acc) => match acc.account() { AccountSpending::DelegationBalance(_, to_spend) => { let staked = statistics From ebea6f3adf05d19f4711c4abf884edfbee5d9308 Mon Sep 17 00:00:00 2001 From: Enrico Rubboli Date: Fri, 16 Oct 2026 00:26:30 +0200 Subject: [PATCH 2/3] test(scanner): sign pool spends only over the live staker balance --- api-server/scanner-lib/src/sync/tests/mod.rs | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/api-server/scanner-lib/src/sync/tests/mod.rs b/api-server/scanner-lib/src/sync/tests/mod.rs index 22b01e032d..30c16e58c5 100644 --- a/api-server/scanner-lib/src/sync/tests/mod.rs +++ b/api-server/scanner-lib/src/sync/tests/mod.rs @@ -719,6 +719,7 @@ fn new_pool_data( // The builder output carries placeholder `NoSignature` witnesses; input 0 is // signed here. fn sign_pool_spend_transaction( + tf: &TestFramework, tx: SignedTransaction, spent_output: TxOutput, pool_id: PoolId, @@ -729,6 +730,13 @@ fn sign_pool_spend_transaction( staking_pk: &PublicKey, rng: &mut impl CryptoRng, ) -> SignedTransaction { + // The sighash of a `ProduceBlockFromStake` spend commits the pool's live + // staker balance, so refuse to sign over anything but the live value. + assert_eq!( + staker_balance, + pool_staker_balance(tf, pool_id), + "the signed staker balance must equal the pool's live staker balance, otherwise the sighash commitment would be stale", + ); assert_eq!( tx.transaction().inputs().len(), 1, @@ -791,14 +799,8 @@ fn decommission_spend_transaction( ) -> SignedTransaction { // `payout` doubles as the sighash `PoolInfo.staker_balance` when // `spent_output` is a `ProduceBlockFromStake` (a `CreateStakePool` spend - // commits just the UTXO); it must equal the pool's live balance, which - // holds because a decommission spend always pays out the pool's full - // balance. - assert_eq!( - payout, - pool_staker_balance(tf, pool_id), - "payout must equal the pool's live staker balance, otherwise a ProduceBlockFromStake spend would be signed over a stale sighash commitment", - ); + // commits just the UTXO); `sign_pool_spend_transaction` asserts that it + // equals the pool's live balance. let transaction = TransactionBuilder::new() .add_input( TxInput::from_utxo(input_source, input_index), @@ -807,6 +809,7 @@ fn decommission_spend_transaction( .add_output(stake_spend_payout_output(chain_config, tf, payout)) .build(); sign_pool_spend_transaction( + tf, transaction, spent_output, pool_id, @@ -1118,6 +1121,7 @@ async fn decommission_and_delegation_spends_update_staked_statistic(#[case] seed )) .build(); let transaction = sign_pool_spend_transaction( + &tf, transaction, block5_reward_out0, pool_id, From ed20da7d92991dedda1204a76c154a83d310408b Mon Sep 17 00:00:00 2001 From: Enrico Rubboli Date: Fri, 16 Oct 2026 10:14:57 +0200 Subject: [PATCH 3/3] fix(scanner): propagate database transaction lifecycle errors --- .../scanner-lib/src/blockchain_state/mod.rs | 44 +++++++++---------- 1 file changed, 21 insertions(+), 23 deletions(-) diff --git a/api-server/scanner-lib/src/blockchain_state/mod.rs b/api-server/scanner-lib/src/blockchain_state/mod.rs index 318aefc1df..9b2a0a7a6b 100644 --- a/api-server/scanner-lib/src/blockchain_state/mod.rs +++ b/api-server/scanner-lib/src/blockchain_state/mod.rs @@ -94,7 +94,7 @@ impl BlockchainState { } pub async fn scan_genesis(&mut self, genesis: &Genesis) -> Result<(), BlockchainStateError> { - let mut db_tx = self.storage.transaction_rw().await.expect("Unable to connect to database"); + let mut db_tx = self.storage.transaction_rw().await?; update_tables_from_block_reward( self.chain_config.clone(), @@ -105,7 +105,7 @@ impl BlockchainState { ) .await?; - db_tx.commit().await.expect("Unable to commit transaction"); + db_tx.commit().await?; logging::log::info!("Database commit completed successfully"); Ok(()) @@ -117,8 +117,8 @@ impl LocalBlockchainState for BlockchainState type Error = BlockchainStateError; async fn best_block(&self) -> Result<(BlockHeight, Id), Self::Error> { - let db_tx = self.storage.transaction_ro().await.expect("Unable to connect to database"); - let best_block = db_tx.get_best_block().await.expect("Unable to get best block"); + let db_tx = self.storage.transaction_ro().await?; + let best_block = db_tx.get_best_block().await?; Ok((best_block.block_height(), best_block.block_id())) } @@ -127,7 +127,7 @@ impl LocalBlockchainState for BlockchainState common_block_height: BlockHeight, blocks: Vec, ) -> Result<(), Self::Error> { - let mut db_tx = self.storage.transaction_rw().await.expect("Unable to connect to database"); + let mut db_tx = self.storage.transaction_rw().await?; // Note: a reorg happened iff the local best block is above the common ancestor with the // new chain, in which case the blocks above the common ancestor are about to be @@ -145,9 +145,7 @@ impl LocalBlockchainState for BlockchainState None }; - disconnect_tables_above_height(&mut db_tx, common_block_height) - .await - .expect("Unable to disconnect tables"); + disconnect_tables_above_height(&mut db_tx, common_block_height).await?; let mut last_event_id: StreamEventId = 0; if let Some(removed_block_ids) = removed_block_ids { @@ -230,10 +228,7 @@ impl LocalBlockchainState for BlockchainState block: WithId::take(block.clone()), tx_additional_infos, }; - db_tx - .set_mainchain_block(block_id, block_height, &block_with_extras) - .await - .expect("Unable to set block"); + db_tx.set_mainchain_block(block_id, block_height, &block_with_extras).await?; let event = StreamEvent::Block { block_id, @@ -251,8 +246,7 @@ impl LocalBlockchainState for BlockchainState block_id, tx_info, ) - .await - .expect("Unable to set transaction"); + .await?; } next_order_number += transactions.len() as u64; @@ -261,8 +255,7 @@ impl LocalBlockchainState for BlockchainState block_id, &BlockAuxData::new(block_id.into(), block_height, block_timestamp), ) - .await - .expect("Unable to set block aux data"); + .await?; let block_fees = total_fees .map_into_block_fees(&self.chain_config, block_height) @@ -282,17 +275,22 @@ impl LocalBlockchainState for BlockchainState // transaction commits; this is what wakes up the stream event pump in the web server. // Note: the guard also skips the notify for the backends that don't support stream // events, since those return the dummy event id 0 from the appends. + // Note: the pruning runs only once per retention window, instead of on every block + // commit; see the `stream_events_next_prune_id` field documentation. + // The watermark only moves after the commit succeeds: a rolled-back transaction + // rolls the pruning back with it, so the next attempt must run it again. + let should_prune = last_event_id > 0 && last_event_id >= self.stream_events_next_prune_id; + if should_prune { + db_tx.prune_stream_events().await?; + } if last_event_id > 0 { - // Note: the pruning runs only once per retention window, instead of on every block - // commit; see the `stream_events_next_prune_id` field documentation. - if last_event_id >= self.stream_events_next_prune_id { - db_tx.prune_stream_events().await?; - self.stream_events_next_prune_id = last_event_id + STREAM_EVENTS_RETENTION_COUNT; - } db_tx.notify_new_stream_events(last_event_id).await?; } - db_tx.commit().await.expect("Unable to commit transaction"); + db_tx.commit().await?; + if should_prune { + self.stream_events_next_prune_id = last_event_id + STREAM_EVENTS_RETENTION_COUNT; + } logging::log::info!("Database commit completed successfully"); Ok(())