feat(ethexe-consensus): mini-announces for instant injected TX promises

ethexe/compute/src/compute.rs

@@ -310,13 +310,21 @@ pub(crate) mod utils {
     ) -> Result<ExecutableData> {
         let block_hash = announce.block_hash;
 
-        let matured_events =
-            find_canonical_events_post_quarantine(db, block_hash, canonical_quarantine)?;
-
-        let events = matured_events
-            .into_iter()
-            .filter_map(|event| event.to_request())
-            .collect();
+        // Check if parent announce is for the same block (mini-announce case).
+        // If so, canonical events were already processed by the parent — skip them.
+        let is_same_block = db
+            .announce(announce.parent)
+            .is_some_and(|parent| parent.block_hash == block_hash);
+        let events = if is_same_block {
+            vec![]
+        } else {
+            let matured_events =
+                find_canonical_events_post_quarantine(db, block_hash, canonical_quarantine)?;
+            matured_events
+                .into_iter()
+                .filter_map(|event| event.to_request())
+                .collect()
+        };
 
         Ok(ExecutableData {
             block: SimpleBlockData {

ethexe/consensus/src/validator/batch/utils.rs

@@ -204,21 +204,21 @@ pub fn calculate_batch_expiry<DB: BlockMetaStorageRO + OnChainStorageRO + Announ
             )
         })?;
 
-    // Amount of announces which we should check to determine if there are not-base announces in the commitment.
-    let Some(announces_to_check_amount) = commitment_delay_limit.checked_sub(head_delay) else {
+    // Number of blocks worth of announces to check for not-base announces in the commitment.
+    let Some(blocks_to_check) = commitment_delay_limit.checked_sub(head_delay) else {
         // No need to set expiry - head announce is old enough, so cannot contain any not-base announces.
         return Ok(None);
     };
 
-    if announces_to_check_amount == 0 {
+    if blocks_to_check == 0 {
         // No need to set expiry - head announce is old enough, so cannot contain any not-base announces.
         return Ok(None);
     }
 
-    let mut oldest_not_base_announce_depth = (!head_announce.is_base()).then_some(0);
+    let mut oldest_not_base_announce_depth = (!head_announce.is_base()).then_some(0u32);
     let mut current_announce_hash = head_announce.parent;
 
-    if announces_to_check_amount == 1 {
+    if blocks_to_check == 1 {
         // If head announce is not base and older than commitment delay limit - 1, then expiry is only 1.
         return Ok(oldest_not_base_announce_depth.map(|_| 1));
     }
@@ -228,8 +228,12 @@ pub fn calculate_batch_expiry<DB: BlockMetaStorageRO + OnChainStorageRO + Announ
         .last_committed_announce
         .ok_or_else(|| anyhow!("last committed announce not found for block {}", block.hash))?;
 
-    // from 1 because we have already checked head announce (note announces_to_check_amount > 1)
-    for i in 1..announces_to_check_amount {
+    // Walk backwards counting block transitions (not announce hops) because
+    // mini-announces chain within the same block. CDL is defined in blocks per S1.
+    let mut blocks_seen = 1u32; // already counted head announce's block
+    let mut prev_block_hash = Some(head_announce.block_hash);
+
+    loop {
         if current_announce_hash == last_committed_announce {
             break;
         }
@@ -238,15 +242,24 @@ pub fn calculate_batch_expiry<DB: BlockMetaStorageRO + OnChainStorageRO + Announ
             .announce(current_announce_hash)
             .ok_or_else(|| anyhow!("Cannot get announce by {current_announce_hash}",))?;
 
+        let is_new_block = prev_block_hash != Some(current_announce.block_hash);
+        if is_new_block {
+            blocks_seen += 1;
+            prev_block_hash = Some(current_announce.block_hash);
+            if blocks_seen >= blocks_to_check {
+                break;
+            }
+        }
+
         if !current_announce.is_base() {
-            oldest_not_base_announce_depth = Some(i);
+            oldest_not_base_announce_depth = Some(blocks_seen);
         }
 
         current_announce_hash = current_announce.parent;
     }
 
     Ok(oldest_not_base_announce_depth
-        .map(|depth| announces_to_check_amount - depth)
+        .map(|depth| blocks_to_check - depth)
         .map(TryInto::try_into)
         .transpose()?)
 }

ethexe/consensus/src/validator/coordinator.rs

@@ -40,6 +40,8 @@ pub struct Coordinator {
     ctx: ValidatorContext,
     validators: BTreeSet<Address>,
     multisigned_batch: MultisignedBatchCommitment,
+    /// Buffered next block to process after submission completes.
+    next_block: Option<SimpleBlockData>,
 }
 
 impl StateHandler for Coordinator {
@@ -55,6 +57,19 @@ impl StateHandler for Coordinator {
         self.ctx
     }
 
+    fn process_new_head(mut self, block: SimpleBlockData) -> Result<ValidatorState> {
+        if self.next_block.is_some() {
+            // Second new head while still waiting for signatures — give up on this batch.
+            // Not enough validators responded within one block window.
+            tracing::warn!("Coordinator received second new head, abandoning batch");
+            Initial::create(self.ctx)?.process_new_head(block)
+        } else {
+            // Buffer the first new head. Process it after submission.
+            self.next_block = Some(block);
+            Ok(self.into())
+        }
+    }
+
     fn process_validation_reply(
         mut self,
         reply: BatchCommitmentValidationReply,
@@ -72,7 +87,7 @@ impl StateHandler for Coordinator {
         }
 
         if self.multisigned_batch.signatures().len() as u64 >= self.ctx.core.signatures_threshold {
-            Self::submission(self.ctx, self.multisigned_batch)
+            Self::submission(self.ctx, self.multisigned_batch, self.next_block)
         } else {
             Ok(self.into())
         }
@@ -110,7 +125,7 @@ impl Coordinator {
             .set(block.header.height);
 
         if multisigned_batch.signatures().len() as u64 >= ctx.core.signatures_threshold {
-            return Self::submission(ctx, multisigned_batch);
+            return Self::submission(ctx, multisigned_batch, None);
         }
 
         let era_index = ctx
@@ -131,13 +146,15 @@ impl Coordinator {
             ctx,
             validators: validators.into_iter().collect(),
             multisigned_batch,
+            next_block: None,
         }
         .into())
     }
 
     pub fn submission(
         ctx: ValidatorContext,
         multisigned_batch: MultisignedBatchCommitment,
+        next_block: Option<SimpleBlockData>,
     ) -> Result<ValidatorState> {
         let (batch, signatures) = multisigned_batch.into_parts();
         let cloned_committer = ctx.core.committer.clone_boxed();
@@ -150,16 +167,21 @@ impl Coordinator {
                         block_hash,
                         batch_digest,
                         tx,
-                    }.into(),
+                    }
+                    .into(),
                     Err(err) => ConsensusEvent::Warning(format!(
                         "Failed to submit commitment for block {block_hash}, digest {batch_digest}: {err}"
-                    ))
+                    )),
                 };
                 Ok(event)
             }
             .boxed(),
         );
