diff --git a/massa-pool-worker/src/operation_pool.rs b/massa-pool-worker/src/operation_pool.rs index e871302d1e..c232485e0c 100644 --- a/massa-pool-worker/src/operation_pool.rs +++ b/massa-pool-worker/src/operation_pool.rs @@ -115,7 +115,14 @@ impl OperationPool { pos_draws } - /// Returns the list of executed ops with a boolean indicating whether they are executed as final. + /// Returns execution markers for ops currently in the pool. + /// + /// Map value semantics: + /// - `true`: executed in final history (durable; safe to evict from the pool) + /// - `false`: executed only in speculative/candidate history (can disappear on rollback; + /// must not drive pool eviction) + /// + /// Ops with no execution record are omitted. fn get_execution_statuses(&self) -> PreHashMap { let op_ids: Vec = self.sorted_ops.iter().map(|op_info| op_info.id).collect(); self.channels @@ -125,7 +132,9 @@ impl OperationPool { .zip(op_ids) .filter_map( |((spec_status, final_status), op_id)| match (spec_status, final_status) { - (Some(_), Some(_)) => Some((op_id, true)), + // Final execution is durable (execution layer also surfaces it as speculative). + (_, Some(_)) => Some((op_id, true)), + // Candidate-history only: keep as non-final for optional scoring, never as eviction. (Some(_), None) => Some((op_id, false)), _ => None, }, @@ -179,14 +188,18 @@ impl OperationPool { retain = op_info.fee.checked_sub(self.config.minimal_fees).is_some(); } - // filter out ops that have been executed in final or candidate slots - // TODO: in the re-execution followup, we should only filter out final-executed ops here (exec_status == Some(true)) + // Filter out ops whose execution is final/durable only. + // Speculative/candidate-only markers (exec_status == false) must not evict: + // they can vanish on rollback, and there is no reinsertion path after drop. if retain { - retain = !exec_statuses.contains_key(&op_info.id); + retain = exec_statuses.get(&op_info.id) != Some(&true); } - // filter out ops that spend more than the sender's balance - if retain { + // Filter out ops that spend more than the sender's balance. + // Skip for ops with a live mark: they are not selectable, and their spend is + // already in the candidate balance — comparing again would evict them before + // any rollback can restore the balance. + if retain && !exec_statuses.contains_key(&op_info.id) { retain = match sender_balances.get(&op_info.creator_address) { Some(v) => &op_info.max_spending <= v, None => false, // filter out ops for which the sender does not exist @@ -204,11 +217,16 @@ impl OperationPool { } /// Eliminate all operations that would cause a sender balance overflow. - /// Assumes that the ops are sorted by ascending score. + /// Assumes that the ops are sorted by descending score (best first). fn eliminate_balance_overflows(&mut self, sender_balances: &PreHashMap) { let mut balance_cache = PreHashMap::default(); let mut removed = PreHashSet::default(); self.sorted_ops.retain(|op_info| { + // Live marks: spend already counted in candidate balance; keep for rollback. + // Marked ops also score last, so without this skip they would be cut first. + if op_info.executed { + return true; + } let balance = balance_cache .entry(op_info.creator_address) .or_insert_with(|| { @@ -253,7 +271,7 @@ impl OperationPool { /// Score the operations fn score_operations( &self, - _exec_statuses: &PreHashMap, + exec_statuses: &PreHashMap, pos_draws: &BTreeSet, ) -> PreHashMap { let now = MassaTime::now(); @@ -315,23 +333,19 @@ impl OperationPool { 0.0 }; - /* TODO: re-execution followup - // If the op was executed previously, there is still an exponentially decaying chance of its block being cancelled - // so that it can be reincluded. - // We approximate it with a constant factor for simplicity since we don't have the inclusion slot for now. - let reexecution_penalty = 1.0 / 1000.0; // re-execution penalty factor + // If the op was executed previously, there is still a chance of its block being + // cancelled so that it can be reincluded. Keep it in the pool (for rollback recovery) + // but score it far below non-executed ops. Block production also skips currently + // executed ops to avoid wasting gas while the mark is live. + let reexecution_penalty = 1.0 / 1000.0; let reexecution_factor = if exec_statuses.contains_key(&op_info.id) { - // executed previously reexecution_penalty } else { - // not executed previously => score 1 1.0 }; - */ // compute the score as being the product of all the factors and the fee - let score = fee_factor * resource_factor * inclusion_factor; - // * reexecution_factor; // TODO: re-execution followup + let score = fee_factor * resource_factor * inclusion_factor * reexecution_factor; // store the score scores.insert(op_info.id, score); @@ -348,6 +362,13 @@ impl OperationPool { // get execution statuses let exec_statuses = self.get_execution_statuses(); + // Cache live execution marks on OperationInfo so get_block_operations (factory + // thread) can skip without touching execution. Staleness is bounded by the + // refresh interval; a rare duplicate is ignored by execution. + for op_info in &mut self.sorted_ops { + op_info.executed = exec_statuses.contains_key(&op_info.id); + } + // get sender balances let sender_balances = self.get_sender_balances(); @@ -470,6 +491,10 @@ impl OperationPool { /// Searches the available operations, and selects the sub-set of operations that: /// - fit inside the block /// - is the most profitable for block producer + /// - are not currently marked executed (speculative or final) + /// + /// Must never query execution: runs on the factory thread under the pool read + /// guard at slot time. Uses the `executed` flag set by refresh() instead. pub fn get_block_operations(&self, slot: &Slot) -> (Vec, Storage) { // init list of selected operation IDs let mut op_ids = Vec::new(); @@ -498,7 +523,7 @@ impl OperationPool { continue; } - // exclude ops that are too large + // exclude ops that use too much resources for remaining capacity if op_info.size > remaining_space { continue; } @@ -508,6 +533,12 @@ impl OperationPool { continue; } + // Skip while refresh() last saw a live execution mark (speculative or final). + // Zero cost on the production path; staleness is at most one refresh interval. + if op_info.executed { + continue; + } + // here we consider the operation as accepted op_ids.push(op_info.id); diff --git a/massa-pool-worker/src/tests/operation_pool_tests.rs b/massa-pool-worker/src/tests/operation_pool_tests.rs index df9a78ce5a..6076ce3379 100644 --- a/massa-pool-worker/src/tests/operation_pool_tests.rs +++ b/massa-pool-worker/src/tests/operation_pool_tests.rs @@ -17,17 +17,32 @@ //! latest period given his own thread. All operation which doesn't fit these //! requirements are "irrelevant" //! +use crate::operation_pool::OperationPool; use crate::tests::tools::OpGenerator; use super::tools::{ create_some_operations, default_mock_execution_controller, pool_test, PoolTestBoilerPlate, }; +use massa_execution_exports::MockExecutionController; use massa_models::{ - address::Address, amount::Amount, config::ENDORSEMENT_COUNT, operation::OperationId, slot::Slot, + address::Address, amount::Amount, config::ENDORSEMENT_COUNT, operation::OperationId, + prehash::PreHashMap, slot::Slot, }; -use massa_pool_exports::PoolConfig; +use massa_pool_exports::{PoolBroadcasts, PoolChannels, PoolConfig}; use massa_pos_exports::{MockSelectorController, Selection}; -use std::{collections::BTreeMap, time::Duration}; +use massa_signature::KeyPair; +use massa_storage::Storage; +use massa_wallet::test_exports::create_test_wallet; +use parking_lot::RwLock; +use std::{ + collections::BTreeMap, + sync::{ + atomic::{AtomicU8, Ordering}, + Arc, + }, + time::Duration, +}; +use tokio::sync::broadcast; // Helper to create a recursive selector mock for operation pool tests fn create_recursive_selector_for_ops(addr: Address) -> MockSelectorController { @@ -63,6 +78,110 @@ fn create_recursive_selector_for_ops(addr: Address) -> MockSelectorController { story } +/// Speculative/candidate-only execution must not permanently remove ops from the pool. +/// They must also be skipped for block production while the mark is live (no gas waste), +/// and become selectable again after a rollback clears the mark. +/// +/// Candidate balance already includes speculative spends, so while marked the remaining +/// balance is below max_spending — balance filters must skip live marks or the op is +/// wrongly evicted before rollback. +#[test] +fn test_refresh_keeps_speculative_only_executed_ops() { + let keypair = KeyPair::generate(0).unwrap(); + let addr = Address::from_public_key(&keypair.get_public_key()); + let creator_thread = addr.get_thread(PoolConfig::default().thread_count); + + // Op spends 9 + fee 1 = 10. Full balance covers it; after speculative execution + // only 1 remains — without skipping balance checks for marked ops, refresh would drop it. + let op_amount = Amount::from_raw(9); + let op_fee = Amount::from_raw(1); + let full_balance = Amount::from_raw(11); + let balance_after_speculative = Amount::from_raw(1); + + // 0 = not executed, 1 = speculative only, 2 = final + let exec_phase = Arc::new(AtomicU8::new(0)); + let phase_for_status = exec_phase.clone(); + let phase_for_balance = exec_phase.clone(); + let mut execution_controller = MockExecutionController::new(); + execution_controller + .expect_get_ops_exec_status() + .returning(move |ops| match phase_for_status.load(Ordering::SeqCst) { + 1 => vec![(Some(true), None); ops.len()], + 2 => vec![(Some(true), Some(true)); ops.len()], + _ => vec![(None, None); ops.len()], + }); + execution_controller + .expect_get_final_and_candidate_balance() + .returning(move |addrs| { + let candidate = if phase_for_balance.load(Ordering::SeqCst) == 1 { + balance_after_speculative + } else { + full_balance + }; + vec![(Some(full_balance), Some(candidate)); addrs.len()] + }); + + let mut addresses = PreHashMap::default(); + addresses.insert(addr, keypair.clone()); + let wallet = Arc::new(RwLock::new(create_test_wallet(Some(addresses)))); + let (endorsement_sender, _) = broadcast::channel(1); + let (operation_sender, _) = broadcast::channel(1); + + let storage = Storage::create_root(); + let mut operation_pool = OperationPool::init( + PoolConfig::default(), + &storage, + PoolChannels { + execution_controller: Box::new(execution_controller), + broadcasts: PoolBroadcasts { + endorsement_sender, + operation_sender, + }, + selector: Box::new(create_recursive_selector_for_ops(addr)), + }, + wallet, + ); + + let ops = create_some_operations( + 1, + &OpGenerator::default() + .creator(keypair) + .expirery(10) + .fee(op_fee) + .amount(op_amount), + ); + let op_id = ops[0].id; + let mut ops_storage = storage.clone_without_refs(); + ops_storage.store_operations(ops); + operation_pool.add_operations(ops_storage); + assert_eq!(operation_pool.len(), 1); + + let target_slot = Slot::new(1, creator_thread); + + // Candidate-history mark + depleted candidate balance: op must remain, not be selected. + exec_phase.store(1, Ordering::SeqCst); + operation_pool.refresh(); + assert_eq!(operation_pool.len(), 1); + assert!(operation_pool.contains(&op_id)); + let (selected, _) = operation_pool.get_block_operations(&target_slot); + assert!( + selected.is_empty(), + "speculatively executed ops must not be selected for blocks" + ); + + // Simulate rollback clearing the speculative mark: balance restored, selectable again. + exec_phase.store(0, Ordering::SeqCst); + operation_pool.refresh(); + assert_eq!(operation_pool.len(), 1); + let (selected, _) = operation_pool.get_block_operations(&target_slot); + assert_eq!(selected, vec![op_id]); + + // Final execution: durable, so refresh may drop them. + exec_phase.store(2, Ordering::SeqCst); + operation_pool.refresh(); + assert_eq!(operation_pool.len(), 0); +} + #[test] fn test_add_operation() { use massa_signature::KeyPair; diff --git a/massa-pool-worker/src/types.rs b/massa-pool-worker/src/types.rs index 30c75a2b51..865cdc975c 100644 --- a/massa-pool-worker/src/types.rs +++ b/massa-pool-worker/src/types.rs @@ -17,6 +17,9 @@ pub struct OperationInfo { /// max amount that the op might spend from the sender's balance pub max_spending: Amount, pub validity_period_range: RangeInclusive, + /// Whether execution currently marks this op as executed (speculative or final). + /// Updated only in refresh() so block production never queries execution. + pub executed: bool, } impl OperationInfo { @@ -37,6 +40,7 @@ impl OperationInfo { thread: op.content_creator_address.get_thread(thread_count), validity_period_range: op.get_validity_range(operation_validity_periods), max_spending: op.get_max_spending(roll_price), + executed: false, } } }