fix(fast-inbox): predict the next block's inbox messages when simulating public calls (A-1388)

A transaction that consumes an L1-to-L2 message sent moments ago failed public simulation until a
block actually consumed the message, because the simulation fork stops at the tip's message tree.
Predict the bundle the next block would take — the sequencer's own bucket selection, so lag
eligibility and the per-block/per-checkpoint caps match what will be built — and append it to the
fork. The prediction treats the next block as non-final (the censorship cutoff only widens
consumption on a checkpoint's last block, which the node cannot know) and derives its cursor from
the fork's own leaf count, so it can never re-insert messages the fork already holds. Best-effort:
unsynced buckets or a missing header leave the simulation on the bare tip, as before.

Author: spalladino

yarn-project/aztec-node/src/aztec-node/node_public_calls_simulator.test.ts

@@ -23,11 +23,13 @@ import {
 } from '@aztec/stdlib/block';
 import type { ProposedCheckpointData } from '@aztec/stdlib/checkpoint';
 import type { ContractDataSource } from '@aztec/stdlib/contract';
+import { EmptyL1RollupConstants } from '@aztec/stdlib/epoch-helpers';
 import { GasFees } from '@aztec/stdlib/gas';
 import type { MerkleTreeWriteOperations, WorldStateSynchronizer } from '@aztec/stdlib/interfaces/server';
+import type { InboxBucket, L1ToL2MessageSource } from '@aztec/stdlib/messaging';
 import { CheckpointHeader } from '@aztec/stdlib/rollup';
 import { mockTx } from '@aztec/stdlib/testing';
-import { AppendOnlyTreeSnapshot } from '@aztec/stdlib/trees';
+import { AppendOnlyTreeSnapshot, MerkleTreeId } from '@aztec/stdlib/trees';
 import {
   BlockHeader,
   type CheckpointGlobalVariables,
@@ -48,6 +50,7 @@ const ROLLUP_ADDRESS = EthAddress.random();
 describe('NodePublicCallsSimulator', () => {
   let blockSource: MockProxy<L2BlockSource>;
   let worldStateSynchronizer: MockProxy<WorldStateSynchronizer>;
+  let l1ToL2MessageSource: MockProxy<L1ToL2MessageSource>;
   let contractDataSource: MockProxy<ContractDataSource>;
   let globalVariableBuilder: MockProxy<GlobalVariableBuilder>;
   let rollupContract: MockProxy<RollupContract>;
@@ -111,6 +114,26 @@ describe('NodePublicCallsSimulator', () => {
     gasFees: GasFees.empty(),
   });
 
+  /**
+   * Mocks the Inbox so the next-block prediction selects a two-message bundle: the fork's message total (0)
+   * resolves to bucket 0, and bucket 1 is lag-eligible and holds both messages.
+   */
+  const mockInboxSelection = () => {
+    const makeBucket = (seq: bigint, totalMsgCount: bigint): InboxBucket => ({
+      seq,
+      inboxRollingHash: Fr.ZERO,
+      totalMsgCount,
+      timestamp: 0n,
+      msgCount: Number(totalMsgCount),
+      lastMessageIndex: totalMsgCount === 0n ? 0n : totalMsgCount - 1n,
+    });
+    const bundle = [new Fr(0x1234), new Fr(0x5678)];
+    l1ToL2MessageSource.getInboxBucketByTotalMsgCount.mockResolvedValue(makeBucket(0n, 0n));
+    l1ToL2MessageSource.getLatestInboxBucketAtOrBefore.mockResolvedValue(makeBucket(1n, 2n));
+    l1ToL2MessageSource.getL1ToL2MessagesBetweenBuckets.mockResolvedValue(bundle);
+    return bundle;
+  };
+
   const lowGasTx = () =>
     mockTx(0x10000, {
       numberOfNonRevertiblePublicCallRequests: 0,
@@ -124,6 +147,7 @@ describe('NodePublicCallsSimulator', () => {
 
     blockSource = mock<L2BlockSource>();
     worldStateSynchronizer = mock<WorldStateSynchronizer>();
+    l1ToL2MessageSource = mock<L1ToL2MessageSource>();
     contractDataSource = mock<ContractDataSource>();
     globalVariableBuilder = mock<GlobalVariableBuilder>();
     rollupContract = mock<RollupContract>();
@@ -136,8 +160,18 @@ describe('NodePublicCallsSimulator', () => {
     (merkleTreeFork as unknown as { [Symbol.asyncDispose]: () => Promise<void> })[Symbol.asyncDispose] = () =>
       Promise.resolve();
     worldStateSynchronizer.fork.mockResolvedValue(merkleTreeFork);
+    merkleTreeFork.getTreeInfo.mockResolvedValue({
+      treeId: MerkleTreeId.L1_TO_L2_MESSAGE_TREE,
+      root: Buffer.alloc(32),
+      size: 0n,
+      depth: 16,
+    });
     blockSource.getPendingChainValidationStatus.mockResolvedValue({ valid: true });
     blockSource.getProposedCheckpointData.mockResolvedValue(undefined);
+    // No Inbox bucket resolves to the fork's message total by default, so the next-block message prediction
+    // bails out and tests see the bare tip state unless they opt into it.
+    l1ToL2MessageSource.getInboxBucketByTotalMsgCount.mockResolvedValue(undefined);
+    epochCache.getL1Constants.mockReturnValue(EmptyL1RollupConstants);
 
     globalVariableBuilder.buildCheckpointGlobalVariables.mockImplementation((_c, _f, slotNumber) =>
       Promise.resolve(checkpointGlobals(slotNumber)),
@@ -163,6 +197,7 @@ describe('NodePublicCallsSimulator', () => {
     simulator = new NodePublicCallsSimulator({
       blockSource,
       worldStateSynchronizer,
+      l1ToL2MessageSource,
       contractDataSource,
       globalVariableBuilder,
       rollupContract,
@@ -216,16 +251,31 @@ describe('NodePublicCallsSimulator', () => {
       expect(rollupContract.getManaTarget).not.toHaveBeenCalled();
     });
 
-    it('does not insert L1-to-L2 messages', async () => {
+    it('appends the message bundle the next block would consume', async () => {
       const tx = await lowGasTx();
       setupMidCheckpoint();
       blockSource.getBlockData.mockImplementation((query: BlockQuery) =>
         Promise.resolve('number' in query ? makeBlockData(query.number, SlotNumber(42)) : undefined),
       );
       mockNextL1Slot(SlotNumber(100));
+      const bundle = mockInboxSelection();
 
       await simulator.simulate(tx);
 
+      expect(merkleTreeFork.appendLeaves).toHaveBeenCalledWith(MerkleTreeId.L1_TO_L2_MESSAGE_TREE, bundle);
+    });
+
+    it('simulates against the tip when the parent Inbox bucket is not synced', async () => {
+      const tx = await lowGasTx();
+      setupMidCheckpoint();
+      blockSource.getBlockData.mockImplementation((query: BlockQuery) =>
+        Promise.resolve('number' in query ? makeBlockData(query.number, SlotNumber(42)) : undefined),
+      );
+      mockNextL1Slot(SlotNumber(100));
+      // Default mock: no bucket resolves the fork's message total.
+
+      await expect(simulator.simulate(tx)).resolves.toBeDefined();
+
       expect(merkleTreeFork.appendLeaves).not.toHaveBeenCalled();
     });
 
@@ -273,18 +323,19 @@ describe('NodePublicCallsSimulator', () => {
       expect(plan?.chainTipsOverride).toEqual({ pending: CheckpointNumber(1), proven: CheckpointNumber(1) });
     });
 
-    it('does not insert L1-to-L2 messages when opening a new checkpoint', async () => {
+    it('appends the message bundle the next block would consume', async () => {
       const tx = await lowGasTx();
       blockSource.getL2Tips.mockResolvedValue(setupBoundary());
       blockSource.getBlockData.mockImplementation((query: BlockQuery) =>
         Promise.resolve('number' in query ? makeBlockData(query.number, SlotNumber(5)) : undefined),
       );
       mockNextL1Slot(SlotNumber(20));
+      const bundle = mockInboxSelection();
 
       await expect(simulator.simulate(tx)).resolves.toBeDefined();
 
-      // Streaming Inbox: the next checkpoint's messages are consumed per block, so none are appended here.
-      expect(merkleTreeFork.appendLeaves).not.toHaveBeenCalled();
+      // Only the first block's worth of messages: a fresh checkpoint starts its per-checkpoint budget at the tip.
+      expect(merkleTreeFork.appendLeaves).toHaveBeenCalledWith(MerkleTreeId.L1_TO_L2_MESSAGE_TREE, bundle);
     });
 
     it('targets parentSlot + 1 and carries the parent overrides when pipelining on a proposed checkpoint', async () => {
@@ -382,6 +433,7 @@ describe('NodePublicCallsSimulator', () => {
       new NodePublicCallsSimulator({
         blockSource,
         worldStateSynchronizer,
+        l1ToL2MessageSource,
         contractDataSource,
         globalVariableBuilder,
         epochCache,

yarn-project/aztec-node/src/aztec-node/node_public_calls_simulator.ts

@@ -1,3 +1,4 @@
+import { INBOX_LAG_SECONDS, MAX_L1_TO_L2_MSGS_PER_BLOCK, MAX_L1_TO_L2_MSGS_PER_CHECKPOINT } from '@aztec/constants';
 import { PROPOSER_PIPELINING_SLOT_OFFSET } from '@aztec/epoch-cache';
 import type { EpochCacheInterface } from '@aztec/epoch-cache';
 import {
@@ -11,14 +12,17 @@ import { EthAddress } from '@aztec/foundation/eth-address';
 import { BadRequestError } from '@aztec/foundation/json-rpc';
 import { type Logger, createLogger } from '@aztec/foundation/log';
 import { DateProvider } from '@aztec/foundation/timer';
+import { type InboxBucketSource, selectInboxBucketForBlock } from '@aztec/sequencer-client';
 import { type AvmSimulator, PublicContractsDB, PublicProcessorFactory } from '@aztec/simulator/server';
 import { CollectionLimitsConfig, PublicSimulatorConfig } from '@aztec/stdlib/avm';
 import { AztecAddress } from '@aztec/stdlib/aztec-address';
 import type { L2BlockSource, L2Tips } from '@aztec/stdlib/block';
 import { type ProposedCheckpointData, buildCheckpointSimulationOverridesPlan } from '@aztec/stdlib/checkpoint';
 import type { ContractDataSource } from '@aztec/stdlib/contract';
-import type { WorldStateSynchronizer } from '@aztec/stdlib/interfaces/server';
+import type { MerkleTreeWriteOperations, WorldStateSynchronizer } from '@aztec/stdlib/interfaces/server';
+import { type L1ToL2MessageSource, appendL1ToL2MessagesToTree, getInboxCutoffTimestamp } from '@aztec/stdlib/messaging';
 import type { CoordinationSignatureContext } from '@aztec/stdlib/p2p';
+import { MerkleTreeId } from '@aztec/stdlib/trees';
 import {
   type GlobalVariableBuilder,
   GlobalVariables,
@@ -30,6 +34,9 @@ import { type TelemetryClient, getTelemetryClient } from '@aztec/telemetry-clien
 
 import { applyPublicDataOverrides } from './public_data_overrides.js';
 
+/** Inbox queries the simulator needs to predict the message bundle the next block would consume. */
+type SimulatorInboxSource = InboxBucketSource & Pick<L1ToL2MessageSource, 'getInboxBucketByTotalMsgCount'>;
+
 /** Config fields the simulator needs — a narrow subset of `AztecNodeConfig`. */
 export interface NodePublicCallsSimulatorConfig {
   /** Maximum total gas limit accepted for an incoming simulation. */
@@ -42,6 +49,8 @@ export interface NodePublicCallsSimulatorConfig {
 export interface NodePublicCallsSimulatorDeps {
   blockSource: L2BlockSource;
   worldStateSynchronizer: WorldStateSynchronizer;
+  /** Inbox bucket queries, used to predict the L1-to-L2 messages the next block will consume. */
+  l1ToL2MessageSource: SimulatorInboxSource;
   contractDataSource: ContractDataSource;
   globalVariableBuilder: GlobalVariableBuilder;
   /**
@@ -73,15 +82,20 @@ export interface NodePublicCallsSimulatorDeps {
  * - **When the next block continues an in-progress checkpoint** (the latest proposed block is ahead of
  *   the proposed-checkpoint frontier): every block in a checkpoint shares the same
  *   `CheckpointGlobalVariables`, so we copy the latest proposed block's globals verbatim and only
- *   bump the block number. No L1 calls, no L1-to-L2 message insertion.
+ *   bump the block number. No L1 calls.
  * - **When the next block opens a new checkpoint** (the latest proposed block coincides with the
  *   proposed-checkpoint frontier): we compute fresh globals for the slot the next block will land in,
  *   applying the same `SimulationOverridesPlan` the sequencer applies so the simulated mana min fee
  *   matches what the sequencer will write into the block header.
+ *
+ * Either way it also predicts the L1-to-L2 message bundle the next block would consume and appends it
+ * to the fork, so a transaction consuming a message that is in the Inbox but not yet in a block
+ * simulates against the state it will actually run in.
  */
 export class NodePublicCallsSimulator {
   private readonly blockSource: L2BlockSource;
   private readonly worldStateSynchronizer: WorldStateSynchronizer;
+  private readonly l1ToL2MessageSource: SimulatorInboxSource;
   private readonly contractDataSource: ContractDataSource;
   private readonly globalVariableBuilder: GlobalVariableBuilder;
   private readonly rollupContract: RollupContract | undefined;
@@ -91,10 +105,12 @@ export class NodePublicCallsSimulator {
   private readonly avmSimulator?: AvmSimulator;
   private readonly telemetry: TelemetryClient;
   private readonly log: Logger;
+  private readonly dateProvider = new DateProvider();
 
   constructor(deps: NodePublicCallsSimulatorDeps) {
     this.blockSource = deps.blockSource;
     this.worldStateSynchronizer = deps.worldStateSynchronizer;
+    this.l1ToL2MessageSource = deps.l1ToL2MessageSource;
     this.contractDataSource = deps.contractDataSource;
     this.globalVariableBuilder = deps.globalVariableBuilder;
     this.rollupContract = deps.rollupContract;
@@ -160,7 +176,7 @@ export class NodePublicCallsSimulator {
     const publicProcessorFactory = new PublicProcessorFactory(
       this.contractDataSource,
       this.avmSimulator,
-      new DateProvider(),
+      this.dateProvider,
       this.telemetry,
       this.log.getBindings(),
     );
@@ -175,12 +191,15 @@ export class NodePublicCallsSimulator {
     // Ensure world-state has caught up with the latest block we loaded from the archiver
     await this.worldStateSynchronizer.syncImmediate(latestBlockNumber);
 
-    // Request a new fork of the world state at the latest block number, and apply any overrides to it before
-    // simulation. The next checkpoint's L1-to-L2 messages are not inserted here: under the streaming Inbox
-    // (AZIP-22 Fast Inbox) a checkpoint's messages are consumed per block, so the simulation runs against the
-    // fork's current tree without predicting the next block's message bundle.
+    // Request a new fork of the world state at the latest block number, then apply the next block's predicted
+    // L1-to-L2 message bundle and any caller overrides to it before simulation.
     await using merkleTreeFork = await this.worldStateSynchronizer.fork(latestBlockNumber);
 
+    await this.appendPredictedL1ToL2Messages(merkleTreeFork, {
+      slotNumber: newGlobalVariables.slotNumber,
+      checkpointStartBlock: atCheckpointBoundary ? undefined : proposedCheckpointLastBlock,
+    });
+
     await applyPublicDataOverrides(merkleTreeFork, overrides?.publicStorage);
 
     const config = PublicSimulatorConfig.from({
@@ -219,17 +238,89 @@ export class NodePublicCallsSimulator {
     );
   }
 
+  /**
+   * Appends the L1-to-L2 message bundle the next block would consume to the simulation fork, so a transaction
+   * consuming a message that has reached the Inbox but no block yet simulates against the state it will run in.
+   * Runs the same bucket selection the sequencer runs (lag eligibility plus the per-block and per-checkpoint caps),
+   * treating the next block as non-final: the censorship cutoff only widens consumption on a checkpoint's last
+   * block, and the node cannot know whether the next block is it.
+   *
+   * Best-effort. Any failure — Inbox buckets not synced yet, a torn archiver snapshot — leaves the fork at the tip
+   * state, which is what the transaction sees if the next block consumes nothing.
+   */
+  private async appendPredictedL1ToL2Messages(
+    fork: MerkleTreeWriteOperations,
+    opts: {
+      /** Slot the next block lands in; anchors the censorship cutoff. */
+      slotNumber: SlotNumber;
+      /** Last block of the checkpoint the next block extends; undefined when the next block opens a checkpoint. */
+      checkpointStartBlock: BlockNumber | undefined;
+    },
+  ): Promise<void> {
+    try {
+      const parentTotalMsgCount = (await fork.getTreeInfo(MerkleTreeId.L1_TO_L2_MESSAGE_TREE)).size;
+      const parentBucket = await this.l1ToL2MessageSource.getInboxBucketByTotalMsgCount(parentTotalMsgCount);
+      if (parentBucket === undefined) {
+        this.log.debug(`Inbox bucket at message total ${parentTotalMsgCount} not synced; simulating against the tip`, {
+          parentTotalMsgCount,
+        });
+        return;
+      }
+
+      // Origin of the per-checkpoint cap: the total consumed as of the checkpoint's parent. A block extending an
+      // in-progress checkpoint reads it off that checkpoint's parent block; a block opening one starts from the tip.
+      const checkpointStartTotalMsgCount =
+        opts.checkpointStartBlock === undefined
+          ? parentTotalMsgCount
+          : await this.getConsumedMessageTotal(opts.checkpointStartBlock);
+      if (checkpointStartTotalMsgCount === undefined) {
+        this.log.debug(`Block ${opts.checkpointStartBlock} has no header on this node; simulating against the tip`);
+        return;
+      }
+
+      const selection = await selectInboxBucketForBlock({
+        messageSource: this.l1ToL2MessageSource,
+        now: BigInt(Math.floor(this.dateProvider.now() / 1000)),
+        lagSeconds: BigInt(INBOX_LAG_SECONDS),
+        parent: { seq: parentBucket.seq, totalMsgCount: parentBucket.totalMsgCount },
+        checkpointStartTotalMsgCount,
+        perBlockCap: MAX_L1_TO_L2_MSGS_PER_BLOCK,
+        perCheckpointCap: MAX_L1_TO_L2_MSGS_PER_CHECKPOINT,
+        isLastBlock: false,
+        cutoffTimestamp: getInboxCutoffTimestamp(opts.slotNumber, this.epochCache.getL1Constants(), INBOX_LAG_SECONDS),
+      });
+      if (!selection.consume || selection.bundle.length === 0) {
+        return;
+      }
+
+      await appendL1ToL2MessagesToTree(fork, selection.bundle);
+      this.log.debug(`Appended ${selection.bundle.length} predicted L1-to-L2 messages to the simulation fork`, {
+        bucketSeq: selection.bucket.seq,
+        messageCount: selection.bundle.length,
+      });
+    } catch (err) {
+      this.log.verbose(`Could not predict the next block's L1-to-L2 messages, simulating against the tip: ${err}`);
+    }
+  }
+
+  /** Cumulative Inbox message total consumed as of `blockNumber`, i.e. its L1-to-L2 message tree leaf count. */
+  private async getConsumedMessageTotal(blockNumber: BlockNumber): Promise<bigint | undefined> {
+    if (blockNumber === BlockNumber.ZERO) {
+      return 0n;
+    }
+    const block = await this.blockSource.getBlockData({ number: blockNumber });
+    return block === undefined ? undefined : BigInt(block.header.state.l1ToL2MessageTree.nextAvailableLeafIndex);
+  }
+
   /**
    * Continues an in-progress checkpoint: the next block extends the checkpoint the latest proposed
    * block belongs to. Every block in a checkpoint shares the same `CheckpointGlobalVariables`, so the
    * next block's globals are the latest proposed block's globals with only the block number bumped —
-   * including the proposer's real coinbase/feeRecipient. No L1 reads and no L1-to-L2 message insertion
-   * happen here.
+   * including the proposer's real coinbase/feeRecipient. No L1 reads happen here.
    *
    * A missing header means the archiver reported a proposed tip via `getL2Tips` but no longer has its
    * data (a torn snapshot). We throw a transient/retryable error rather than treating the next block as
-   * opening a new checkpoint: the fork at `latestBlockNumber` already contains the ongoing checkpoint's
-   * L1-to-L2 messages, so inserting the next checkpoint's messages would append them a second time.
+   * opening a new checkpoint, whose globals would be built for the wrong slot.
    */
   private async copyGlobalVariablesFromLatestProposedBlock(
     latestBlockNumber: BlockNumber,

yarn-project/aztec-node/src/aztec-node/server.ts

@@ -241,6 +241,7 @@ export class AztecNodeService implements AztecNode, AztecNodeAdmin, AztecNodeDeb
     this.nodePublicCallsSimulator = new NodePublicCallsSimulator({
       blockSource: this.blockSource,
       worldStateSynchronizer: this.worldStateSynchronizer,
+      l1ToL2MessageSource: this.l1ToL2MessageSource,
       contractDataSource: this.contractDataSource,
       globalVariableBuilder: this.globalVariableBuilder,
       rollupContract: this.rollupContract,

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

@@ -6,3 +6,10 @@ export { Sequencer, SequencerState, type SequencerEvents } from './sequencer/ind
 // Used by the node to simulate public parts of transactions. Should these be moved to a shared library?
 // ISSUE(#9832)
 export * from './global_variable_builder/index.js';
+export {
+  type ConsumedBucketCursor,
+  type InboxBucketSelection,
+  type InboxBucketSource,
+  type SelectInboxBucketInput,
+  selectInboxBucketForBlock,
+} from './sequencer/inbox_bucket_selector.js';