feat(pipeline): minimize deadzone w cross slot attesting

yarn-project/end-to-end/src/e2e_epochs/epochs_mbps.pipeline.parallel.test.ts

@@ -30,10 +30,10 @@ import { EpochsTestContext } from './epochs_test.js';
 jest.setTimeout(1000 * 60 * 20);
 
 const NODE_COUNT = 4;
-const EXPECTED_BLOCKS_PER_CHECKPOINT = 3;
+const EXPECTED_BLOCKS_PER_CHECKPOINT = 8;
 
 // Send enough transactions to trigger multiple blocks within a checkpoint assuming 2 txs per block.
-const TX_COUNT = 10;
+const TX_COUNT = 24;
 
 /**
  * E2E tests for proposer pipelining with Multiple Blocks Per Slot (MBPS).
@@ -72,16 +72,18 @@ describe('e2e_epochs/epochs_mbps_pipeline', () => {
       initialValidators: validators,
       enableProposerPipelining: true, // <- yehaw
       mockGossipSubNetwork: true,
+      mockGossipSubNetworkLatency: 500, // 200 ms delay in message prop - adverse network conditions
       disableAnvilTestWatcher: true,
       startProverNode: true,
-      perBlockAllocationMultiplier: 1,
+      perBlockAllocationMultiplier: 8,
       aztecEpochDuration: 4,
       enforceTimeTable: true,
-      ethereumSlotDuration: 4,
-      aztecSlotDuration: 36,
+      ethereumSlotDuration: 12,
+      aztecSlotDuration: 72,
       blockDurationMs: 8000,
-      l1PublishingTime: 2,
-      attestationPropagationTime: 0.5,
+      // maxDABlockGas: 786432, // Set max DA block gas to be the same as the checkpoint
+      // l1PublishingTime: 2,
+      // attestationPropagationTime: 1,
       aztecTargetCommitteeSize: 3,
       ...setupOpts,
       pxeOpts: { syncChainTip },

yarn-project/end-to-end/src/fixtures/setup.ts

@@ -177,6 +177,8 @@ export type SetupOptions = {
   proverNodeConfig?: Partial<ProverNodeConfig>;
   /** Whether to use a mock gossip sub network for p2p clients. */
   mockGossipSubNetwork?: boolean;
+  /** Whether to add simulated latency to the mock gossipsub network (in ms) */
+  mockGossipSubNetworkLatency?: number;
   /** Whether to disable the anvil test watcher (can still be manually started) */
   disableAnvilTestWatcher?: boolean;
   /** Whether to enable anvil automine during deployment of L1 contracts (consider defaulting this to true). */
@@ -464,7 +466,7 @@ export async function setup(
     let p2pClientDeps: P2PClientDeps | undefined = undefined;
 
     if (opts.mockGossipSubNetwork) {
-      mockGossipSubNetwork = new MockGossipSubNetwork();
+      mockGossipSubNetwork = new MockGossipSubNetwork(opts.mockGossipSubNetworkLatency);
       p2pClientDeps = { p2pServiceFactory: getMockPubSubP2PServiceFactory(mockGossipSubNetwork) };
     }
 

yarn-project/ethereum/src/contracts/rollup.test.ts

@@ -117,6 +117,27 @@ describe('Rollup', () => {
     });
   });
 
