fix(sequencer): set own proposed checkpoint locally instead of via p2p loopback (#23659)

## Motivation

The proposer relied on looping its own checkpoint proposal back through
the p2p receive path to advance its local proposed-checkpoint tip before
propagating. Under `broadcastInvalidBlockProposal` the broadcast
checkpoint archive is deliberately corrupted, so the loopback handler's
archive-based block lookup (`getBlockData({ archive })`) found nothing
and retried until the next slot. By the time the proposer returned from
broadcast, propagation had slipped past the p2p validator's stale window
— producing intermittent failures (e.g. peers rejecting a late slot
proposal).

## Approach

The proposer's optimistic proposed-checkpoint tip is the proposer's own
local state, so it is now set directly in the sequencer's checkpoint
proposal job rather than via a p2p loopback. The job adds the proposed
checkpoint to the archiver from local checkpoint data (block numbers and
counts, never the possibly-corrupted broadcast archive) immediately
before gossiping, failing closed if the local insert fails. Because
every block is already added to the archiver's FIFO queue (and awaited)
during block building, the checkpoint insert needs no retry. The
`notifyOwnCheckpointProposal` loopback is removed entirely, so the path
is identical whether p2p is enabled or not.

## Changes

- **stdlib**: New `ProposedCheckpointSink` interface alongside
`L2BlockSink`.
- **sequencer-client**: `CheckpointProposalJob` now pushes the proposed
checkpoint to the archiver from local data before broadcast, gated on
proposer pipelining and skipped when block-push is disabled
(`skipPushProposedBlocksToArchiver`, fisherman mode); widened the
sequencer/client `l2BlockSource` types to `ProposedCheckpointSink`.
- **p2p**: Removed `notifyOwnCheckpointProposal` from the `P2PService`
interface, the libp2p and dummy services, and the
`P2PClient.broadcastCheckpointProposal` call site (own proposals are
still stored in the attestation pool before propagation).
- **validator-client**: The all-nodes own-proposal branch now skips
validation and returns; removed the now-dead
`setProposedCheckpointFromBlocks` and narrowed the archiver `Pick`.
- **tests**: Added job tests (push-from-local-data and
order-before-gossip, abort-on-push-failure,
no-push-when-pipelining-disabled, fisherman) and a proposal_handler
own-proposal test; removed the obsolete libp2p loopback test and the e2e
slash-test stub; widened affected mock types.

Author: spalladino

yarn-project/end-to-end/src/e2e_p2p/duplicate_attestation_slash.test.ts

@@ -213,18 +213,6 @@ describe('e2e_p2p_duplicate_attestation_slash', () => {
 
     nodes = [maliciousNode1, maliciousNode2, honestNode1, honestNode2];
 
-    // Stub the proposer's own-checkpoint-proposal loopback on the malicious nodes. The default
-    // path awaits a local handleCheckpointProposal → validateCheckpointProposal that retries
-    // until the proposed block lands in the archiver — but skipPushProposedBlocksToArchiver
-    // means it never does, so the await hangs until the retry deadline (~one slot). By the
-    // time the proposer returns from broadcast, the wallclock is in the target slot and the
-    // staleness gate refuses the self-attestation, so no duplicate attestations are ever
-    // broadcast.
-    for (const node of [maliciousNode1, maliciousNode2]) {
-      const p2pService: any = (node as any).p2pClient.p2pService;
-      jest.spyOn(p2pService, 'notifyOwnCheckpointProposal').mockResolvedValue(undefined);
-    }
-
     // Wait for P2P mesh on all needed topics before starting sequencers
     await t.waitForP2PMeshConnectivity(nodes, NUM_VALIDATORS, 30, 0.1, [
       TopicType.tx,

yarn-project/p2p/src/client/p2p_client.ts

@@ -377,8 +377,6 @@ export class P2PClient extends WithTracer implements P2P {
         throw new Error(`Attempted to broadcast a duplicate checkpoint proposal for slot ${proposal.slotNumber}`);
       }
     }
-    // Gossipsub doesn't deliver own messages, so fire the all-nodes handler locally
-    await this.p2pService.notifyOwnCheckpointProposal(checkpointCore);
     return this.p2pService.propagate(proposal);
   }
 

yarn-project/p2p/src/services/dummy_service.ts

@@ -1,6 +1,6 @@
 import type { EthAddress } from '@aztec/foundation/eth-address';
 import type { PeerInfo } from '@aztec/stdlib/interfaces/server';
-import type { CheckpointProposalCore, Gossipable, PeerErrorSeverity, TopicType } from '@aztec/stdlib/p2p';
+import type { Gossipable, PeerErrorSeverity, TopicType } from '@aztec/stdlib/p2p';
 import { Tx, TxHash } from '@aztec/stdlib/tx';
 
 import type { PeerId } from '@libp2p/interface';
@@ -93,12 +93,6 @@ export class DummyP2PService implements P2PService {
     this.allNodesCheckpointReceivedCallback = callback;
   }
 
-  // Mirror libp2p's own-proposal loopback so the proposer's pipelined `canProposeAt` override sees its own
-  // in-flight parent checkpoint when running in p2p-disabled (single-node e2e) mode.
-  public async notifyOwnCheckpointProposal(checkpoint: CheckpointProposalCore): Promise<void> {
-    await this.allNodesCheckpointReceivedCallback?.(checkpoint, undefined as unknown as PeerId);
-  }
-
   /**
    * Register a callback for when a duplicate proposal is detected
    */

yarn-project/p2p/src/services/libp2p/libp2p_service.test.ts

@@ -1071,16 +1071,6 @@ describe('LibP2PService', () => {
       expect(reportMessageValidationResultSpy).toHaveBeenCalledWith('msg-1', MOCK_PEER_ID, TopicValidatorResult.Reject);
     });
 
-    it('notifyOwnCheckpointProposal fires allNodesCheckpointReceivedCallback', async () => {
-      const checkpointHeader = makeCheckpointHeader(1, { slotNumber: targetSlot });
-      const proposal = await makeCheckpointProposal({ signer, checkpointHeader });
-
-      await service.notifyOwnCheckpointProposal(proposal.toCore());
-
-      expect(allNodesCheckpointReceivedCallback).toHaveBeenCalledTimes(1);
-      expect(allNodesCheckpointReceivedCallback).toHaveBeenCalledWith(expect.any(Object), expect.anything());
-    });
-
     // Regression for A-1013: payloads sharing (slot, archive) but differing on feeAssetPriceModifier
     // used to dedup by archive only and silently drop the second one. The pool now dedups by
     // signed-payload hash, so the equivocation surfaces.

yarn-project/p2p/src/services/libp2p/libp2p_service.ts

@@ -722,10 +722,6 @@ export class LibP2PService extends WithTracer implements P2PService {
     this.allNodesCheckpointReceivedCallback = callback;
   }
 
-  public async notifyOwnCheckpointProposal(checkpoint: CheckpointProposalCore): Promise<void> {
-    await this.allNodesCheckpointReceivedCallback(checkpoint, this.node.peerId);
-  }
-
   /**
    * Registers a callback to be invoked when a duplicate proposal is detected.
    * This callback is triggered on the first duplicate (when count goes from 1 to 2).

yarn-project/p2p/src/services/service.ts

@@ -102,9 +102,6 @@ export interface P2PService {
 
   registerAllNodesCheckpointReceivedCallback(callback: P2PCheckpointReceivedCallback): void;
 
-  /** Fires the all-nodes checkpoint callback for our own proposal (gossipsub doesn't deliver own messages). */
-  notifyOwnCheckpointProposal(checkpoint: CheckpointProposalCore): Promise<void>;
-
   /**
    * Registers a callback invoked when a duplicate proposal is detected (equivocation).
    * The callback is triggered on the first duplicate (when count goes from 1 to 2).

yarn-project/sequencer-client/src/client/sequencer-client.ts

@@ -12,7 +12,7 @@ import type { DateProvider } from '@aztec/foundation/timer';
 import type { KeystoreManager } from '@aztec/node-keystore';
 import type { P2P } from '@aztec/p2p';
 import type { SlasherClientInterface } from '@aztec/slasher';
-import type { L2BlockSink, L2BlockSource } from '@aztec/stdlib/block';
+import type { L2BlockSink, L2BlockSource, ProposedCheckpointSink } from '@aztec/stdlib/block';
 import type { ValidatorClientFullConfig, WorldStateSynchronizer } from '@aztec/stdlib/interfaces/server';
 import type { L1ToL2MessageSource } from '@aztec/stdlib/messaging';
 import { L1Metrics, type TelemetryClient } from '@aztec/telemetry-client';
@@ -56,7 +56,7 @@ export class SequencerClient {
       worldStateSynchronizer: WorldStateSynchronizer;
       slasherClient: SlasherClientInterface | undefined;
       checkpointsBuilder: FullNodeCheckpointsBuilder;
-      l2BlockSource: L2BlockSource & L2BlockSink;
+      l2BlockSource: L2BlockSource & L2BlockSink & ProposedCheckpointSink;
       l1ToL2MessageSource: L1ToL2MessageSource;
       telemetry: TelemetryClient;
       publisherFactory?: SequencerPublisherFactory;

yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts

@@ -24,6 +24,7 @@ import {
   L2Block,
   type L2BlockSink,
   type L2BlockSource,
+  type ProposedCheckpointSink,
   type ValidateCheckpointResult,
 } from '@aztec/stdlib/block';
 import {
@@ -87,7 +88,7 @@ describe('CheckpointProposalJob', () => {
   let checkpointBuilder: MockCheckpointBuilder;
   let l1ToL2MessageSource: MockProxy<L1ToL2MessageSource>;
   let l2BlockSource: MockProxy<L2BlockSource>;
-  let blockSink: MockProxy<L2BlockSink>;
+  let blockSink: MockProxy<L2BlockSink & ProposedCheckpointSink>;
   let slasherClient: MockProxy<SlasherClientInterface>;
   let dateProvider: TestDateProvider;
   let metrics: MockProxy<SequencerMetrics>;
@@ -244,8 +245,9 @@ describe('CheckpointProposalJob', () => {
     l2BlockSource = mock<L2BlockSource>();
     l2BlockSource.getCheckpointsData.mockResolvedValue([]);
 
-    blockSink = mock<L2BlockSink>();
+    blockSink = mock<L2BlockSink & ProposedCheckpointSink>();
     blockSink.addBlock.mockResolvedValue(undefined);
+    blockSink.addProposedCheckpoint.mockResolvedValue(undefined);
 
     validatorClient = mock<ValidatorClient>();
     validatorClient.collectAttestations.mockImplementation(() => Promise.resolve([]));
@@ -1134,6 +1136,40 @@ describe('CheckpointProposalJob', () => {
       expect(mismatchEvents).toHaveLength(0);
     });
 
+    it('pushes the proposed checkpoint to the archiver from local data before broadcasting', async () => {
+      const pipelinedJob = await createPipelinedJobWithBlock(proposedParent);
+      mockL2BlockSource({ checkpointedNumber: CheckpointNumber(1), checkpointedHash: parentCheckpointHash });
+
+      await pipelinedJob.executeAndAwait();
+
+      // Built from local checkpoint data: startBlock = syncedToBlockNumber + 1, blockCount = blocks built,
+      // checkpointNumber from the job — never derived from the (possibly corrupted) broadcast proposal archive.
+      expect(blockSink.addProposedCheckpoint).toHaveBeenCalledTimes(1);
+      expect(blockSink.addProposedCheckpoint).toHaveBeenCalledWith(
+        expect.objectContaining({
+          checkpointNumber: CheckpointNumber(2),
+          startBlock: BlockNumber(lastBlockNumber + 1),
+          blockCount: 1,
+        }),
+      );
+      // The proposed checkpoint must be pushed locally before the proposal is gossiped.
+      expect(blockSink.addProposedCheckpoint.mock.invocationCallOrder[0]).toBeLessThan(
+        p2p.broadcastCheckpointProposal.mock.invocationCallOrder[0],
+      );
+    });
+
+    it('aborts the checkpoint without broadcasting when the proposed checkpoint push fails', async () => {
+      blockSink.addProposedCheckpoint.mockRejectedValue(new Error('proposed checkpoint slot expired'));
+      const pipelinedJob = await createPipelinedJobWithBlock(proposedParent);
+      mockL2BlockSource({ checkpointedNumber: CheckpointNumber(1), checkpointedHash: parentCheckpointHash });
+
+      const checkpoint = await pipelinedJob.execute();
+
+      expect(checkpoint).toBeUndefined();
+      expect(blockSink.addProposedCheckpoint).toHaveBeenCalledTimes(1);
+      expect(p2p.broadcastCheckpointProposal).not.toHaveBeenCalled();
+    });
+
     it('skips proposal with archiver-sync-timeout when archiver does not sync in time', async () => {
       const pipelinedJob = await createPipelinedJobWithBlock(proposedParent);
       l2BlockSource.getSyncedL2SlotNumber.mockResolvedValue(SlotNumber(0));
@@ -1705,6 +1741,21 @@ describe('CheckpointProposalJob', () => {
       expect(checkpointBuilder.buildBlockCalls).toHaveLength(1);
       // But must NOT push to the archiver — that was the bug causing reorgs on mainnet
       expect(blockSink.addBlock).not.toHaveBeenCalled();
+      expect(blockSink.addProposedCheckpoint).not.toHaveBeenCalled();
+    });
+
+    it('does not push the proposed checkpoint when pipelining is disabled', async () => {
+      // The proposed-checkpoint tip is a pipelining-only concept, so a non-pipelining proposer
+      // must still broadcast but must not advance it.
+      epochCache.isProposerPipeliningEnabled.mockReturnValue(false);
+      const { txs, block } = await setupTxsAndBlock(p2p, globalVariables, 1, chainId);
+      checkpointBuilder.seedBlocks([block], [txs]);
+      validatorClient.collectAttestations.mockResolvedValue(getAttestations(block));
+
+      await job.executeAndAwait();
+
+      expect(blockSink.addProposedCheckpoint).not.toHaveBeenCalled();
+      expect(p2p.broadcastCheckpointProposal).toHaveBeenCalledTimes(1);
     });
 
     it('handles empty committee gracefully', async () => {

yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.timing.test.ts

@@ -10,7 +10,7 @@ import type { TypedEventEmitter } from '@aztec/foundation/types';
 import { type P2P, P2PClientState } from '@aztec/p2p';
 import type { SlasherClientInterface } from '@aztec/slasher';
 import { AztecAddress } from '@aztec/stdlib/aztec-address';
-import type { L2Block, L2BlockSink, L2BlockSource } from '@aztec/stdlib/block';
+import type { L2Block, L2BlockSink, L2BlockSource, ProposedCheckpointSink } from '@aztec/stdlib/block';
 import type { L1RollupConstants } from '@aztec/stdlib/epoch-helpers';
 import { GasFees } from '@aztec/stdlib/gas';
 import type {
@@ -206,7 +206,7 @@ describe('CheckpointProposalJob Timing Tests', () => {
   let worldState: MockProxy<WorldStateSynchronizer>;
   let l1ToL2MessageSource: MockProxy<L1ToL2MessageSource>;
   let l2BlockSource: MockProxy<L2BlockSource>;
-  let blockSink: MockProxy<L2BlockSink>;
+  let blockSink: MockProxy<L2BlockSink & ProposedCheckpointSink>;
   let slasherClient: MockProxy<SlasherClientInterface>;
   let metrics: MockProxy<SequencerMetrics>;
   let checkpointMetrics: MockProxy<CheckpointProposalJobMetricsRecorder>;
@@ -438,7 +438,7 @@ describe('CheckpointProposalJob Timing Tests', () => {
     l2BlockSource = mock<L2BlockSource>();
     l2BlockSource.getCheckpointsData.mockResolvedValue([]);
 
-    blockSink = mock<L2BlockSink>();
+    blockSink = mock<L2BlockSink & ProposedCheckpointSink>();
     blockSink.addBlock.mockResolvedValue(undefined);
 
     validatorClient = mock<ValidatorClient>();

yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts

@@ -31,6 +31,7 @@ import {
   type L2BlockSink,
   type L2BlockSource,
   MaliciousCommitteeAttestationsAndSigners,
+  type ProposedCheckpointSink,
   type ValidateCheckpointResult,
 } from '@aztec/stdlib/block';
 import {
@@ -137,7 +138,7 @@ export class CheckpointProposalJob implements Traceable {
     private readonly l1ToL2MessageSource: L1ToL2MessageSource,
     private readonly l2BlockSource: L2BlockSource,
     private readonly checkpointsBuilder: FullNodeCheckpointsBuilder,
-    private readonly blockSink: L2BlockSink,
+    private readonly blockSink: L2BlockSink & ProposedCheckpointSink,
     private readonly l1Constants: SequencerRollupConstants,
     private readonly signatureContext: CoordinationSignatureContext,
     protected config: ResolvedSequencerConfig,
@@ -774,6 +775,13 @@ export class CheckpointProposalJob implements Traceable {
         checkpointProposalOptions,
       );
 
+      // Advance our own optimistic proposed-checkpoint tip locally before gossiping. Gossipsub
+      // doesn't echo our own messages back, so this is how the proposer makes its own proposed
+      // checkpoint visible for pipelining the next slot. Built from local checkpoint data — never
+      // from the broadcast proposal archive, which may be deliberately corrupted under test flags.
+      // Fail closed: if this throws, the outer catch aborts the slot before gossiping.
+      await this.syncProposedCheckpointToArchiver(checkpoint, blocksInCheckpoint.length, feeAssetPriceModifier);
+
       const blockProposedAt = this.dateProvider.now();
       if (this.config.skipBroadcastCheckpointProposal) {
         // Test-only: suppress the CheckpointProposal so peers never see a proposed checkpoint for
@@ -1489,6 +1497,45 @@ export class CheckpointProposalJob implements Traceable {
     await this.blockSink.addBlock(block);
   }
 
+  /**
+   * Adds the proposed checkpoint to the archiver so the proposer's optimistic proposed-checkpoint
+   * tip advances locally. Gossip doesn't echo our own messages back, so without this the proposer
+   * would never see its own proposed checkpoint and couldn't pipeline the next slot.
+   *
+   * Only runs under proposer pipelining (the proposed tip is a pipelining-only concept) and is
+   * skipped whenever proposed blocks aren't pushed (`skipPushProposedBlocksToArchiver`, fisherman
+   * mode): the archiver derives the checkpoint archive from its stored blocks, so without them the
+   * push would fail. All blocks were already added (and awaited) during block building, so this
+   * needs no retry — they are guaranteed present by the time we get here.
+   */
+  private async syncProposedCheckpointToArchiver(
+    checkpoint: Checkpoint,
+    blockCount: number,
+    feeAssetPriceModifier: bigint,
+  ): Promise<void> {
+    if (this.config.skipPushProposedBlocksToArchiver || this.config.fishermanMode) {
+      return;
+    }
+    if (!this.epochCache.isProposerPipeliningEnabled()) {
+      return;
+    }
+    const startBlock = BlockNumber(this.syncedToBlockNumber + 1);
+    this.log.debug(`Syncing proposed checkpoint ${this.checkpointNumber} to archiver`, {
+      checkpointNumber: this.checkpointNumber,
+      slot: this.targetSlot,
+      startBlock,
+      blockCount,
+    });
+    await this.blockSink.addProposedCheckpoint({
+      header: checkpoint.header,
+      checkpointNumber: this.checkpointNumber,
+      startBlock,
+      blockCount,
+      totalManaUsed: checkpoint.header.totalManaUsed.toBigInt(),
+      feeAssetPriceModifier,
+    });
+  }
+
   /** Runs fee analysis and logs checkpoint outcome as fisherman */
   private async handleCheckpointEndAsFisherman(checkpoint: Checkpoint | undefined) {
     // Perform L1 fee analysis before clearing requests