fix(archiver): sync proper txs from blob data

When syncing from L1 blob data, we were using the txs from the first
block for all blocks. This didn't pop up earlier because tests either
still work in single block per slot, or don't use txs, or reexecute so
they don't need to sync from L1.

Also adds an extra step to the epochs-mbps test to have a non-validator
node sync the slots from L1 from scratch, which was helpful in
reproducing the issue.

Author: spalladino

yarn-project/archiver/src/l1/data_retrieval.test.ts

@@ -0,0 +1,142 @@
+import { type BlockBlobData, type CheckpointBlobData, makeBlockEndBlobData } from '@aztec/blob-lib/encoding';
+import { BlockNumber, CheckpointNumber } from '@aztec/foundation/branded-types';
+import { Fr } from '@aztec/foundation/curves/bn254';
+import { Body, CommitteeAttestation } from '@aztec/stdlib/block';
+import { L1PublishedData } from '@aztec/stdlib/checkpoint';
+import { CheckpointHeader } from '@aztec/stdlib/rollup';
+
+import { type RetrievedCheckpoint, retrievedToPublishedCheckpoint } from './data_retrieval.js';
+
+describe('data_retrieval', () => {
+  describe('retrievedToPublishedCheckpoint', () => {
+    it('handles multi-block checkpoint', async () => {
+      // Create 3 different bodies with distinct transaction effects
+      const body1 = await Body.random({ txsPerBlock: 2 });
+      const body2 = await Body.random({ txsPerBlock: 2 });
+      const body3 = await Body.random({ txsPerBlock: 2 });
+
+      // Convert to BlockBlobData
+      const block1BlobData = makeBlockBlobDataFromBody(body1, BlockNumber(1), true, 1000);
+      const block2BlobData = makeBlockBlobDataFromBody(body2, BlockNumber(2), false, 2000);
+      const block3BlobData = makeBlockBlobDataFromBody(body3, BlockNumber(3), false, 3000);
+
+      // Calculate total blob fields for checkpoint end marker
+      const numBlobFields = 100; // Approximate, doesn't need to be exact for this test
+
+      const checkpointBlobData: CheckpointBlobData = {
+        blocks: [block1BlobData, block2BlobData, block3BlobData],
+        checkpointEndMarker: { numBlobFields },
+      };
+
+      const archiveRoot = Fr.random();
+
+      const retrievedCheckpoint: RetrievedCheckpoint = {
+        checkpointNumber: CheckpointNumber(1),
+        archiveRoot,
+        header: CheckpointHeader.random(),
+        checkpointBlobData,
+        l1: new L1PublishedData(1n, 1000n, '0x1234'),
+        chainId: new Fr(1),
+        version: new Fr(1),
+        attestations: [CommitteeAttestation.empty()],
+      };
+
+      const publishedCheckpoint = await retrievedToPublishedCheckpoint(retrievedCheckpoint);
+
+      // Verify we got 3 blocks
+      expect(publishedCheckpoint.checkpoint.blocks).toHaveLength(3);
+
+      // Verify each block has the correct number of txs
+      for (const block of publishedCheckpoint.checkpoint.blocks) {
+        expect(block.body.txEffects).toHaveLength(2);
+      }
+
+      // The critical assertion: each block should have the tx hashes from its corresponding body
+      const reconstructedBlock1 = publishedCheckpoint.checkpoint.blocks[0];
+      const reconstructedBlock2 = publishedCheckpoint.checkpoint.blocks[1];
+      const reconstructedBlock3 = publishedCheckpoint.checkpoint.blocks[2];
+
+      // Block 1 should have body1's tx hashes
+      expect(reconstructedBlock1.body.txEffects.map(tx => tx.txHash.toString())).toEqual(
+        body1.txEffects.map(tx => tx.txHash.toString()),
+      );
+
+      // Block 2 should have body2's tx hashes
+      expect(reconstructedBlock2.body.txEffects.map(tx => tx.txHash.toString())).toEqual(
+        body2.txEffects.map(tx => tx.txHash.toString()),
+      );
+
+      // Block 3 should have body3's tx hashes
+      expect(reconstructedBlock3.body.txEffects.map(tx => tx.txHash.toString())).toEqual(
+        body3.txEffects.map(tx => tx.txHash.toString()),
+      );
+
+      // Also verify blocks are distinct from each other
+      expect(reconstructedBlock1.body.txEffects.map(tx => tx.txHash.toString())).not.toEqual(
+        reconstructedBlock2.body.txEffects.map(tx => tx.txHash.toString()),
+      );
+      expect(reconstructedBlock1.body.txEffects.map(tx => tx.txHash.toString())).not.toEqual(
+        reconstructedBlock3.body.txEffects.map(tx => tx.txHash.toString()),
+      );
+    });
+
+    it('handles single-block checkpoint', async () => {
+      const body1 = await Body.random({ txsPerBlock: 3 });
+      const block1BlobData = makeBlockBlobDataFromBody(body1, BlockNumber(1), true, 5000);
+
+      const checkpointBlobData: CheckpointBlobData = {
+        blocks: [block1BlobData],
+        checkpointEndMarker: { numBlobFields: 50 },
+      };
+
+      const archiveRoot = Fr.random();
+
+      const retrievedCheckpoint: RetrievedCheckpoint = {
+        checkpointNumber: CheckpointNumber(1),
+        archiveRoot,
+        header: CheckpointHeader.random(),
+        checkpointBlobData,
+        l1: new L1PublishedData(1n, 1000n, '0x1234'),
+        chainId: new Fr(1),
+        version: new Fr(1),
+        attestations: [],
+      };
+
+      const publishedCheckpoint = await retrievedToPublishedCheckpoint(retrievedCheckpoint);
+
+      expect(publishedCheckpoint.checkpoint.blocks).toHaveLength(1);
+      expect(publishedCheckpoint.checkpoint.blocks[0].body.txEffects).toHaveLength(3);
+
+      // Verify tx hashes match the original body
+      expect(publishedCheckpoint.checkpoint.blocks[0].body.txEffects.map(tx => tx.txHash.toString())).toEqual(
+        body1.txEffects.map(tx => tx.txHash.toString()),
+      );
+    });
+  });
+});
+
+/**
+ * Helper to create a BlockBlobData from a Body. This ensures the blob data is compatible
+ * with Body.fromTxBlobData.
+ */
+function makeBlockBlobDataFromBody(
+  body: Body,
+  blockNumber: BlockNumber,
+  isFirstBlock: boolean,
+  seed: number,
+): BlockBlobData {
+  const blockEndBlobData = makeBlockEndBlobData({
+    seed,
+    isFirstBlock,
+    blockEndMarker: {
+      numTxs: body.txEffects.length,
+      blockNumber,
+      timestamp: BigInt(1000 + blockNumber),
+    },
+  });
+
+  return {
+    txs: body.toTxBlobData(),
+    ...blockEndBlobData,
+  };
+}

