[F97] feat: keep speculative ops in pool, but hard skip them

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<OperationId, bool> {
         let op_ids: Vec<OperationId> = 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<Address, Amount>) {
         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<OperationId, bool>,
+        exec_statuses: &PreHashMap<OperationId, bool>,
         pos_draws: &BTreeSet<Slot>,
     ) -> PreHashMap<OperationId, f32> {
         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<OperationId>, 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);
 

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;

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<u64>,
+    /// 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,
         }
     }
 }