test(e2e): unskip pipelining related e2e tests (#23642)

## Summary
- fix the L1 reorg message tests to wait for pre-reorg message
visibility instead of readiness where readiness is not the behavior
under test
- stabilize the L1 reorg pending-chain prune test by reorging back
before the checkpoint publish block
- target a future pipelined submission slot in the MBPS prune test
before checkpoint publishing is disabled
- keep recently unskipped tests listed as flake patterns without skip:
true
- leave blacklist token contract e2e tests on AUTOMINE_E2E_OPTS and
remove stale pipelining migration TODOs

## Verification
- yarn lint end-to-end
- git diff --check
- ANVIL_PORT=8571 e2e_epochs/epochs_l1_reorgs.parallel.test.ts -t
'updates L1 to L2 messages changed due to an L1 reorg' passed locally
- ANVIL_PORT=8572 e2e_epochs/epochs_mbps.pipeline.parallel.test.ts -t
'prunes uncheckpointed blocks when proposer fails to deliver' passed
locally
- Raman: escalated red/green local runs passed for
e2e_epochs/epochs_l1_reorgs.parallel.test.ts -t 'handles missed message
inserted by an L1 reorg'
- Raman: escalated red/green local runs passed for
e2e_epochs/epochs_l1_reorgs.parallel.test.ts -t 'prunes blocks from
pending chain removed from L1 due to an L1 reorg'
- Raman: yarn build passed

Author: spalladino

.test_patterns.yml

@@ -185,15 +185,7 @@ tests:
     error_regex: "ContractFunctionExecutionError: The contract function"
     owners:
       - *mitch
-  # Under proposer pipelining each validator votes in its own slot and the votes
-  # don't aggregate into the same round, so the slashing quorum (3) is never
-  # reached within the 414s budget; the test consistently times out at the docker
-  # outer 600s (exit 124). The publisher refactor lands all vote-offenses tx's
-  # on L1 successfully — voteCount on the slasher proposer simply stays at 1
-  # per round. This is a slashing-payload aggregation issue independent of
-  # publisher work; skip until the slashing team addresses it separately.
   - regex: "e2e_p2p/valid_epoch_pruned_slash.test.ts"
-    skip: true
     owners:
       - *mitch
       - *palla
@@ -276,14 +268,10 @@ tests:
       - *adam
 
   - regex: "src/e2e_epochs/epochs_l1_reorgs.parallel.test.ts"
-    skip: true
     owners:
       - *phil
 
-  # Consistently times out — prune detection timing too tight for CI resources
-  # See https://github.com/AztecProtocol/aztec-packages/pull/22392
   - regex: "epochs_mbps.pipeline.parallel.test.ts.*prunes uncheckpointed"
-    skip: true
     owners:
       - *sean
 
@@ -383,15 +371,7 @@ tests:
     owners:
       - *palla
 
-  # HA full suite is unstable under HA pipelining: multiple distinct failure
-  # modes (governance vote coord, work distribution, afterAll teardown hook
-  # timeouts, peer races on checkpoint proposals — see #23344, #23541,
-  # ci.aztec-labs.com/136431da99834194). Flagging individual error_regex
-  # entries and even the whole-suite flake match still let it dequeue PRs
-  # because ci3 retries once and the suite fails both attempts. Skip
-  # outright on merge-train/spartan until HA pipelining stabilises.
   - regex: "yarn-project/end-to-end/scripts/run_test.sh ha src/composed/ha/e2e_ha_full.test.ts"
-    skip: true
     owners:
       - *spyros
 

yarn-project/end-to-end/src/composed/ha/e2e_ha_full.test.ts

@@ -27,6 +27,7 @@ import { GovernanceProposerAbi } from '@aztec/l1-artifacts/GovernanceProposerAbi
 import { StatefulTestContractArtifact } from '@aztec/noir-test-contracts.js/StatefulTest';
 import { type AttestationInfo, getAttestationInfoFromPublishedCheckpoint } from '@aztec/stdlib/block';
 import { Checkpoint } from '@aztec/stdlib/checkpoint';
+import { TopicType } from '@aztec/stdlib/p2p';
 import { OffenseType } from '@aztec/stdlib/slashing';
 import { TxStatus } from '@aztec/stdlib/tx';
 import type { GenesisData } from '@aztec/stdlib/world-state';
@@ -263,6 +264,30 @@ describe('HA Full Setup', () => {
     }
 
     logger.info(`All ${NODE_COUNT} HA peer nodes started and coordinating via PostgreSQL database`);
+    logger.info('Waiting for HA peer nodes to join the tx gossip mesh before deploying the test account');
+    await retryUntil(
+      async () => {
+        const meshStates = await Promise.all(
+          haNodeServices.map(async (service, nodeIndex) => {
+            const p2p = service.getP2P();
+            const [peers, txMeshPeerCount] = await Promise.all([
+              p2p.getPeers(),
+              p2p.getGossipMeshPeerCount(TopicType.tx),
+            ]);
+
+            return { nodeIndex, peerCount: peers.length, txMeshPeerCount };
+          }),
+        );
+
+        logger.debug('HA tx gossip mesh status', { meshStates });
+        return meshStates.every(({ peerCount, txMeshPeerCount }) => peerCount > 0 && txMeshPeerCount > 0)
+          ? true
+          : undefined;
+      },
+      'HA tx gossip mesh readiness',
+      60,
+      1,
+    );
 
     // Now deploy the account - blocks can be built by the HA nodes
     logger.info('Deploying test account now that validators are running');

yarn-project/end-to-end/src/e2e_blacklist_token_contract/burn.test.ts

@@ -11,9 +11,6 @@ describe('e2e_blacklist_token_contract burn', () => {
   let { asset, tokenSim, wallet, adminAddress, otherAddress, blacklistedAddress } = t;
 
   beforeAll(async () => {
-    // TODO(kill-non-pipelined): re-enable pipelining once B1 (world-state fork lifecycle) is
-    // fixed — BlacklistTokenContractTest.applyBaseSetup runs two 86400s warps which time out
-    // mineBlock under pipelining. See PIPELINING_GOTCHAS.md.
     await t.setup({ ...AUTOMINE_E2E_OPTS });
     // Beware that we are adding the wallet as minter here, which is very slow because it needs multiple blocks.
     await t.applyMint();

yarn-project/end-to-end/src/e2e_blacklist_token_contract/minting.test.ts

@@ -11,9 +11,6 @@ describe('e2e_blacklist_token_contract mint', () => {
   let { asset, tokenSim, adminAddress, otherAddress, blacklistedAddress } = t;
 
   beforeAll(async () => {
-    // TODO(kill-non-pipelined): re-enable pipelining once B1 (world-state fork lifecycle) is
-    // fixed — BlacklistTokenContractTest.applyBaseSetup runs two 86400s warps which time out
-    // mineBlock under pipelining. See PIPELINING_GOTCHAS.md.
     await t.setup({ ...AUTOMINE_E2E_OPTS });
     // Beware that we are adding the admin as minter here, which is very slow because it needs multiple blocks.
     await t.applyMint();

yarn-project/end-to-end/src/e2e_blacklist_token_contract/shielding.test.ts

@@ -10,9 +10,6 @@ describe('e2e_blacklist_token_contract shield + redeem_shield', () => {
   let { asset, tokenSim, wallet, adminAddress, otherAddress, blacklistedAddress } = t;
 
   beforeAll(async () => {
-    // TODO(kill-non-pipelined): re-enable pipelining once B1 (world-state fork lifecycle) is
-    // fixed — BlacklistTokenContractTest.applyBaseSetup runs two 86400s warps which time out
-    // mineBlock under pipelining. See PIPELINING_GOTCHAS.md.
     await t.setup({ ...AUTOMINE_E2E_OPTS });
     await t.applyMint(); // Beware that we are adding the admin as minter here
     // Have to destructure again to ensure we have latest refs.

yarn-project/end-to-end/src/e2e_blacklist_token_contract/transfer_private.test.ts

@@ -10,9 +10,6 @@ describe('e2e_blacklist_token_contract transfer private', () => {
   let { asset, tokenSim, wallet, adminAddress, otherAddress, blacklistedAddress } = t;
 
   beforeAll(async () => {
-    // TODO(kill-non-pipelined): re-enable pipelining once B1 (world-state fork lifecycle) is
-    // fixed — BlacklistTokenContractTest.applyBaseSetup runs two 86400s warps which time out
-    // mineBlock under pipelining. See PIPELINING_GOTCHAS.md.
     await t.setup({ ...AUTOMINE_E2E_OPTS });
     // Beware that we are adding the admin as minter here, which is very slow because it needs multiple blocks.
     await t.applyMint();

yarn-project/end-to-end/src/e2e_blacklist_token_contract/transfer_public.test.ts

@@ -9,9 +9,6 @@ describe('e2e_blacklist_token_contract transfer public', () => {
   let { asset, tokenSim, wallet, adminAddress, otherAddress, blacklistedAddress } = t;
 
   beforeAll(async () => {
-    // TODO(kill-non-pipelined): re-enable pipelining once B1 (world-state fork lifecycle) is
-    // fixed — BlacklistTokenContractTest.applyBaseSetup runs two 86400s warps which time out
-    // mineBlock under pipelining. See PIPELINING_GOTCHAS.md.
     await t.setup({ ...AUTOMINE_E2E_OPTS });
     // Beware that we are adding the admin as minter here, which is very slow because it needs multiple blocks.
     await t.applyMint();

yarn-project/end-to-end/src/e2e_blacklist_token_contract/unshielding.test.ts

@@ -10,9 +10,6 @@ describe('e2e_blacklist_token_contract unshielding', () => {
   let { asset, tokenSim, wallet, adminAddress, otherAddress, blacklistedAddress } = t;
 
   beforeAll(async () => {
-    // TODO(kill-non-pipelined): re-enable pipelining once B1 (world-state fork lifecycle) is
-    // fixed — BlacklistTokenContractTest.applyBaseSetup runs two 86400s warps which time out
-    // mineBlock under pipelining. See PIPELINING_GOTCHAS.md.
     await t.setup({ ...AUTOMINE_E2E_OPTS });
     // Beware that we are adding the admin as minter here, which is very slow because it needs multiple blocks.
     await t.applyMint();

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

@@ -25,6 +25,7 @@ import { keccak256, parseTransaction } from 'viem';
 
 import { sendL1ToL2Message } from '../fixtures/l1_to_l2_messaging.js';
 import type { EndToEndContext } from '../fixtures/utils.js';
+import { waitForL1ToL2MessageSeen } from '../shared/wait_for_l1_to_l2_message.js';
 import { proveInteraction } from '../test-wallet/utils.js';
 import { EpochsTestContext } from './epochs_test.js';
 
@@ -350,12 +351,12 @@ describe('e2e_epochs/epochs_l1_reorgs', () => {
       const CHECKPOINT_NUMBER = CheckpointNumber(initialCheckpoint + 3);
       await test.waitUntilCheckpointNumber(CHECKPOINT_NUMBER, L2_SLOT_DURATION_IN_S * 10);
       expect(monitor.checkpointNumber).toEqual(CHECKPOINT_NUMBER);
+      const l1BlockNumber = monitor.l1BlockNumber;
       // Stop the sequencer immediately so any in-flight pipelined publish for CHECKPOINT_NUMBER+1
-      // doesn't extend l1BlockNumber before we capture it. setConfig alone is not enough under
+      // doesn't extend the reorg range before we calculate it. setConfig alone is not enough under
       // pipelining because already-constructed jobs snapshot the old config.
       await context.sequencer!.stop();
       logger.warn(`Sequencer stopped`);
-      const l1BlockNumber = monitor.l1BlockNumber;
       // Wait for node to sync to the checkpoint.
       await retryUntil(() => getCheckpointNumber(node).then(b => b === CHECKPOINT_NUMBER), 'node sync', 10, 0.1);
       logger.warn(`Reached checkpoint ${CHECKPOINT_NUMBER}`);
@@ -364,8 +365,7 @@ describe('e2e_epochs/epochs_l1_reorgs', () => {
       await test.assertMultipleBlocksPerSlot(2);
 
       // Remove the L2 block from L1
-      const l1BlocksToReorg = monitor.l1BlockNumber - l1BlockNumber + 1;
-      await context.cheatCodes.eth.reorgWithReplacement(l1BlocksToReorg);
+      await context.cheatCodes.eth.reorgTo(l1BlockNumber - 1);
       expect(await monitor.run(true).then(monitor => monitor.checkpointNumber)).toEqual(
         CheckpointNumber(CHECKPOINT_NUMBER - 1),
       );
@@ -472,7 +472,7 @@ describe('e2e_epochs/epochs_l1_reorgs', () => {
       });
       logger.warn(`Sent messages on L1 blocks ${msgs.map(m => m.txReceipt.blockNumber)}`);
 
-      await waitForL1ToL2MessageReady(node, msgs.at(-1)!.msgHash, {
+      await waitForL1ToL2MessageSeen(node, msgs.at(-1)!.msgHash, {
         timeoutSeconds: msgs.length * L1_BLOCK_TIME_IN_S * 2,
       });
 
@@ -485,7 +485,7 @@ describe('e2e_epochs/epochs_l1_reorgs', () => {
       logger.warn(`Sent new message on L1 block ${newMsg.txReceipt.blockNumber}`);
 
       // New msg gets synced, and old one is out
-      await waitForL1ToL2MessageReady(node, newMsg.msgHash, { timeoutSeconds: L1_BLOCK_TIME_IN_S * 6 });
+      await waitForL1ToL2MessageReady(node, newMsg.msgHash, { timeoutSeconds: L2_SLOT_DURATION_IN_S * 5 });
       expect(await isL1ToL2MessageReady(node, msgs[0].msgHash)).toBe(true);
       expect(await isL1ToL2MessageReady(node, msgs.at(-1)!.msgHash)).toBe(false);
 
@@ -502,7 +502,7 @@ describe('e2e_epochs/epochs_l1_reorgs', () => {
       logger.warn(`Sending first cross chain message`);
       const firstMsg = await sendMessage();
       logger.warn(`Sent first message on L1 block ${firstMsg.txReceipt.blockNumber}`);
-      await waitForL1ToL2MessageReady(node, firstMsg.msgHash, { timeoutSeconds: L1_BLOCK_TIME_IN_S * 3 });
+      await waitForL1ToL2MessageSeen(node, firstMsg.msgHash, { timeoutSeconds: L1_BLOCK_TIME_IN_S * 3 });
       logger.warn(`Synced first message`);
 
       // Next message shall not land
@@ -524,11 +524,12 @@ describe('e2e_epochs/epochs_l1_reorgs', () => {
       const reorgDepth = (await monitor.run(true).then(m => m.l1BlockNumber)) - l1BlockNumber;
       await context.cheatCodes.eth.reorgWithReplacement(reorgDepth, [[l1ClientDelayer.getCancelledTxs()[0]]]);
       const secondMsg = await secondMsgPromise;
+      await waitForL1ToL2MessageSeen(node, secondMsg.msgHash, { timeoutSeconds: L1_BLOCK_TIME_IN_S * 3 });
 
       // Archiver should see the new message and should be able to accept a third one on top, without any rolling hash issues
       logger.warn(`Reorged-in second message on L1 block ${secondMsg.txReceipt.blockNumber}. Sending third message.`);
       const thirdMsg = await sendMessage();
-      await waitForL1ToL2MessageReady(node, thirdMsg.msgHash, { timeoutSeconds: L1_BLOCK_TIME_IN_S * 3 });
+      await waitForL1ToL2MessageSeen(node, thirdMsg.msgHash, { timeoutSeconds: L1_BLOCK_TIME_IN_S * 3 });
 
       // Verify multi-block checkpoints were built
       await test.assertMultipleBlocksPerSlot(2);

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

@@ -325,12 +325,12 @@ describe('e2e_epochs/epochs_mbps_pipeline', () => {
     await test.waitUntilCheckpointNumber(CheckpointNumber(1), checkpointTimeout);
     const checkpointedBlockNumber = await archiver.getBlockNumber();
     logger.warn(`Baseline established: checkpoint 1 reached at block ${checkpointedBlockNumber}`);
-    // Find the next proposer and prevent it from publishing checkpoints
+    // Target a submission slot whose pipelined build has not started yet.
     const { slot: currentSlot } = test.epochCache.getEpochAndSlotNow();
     const { proposerIndex, slot: proposerSlotToNotPublish } = await findNextProposerIndex(
       test.epochCache,
       validators,
-      SlotNumber(currentSlot + 1),
+      SlotNumber(currentSlot + 2),
     );
     logger.warn(
       `Will skip checkpoint publishing for proposer ${proposerIndex} in slot ${proposerSlotToNotPublish} - current slot ${currentSlot}`,