yarn-project/archiver/src/l1/data_retrieval.ts

@@ -100,7 +100,7 @@ export async function retrievedToPublishedCheckpoint({
       }),
     });
 
-    const body = Body.fromTxBlobData(checkpointBlobData.blocks[0].txs);
+    const body = Body.fromTxBlobData(blockBlobData.txs);
 
     const blobFields = encodeBlockBlobData(blockBlobData);
     await spongeBlob.absorb(blobFields);

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

@@ -364,10 +364,10 @@ describe('e2e_epochs/epochs_mbps', () => {
     await assertMultipleBlocksPerSlot(2, logger);
   });
 
-  it('builds multiple blocks per slot and non-validator re-executes and stores proposed multi-block slots', async () => {
+  it('builds multiple blocks per slot and non-validators re-execute and sync multi-block slots', async () => {
     await setupTest({ syncChainTip: 'proposed', minTxsPerBlock: 1, maxTxsPerBlock: 1 });
 
-    logger.warn(`Creating non-validator node`);
+    logger.warn(`Creating non-validator reexecuting node`);
     const nonValidatorNode = await test.createNonValidatorNode({
       alwaysReexecuteBlockProposals: true,
       skipPushProposedBlocksToArchiver: false,
@@ -413,7 +413,7 @@ describe('e2e_epochs/epochs_mbps', () => {
       0.5,
     );
 
-    // ensure the proposed multi-block slot has valid effects
+    // Ensure the proposed multi-block slot has valid effects
     expect(multiBlockSlotNumber).toBeDefined();
     const blocksInSlot = await nonValidatorArchiver.getBlocksForSlot(SlotNumber(multiBlockSlotNumber!));
     expect(blocksInSlot.length).toBeGreaterThanOrEqual(2);
@@ -424,15 +424,26 @@ describe('e2e_epochs/epochs_mbps', () => {
     const effectsInSlot = await Promise.all(txHashesInSlot.map(txHash => nonValidatorArchiver.getTxEffect(txHash)));
     expect(effectsInSlot.every(effect => effect !== undefined)).toBe(true);
 
+    // Wait until the node syncs to the checkpointed block successfully
     const maxBlockNumberInSlot = Math.max(...blocksInSlot.map(block => block.number));
     await retryUntil(
-      async () => {
-        const tips = await nonValidatorArchiver.getL2Tips();
-        return tips.checkpointed.block.number >= maxBlockNumberInSlot;
-      },
+      async () => (await nonValidatorArchiver.getL2Tips()).checkpointed.block.number >= maxBlockNumberInSlot!,
       'non-validator node to sync checkpointed block',
       test.L2_SLOT_DURATION_IN_S * 5,
       0.5,
     );
+
+    // Start a new node an make sure it can sync from scratch including the multi-block slot
+    logger.warn(`Creating non-validator syncing node`);
+    const nonValidatorSyncingNode = await test.createNonValidatorNode({
+      alwaysReexecuteBlockProposals: false,
+    });
+    await retryUntil(
+      async () =>
+        (await nonValidatorSyncingNode.getBlockSource().getL2Tips()).checkpointed.block.number >= maxBlockNumberInSlot!,
+      'non-validator syncing node to sync checkpointed block',
+      test.L2_SLOT_DURATION_IN_S * 10,
+      0.5,
+    );
   });
 });

yarn-project/world-state/src/synchronizer/server_world_state_synchronizer.ts

@@ -300,7 +300,7 @@ export class ServerWorldStateSynchronizer
    * @returns Whether the block handled was produced by this same node.
    */
   private async handleL2Blocks(l2Blocks: L2Block[]) {
-    this.log.trace(`Handling L2 blocks ${l2Blocks[0].number} to ${l2Blocks.at(-1)!.number}`);
+    this.log.debug(`Handling L2 blocks ${l2Blocks[0].number} to ${l2Blocks.at(-1)!.number}`);
 
     // Fetch the L1->L2 messages for the first block in a checkpoint.
     const messagesForBlocks = new Map<BlockNumber, Fr[]>();
@@ -341,10 +341,12 @@ export class ServerWorldStateSynchronizer
    * @returns Whether the block handled was produced by this same node.
    */
   private async handleL2Block(l2Block: L2Block, l1ToL2Messages: Fr[]): Promise<WorldStateStatusFull> {
-    this.log.trace(`Pushing L2 block ${l2Block.number} to merkle tree db `, {
+    this.log.debug(`Pushing L2 block ${l2Block.number} to merkle tree db `, {
       blockNumber: l2Block.number,
       blockHash: await l2Block.hash().then(h => h.toString()),
       l1ToL2Messages: l1ToL2Messages.map(msg => msg.toString()),
+      blockHeader: l2Block.header.toInspect(),
+      blockStats: l2Block.getStats(),
     });
     const result = await this.merkleTreeDb.handleL2BlockAndMessages(l2Block, l1ToL2Messages);