-        Initial::create(ctx)
+        let state = Initial::create(ctx)?;
+        match next_block {
+            Some(block) => state.process_new_head(block),
+            None => Ok(state),
+        }
     }
 }
 

ethexe/consensus/src/validator/initial.rs

@@ -438,11 +438,13 @@ mod tests {
 
         let ctx = state.into_context();
         assert_eq!(ctx.output, vec![]);
-        for i in last - 5..last - 5 + ctx.core.commitment_delay_limit as usize {
+        // With block-aware CDL, non-base announces are eligible for CDL blocks per S1
+        // (`<= commitment_delay_limit`), so sibling branches survive one block longer.
+        for i in last - 5..=last - 5 + ctx.core.commitment_delay_limit as usize {
             let announces = ctx.core.db.block_meta(chain.blocks[i].hash).announces;
             assert_eq!(announces.unwrap().len(), 2);
         }
-        for i in last - 5 + ctx.core.commitment_delay_limit as usize..=last {
+        for i in last - 5 + ctx.core.commitment_delay_limit as usize + 1..=last {
             let announces = ctx.core.db.block_meta(chain.blocks[i].hash).announces;
             assert_eq!(announces.unwrap().len(), 1);
         }

ethexe/consensus/src/validator/mock.rs

@@ -74,6 +74,7 @@ impl ElectionProvider for MockEthereum {
 #[async_trait]
 pub trait WaitFor {
     async fn wait_for_event(self) -> Result<(ValidatorState, ConsensusEvent)>;
+    #[allow(dead_code)]
     async fn wait_for_state<F>(self, f: F) -> Result<ValidatorState>
     where
         F: Fn(&ValidatorState) -> bool + Unpin + Send;