fix(sequencer): skip persistence when a produced block loses the competing-write race

lez/sequencer/core/src/lib.rs

@@ -33,7 +33,7 @@ use storage::sequencer::{
 };
 
 use crate::{
-    block_publisher::{BlockPublisherTrait, Ed25519PublicKey, ZoneSdkPublisher},
+    block_publisher::{BlockPublisherTrait, Ed25519PublicKey, MsgId, ZoneSdkPublisher},
     block_store::SequencerStore,
 };
 
@@ -390,24 +390,65 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             .await
             .context("Failed to publish block to Bedrock")?;
 
-        // Apply our own block to the head with the MsgId the publish assigned it,
-        // so the head advances and the later adopted redelivery dedups.
-        let head_state = {
-            let mut chain = self.chain.lock().expect("chain state mutex poisoned");
-            chain.apply_adopted(this_msg, &block);
-            chain.head_state().clone()
-        };
-
-        self.store.update(
+        self.record_produced_block(
+            this_msg,
             &block,
             &deposit_event_ids,
             withdrawal_reconciliation_keys,
-            &head_state,
         )?;
 
         Ok(block.header.block_id)
     }
 
+    /// Applies our own freshly-published block to the head with the [`MsgId`] the
+    /// publish assigned it, so the head advances and the later adopted
+    /// redelivery dedups, then persists it.
+    ///
+    /// Persistence is gated on the block actually becoming the head: if a peer
+    /// block won this height while we were publishing (`AlreadyApplied`, or
+    /// `Parked` when the head reorged to a different parent), the canonical
+    /// block is persisted by the follow path instead, and our invalidated
+    /// inscription comes back via `orphaned`.
+    fn record_produced_block(
+        &mut self,
+        this_msg: MsgId,
+        block: &Block,
+        deposit_event_ids: &[HashType],
+        withdrawal_reconciliation_keys: Vec<WithdrawalReconciliationKey>,
+    ) -> Result<()> {
+        let head_state = {
+            let mut chain = self.chain.lock().expect("chain state mutex poisoned");
+            match chain.apply_adopted(this_msg, block) {
+                AcceptOutcome::Applied => Some(chain.head_state().clone()),
+                AcceptOutcome::AlreadyApplied => {
+                    warn!(
+                        "Produced block {} lost a competing-write race, skipping persistence",
+                        block.header.block_id
+                    );
+                    None
+                }
+                AcceptOutcome::Parked(err) | AcceptOutcome::RetryableFailure(err) => {
+                    warn!(
+                        "Produced block {} no longer chains on the head, skipping persistence: {err}",
+                        block.header.block_id
+                    );
+                    None
+                }
+            }
+        };
+
+        if let Some(head_state) = head_state {
+            self.store.update(
+                block,
+                deposit_event_ids,
+                withdrawal_reconciliation_keys,
+                &head_state,
+            )?;
+        }
+
+        Ok(())
+    }
+
     /// Validates and applies a single mempool transaction to the current state.
     /// Returns `Ok(true)` if the transaction was valid and applied, `Ok(false)` if
     /// it was skipped due to validation failure.

lez/sequencer/core/src/tests.rs

@@ -1567,3 +1567,57 @@ async fn configure_channel_delegates_to_publisher() {
     assert_eq!(calls[0].configuration_threshold, 1);
     assert_eq!(calls[0].withdraw_threshold, 1);
 }
+
+#[tokio::test]
+async fn record_produced_block_skips_persistence_on_lost_race() {
+    let config = setup_sequencer_config();
+    let (mut sequencer, _mempool_handle) =
+        SequencerCoreWithMockClients::start_from_config(config).await;
+    let genesis_meta = sequencer.store.latest_block_meta().unwrap();
+
+    // A peer block wins height 2 while "our" block is in flight.
+    let peer_block = common::test_utils::produce_dummy_block(2, Some(genesis_meta.hash), vec![]);
+    sequencer
+        .chain()
+        .lock()
+        .expect("chain mutex poisoned")
+        .apply_adopted(MsgId::from([9; 32]), &peer_block);
+
+    // Our competing block at the same height: same parent, different content.
+    let tx = common::test_utils::produce_dummy_empty_transaction();
+    let our_block = common::test_utils::produce_dummy_block(2, Some(genesis_meta.hash), vec![tx]);
+    sequencer
+        .record_produced_block(
+            MsgId::from(our_block.header.hash.0),
+            &our_block,
+            &[],
+            vec![],
+        )
+        .unwrap();
+
+    // The lost-race block must not reach the store; the head keeps the peer block.
+    assert!(sequencer.store.get_block_at_id(2).unwrap().is_none());
+    let head_tip = sequencer
+        .chain()
+        .lock()
+        .expect("chain mutex poisoned")
+        .head_tip()
+        .expect("head tip");
+    assert_eq!(head_tip.hash, peer_block.header.hash);
+}
+
+#[tokio::test]
+async fn record_produced_block_skips_persistence_when_block_no_longer_chains() {
+    let config = setup_sequencer_config();
+    let (mut sequencer, _mempool_handle) =
+        SequencerCoreWithMockClients::start_from_config(config).await;
+
+    // The head reorged under us: our block's parent is no longer the tip.
+    let stale = common::test_utils::produce_dummy_block(2, Some(HashType([9; 32])), vec![]);
+    sequencer
+        .record_produced_block(MsgId::from(stale.header.hash.0), &stale, &[], vec![])
+        .unwrap();
+
+    assert!(sequencer.store.get_block_at_id(2).unwrap().is_none());
+    assert_eq!(sequencer.chain_height(), 1, "head is unchanged");
+}