test(fast-inbox): reconstruct L1-to-L2 messages in the l1 publisher integration mock (A-1388)

The streaming flip makes world-state sync reconstruct L1-to-L2 messages per block via
the archiver's bucket queries, but the integration test's block-source mock did not
implement them, so block 1 committed with zero messages and the next fork threw. Back
the mock with MockL1ToL2MessageSource and register a bucket per published block.

Author: spalladino

yarn-project/sequencer-client/src/publisher/l1_publisher.integration.test.ts

@@ -1,4 +1,5 @@
 import type { ArchiverDataSource } from '@aztec/archiver';
+import { MockL1ToL2MessageSource } from '@aztec/archiver/test';
 import { AztecAddress } from '@aztec/aztec.js/addresses';
 import { Fr } from '@aztec/aztec.js/fields';
 import { createLogger } from '@aztec/aztec.js/log';
@@ -144,6 +145,11 @@ describe('L1Publisher integration', () => {
 
   let builderDb: NativeWorldStateService;
 
+  // Backs the blockSource mock's streaming L1->L2 message queries. The world-state synchronizer reconstructs each
+  // block's consumed message bundle from Inbox buckets (AZIP-22 Fast Inbox) when it syncs a block back, so the test
+  // registers one bucket per published block here (see buildAndPublishBlock).
+  let messageSource: MockL1ToL2MessageSource;
+
   // The header of the last block
   let prevHeader: BlockHeader;
 
@@ -280,6 +286,21 @@ describe('L1Publisher integration', () => {
       checkpointNumber: CheckpointNumber.ZERO,
       indexWithinCheckpoint: IndexWithinCheckpoint(0),
     };
+    // Seed the genesis sentinel bucket (seq 0, no messages) so the world-state synchronizer can resolve a
+    // totalMsgCount of 0 to a bucket when reconstructing the first block's message bundle.
+    messageSource = new MockL1ToL2MessageSource(0);
+    messageSource.setInboxBucket(
+      {
+        seq: 0n,
+        inboxRollingHash: Fr.ZERO,
+        totalMsgCount: 0n,
+        timestamp: 0n,
+        msgCount: 0,
+        lastMessageIndex: 0n,
+        isOpen: false,
+      },
+      [],
+    );
     blockSource = mock<ArchiverDataSource>({
       getBlocks(query: BlocksQuery) {
         if (!('from' in query)) {
@@ -355,6 +376,14 @@ describe('L1Publisher integration', () => {
       getBlockNumber(): Promise<BlockNumber> {
         return Promise.resolve(BlockNumber(blocks.at(-1)?.number ?? BlockNumber.ZERO));
       },
+      // Streaming L1->L2 message reconstruction (AZIP-22 Fast Inbox): the world-state synchronizer resolves each
+      // block's consumed message bundle from the Inbox buckets registered per published block in buildAndPublishBlock.
+      getInboxBucketByTotalMsgCount(totalMsgCount: bigint) {
+        return messageSource.getInboxBucketByTotalMsgCount(totalMsgCount);
+      },
+      getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint) {
+        return messageSource.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive);
+      },
     });
 
     const worldStateConfig: WorldStateConfig = {
@@ -594,6 +623,24 @@ describe('L1Publisher integration', () => {
         previousInboxRollingHash = checkpoint.header.inboxRollingHash;
         const block = checkpoint.blocks[0];
 
+        // Register this block's consumed messages as an Inbox bucket so the world-state synchronizer can rebuild the
+        // per-block bundle when it syncs the block back on the next iteration. One bucket per block: its sequence is
+        // the block number and its cumulative total is the block's L1->L2 tree leaf count, which is exactly the value
+        // world-state looks the bucket up by (AZIP-22 Fast Inbox streaming reconstruction).
+        const cumulativeMsgCount = BigInt(block.header.state.l1ToL2MessageTree.nextAvailableLeafIndex);
+        messageSource.setInboxBucket(
+          {
+            seq: BigInt(block.number),
+            inboxRollingHash: checkpoint.header.inboxRollingHash,
+            totalMsgCount: cumulativeMsgCount,
+            timestamp,
+            msgCount: currentL1ToL2Messages.length,
+            lastMessageIndex: cumulativeMsgCount - 1n,
+            isOpen: false,
+          },
+          currentL1ToL2Messages,
+        );
+
         const totalManaUsed = txs.reduce((acc, tx) => acc.add(new Fr(tx.gasUsed.billedGas.l2Gas)), Fr.ZERO);
         expect(totalManaUsed.toBigInt()).toEqual(block.header.totalManaUsed.toBigInt());