+  describe('makeArchiveOverride', () => {
+    it('creates state override that correctly sets archive for a checkpoint number', async () => {
+      const checkpointNumber = CheckpointNumber(5);
+      const expectedArchive = Fr.random();
+
+      // Create the override
+      const stateOverride = rollup.makeArchiveOverride(checkpointNumber, expectedArchive);
+
+      // Test the override using simulateContract to read archiveAt(checkpointNumber)
+      const { result: overriddenArchive } = await publicClient.simulateContract({
+        address: rollupAddress,
+        abi: RollupAbi as Abi,
+        functionName: 'archiveAt',
+        args: [BigInt(checkpointNumber)],
+        stateOverride,
+      });
+
+      expect(Fr.fromString(overriddenArchive as string).equals(expectedArchive)).toBe(true);
+    });
+  });
+
   describe('getSlashingProposer', () => {
     it('returns a slashing proposer', async () => {
       const slashingProposer = await rollup.getSlashingProposer();

yarn-project/p2p/src/msg_validators/attestation_validator/attestation_validator.test.ts

@@ -74,6 +74,72 @@ describe('CheckpointAttestationValidator', () => {
     expect(result).toEqual({ result: 'ignore' });
   });
 
+  it('accepts attestation for current slot within pipelining grace period', async () => {
+    // Proposal is for slot 98 (current wallclock slot), but targetSlot is 99 (pipelining)
+    const header = CheckpointHeader.random({ slotNumber: SlotNumber(98) });
+    const mockAttestation = makeCheckpointAttestation({
+      header,
+      attesterSigner: attester,
+      proposerSigner: proposer,
+    });
+
+    epochCache.getTargetAndNextSlot.mockReturnValue({
+      targetSlot: SlotNumber(99),
+      nextSlot: SlotNumber(100),
+    });
+    epochCache.getSlotNow.mockReturnValue(SlotNumber(98));
+    epochCache.isProposerPipeliningEnabled.mockReturnValue(true);
+    epochCache.getL1Constants.mockReturnValue({
+      ethereumSlotDuration: 12,
+    } as any);
+
+    // Within grace period: 1000ms elapsed < 6000ms
+    epochCache.getEpochAndSlotNow.mockReturnValue({
+      epoch: EpochNumber(1),
+      slot: SlotNumber(98),
+      ts: 1000n,
+      nowMs: 1001000n, // 1000ms elapsed
+    });
+    epochCache.isInCommittee.mockResolvedValue(true);
+    epochCache.getProposerAttesterAddressInSlot.mockResolvedValue(proposer.address);
+
+    const result = await validator.validate(mockAttestation);
+    expect(result).toEqual({ result: 'accept' });
+  });
+
+  it('rejects attestation for current slot outside pipelining grace period', async () => {
+    // Proposal is for slot 97 (one behind current wallclock slot 98), targetSlot is 99 (pipelining)
+    const header = CheckpointHeader.random({ slotNumber: SlotNumber(97) });
+    const mockAttestation = makeCheckpointAttestation({
+      header,
+      attesterSigner: attester,
+      proposerSigner: proposer,
+    });
+
+    epochCache.getTargetAndNextSlot.mockReturnValue({
+      targetSlot: SlotNumber(99),
+      nextSlot: SlotNumber(100),
+    });
+    epochCache.getTargetSlot.mockReturnValue(SlotNumber(99));
+    epochCache.getSlotNow.mockReturnValue(SlotNumber(98));
+    epochCache.isProposerPipeliningEnabled.mockReturnValue(true);
+    epochCache.getL1Constants.mockReturnValue({
+      ethereumSlotDuration: 12,
+    } as any);
+
+    // Outside grace period AND outside clock tolerance: 7000ms elapsed
+    epochCache.getEpochAndSlotNow.mockReturnValue({
+      epoch: EpochNumber(1),
+      slot: SlotNumber(99),
+      ts: 1000n,
+      nowMs: 1007000n, // 7000ms elapsed
+    });
+    epochCache.isInCommittee.mockResolvedValue(true);
+
+    const result = await validator.validate(mockAttestation);
+    expect(result).toEqual({ result: 'reject', severity: PeerErrorSeverity.HighToleranceError });
+  });
+
   it('returns high tolerance error if attester is not in committee', async () => {
     const header = CheckpointHeader.random({ slotNumber: SlotNumber(100) });
     const mockAttestation = makeCheckpointAttestation({

yarn-project/p2p/src/msg_validators/attestation_validator/attestation_validator.ts

@@ -8,7 +8,7 @@ import {
   type ValidationResult,
 } from '@aztec/stdlib/p2p';
 
-import { isWithinClockTolerance } from '../clock_tolerance.js';
+import { isWithinClockTolerance, isWithinPipeliningGracePeriod } from '../clock_tolerance.js';
 
 export class CheckpointAttestationValidator implements P2PValidator<CheckpointAttestation> {
   protected epochCache: EpochCacheInterface;
@@ -23,19 +23,23 @@ export class CheckpointAttestationValidator implements P2PValidator<CheckpointAt
     const slotNumber = message.payload.header.slotNumber;
 
     try {
-      // Use target slots since proposals target pipeline slots (slot + 1 when pipelining)
+      // Use target slots since proposals target pipeline slots (slot + 1 when pipelining).
       const { targetSlot, nextSlot } = this.epochCache.getTargetAndNextSlot();
 
       if (slotNumber !== targetSlot && slotNumber !== nextSlot) {
-        // Check if message is for previous slot and within clock tolerance
-        if (!isWithinClockTolerance(slotNumber, targetSlot, this.epochCache)) {
+        // When pipelining, accept attestations for the current slot (built in the previous slot)
+        // if we're within the first ethereumSlotDuration/2 seconds of the slot.
+        if (isWithinPipeliningGracePeriod(slotNumber, this.epochCache)) {
+          // Fall through to remaining validation (signature, committee, etc.)
+        } else if (!isWithinClockTolerance(slotNumber, targetSlot, this.epochCache)) {
           this.logger.warn(
             `Checkpoint attestation slot ${slotNumber} is not current (${targetSlot}) or next (${nextSlot}) slot`,
           );
           return { result: 'reject', severity: PeerErrorSeverity.HighToleranceError };
+        } else {
+          this.logger.debug(`Ignoring checkpoint attestation for previous slot ${slotNumber} within clock tolerance`);
+          return { result: 'ignore' };
         }
-        this.logger.debug(`Ignoring checkpoint attestation for previous slot ${slotNumber} within clock tolerance`);
-        return { result: 'ignore' };
       }
 
       // Verify the signature is valid

yarn-project/p2p/src/msg_validators/clock_tolerance.test.ts

@@ -3,7 +3,11 @@ import { SlotNumber } from '@aztec/foundation/branded-types';
 
 import { mock } from 'jest-mock-extended';
 
-import { MAXIMUM_GOSSIP_CLOCK_DISPARITY_MS, isWithinClockTolerance } from './clock_tolerance.js';
+import {
+  MAXIMUM_GOSSIP_CLOCK_DISPARITY_MS,
+  isWithinClockTolerance,
+  isWithinPipeliningGracePeriod,
+} from './clock_tolerance.js';
 
 describe('clock_tolerance', () => {
   describe('MAXIMUM_GOSSIP_CLOCK_DISPARITY_MS', () => {
@@ -204,4 +208,98 @@ describe('clock_tolerance', () => {
       expect(isWithinClockTolerance(messageSlot, currentSlot, epochCache)).toBe(false);
     });
   });
+
+  describe('isWithinPipeliningGracePeriod', () => {
+    let epochCache: ReturnType<typeof mock<EpochCacheInterface>>;
+
+    beforeEach(() => {
+      epochCache = mock<EpochCacheInterface>();
+      epochCache.getSlotNow.mockReturnValue(SlotNumber(100));
+      epochCache.isProposerPipeliningEnabled.mockReturnValue(true);
+    });
+
+    it('returns true when pipelining enabled, message is for current slot, and within grace period', () => {
+      // Grace period = DEFAULT_P2P_PROPAGATION_TIME * 1000 = 2000ms
+      epochCache.getEpochAndSlotNow.mockReturnValue({
+        epoch: 1 as any,
+        slot: SlotNumber(100),
+        ts: 1000n,
+        nowMs: 1001000n, // 1000ms elapsed, within 2000ms grace period
+      });
+
+      expect(isWithinPipeliningGracePeriod(SlotNumber(100), epochCache)).toBe(true);
+    });
+
+    it('returns true at exactly 0ms elapsed', () => {
+      epochCache.getEpochAndSlotNow.mockReturnValue({
+        epoch: 1 as any,
+        slot: SlotNumber(100),
+        ts: 1000n,
+        nowMs: 1000000n, // 0ms elapsed
+      });
+
+      expect(isWithinPipeliningGracePeriod(SlotNumber(100), epochCache)).toBe(true);
+    });
+
+    it('returns false when elapsed time exceeds grace period', () => {
+      // 3000ms elapsed > 2000ms grace period
+      epochCache.getEpochAndSlotNow.mockReturnValue({
+        epoch: 1 as any,
+        slot: SlotNumber(100),
+        ts: 1000n,
+        nowMs: 1003000n, // 3000ms elapsed
+      });
+
+      expect(isWithinPipeliningGracePeriod(SlotNumber(100), epochCache)).toBe(false);
+    });
+
+    it('returns false at exactly the grace period boundary', () => {
+      // 2000ms elapsed = DEFAULT_P2P_PROPAGATION_TIME * 1000 (not strictly less than)
+      epochCache.getEpochAndSlotNow.mockReturnValue({
+        epoch: 1 as any,
+        slot: SlotNumber(100),
+        ts: 1000n,
+        nowMs: 1002000n, // 2000ms elapsed
+      });
+
+      expect(isWithinPipeliningGracePeriod(SlotNumber(100), epochCache)).toBe(false);
+    });
+
+    it('returns false when pipelining is disabled', () => {
+      epochCache.isProposerPipeliningEnabled.mockReturnValue(false);
+
+      epochCache.getEpochAndSlotNow.mockReturnValue({
+        epoch: 1 as any,
+        slot: SlotNumber(100),
+        ts: 1000n,
+        nowMs: 1001000n, // 1000ms elapsed, within grace period
+      });
+
+      expect(isWithinPipeliningGracePeriod(SlotNumber(100), epochCache)).toBe(false);
+    });
+
+    it('returns false when message is not for current slot', () => {
+      epochCache.getEpochAndSlotNow.mockReturnValue({
+        epoch: 1 as any,
+        slot: SlotNumber(100),
+        ts: 1000n,
+        nowMs: 1001000n,
+      });
+
+      // Message for slot 99, current slot is 100
+      expect(isWithinPipeliningGracePeriod(SlotNumber(99), epochCache)).toBe(false);
+    });
+
+    it('returns false when message is for a future slot', () => {
+      epochCache.getEpochAndSlotNow.mockReturnValue({
+        epoch: 1 as any,
+        slot: SlotNumber(100),
+        ts: 1000n,
+        nowMs: 1001000n,
+      });
+
+      // Message for slot 101, current slot is 100
+      expect(isWithinPipeliningGracePeriod(SlotNumber(101), epochCache)).toBe(false);
+    });
+  });
 });

yarn-project/p2p/src/msg_validators/clock_tolerance.ts

@@ -1,5 +1,6 @@
 import type { EpochCacheInterface } from '@aztec/epoch-cache';
 import { SlotNumber } from '@aztec/foundation/branded-types';
+import { DEFAULT_P2P_PROPAGATION_TIME } from '@aztec/stdlib/timetable';
 
 /**
  * Maximum clock disparity tolerance for P2P message validation (in milliseconds).
@@ -50,3 +51,33 @@ export function isWithinClockTolerance(
 
   return elapsedMs < MAXIMUM_GOSSIP_CLOCK_DISPARITY_MS;
 }
+
+/**
+ * Checks if a message should be accepted under the pipelining grace period.
+ *
+ * When pipelining is enabled, `targetSlot = slotNow + 1`. A proposal built in slot N-1
+ * for slot N arrives when validators are in slot N, so their `targetSlot = N+1`.
+ * This function accepts proposals for the current wallclock slot if we're within the
+ * first `DEFAULT_P2P_PROPAGATION_TIME` seconds of the slot (the pipelining grace period).
+ *
+ * @param messageSlot - The slot number from the received message
+ * @param epochCache - EpochCache to get timing and pipelining state
+ * @returns true if pipelining is enabled, the message is for the current slot, and we're within the grace period
+ */
+export function isWithinPipeliningGracePeriod(messageSlot: SlotNumber, epochCache: EpochCacheInterface): boolean {
+  if (!epochCache.isProposerPipeliningEnabled()) {
+    return false;
+  }
+
+  const currentSlot = epochCache.getSlotNow();
+  if (messageSlot !== currentSlot) {
+    return false;
+  }
+
+  const { ts: slotStartTs, nowMs } = epochCache.getEpochAndSlotNow();
+  const slotStartMs = slotStartTs * 1000n;
+  const elapsedMs = Number(nowMs - slotStartMs);
+  const gracePeriodMs = DEFAULT_P2P_PROPAGATION_TIME * 1000;
+
+  return elapsedMs < gracePeriodMs;
+}