fix(fast-inbox): apply streaming cutoff floor on the cap-terminated final block and defer bundle accumulation (A-1382)

- The censorship cutoff floor now applies on whichever block ends the checkpoint,
  including the block that reaches the per-checkpoint block cap, not only the
  timetable's last sub-slot, so the checkpoint always consumes through the cutoff.
- The lightweight builder accumulates a block's streaming message bundle only after
  the block is fully built, so a mid-build failure leaves the checkpoint's message
  list (and its inHash/rolling hash) consistent with the blocks actually built.

Author: spalladino

yarn-project/prover-client/src/light/lightweight_checkpoint_builder.ts

@@ -211,15 +211,15 @@ export class LightweightCheckpointBuilder {
     // Streaming Inbox (AZIP-22 Fast Inbox): insert this block's L1-to-L2 message bundle before reading the end state,
     // so the block header's L1-to-L2 tree snapshot reflects it. First-in-checkpoint bundles are padded to
     // NUMBER_OF_L1_L2_MESSAGES_PER_ROLLUP (matching the legacy per-checkpoint insertion and the world-state
-    // synchronizer); non-first bundles are appended compactly. The rolling hash and inHash are then recomputed over
-    // the accumulated logical (unpadded) messages at checkpoint completion.
+    // synchronizer); non-first bundles are appended compactly. The logical (unpadded) messages are accumulated only
+    // once the block is fully built (below), so a mid-build failure does not pollute the checkpoint's inHash/rolling
+    // hash; the rolling hash and inHash are recomputed over them at checkpoint completion.
     if (opts.l1ToL2Messages !== undefined) {
       if (isFirstBlock) {
         await appendL1ToL2MessagesToTree(this.db, opts.l1ToL2Messages);
       } else {
         await this.db.appendLeaves(MerkleTreeId.L1_TO_L2_MESSAGE_TREE, opts.l1ToL2Messages);
       }
-      this.l1ToL2Messages.push(...opts.l1ToL2Messages);
     }
 
     const [msGetEndState, endState] = await elapsed(() => this.db.getStateReference());
@@ -257,6 +257,12 @@ export class LightweightCheckpointBuilder {
     const block = new L2Block(newArchive, header, body, this.checkpointNumber, indexWithinCheckpoint);
     this.blocks.push(block);
 
+    // Accumulate the streaming bundle now that the block is fully built, so a mid-build throw above leaves the
+    // checkpoint's message list (and thus its inHash/rolling hash) consistent with the blocks actually built.
+    if (opts.l1ToL2Messages !== undefined) {
+      this.l1ToL2Messages.push(...opts.l1ToL2Messages);
+    }
+
     const [msSpongeAbsorb] = await elapsed(() => this.spongeBlob.absorb(blockBlobFields));
     timings.spongeAbsorb = msSpongeAbsorb;
     this.blobFields.push(...blockBlobFields);

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

@@ -950,9 +950,12 @@ export class CheckpointProposalJob implements Traceable {
       // Streaming Inbox: select this block's message bundle against the current (not-yet-advanced) consumption
       // cursor. The state is only advanced once the block builds successfully, so a failed build (retried in a
       // later sub-slot) re-derives the bundle rather than losing it. The builder inserts the bundle and rolls it
-      // back with the fork on failure.
+      // back with the fork on failure. The censorship cutoff floor must apply on whichever block ends the
+      // checkpoint, which includes the block that reaches the per-checkpoint block cap, not just the timetable's
+      // last sub-slot.
+      const isCheckpointFinalBlock = timingInfo.isLastBlock || blocksBuilt + 1 >= this.config.maxBlocksPerCheckpoint;
       const selection = streamingState
-        ? await this.selectStreamingBundle(streamingState, timingInfo.isLastBlock, nowSeconds)
+        ? await this.selectStreamingBundle(streamingState, isCheckpointFinalBlock, nowSeconds)
         : undefined;
       const streamingBundle = streamingState ? (selection && selection.consume ? selection.bundle : []) : undefined;