From 41992ab410cbab605f868112e817599bf5c30bfc Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Sat, 18 Jul 2026 03:17:31 -0300 Subject: [PATCH 1/6] feat(fast-inbox): archiver syncs Inbox buckets (A-1379) --- .../archiver/src/archiver-sync.test.ts | 30 +++ .../archiver/src/l1/data_retrieval.ts | 3 + .../archiver/src/modules/data_source_base.ts | 14 +- .../archiver/src/store/data_stores.ts | 2 +- .../archiver/src/store/message_store.test.ts | 162 ++++++++++++++ .../archiver/src/store/message_store.ts | 197 +++++++++++++++++- .../archiver/src/structs/inbox_message.ts | 29 ++- .../archiver/src/test/fake_l1_state.ts | 64 +++++- .../archiver/src/test/mock_archiver.ts | 18 +- .../src/test/mock_l1_to_l2_message_source.ts | 27 ++- .../archiver/src/test/mock_structs.ts | 14 +- yarn-project/ethereum/src/contracts/inbox.ts | 65 ++++-- .../stdlib/src/interfaces/archiver.test.ts | 45 ++++ .../stdlib/src/interfaces/archiver.ts | 10 + .../stdlib/src/messaging/inbox_bucket.ts | 43 ++++ yarn-project/stdlib/src/messaging/index.ts | 1 + .../src/messaging/l1_to_l2_message_source.ts | 25 +++ 17 files changed, 712 insertions(+), 37 deletions(-) create mode 100644 yarn-project/stdlib/src/messaging/inbox_bucket.ts diff --git a/yarn-project/archiver/src/archiver-sync.test.ts b/yarn-project/archiver/src/archiver-sync.test.ts index c8595cfe9e43..cf940c547593 100644 --- a/yarn-project/archiver/src/archiver-sync.test.ts +++ b/yarn-project/archiver/src/archiver-sync.test.ts @@ -259,6 +259,36 @@ describe('Archiver Sync', () => { expect( (await archiver.getCheckpoints({ from: CheckpointNumber(1), limit: 100 })).map(b => b.checkpoint.number), ).toEqual([1, 2, 3]); + + // Inbox buckets: each of the three L1 message blocks opened its own bucket, in insertion order. + const t1 = fake.getTimestampAtL1Block(98n); + const t2 = fake.getTimestampAtL1Block(2504n); + const t3 = fake.getTimestampAtL1Block(2511n); + + expect(await archiver.getInboxBucket(1n)).toMatchObject({ + seq: 1n, + msgCount: 3, + totalMsgCount: 3n, + timestamp: t1, + }); + expect(await archiver.getInboxBucket(3n)).toMatchObject({ + seq: 3n, + msgCount: 3, + totalMsgCount: 9n, + timestamp: t3, + isOpen: true, + }); + expect((await archiver.getInboxBucket(1n))!.isOpen).toBe(false); + + // At-or-before lookups resolve the latest bucket not opened after the given timestamp. + expect((await archiver.getLatestInboxBucketAtOrBefore(t3))!.seq).toEqual(3n); + expect((await archiver.getLatestInboxBucketAtOrBefore(t2))!.seq).toEqual(2n); + expect(await archiver.getLatestInboxBucketAtOrBefore(t1 - 1n)).toBeUndefined(); + + // Messages between buckets, in insertion order. + expect(await archiver.getL1ToL2MessagesBetweenBuckets(0n, 3n)).toEqual([...msgs1, ...msgs2, ...msgs3]); + expect(await archiver.getL1ToL2MessagesBetweenBuckets(1n, 2n)).toEqual(msgs2); + expect(await archiver.getL1ToL2MessagesBetweenBuckets(2n, 3n)).toEqual(msgs3); }, 30_000); it('ignores checkpoint 3 because it has been pruned', async () => { diff --git a/yarn-project/archiver/src/l1/data_retrieval.ts b/yarn-project/archiver/src/l1/data_retrieval.ts index 18282920db60..8b245f24fb76 100644 --- a/yarn-project/archiver/src/l1/data_retrieval.ts +++ b/yarn-project/archiver/src/l1/data_retrieval.ts @@ -387,6 +387,9 @@ function mapLogInboxMessage(log: MessageSentLog): InboxMessage { l1BlockHash: log.l1BlockHash, checkpointNumber: log.args.checkpointNumber, rollingHash: log.args.rollingHash, + inboxRollingHash: log.args.inboxRollingHash, + bucketSeq: log.args.bucketSeq, + bucketTimestamp: log.l1BlockTimestamp, }; } diff --git a/yarn-project/archiver/src/modules/data_source_base.ts b/yarn-project/archiver/src/modules/data_source_base.ts index 6cde2b44b0d8..10da66d12bc4 100644 --- a/yarn-project/archiver/src/modules/data_source_base.ts +++ b/yarn-project/archiver/src/modules/data_source_base.ts @@ -40,7 +40,7 @@ import { } from '@aztec/stdlib/epoch-helpers'; import type { L2LogsSource } from '@aztec/stdlib/interfaces/server'; import type { LogResult, PrivateLogsQuery, PublicLogsQuery } from '@aztec/stdlib/logs'; -import type { L1ToL2MessageSource, L2ToL1MembershipWitness } from '@aztec/stdlib/messaging'; +import type { InboxBucket, L1ToL2MessageSource, L2ToL1MembershipWitness } from '@aztec/stdlib/messaging'; import { AppendOnlyTreeSnapshot } from '@aztec/stdlib/trees'; import type { BlockHeader, IndexedTxEffect, TxHash } from '@aztec/stdlib/tx'; import type { UInt64 } from '@aztec/stdlib/types'; @@ -324,6 +324,18 @@ export abstract class ArchiverDataSourceBase return this.stores.messages.getL1ToL2MessageIndex(l1ToL2Message); } + public getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise { + return this.stores.messages.getLatestInboxBucketAtOrBefore(timestamp); + } + + public getInboxBucket(seq: bigint): Promise { + return this.stores.messages.getInboxBucket(seq); + } + + public getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { + return this.stores.messages.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive); + } + private async getPublishedCheckpointFromCheckpointData(checkpoint: CheckpointData): Promise { const blocksForCheckpoint = await this.stores.blocks.getBlocksForCheckpoint(checkpoint.checkpointNumber); if (!blocksForCheckpoint) { diff --git a/yarn-project/archiver/src/store/data_stores.ts b/yarn-project/archiver/src/store/data_stores.ts index d34d9497b152..f463e06735ba 100644 --- a/yarn-project/archiver/src/store/data_stores.ts +++ b/yarn-project/archiver/src/store/data_stores.ts @@ -13,7 +13,7 @@ import { FunctionNamesCache } from './function_names_cache.js'; import { LogStore } from './log_store.js'; import { MessageStore } from './message_store.js'; -export const ARCHIVER_DB_VERSION = 7; +export const ARCHIVER_DB_VERSION = 8; /** * Represents the latest L1 block processed by the archiver for various objects in L2. diff --git a/yarn-project/archiver/src/store/message_store.test.ts b/yarn-project/archiver/src/store/message_store.test.ts index ef2ca57b9013..69fe30a75091 100644 --- a/yarn-project/archiver/src/store/message_store.test.ts +++ b/yarn-project/archiver/src/store/message_store.test.ts @@ -1,6 +1,7 @@ import { NUMBER_OF_L1_L2_MESSAGES_PER_ROLLUP } from '@aztec/constants'; import { CheckpointNumber } from '@aztec/foundation/branded-types'; import { Buffer16, Buffer32 } from '@aztec/foundation/buffer'; +import { Fr } from '@aztec/foundation/curves/bn254'; import { toArray } from '@aztec/foundation/iterable'; import { openTmpStore } from '@aztec/kv-store/lmdb-v2'; import { Checkpoint, type PublishedCheckpoint } from '@aztec/stdlib/checkpoint'; @@ -136,6 +137,7 @@ describe('MessageStore', () => { const msgs2 = makeInboxMessages(3, { initialCheckpointNumber: CheckpointNumber(20), initialHash: msgs1.at(-1)!.rollingHash, + initialInboxHash: msgs1.at(-1)!.inboxRollingHash, }); await messageStore.addL1ToL2Messages(msgs1); @@ -327,4 +329,164 @@ describe('MessageStore', () => { }); }); }); + + describe('Inbox buckets', () => { + // Builds `count` consecutive valid messages in a single checkpoint, then reassigns their bucket sequence and + // timestamp per the given per-message spec so we can exercise multi-message and rollover buckets. + const makeBucketedMessages = (spec: { seq: bigint; timestamp: bigint }[]): InboxMessage[] => { + const msgs = makeInboxMessages(spec.length, { + initialCheckpointNumber: CheckpointNumber(1), + messagesPerCheckpoint: spec.length, + }); + msgs.forEach((msg, i) => { + msg.bucketSeq = spec[i].seq; + msg.bucketTimestamp = spec[i].timestamp; + }); + return msgs; + }; + + // Three buckets over six messages: bucket 1 = [0,1,2], bucket 2 = [3,4], bucket 3 = [5]. + const threeBucketSpec = [ + { seq: 1n, timestamp: 100n }, + { seq: 1n, timestamp: 100n }, + { seq: 1n, timestamp: 100n }, + { seq: 2n, timestamp: 200n }, + { seq: 2n, timestamp: 200n }, + { seq: 3n, timestamp: 300n }, + ]; + + it('snapshots buckets as messages are inserted', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2Messages(msgs); + + expect(await messageStore.getInboxBucket(1n)).toEqual({ + seq: 1n, + inboxRollingHash: msgs[2].inboxRollingHash, + totalMsgCount: 3n, + timestamp: 100n, + msgCount: 3, + lastMessageIndex: msgs[2].index, + isOpen: false, + }); + expect(await messageStore.getInboxBucket(2n)).toEqual({ + seq: 2n, + inboxRollingHash: msgs[4].inboxRollingHash, + totalMsgCount: 5n, + timestamp: 200n, + msgCount: 2, + lastMessageIndex: msgs[4].index, + isOpen: false, + }); + expect(await messageStore.getInboxBucket(3n)).toEqual({ + seq: 3n, + inboxRollingHash: msgs[5].inboxRollingHash, + totalMsgCount: 6n, + timestamp: 300n, + msgCount: 1, + lastMessageIndex: msgs[5].index, + isOpen: true, + }); + expect(await messageStore.getInboxBucket(4n)).toBeUndefined(); + }); + + it('continues a bucket that spans two insertion batches', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2Messages(msgs.slice(0, 2)); + await messageStore.addL1ToL2Messages(msgs.slice(2)); + + // Bucket 1 keeps accumulating across the batch boundary rather than restarting its message count. + expect(await messageStore.getInboxBucket(1n)).toMatchObject({ msgCount: 3, totalMsgCount: 3n }); + expect(await messageStore.getInboxBucket(3n)).toMatchObject({ msgCount: 1, totalMsgCount: 6n }); + }); + + it('throws if the consensus rolling hash is not correct', async () => { + const msgs = makeInboxMessages(5); + msgs[1].inboxRollingHash = Fr.random(); + await expect(messageStore.addL1ToL2Messages(msgs)).rejects.toThrow(MessageStoreError); + }); + + it('resolves the latest bucket at or before a timestamp', async () => { + await messageStore.addL1ToL2Messages(makeBucketedMessages(threeBucketSpec)); + + expect((await messageStore.getLatestInboxBucketAtOrBefore(100n))!.seq).toEqual(1n); + expect((await messageStore.getLatestInboxBucketAtOrBefore(150n))!.seq).toEqual(1n); + expect((await messageStore.getLatestInboxBucketAtOrBefore(300n))!.seq).toEqual(3n); + expect((await messageStore.getLatestInboxBucketAtOrBefore(10_000n))!.seq).toEqual(3n); + expect(await messageStore.getLatestInboxBucketAtOrBefore(99n)).toBeUndefined(); + }); + + it('resolves rollover buckets that share a timestamp to the highest sequence', async () => { + // Buckets 2 and 3 share timestamp 200 (a full bucket rolling over within the same L1 block). + const msgs = makeBucketedMessages([ + { seq: 1n, timestamp: 100n }, + { seq: 2n, timestamp: 200n }, + { seq: 3n, timestamp: 200n }, + ]); + await messageStore.addL1ToL2Messages(msgs); + + expect((await messageStore.getLatestInboxBucketAtOrBefore(200n))!.seq).toEqual(3n); + }); + + it('returns messages between buckets in insertion order', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2Messages(msgs); + const leaves = msgs.map(m => m.leaf); + + expect(await messageStore.getL1ToL2MessagesBetweenBuckets(0n, 3n)).toEqual(leaves); + expect(await messageStore.getL1ToL2MessagesBetweenBuckets(1n, 2n)).toEqual(leaves.slice(3, 5)); + expect(await messageStore.getL1ToL2MessagesBetweenBuckets(2n, 3n)).toEqual(leaves.slice(5)); + // An empty (fromExclusive, toInclusive] range yields no messages. + expect(await messageStore.getL1ToL2MessagesBetweenBuckets(3n, 3n)).toEqual([]); + // Unknown upper bucket yields no messages. + expect(await messageStore.getL1ToL2MessagesBetweenBuckets(0n, 9n)).toEqual([]); + }); + + it('rewinds buckets when messages are removed', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2Messages(msgs); + + // Remove the last two messages (msgs[4] in bucket 2, msgs[5] in bucket 3), splitting bucket 2. + await messageStore.removeL1ToL2Messages(msgs[4].index); + + expect(await messageStore.getInboxBucket(3n)).toBeUndefined(); + expect(await messageStore.getInboxBucket(2n)).toEqual({ + seq: 2n, + inboxRollingHash: msgs[3].inboxRollingHash, + totalMsgCount: 4n, + timestamp: 200n, + msgCount: 1, + lastMessageIndex: msgs[3].index, + isOpen: true, + }); + expect(await messageStore.getInboxBucket(1n)).toMatchObject({ msgCount: 3, totalMsgCount: 3n, isOpen: false }); + + // Bucket 3's timestamp index entry is gone, so an at-or-before lookup falls back to bucket 2. + expect((await messageStore.getLatestInboxBucketAtOrBefore(300n))!.seq).toEqual(2n); + }); + + it('rewinds a rollover bucket sharing a timestamp with the surviving boundary', async () => { + const msgs = makeBucketedMessages([ + { seq: 1n, timestamp: 100n }, + { seq: 2n, timestamp: 200n }, + { seq: 3n, timestamp: 200n }, + ]); + await messageStore.addL1ToL2Messages(msgs); + + // Removing the last message deletes bucket 3, whose timestamp (200) is shared with the surviving bucket 2. + await messageStore.removeL1ToL2Messages(msgs[2].index); + + expect(await messageStore.getInboxBucket(3n)).toBeUndefined(); + expect((await messageStore.getLatestInboxBucketAtOrBefore(200n))!.seq).toEqual(2n); + }); + + it('clears all buckets when every message is removed', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2Messages(msgs); + + await messageStore.removeL1ToL2Messages(msgs[0].index); + + expect(await messageStore.getInboxBucket(1n)).toBeUndefined(); + expect(await messageStore.getLatestInboxBucketAtOrBefore(300n)).toBeUndefined(); + }); + }); }); diff --git a/yarn-project/archiver/src/store/message_store.ts b/yarn-project/archiver/src/store/message_store.ts index 5aaf949ee3b0..1513df52a905 100644 --- a/yarn-project/archiver/src/store/message_store.ts +++ b/yarn-project/archiver/src/store/message_store.ts @@ -4,7 +4,7 @@ import { Buffer16, Buffer32 } from '@aztec/foundation/buffer'; import { Fr } from '@aztec/foundation/curves/bn254'; import { toArray } from '@aztec/foundation/iterable'; import { createLogger } from '@aztec/foundation/log'; -import { BufferReader, serializeToBuffer } from '@aztec/foundation/serialize'; +import { BufferReader, bigintToUInt64BE, numToUInt32BE, serializeToBuffer } from '@aztec/foundation/serialize'; import { type AztecAsyncKVStore, type AztecAsyncMap, @@ -12,7 +12,7 @@ import { type CustomRange, mapRange, } from '@aztec/kv-store'; -import { InboxLeaf } from '@aztec/stdlib/messaging'; +import { type InboxBucket, InboxLeaf, updateInboxRollingHash } from '@aztec/stdlib/messaging'; import { L1ToL2MessagesNotReadyError } from '../errors.js'; import { @@ -22,6 +22,38 @@ import { updateRollingHash, } from '../structs/inbox_message.js'; +/** + * Persisted snapshot of an Inbox rolling-hash bucket. Mirrors the fields the on-chain Inbox tracks per bucket, + * plus the last absorbed message index so the between-buckets query can range-scan messages directly. + */ +type BucketSnapshot = { + inboxRollingHash: Fr; + totalMsgCount: bigint; + timestamp: bigint; + msgCount: number; + lastMessageIndex: bigint; +}; + +function serializeBucketSnapshot(snapshot: BucketSnapshot): Buffer { + return serializeToBuffer([ + snapshot.inboxRollingHash, + bigintToUInt64BE(snapshot.totalMsgCount), + bigintToUInt64BE(snapshot.timestamp), + numToUInt32BE(snapshot.msgCount), + bigintToUInt64BE(snapshot.lastMessageIndex), + ]); +} + +function deserializeBucketSnapshot(buffer: Buffer): BucketSnapshot { + const reader = BufferReader.asReader(buffer); + const inboxRollingHash = reader.readObject(Fr); + const totalMsgCount = reader.readUInt64(); + const timestamp = reader.readUInt64(); + const msgCount = reader.readNumber(); + const lastMessageIndex = reader.readUInt64(); + return { inboxRollingHash, totalMsgCount, timestamp, msgCount, lastMessageIndex }; +} + export class MessageStoreError extends Error { constructor( message: string, @@ -45,6 +77,10 @@ export class MessageStore { #inboxTreeInProgress: AztecAsyncSingleton; /** Stores the L1 finalized block as of the last successful message sync. */ #messagesFinalizedL1Block: AztecAsyncSingleton; + /** Maps from Inbox bucket sequence number to its serialized snapshot. */ + #inboxBuckets: AztecAsyncMap; + /** Maps from a bucket's L1 timestamp (key) to the highest bucket sequence number opened at that timestamp. */ + #bucketTimestampToSeq: AztecAsyncMap; #log = createLogger('archiver:message_store'); @@ -55,6 +91,8 @@ export class MessageStore { this.#totalMessageCount = db.openSingleton('archiver_l1_to_l2_message_count'); this.#inboxTreeInProgress = db.openSingleton('archiver_inbox_tree_in_progress'); this.#messagesFinalizedL1Block = db.openSingleton('archiver_messages_finalized_l1_block'); + this.#inboxBuckets = db.openMap('archiver_inbox_buckets'); + this.#bucketTimestampToSeq = db.openMap('archiver_inbox_bucket_timestamps'); } public async getTotalL1ToL2MessageCount(): Promise { @@ -112,6 +150,14 @@ export class MessageStore { let lastMessage = await this.getLastMessage(); let messageCount = 0; + // Running cumulative message count and in-progress bucket state, threaded across the batch so we can snapshot + // each Inbox bucket as its messages are inserted. Seeded from the last stored message so a bucket that spans + // two batches keeps accumulating. + let cumulativeTotal = await this.getTotalL1ToL2MessageCount(); + let currentBucketSeq: bigint | undefined = lastMessage?.bucketSeq; + let currentBucketMsgCount = + currentBucketSeq !== undefined ? ((await this.getBucketSnapshotBySeq(currentBucketSeq))?.msgCount ?? 0) : 0; + for (const message of messages) { // Check messages are inserted in increasing order, but allow reinserting messages. if (lastMessage && message.index <= lastMessage.index) { @@ -141,6 +187,20 @@ export class MessageStore { ); } + // Check the full-width consensus rolling hash is valid (AZIP-22 Fast Inbox). Runs alongside the legacy + // 128-bit check above until the streaming inbox flips on and the legacy hash is removed. + const previousInboxRollingHash = lastMessage?.inboxRollingHash ?? Fr.ZERO; + const expectedInboxRollingHash = updateInboxRollingHash(previousInboxRollingHash, message.leaf); + if (!expectedInboxRollingHash.equals(message.inboxRollingHash)) { + throw new MessageStoreError( + `Invalid inbox rolling hash for incoming L1 to L2 message ${message.leaf.toString()} ` + + `with index ${message.index} ` + + `(expected ${expectedInboxRollingHash.toString()} from previous hash ${previousInboxRollingHash.toString()} ` + + `but got ${message.inboxRollingHash.toString()})`, + message, + ); + } + // Check index corresponds to the checkpoint number. const [expectedStart, expectedEnd] = InboxLeaf.indexRangeForCheckpoint(message.checkpointNumber); if (message.index < expectedStart || message.index >= expectedEnd) { @@ -180,6 +240,23 @@ export class MessageStore { await this.#l1ToL2Messages.set(this.indexToKey(message.index), serializeInboxMessage(message)); await this.#l1ToL2MessageIndices.set(this.leafToIndexKey(message.leaf), message.index); messageCount++; + + // Snapshot the bucket this message was absorbed into. A message opens a new bucket whenever its bucket + // sequence differs from the one currently being accumulated; otherwise it extends the current bucket. + cumulativeTotal += 1n; + if (currentBucketSeq === undefined || message.bucketSeq !== currentBucketSeq) { + currentBucketSeq = message.bucketSeq; + currentBucketMsgCount = 0; + } + currentBucketMsgCount += 1; + await this.writeBucketSnapshot(message.bucketSeq, { + inboxRollingHash: message.inboxRollingHash, + totalMsgCount: cumulativeTotal, + timestamp: message.bucketTimestamp, + msgCount: currentBucketMsgCount, + lastMessageIndex: message.index, + }); + this.#log.trace(`Inserted L1 to L2 message ${message.leaf} with index ${message.index} into the store`); lastMessage = message; } @@ -281,10 +358,126 @@ export class MessageStore { deleteCount++; } await this.increaseTotalMessageCount(-deleteCount); + await this.rewindBucketsAfterRemoval(); this.#log.warn(`Deleted ${deleteCount} L1 to L2 messages from index ${startIndex} from the store`); }); } + /** + * Rewinds the Inbox bucket snapshots to match the messages remaining after a removal. Buckets whose messages + * were all removed are deleted, and the boundary bucket (the one holding the last surviving message) is + * recomputed from its remaining messages, since a checkpoint-aligned removal can split a bucket. Must run + * inside the removal transaction, after the total message count has been updated. + */ + private async rewindBucketsAfterRemoval(): Promise { + const lastRemaining = await this.getLastMessage(); + const boundarySeq = lastRemaining?.bucketSeq; + + // Delete snapshots (and their timestamp index entries) for buckets entirely past the surviving tip. + const deleteFromKey = boundarySeq === undefined ? 0 : this.bucketSeqToKey(boundarySeq) + 1; + for await (const [seqKey, snapBuffer] of this.#inboxBuckets.entriesAsync({ start: deleteFromKey })) { + const snapshot = deserializeBucketSnapshot(snapBuffer); + await this.#bucketTimestampToSeq.delete(this.timestampToKey(snapshot.timestamp)); + await this.#inboxBuckets.delete(seqKey); + } + + // Recompute the boundary bucket from its surviving messages. This also restores its timestamp index entry if a + // just-deleted rollover bucket shared the timestamp. + if (lastRemaining !== undefined && boundarySeq !== undefined) { + let msgCount = 0; + for await (const msg of this.iterateL1ToL2Messages({ reverse: true })) { + if (msg.bucketSeq !== boundarySeq) { + break; + } + msgCount += 1; + } + await this.writeBucketSnapshot(boundarySeq, { + inboxRollingHash: lastRemaining.inboxRollingHash, + totalMsgCount: await this.getTotalL1ToL2MessageCount(), + timestamp: lastRemaining.bucketTimestamp, + msgCount, + lastMessageIndex: lastRemaining.index, + }); + } + } + + /** + * Returns the Inbox bucket with the given sequence number, or undefined if it has not been synced (AZIP-22 Fast + * Inbox). + */ + public async getInboxBucket(seq: bigint): Promise { + const snapshot = await this.getBucketSnapshotBySeq(seq); + return snapshot && this.toInboxBucket(seq, snapshot); + } + + /** + * Returns the latest Inbox bucket opened at or before the given L1 timestamp, or undefined if every synced bucket + * was opened strictly after it (AZIP-22 Fast Inbox). + */ + public async getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise { + // Bucket timestamps are non-decreasing in sequence number, and the index holds the highest sequence per + // timestamp. A reverse scan bounded above (inclusively) by the requested timestamp yields, first, the value + // at the largest timestamp at-or-before it — the bucket sequence we want. + const [seq] = await toArray( + this.#bucketTimestampToSeq.valuesAsync({ end: this.timestampToKey(timestamp), reverse: true, limit: 1 }), + ); + return seq === undefined ? undefined : this.getInboxBucket(BigInt(seq)); + } + + /** + * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion + * order (AZIP-22 Fast Inbox). Returns an empty array if the upper bucket has not been synced. + */ + public async getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { + const toBucket = await this.getBucketSnapshotBySeq(toInclusive); + if (toBucket === undefined) { + return []; + } + const fromBucket = fromExclusive > 0n ? await this.getBucketSnapshotBySeq(fromExclusive) : undefined; + const startIndex = fromBucket ? fromBucket.lastMessageIndex + 1n : 0n; + const endIndexExclusive = toBucket.lastMessageIndex + 1n; + + const leaves: Fr[] = []; + for await (const msgBuffer of this.#l1ToL2Messages.valuesAsync({ + start: this.indexToKey(startIndex), + end: this.indexToKey(endIndexExclusive), + })) { + leaves.push(deserializeInboxMessage(msgBuffer).leaf); + } + return leaves; + } + + private async getBucketSnapshotBySeq(seq: bigint): Promise { + const buffer = await this.#inboxBuckets.getAsync(this.bucketSeqToKey(seq)); + return buffer && deserializeBucketSnapshot(buffer); + } + + private async writeBucketSnapshot(seq: bigint, snapshot: BucketSnapshot): Promise { + await this.#inboxBuckets.set(this.bucketSeqToKey(seq), serializeBucketSnapshot(snapshot)); + await this.#bucketTimestampToSeq.set(this.timestampToKey(snapshot.timestamp), this.bucketSeqToKey(seq)); + } + + private async toInboxBucket(seq: bigint, snapshot: BucketSnapshot): Promise { + const lastMessage = await this.getLastMessage(); + return { + seq, + inboxRollingHash: snapshot.inboxRollingHash, + totalMsgCount: snapshot.totalMsgCount, + timestamp: snapshot.timestamp, + msgCount: snapshot.msgCount, + lastMessageIndex: snapshot.lastMessageIndex, + isOpen: lastMessage?.bucketSeq === seq, + }; + } + + private bucketSeqToKey(seq: bigint): number { + return Number(seq); + } + + private timestampToKey(timestamp: bigint): number { + return Number(timestamp); + } + public rollbackL1ToL2MessagesToCheckpoint(targetCheckpointNumber: CheckpointNumber): Promise { this.#log.debug(`Deleting L1 to L2 messages up to target checkpoint ${targetCheckpointNumber}`); const startIndex = InboxLeaf.smallestIndexForCheckpoint(CheckpointNumber(targetCheckpointNumber + 1)); diff --git a/yarn-project/archiver/src/structs/inbox_message.ts b/yarn-project/archiver/src/structs/inbox_message.ts index 9f981b505e9b..2e88e69cfe78 100644 --- a/yarn-project/archiver/src/structs/inbox_message.ts +++ b/yarn-project/archiver/src/structs/inbox_message.ts @@ -10,7 +10,14 @@ export type InboxMessage = { checkpointNumber: CheckpointNumber; l1BlockNumber: bigint; l1BlockHash: Buffer32; + /** Legacy 128-bit keccak rolling hash of all messages inserted up to and including this one. */ rollingHash: Buffer16; + /** Consensus rolling hash (truncated sha256 chain) of all messages up to and including this one (AZIP-22 Fast Inbox). */ + inboxRollingHash: Fr; + /** Sequence number of the Inbox bucket this message was absorbed into (AZIP-22 Fast Inbox). */ + bucketSeq: bigint; + /** L1 block timestamp at which this message's bucket was opened; the bucket's recency key, in seconds. */ + bucketTimestamp: bigint; }; export function updateRollingHash(currentRollingHash: Buffer16, leaf: Fr): Buffer16 { @@ -23,9 +30,12 @@ export function serializeInboxMessage(message: InboxMessage): Buffer { bigintToUInt64BE(message.index), message.leaf, message.l1BlockHash, - numToUInt32BE(Number(message.l1BlockNumber)), + bigintToUInt64BE(message.l1BlockNumber), numToUInt32BE(message.checkpointNumber), message.rollingHash, + message.inboxRollingHash, + bigintToUInt64BE(message.bucketSeq), + bigintToUInt64BE(message.bucketTimestamp), ]); } @@ -34,8 +44,21 @@ export function deserializeInboxMessage(buffer: Buffer): InboxMessage { const index = reader.readUInt64(); const leaf = reader.readObject(Fr); const l1BlockHash = reader.readObject(Buffer32); - const l1BlockNumber = BigInt(reader.readNumber()); + const l1BlockNumber = reader.readUInt64(); const checkpointNumber = CheckpointNumber(reader.readNumber()); const rollingHash = reader.readObject(Buffer16); - return { index, leaf, l1BlockHash, l1BlockNumber, checkpointNumber, rollingHash }; + const inboxRollingHash = reader.readObject(Fr); + const bucketSeq = reader.readUInt64(); + const bucketTimestamp = reader.readUInt64(); + return { + index, + leaf, + l1BlockHash, + l1BlockNumber, + checkpointNumber, + rollingHash, + inboxRollingHash, + bucketSeq, + bucketTimestamp, + }; } diff --git a/yarn-project/archiver/src/test/fake_l1_state.ts b/yarn-project/archiver/src/test/fake_l1_state.ts index d5e5bf6e2197..28226f12020b 100644 --- a/yarn-project/archiver/src/test/fake_l1_state.ts +++ b/yarn-project/archiver/src/test/fake_l1_state.ts @@ -14,7 +14,7 @@ import { RollupAbi } from '@aztec/l1-artifacts'; import { CommitteeAttestation, CommitteeAttestationsAndSigners, L2Block } from '@aztec/stdlib/block'; import { Checkpoint } from '@aztec/stdlib/checkpoint'; import { getSlotAtTimestamp } from '@aztec/stdlib/epoch-helpers'; -import { InboxLeaf } from '@aztec/stdlib/messaging'; +import { InboxLeaf, updateInboxRollingHash } from '@aztec/stdlib/messaging'; import { ConsensusPayload, getHashedSignaturePayloadTypedData } from '@aztec/stdlib/p2p'; import { mockCheckpointAndMessages } from '@aztec/stdlib/testing'; import { AppendOnlyTreeSnapshot } from '@aztec/stdlib/trees'; @@ -109,8 +109,13 @@ type MessageData = { index: bigint; leaf: Fr; rollingHash: Buffer16; + inboxRollingHash: Fr; + bucketSeq: bigint; }; +// Mirror of the on-chain per-bucket message cap: further messages in the same L1 block spill into the next bucket. +const MAX_MSGS_PER_BUCKET = 256; + /** * Stateful fake for L1 data used by the archiver. * @@ -138,6 +143,11 @@ export class FakeL1State { private checkpoints: CheckpointData[] = []; private messages: MessageData[] = []; private messagesRollingHash: Buffer16 = Buffer16.ZERO; + // Consensus rolling-hash and bucket-ring state, mirroring the on-chain Inbox (AZIP-22 Fast Inbox). + private messagesConsensusRollingHash: Fr = Fr.ZERO; + private currentBucketSeq: bigint = 0n; + private currentBucketTimestamp: bigint = 0n; + private currentBucketMsgCount: number = 0; private lastArchive: AppendOnlyTreeSnapshot; private provenCheckpointNumber: CheckpointNumber = CheckpointNumber(0); private targetCommitteeSize: number = 0; @@ -166,9 +176,11 @@ export class FakeL1State { * Use this method only when you need to add messages without creating a checkpoint (e.g., for reorg tests). */ addMessages(checkpointNumber: CheckpointNumber, l1BlockNumber: bigint, messageLeaves: Fr[]): void { + const timestamp = this.getTimestampAtL1Block(l1BlockNumber); messageLeaves.forEach((leaf, i) => { const index = InboxLeaf.smallestIndexForCheckpoint(checkpointNumber) + BigInt(i); this.messagesRollingHash = updateRollingHash(this.messagesRollingHash, leaf); + const { bucketSeq, inboxRollingHash } = this.absorbIntoBucket(leaf, timestamp); this.messages.push({ l1BlockNumber, @@ -176,10 +188,47 @@ export class FakeL1State { index, leaf, rollingHash: this.messagesRollingHash, + inboxRollingHash, + bucketSeq, }); }); } + /** Mirrors the on-chain `_absorbIntoBucket`: opens a new bucket on a strictly larger timestamp or a full bucket. */ + private absorbIntoBucket(leaf: Fr, timestamp: bigint): { bucketSeq: bigint; inboxRollingHash: Fr } { + if ( + this.currentBucketSeq === 0n || + this.currentBucketTimestamp < timestamp || + this.currentBucketMsgCount === MAX_MSGS_PER_BUCKET + ) { + this.currentBucketSeq += 1n; + this.currentBucketTimestamp = timestamp; + this.currentBucketMsgCount = 0; + } + this.messagesConsensusRollingHash = updateInboxRollingHash(this.messagesConsensusRollingHash, leaf); + this.currentBucketMsgCount += 1; + return { bucketSeq: this.currentBucketSeq, inboxRollingHash: this.messagesConsensusRollingHash }; + } + + /** Rebuilds all per-message derived state (rolling hashes and bucket assignments) after the message set changes. */ + private recomputeDerivedMessageState(): void { + this.messagesRollingHash = Buffer16.ZERO; + this.messagesConsensusRollingHash = Fr.ZERO; + this.currentBucketSeq = 0n; + this.currentBucketTimestamp = 0n; + this.currentBucketMsgCount = 0; + for (const msg of this.messages) { + this.messagesRollingHash = updateRollingHash(this.messagesRollingHash, msg.leaf); + const { bucketSeq, inboxRollingHash } = this.absorbIntoBucket( + msg.leaf, + this.getTimestampAtL1Block(msg.l1BlockNumber), + ); + msg.rollingHash = this.messagesRollingHash; + msg.inboxRollingHash = inboxRollingHash; + msg.bucketSeq = bucketSeq; + } + } + /** * Creates blocks for a checkpoint without adding them to L1 state. * Useful for creating blocks to pass to addBlock() for testing provisional block handling. @@ -354,8 +403,7 @@ export class FakeL1State { */ removeMessagesAfter(totalIndex: number): void { this.messages = this.messages.slice(0, totalIndex); - // Recalculate rolling hash - this.messagesRollingHash = this.messages.reduce((hash, msg) => updateRollingHash(hash, msg.leaf), Buffer16.ZERO); + this.recomputeDerivedMessageState(); } /** @@ -623,13 +671,14 @@ export class FakeL1State { l1BlockNumber: msg.l1BlockNumber, l1BlockHash: Buffer32.fromBigInt(msg.l1BlockNumber), l1TransactionHash: `0x${msg.l1BlockNumber.toString(16)}` as `0x${string}`, + l1BlockTimestamp: this.getTimestampAtL1Block(msg.l1BlockNumber), args: { checkpointNumber: msg.checkpointNumber, index: msg.index, leaf: msg.leaf, rollingHash: msg.rollingHash, - inboxRollingHash: Fr.ZERO, - bucketSeq: 0n, + inboxRollingHash: msg.inboxRollingHash, + bucketSeq: msg.bucketSeq, }, })); } @@ -648,13 +697,14 @@ export class FakeL1State { l1BlockNumber: msg.l1BlockNumber, l1BlockHash: Buffer32.fromBigInt(msg.l1BlockNumber), l1TransactionHash: `0x${msg.l1BlockNumber.toString(16)}` as `0x${string}`, + l1BlockTimestamp: this.getTimestampAtL1Block(msg.l1BlockNumber), args: { checkpointNumber: msg.checkpointNumber, index: msg.index, leaf: msg.leaf, rollingHash: msg.rollingHash, - inboxRollingHash: Fr.ZERO, - bucketSeq: 0n, + inboxRollingHash: msg.inboxRollingHash, + bucketSeq: msg.bucketSeq, }, }; } diff --git a/yarn-project/archiver/src/test/mock_archiver.ts b/yarn-project/archiver/src/test/mock_archiver.ts index bcdcc3928d96..4449f9eff9be 100644 --- a/yarn-project/archiver/src/test/mock_archiver.ts +++ b/yarn-project/archiver/src/test/mock_archiver.ts @@ -2,7 +2,7 @@ import type { CheckpointNumber } from '@aztec/foundation/branded-types'; import type { Fr } from '@aztec/foundation/curves/bn254'; import type { L2BlockSource } from '@aztec/stdlib/block'; import type { Checkpoint } from '@aztec/stdlib/checkpoint'; -import type { L1ToL2MessageSource } from '@aztec/stdlib/messaging'; +import type { InboxBucket, L1ToL2MessageSource } from '@aztec/stdlib/messaging'; import { MockL1ToL2MessageSource } from './mock_l1_to_l2_message_source.js'; import { MockL2BlockSource } from './mock_l2_block_source.js'; @@ -17,6 +17,10 @@ export class MockArchiver extends MockL2BlockSource implements L2BlockSource, L1 this.messageSource.setL1ToL2Messages(checkpointNumber, msgs); } + public setInboxBucket(bucket: InboxBucket, msgs: Fr[] = []) { + this.messageSource.setInboxBucket(bucket, msgs); + } + getL1ToL2Messages(checkpointNumber: CheckpointNumber): Promise { return this.messageSource.getL1ToL2Messages(checkpointNumber); } @@ -24,6 +28,18 @@ export class MockArchiver extends MockL2BlockSource implements L2BlockSource, L1 getL1ToL2MessageIndex(_l1ToL2Message: Fr): Promise { return this.messageSource.getL1ToL2MessageIndex(_l1ToL2Message); } + + getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise { + return this.messageSource.getLatestInboxBucketAtOrBefore(timestamp); + } + + getInboxBucket(seq: bigint): Promise { + return this.messageSource.getInboxBucket(seq); + } + + getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { + return this.messageSource.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive); + } } /** diff --git a/yarn-project/archiver/src/test/mock_l1_to_l2_message_source.ts b/yarn-project/archiver/src/test/mock_l1_to_l2_message_source.ts index f3b72a1c940f..75b6ee1cb859 100644 --- a/yarn-project/archiver/src/test/mock_l1_to_l2_message_source.ts +++ b/yarn-project/archiver/src/test/mock_l1_to_l2_message_source.ts @@ -1,13 +1,15 @@ import { BlockNumber, CheckpointNumber } from '@aztec/foundation/branded-types'; import { Fr } from '@aztec/foundation/curves/bn254'; import type { CheckpointId, L2BlockId, L2TipId, L2Tips } from '@aztec/stdlib/block'; -import type { L1ToL2MessageSource } from '@aztec/stdlib/messaging'; +import type { InboxBucket, L1ToL2MessageSource } from '@aztec/stdlib/messaging'; /** * A mocked implementation of L1ToL2MessageSource to be used in tests. */ export class MockL1ToL2MessageSource implements L1ToL2MessageSource { private messagesPerCheckpoint = new Map(); + private buckets = new Map(); + private messagesPerBucket = new Map(); constructor(private blockNumber: number) {} @@ -15,6 +17,11 @@ export class MockL1ToL2MessageSource implements L1ToL2MessageSource { this.messagesPerCheckpoint.set(checkpointNumber, msgs); } + public setInboxBucket(bucket: InboxBucket, msgs: Fr[] = []) { + this.buckets.set(bucket.seq, bucket); + this.messagesPerBucket.set(bucket.seq, msgs); + } + public setBlockNumber(blockNumber: number) { this.blockNumber = blockNumber; } @@ -27,6 +34,24 @@ export class MockL1ToL2MessageSource implements L1ToL2MessageSource { throw new Error('Method not implemented.'); } + getInboxBucket(seq: bigint): Promise { + return Promise.resolve(this.buckets.get(seq)); + } + + getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise { + const atOrBefore = [...this.buckets.values()] + .filter(bucket => bucket.timestamp <= timestamp) + .sort((a, b) => Number(a.seq - b.seq)); + return Promise.resolve(atOrBefore.at(-1)); + } + + getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { + const seqs = [...this.messagesPerBucket.keys()] + .filter(seq => seq > fromExclusive && seq <= toInclusive) + .sort((a, b) => Number(a - b)); + return Promise.resolve(seqs.flatMap(seq => this.messagesPerBucket.get(seq) ?? [])); + } + getBlockNumber() { return Promise.resolve(BlockNumber(this.blockNumber)); } diff --git a/yarn-project/archiver/src/test/mock_structs.ts b/yarn-project/archiver/src/test/mock_structs.ts index 91145fcfcb11..1f3b78b58e6a 100644 --- a/yarn-project/archiver/src/test/mock_structs.ts +++ b/yarn-project/archiver/src/test/mock_structs.ts @@ -15,7 +15,7 @@ import { AztecAddress } from '@aztec/stdlib/aztec-address'; import { CommitteeAttestation, L2Block } from '@aztec/stdlib/block'; import { Checkpoint, L1PublishedData, PublishedCheckpoint } from '@aztec/stdlib/checkpoint'; import { PrivateLog, PublicLog, SiloedTag, Tag } from '@aztec/stdlib/logs'; -import { InboxLeaf } from '@aztec/stdlib/messaging'; +import { InboxLeaf, updateInboxRollingHash } from '@aztec/stdlib/messaging'; import { orderAttestations } from '@aztec/stdlib/p2p'; import { CheckpointHeader } from '@aztec/stdlib/rollup'; import { makeCheckpointAttestationFromCheckpoint } from '@aztec/stdlib/testing'; @@ -34,6 +34,10 @@ export function makeInboxMessage( const { leaf = Fr.random() } = overrides; const { rollingHash = updateRollingHash(previousRollingHash, leaf) } = overrides; const { index = InboxLeaf.smallestIndexForCheckpoint(checkpointNumber) } = overrides; + const { inboxRollingHash = updateInboxRollingHash(Fr.ZERO, leaf) } = overrides; + // Default each message to its own bucket, keyed monotonically off its global index. + const { bucketSeq = index + 1n } = overrides; + const { bucketTimestamp = index + 1n } = overrides; return { index, @@ -42,6 +46,9 @@ export function makeInboxMessage( l1BlockNumber, l1BlockHash, rollingHash, + inboxRollingHash, + bucketSeq, + bucketTimestamp, }; } @@ -49,6 +56,7 @@ export function makeInboxMessages( totalCount: number, opts: { initialHash?: Buffer16; + initialInboxHash?: Fr; initialCheckpointNumber?: CheckpointNumber; messagesPerCheckpoint?: number; overrideFn?: (msg: InboxMessage, index: number) => InboxMessage; @@ -56,6 +64,7 @@ export function makeInboxMessages( ): InboxMessage[] { const { initialHash = Buffer16.ZERO, + initialInboxHash = Fr.ZERO, overrideFn = msg => msg, initialCheckpointNumber = CheckpointNumber(1), messagesPerCheckpoint = 1, @@ -63,6 +72,7 @@ export function makeInboxMessages( const messages: InboxMessage[] = []; let rollingHash = initialHash; + let inboxRollingHash = initialInboxHash; for (let i = 0; i < totalCount; i++) { const msgIndex = i % messagesPerCheckpoint; const checkpointNumber = CheckpointNumber.fromBigInt( @@ -74,10 +84,12 @@ export function makeInboxMessages( leaf, checkpointNumber, index: InboxLeaf.smallestIndexForCheckpoint(checkpointNumber) + BigInt(msgIndex), + inboxRollingHash: updateInboxRollingHash(inboxRollingHash, leaf), }), i, ); rollingHash = message.rollingHash; + inboxRollingHash = message.inboxRollingHash; messages.push(message); } return messages; diff --git a/yarn-project/ethereum/src/contracts/inbox.ts b/yarn-project/ethereum/src/contracts/inbox.ts index 23d7f6ce3df4..75040935053b 100644 --- a/yarn-project/ethereum/src/contracts/inbox.ts +++ b/yarn-project/ethereum/src/contracts/inbox.ts @@ -26,8 +26,11 @@ export type MessageSentArgs = { bucketSeq: bigint; }; -/** Log type for MessageSent events. */ -export type MessageSentLog = L1EventLog; +/** Log type for MessageSent events, enriched with the emitting L1 block's timestamp (the bucket recency key). */ +export type MessageSentLog = L1EventLog & { + /** Timestamp (in seconds) of the L1 block that emitted the event; the key of the message's Inbox bucket. */ + l1BlockTimestamp: bigint; +}; export class InboxContract { private readonly inbox: GetContractReturnType; @@ -81,10 +84,11 @@ export class InboxContract { /** Fetches MessageSent events within the given block range. */ async getMessageSentEvents(fromBlock: bigint, toBlock: bigint): Promise { - const logs = await this.inbox.getEvents.MessageSent({}, { fromBlock, toBlock }); - return logs - .filter(log => log.blockNumber! >= fromBlock && log.blockNumber! <= toBlock) - .map(log => this.mapMessageSentLog(log)); + const logs = (await this.inbox.getEvents.MessageSent({}, { fromBlock, toBlock })).filter( + log => log.blockNumber! >= fromBlock && log.blockNumber! <= toBlock, + ); + const timestamps = await this.getBlockTimestamps(logs.map(log => log.blockNumber!)); + return logs.map(log => this.mapMessageSentLog(log, timestamps.get(log.blockNumber!)!)); } /** Fetches MessageSent events for a specific message hash around a specific block. */ @@ -97,26 +101,47 @@ export class InboxContract { { hash: msgHash }, { fromBlock: maxBigint(aroundL1BlockNumber - 5n, 1n), toBlock: aroundL1BlockNumber + 5n }, ); - return log && this.mapMessageSentLog(log); + if (!log) { + return log as unknown as MessageSentLog; + } + const [timestamp] = (await this.getBlockTimestamps([log.blockNumber!])).values(); + return this.mapMessageSentLog(log, timestamp); } - private mapMessageSentLog(log: { - blockNumber: bigint | null; - blockHash: `0x${string}` | null; - transactionHash: `0x${string}` | null; - args: { - index?: bigint; - hash?: `0x${string}`; - checkpointNumber?: bigint; - rollingHash?: `0x${string}`; - inboxRollingHash?: `0x${string}`; - bucketSeq?: bigint; - }; - }): MessageSentLog { + /** Fetches the timestamp of each distinct L1 block number, so each MessageSent log can carry its bucket key. */ + private async getBlockTimestamps(blockNumbers: bigint[]): Promise> { + const uniqueBlockNumbers = [...new Set(blockNumbers)]; + const timestamps = new Map(); + await Promise.all( + uniqueBlockNumbers.map(async blockNumber => { + const block = await this.client.getBlock({ blockNumber, includeTransactions: false }); + timestamps.set(blockNumber, block.timestamp); + }), + ); + return timestamps; + } + + private mapMessageSentLog( + log: { + blockNumber: bigint | null; + blockHash: `0x${string}` | null; + transactionHash: `0x${string}` | null; + args: { + index?: bigint; + hash?: `0x${string}`; + checkpointNumber?: bigint; + rollingHash?: `0x${string}`; + inboxRollingHash?: `0x${string}`; + bucketSeq?: bigint; + }; + }, + l1BlockTimestamp: bigint, + ): MessageSentLog { return { l1BlockNumber: log.blockNumber!, l1BlockHash: Buffer32.fromString(log.blockHash!), l1TransactionHash: log.transactionHash!, + l1BlockTimestamp, args: { index: log.args.index!, leaf: Fr.fromString(log.args.hash!), diff --git a/yarn-project/stdlib/src/interfaces/archiver.test.ts b/yarn-project/stdlib/src/interfaces/archiver.test.ts index 5ace4e328afe..2b005e726b23 100644 --- a/yarn-project/stdlib/src/interfaces/archiver.test.ts +++ b/yarn-project/stdlib/src/interfaces/archiver.test.ts @@ -36,6 +36,7 @@ import { type LogResult, randomLogResult } from '../logs/log_result.js'; import type { PrivateLogsQuery, PublicLogsQuery } from '../logs/logs_query.js'; import { SiloedTag } from '../logs/siloed_tag.js'; import { Tag } from '../logs/tag.js'; +import type { InboxBucket } from '../messaging/inbox_bucket.js'; import { CheckpointHeader } from '../rollup/checkpoint_header.js'; import { getTokenContractArtifact } from '../tests/fixtures.js'; import { AppendOnlyTreeSnapshot } from '../trees/append_only_tree_snapshot.js'; @@ -213,6 +214,21 @@ describe('ArchiverApiSchema', () => { expect(result).toBe(1n); }); + it('getLatestInboxBucketAtOrBefore', async () => { + const result = await context.client.getLatestInboxBucketAtOrBefore(123n); + expect(result).toMatchObject({ seq: 1n, msgCount: 3, totalMsgCount: 3n, isOpen: false }); + }); + + it('getInboxBucket', async () => { + const result = await context.client.getInboxBucket(2n); + expect(result).toMatchObject({ seq: 2n, msgCount: 3, isOpen: true }); + }); + + it('getL1ToL2MessagesBetweenBuckets', async () => { + const result = await context.client.getL1ToL2MessagesBetweenBuckets(0n, 3n); + expect(result).toEqual([expect.any(Fr)]); + }); + it('registerContractFunctionSignatures', async () => { await context.client.registerContractFunctionSignatures(['test()']); }); @@ -551,6 +567,35 @@ class MockArchiver implements ArchiverApi { expect(l1ToL2Message).toBeInstanceOf(Fr); return Promise.resolve(1n); } + getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise { + expect(typeof timestamp).toEqual('bigint'); + return Promise.resolve({ + seq: 1n, + inboxRollingHash: Fr.random(), + totalMsgCount: 3n, + timestamp: 100n, + msgCount: 3, + lastMessageIndex: 2n, + isOpen: false, + }); + } + getInboxBucket(seq: bigint): Promise { + expect(typeof seq).toEqual('bigint'); + return Promise.resolve({ + seq, + inboxRollingHash: Fr.random(), + totalMsgCount: 3n, + timestamp: 100n, + msgCount: 3, + lastMessageIndex: 2n, + isOpen: true, + }); + } + getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { + expect(typeof fromExclusive).toEqual('bigint'); + expect(typeof toInclusive).toEqual('bigint'); + return Promise.resolve([Fr.random()]); + } getL1Constants(): Promise { return Promise.resolve(EmptyL1RollupConstants); } diff --git a/yarn-project/stdlib/src/interfaces/archiver.ts b/yarn-project/stdlib/src/interfaces/archiver.ts index 362da31e3a88..d1934e803b7b 100644 --- a/yarn-project/stdlib/src/interfaces/archiver.ts +++ b/yarn-project/stdlib/src/interfaces/archiver.ts @@ -25,6 +25,7 @@ import { import { L1RollupConstantsSchema } from '../epoch-helpers/index.js'; import { LogResultSchema } from '../logs/log_result.js'; import { PrivateLogsQuerySchema, PublicLogsQuerySchema } from '../logs/logs_query.js'; +import { InboxBucketSchema } from '../messaging/inbox_bucket.js'; import type { L1ToL2MessageSource } from '../messaging/l1_to_l2_message_source.js'; import { L2ToL1MembershipWitnessSchema } from '../messaging/l2_to_l1_membership.js'; import { optional, schemas } from '../schemas/schemas.js'; @@ -137,6 +138,15 @@ export const ArchiverApiSchema: ApiSchemaFor = { registerContractFunctionSignatures: z.function({ input: z.tuple([z.array(z.string())]), output: z.void() }), getL1ToL2Messages: z.function({ input: z.tuple([CheckpointNumberSchema]), output: z.array(schemas.Fr) }), getL1ToL2MessageIndex: z.function({ input: z.tuple([schemas.Fr]), output: schemas.BigInt.optional() }), + getLatestInboxBucketAtOrBefore: z.function({ + input: z.tuple([schemas.BigInt]), + output: InboxBucketSchema.optional(), + }), + getInboxBucket: z.function({ input: z.tuple([schemas.BigInt]), output: InboxBucketSchema.optional() }), + getL1ToL2MessagesBetweenBuckets: z.function({ + input: z.tuple([schemas.BigInt, schemas.BigInt]), + output: z.array(schemas.Fr), + }), getDebugFunctionName: z.function({ input: z.tuple([schemas.AztecAddress, schemas.FunctionSelector]), output: optional(z.string()), diff --git a/yarn-project/stdlib/src/messaging/inbox_bucket.ts b/yarn-project/stdlib/src/messaging/inbox_bucket.ts new file mode 100644 index 000000000000..5dd08e09d2b9 --- /dev/null +++ b/yarn-project/stdlib/src/messaging/inbox_bucket.ts @@ -0,0 +1,43 @@ +import { Fr } from '@aztec/foundation/curves/bn254'; +import { schemas } from '@aztec/foundation/schemas'; + +import { z } from 'zod'; + +/** + * Snapshot of an Inbox rolling-hash bucket as tracked by the archiver (AZIP-22 Fast Inbox). + * + * A bucket accumulates the message leaves inserted into the Inbox within a single L1 block (up to a per-bucket + * maximum, after which further messages in the same block spill into the next bucket). Buckets are identified by + * a dense, monotonically increasing sequence number and keyed for recency lookups by the L1 block timestamp at + * which they were opened. Mirrors the on-chain `InboxBucket` struct plus the fields the archiver derives while + * syncing. + */ +export type InboxBucket = { + /** Dense, monotonically increasing sequence number of this bucket in the Inbox ring. */ + seq: bigint; + /** Consensus rolling hash (truncated sha256 chain) after the last message absorbed into this bucket. */ + inboxRollingHash: Fr; + /** Cumulative number of messages inserted into the Inbox up to and including this bucket. */ + totalMsgCount: bigint; + /** L1 block timestamp at which this bucket was opened; the recency key for lag/cutoff comparisons, in seconds. */ + timestamp: bigint; + /** Number of messages absorbed into this bucket. */ + msgCount: number; + /** Global leaf index of the last message absorbed into this bucket. */ + lastMessageIndex: bigint; + /** + * Whether this is the latest bucket the archiver has synced. A latest bucket may still grow as more messages + * arrive on L1; earlier buckets are complete. Consumers that need a settled bucket apply a lag on the timestamp. + */ + isOpen: boolean; +}; + +export const InboxBucketSchema = z.object({ + seq: schemas.BigInt, + inboxRollingHash: Fr.schema, + totalMsgCount: schemas.BigInt, + timestamp: schemas.BigInt, + msgCount: schemas.Integer, + lastMessageIndex: schemas.BigInt, + isOpen: z.boolean(), +}) satisfies z.ZodType; diff --git a/yarn-project/stdlib/src/messaging/index.ts b/yarn-project/stdlib/src/messaging/index.ts index 8728940e405a..da0c32490353 100644 --- a/yarn-project/stdlib/src/messaging/index.ts +++ b/yarn-project/stdlib/src/messaging/index.ts @@ -1,5 +1,6 @@ export * from './append_l1_to_l2_messages.js'; export * from './in_hash.js'; +export * from './inbox_bucket.js'; export * from './inbox_leaf.js'; export * from './inbox_rolling_hash.js'; export * from './l1_to_l2_message_bundle.js'; diff --git a/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts b/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts index 04572c4eb99c..4e3a01172288 100644 --- a/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts +++ b/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts @@ -2,6 +2,7 @@ import type { CheckpointNumber } from '@aztec/foundation/branded-types'; import type { Fr } from '@aztec/foundation/curves/bn254'; import type { L2Tips } from '../block/l2_block_source.js'; +import type { InboxBucket } from './inbox_bucket.js'; /** * Interface of classes allowing for the retrieval of L1 to L2 messages. @@ -23,6 +24,30 @@ export interface L1ToL2MessageSource { */ getL1ToL2MessageIndex(l1ToL2Message: Fr): Promise; + /** + * Returns the latest Inbox bucket opened at or before the given L1 timestamp, or undefined if no such bucket + * exists (i.e., every synced bucket was opened strictly after the timestamp). Used by the sequencer/validator + * to resolve the censorship cutoff and message-lag boundaries (AZIP-22 Fast Inbox). + * @param timestamp - The L1 timestamp (in seconds) to look up at-or-before. + */ + getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise; + + /** + * Returns the Inbox bucket with the given sequence number, or undefined if it has not been synced (AZIP-22 Fast + * Inbox). Validators use this to resolve the bucket a proposal references and check its rolling hash. + * @param seq - The bucket sequence number. + */ + getInboxBucket(seq: bigint): Promise; + + /** + * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion + * order, for streaming message-bundle derivation (AZIP-22 Fast Inbox). Returns an empty array if the upper + * bucket has not been synced. + * @param fromExclusive - The lower bucket sequence bound, exclusive (0 means from the start of the Inbox). + * @param toInclusive - The upper bucket sequence bound, inclusive. + */ + getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise; + /** * Returns the tips of the L2 chain. */ From 3630ec2c5e7fa45ebe897a1c4e191b38225640df Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Sat, 18 Jul 2026 03:26:06 -0300 Subject: [PATCH 2/6] fix(fast-inbox): guard between-buckets lower bound and bound timestamp fetch (A-1379) --- .../archiver/src/store/message_store.test.ts | 2 ++ .../archiver/src/store/message_store.ts | 13 +++++++++++-- yarn-project/ethereum/src/contracts/inbox.ts | 18 +++++++++++------- 3 files changed, 24 insertions(+), 9 deletions(-) diff --git a/yarn-project/archiver/src/store/message_store.test.ts b/yarn-project/archiver/src/store/message_store.test.ts index 69fe30a75091..e78b9a602be7 100644 --- a/yarn-project/archiver/src/store/message_store.test.ts +++ b/yarn-project/archiver/src/store/message_store.test.ts @@ -439,6 +439,8 @@ describe('MessageStore', () => { expect(await messageStore.getL1ToL2MessagesBetweenBuckets(3n, 3n)).toEqual([]); // Unknown upper bucket yields no messages. expect(await messageStore.getL1ToL2MessagesBetweenBuckets(0n, 9n)).toEqual([]); + // An unknown nonzero lower bucket is treated as unavailable, not as genesis. + expect(await messageStore.getL1ToL2MessagesBetweenBuckets(4n, 3n)).toEqual([]); }); it('rewinds buckets when messages are removed', async () => { diff --git a/yarn-project/archiver/src/store/message_store.ts b/yarn-project/archiver/src/store/message_store.ts index 1513df52a905..543aebd984e2 100644 --- a/yarn-project/archiver/src/store/message_store.ts +++ b/yarn-project/archiver/src/store/message_store.ts @@ -433,8 +433,17 @@ export class MessageStore { if (toBucket === undefined) { return []; } - const fromBucket = fromExclusive > 0n ? await this.getBucketSnapshotBySeq(fromExclusive) : undefined; - const startIndex = fromBucket ? fromBucket.lastMessageIndex + 1n : 0n; + // A nonzero lower bound must reference a synced bucket; otherwise the range is unavailable (rather than + // defaulting to genesis, which would silently over-return). Sequence 0 is the genesis base case: start from + // the beginning of the Inbox. + let startIndex = 0n; + if (fromExclusive > 0n) { + const fromBucket = await this.getBucketSnapshotBySeq(fromExclusive); + if (fromBucket === undefined) { + return []; + } + startIndex = fromBucket.lastMessageIndex + 1n; + } const endIndexExclusive = toBucket.lastMessageIndex + 1n; const leaves: Fr[] = []; diff --git a/yarn-project/ethereum/src/contracts/inbox.ts b/yarn-project/ethereum/src/contracts/inbox.ts index 75040935053b..1b192b4eb93a 100644 --- a/yarn-project/ethereum/src/contracts/inbox.ts +++ b/yarn-project/ethereum/src/contracts/inbox.ts @@ -1,3 +1,4 @@ +import { asyncPool } from '@aztec/foundation/async-pool'; import { maxBigint } from '@aztec/foundation/bigint'; import { CheckpointNumber } from '@aztec/foundation/branded-types'; import { Buffer16, Buffer32 } from '@aztec/foundation/buffer'; @@ -108,16 +109,19 @@ export class InboxContract { return this.mapMessageSentLog(log, timestamp); } - /** Fetches the timestamp of each distinct L1 block number, so each MessageSent log can carry its bucket key. */ + /** + * Fetches the timestamp of each distinct L1 block number, so each MessageSent log can carry its bucket key. + * Fetched with bounded concurrency to keep a large sync batch from fanning out unbounded RPC requests. Blocks + * are resolved by number rather than hash so a concurrent L1 reorg does not throw; a resulting cross-fork + * timestamp is transient and re-corrected by the archiver's rolling-hash reorg detection on the next sync. + */ private async getBlockTimestamps(blockNumbers: bigint[]): Promise> { const uniqueBlockNumbers = [...new Set(blockNumbers)]; const timestamps = new Map(); - await Promise.all( - uniqueBlockNumbers.map(async blockNumber => { - const block = await this.client.getBlock({ blockNumber, includeTransactions: false }); - timestamps.set(blockNumber, block.timestamp); - }), - ); + await asyncPool(10, uniqueBlockNumbers, async blockNumber => { + const block = await this.client.getBlock({ blockNumber, includeTransactions: false }); + timestamps.set(blockNumber, block.timestamp); + }); return timestamps; } From d4afda508e2ed27637b82da92aad6b3be41de636 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Mon, 27 Jul 2026 14:32:59 -0300 Subject: [PATCH 3/6] fix(fast-inbox): resolve message-sent L1 block timestamps by hash (A-1379) A message's bucket timestamp must come from the same fork as the message itself. Resolving the emitting L1 block by number returns whatever block sits at that height when the lookup runs, so a reorg between the log query and the timestamp fetch silently stores a cross-fork timestamp. Resolving by the log's block hash pins the fork and turns that race into a failed lookup, which the archiver's L1 synchronizer retries on its next poll. --- yarn-project/ethereum/src/contracts/inbox.ts | 28 +++++++++++--------- 1 file changed, 15 insertions(+), 13 deletions(-) diff --git a/yarn-project/ethereum/src/contracts/inbox.ts b/yarn-project/ethereum/src/contracts/inbox.ts index 1b192b4eb93a..1d2a732ace8d 100644 --- a/yarn-project/ethereum/src/contracts/inbox.ts +++ b/yarn-project/ethereum/src/contracts/inbox.ts @@ -88,8 +88,8 @@ export class InboxContract { const logs = (await this.inbox.getEvents.MessageSent({}, { fromBlock, toBlock })).filter( log => log.blockNumber! >= fromBlock && log.blockNumber! <= toBlock, ); - const timestamps = await this.getBlockTimestamps(logs.map(log => log.blockNumber!)); - return logs.map(log => this.mapMessageSentLog(log, timestamps.get(log.blockNumber!)!)); + const timestamps = await this.getBlockTimestamps(logs.map(log => log.blockHash!)); + return logs.map(log => this.mapMessageSentLog(log, timestamps.get(log.blockHash!)!)); } /** Fetches MessageSent events for a specific message hash around a specific block. */ @@ -105,22 +105,24 @@ export class InboxContract { if (!log) { return log as unknown as MessageSentLog; } - const [timestamp] = (await this.getBlockTimestamps([log.blockNumber!])).values(); + const [timestamp] = (await this.getBlockTimestamps([log.blockHash!])).values(); return this.mapMessageSentLog(log, timestamp); } /** - * Fetches the timestamp of each distinct L1 block number, so each MessageSent log can carry its bucket key. - * Fetched with bounded concurrency to keep a large sync batch from fanning out unbounded RPC requests. Blocks - * are resolved by number rather than hash so a concurrent L1 reorg does not throw; a resulting cross-fork - * timestamp is transient and re-corrected by the archiver's rolling-hash reorg detection on the next sync. + * Fetches the timestamp of each distinct L1 block, so each MessageSent log can carry its bucket key. Blocks are + * resolved by hash, which pins them to the same fork the logs were read from: resolving by number would silently + * return the same-height block of another fork if the chain reorgs between the log query and this lookup, storing a + * timestamp that never applied to the message. By hash, such a reorg fails the lookup instead, and the caller + * retries against the reorged chain. Fetched with bounded concurrency to keep a large sync batch from fanning out + * unbounded RPC requests. */ - private async getBlockTimestamps(blockNumbers: bigint[]): Promise> { - const uniqueBlockNumbers = [...new Set(blockNumbers)]; - const timestamps = new Map(); - await asyncPool(10, uniqueBlockNumbers, async blockNumber => { - const block = await this.client.getBlock({ blockNumber, includeTransactions: false }); - timestamps.set(blockNumber, block.timestamp); + private async getBlockTimestamps(blockHashes: Hex[]): Promise> { + const uniqueBlockHashes = [...new Set(blockHashes)]; + const timestamps = new Map(); + await asyncPool(10, uniqueBlockHashes, async blockHash => { + const block = await this.client.getBlock({ blockHash, includeTransactions: false }); + timestamps.set(blockHash, block.timestamp); }); return timestamps; } From a5811ec57c515e9e9a5ccee66209a08322c4e744 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Mon, 27 Jul 2026 15:09:30 -0300 Subject: [PATCH 4/6] refactor(fast-inbox): store inbox buckets as complete units in the archiver (A-1379) The message store used to snapshot a bucket incrementally, once per absorbed message, threading the in-progress bucket state across insertion batches. Every bucket is opened and closed within a single L1 block and the synchronizer retrieves Inbox logs in whole-L1-block ranges, so a bucket's messages always arrive together: group the batch by bucket sequence and derive one snapshot per bucket from its complete message set instead. The completeness that makes this correct is now enforced rather than assumed. A batch that continues a stored bucket from the middle, or that opens a bucket older than the newest stored one, is rejected: either would produce a snapshot disagreeing with the messages it covers. Re-delivering a stored bucket from its first message stays allowed, since an L1 reorg replaces a bucket's tail and the re-sync that follows replays the whole L1 block. Renames `addL1ToL2Messages` to `addL1ToL2MessageBuckets` to say what the method now requires of its caller, drops `isOpen` from `InboxBucket` (it had no consumers outside the store that produced it), and records each bucket's L1 block number and first message index in the snapshot for rollbacks and range queries to work off bucket records alone. --- .../archiver/src/archiver-sync.test.ts | 2 - .../archiver/src/modules/l1_synchronizer.ts | 7 +- .../archiver/src/store/message_store.test.ts | 136 +++++++++------ .../archiver/src/store/message_store.ts | 158 ++++++++++++++---- .../stdlib/src/interfaces/archiver.test.ts | 6 +- .../stdlib/src/messaging/inbox_bucket.ts | 6 - .../src/validator.integration.test.ts | 8 +- 7 files changed, 224 insertions(+), 99 deletions(-) diff --git a/yarn-project/archiver/src/archiver-sync.test.ts b/yarn-project/archiver/src/archiver-sync.test.ts index cf940c547593..6237c468dd3d 100644 --- a/yarn-project/archiver/src/archiver-sync.test.ts +++ b/yarn-project/archiver/src/archiver-sync.test.ts @@ -276,9 +276,7 @@ describe('Archiver Sync', () => { msgCount: 3, totalMsgCount: 9n, timestamp: t3, - isOpen: true, }); - expect((await archiver.getInboxBucket(1n))!.isOpen).toBe(false); // At-or-before lookups resolve the latest bucket not opened after the given timestamp. expect((await archiver.getLatestInboxBucketAtOrBefore(t3))!.seq).toEqual(3n); diff --git a/yarn-project/archiver/src/modules/l1_synchronizer.ts b/yarn-project/archiver/src/modules/l1_synchronizer.ts index 1739dffccde2..8fbafdf7f70b 100644 --- a/yarn-project/archiver/src/modules/l1_synchronizer.ts +++ b/yarn-project/archiver/src/modules/l1_synchronizer.ts @@ -495,7 +495,10 @@ export class ArchiverL1Synchronizer implements Traceable { ); } - /** Retrieves L1 to L2 messages from L1 in batches and stores them. */ + /** + * Retrieves L1 to L2 messages from L1 in batches and stores them. Batches must span whole L1 blocks so that every + * message of an Inbox bucket is stored in a single call, which the message store requires. + */ private async retrieveAndStoreMessages(fromL1Block: bigint, toL1Block: bigint): Promise { let searchStartBlock: bigint = 0n; let searchEndBlock: bigint = fromL1Block; @@ -508,7 +511,7 @@ export class ArchiverL1Synchronizer implements Traceable { this.log.trace(`Retrieving L1 to L2 messages in L1 blocks ${searchStartBlock}-${searchEndBlock}`); const messages = await retrieveL1ToL2Messages(this.inbox, searchStartBlock, searchEndBlock); const timer = new Timer(); - await this.stores.messages.addL1ToL2Messages(messages); + await this.stores.messages.addL1ToL2MessageBuckets(messages); const perMsg = timer.ms() / messages.length; this.instrumentation.processNewMessages(messages.length, perMsg); for (const msg of messages) { diff --git a/yarn-project/archiver/src/store/message_store.test.ts b/yarn-project/archiver/src/store/message_store.test.ts index e78b9a602be7..bfebd4121177 100644 --- a/yarn-project/archiver/src/store/message_store.test.ts +++ b/yarn-project/archiver/src/store/message_store.test.ts @@ -5,10 +5,11 @@ import { Fr } from '@aztec/foundation/curves/bn254'; import { toArray } from '@aztec/foundation/iterable'; import { openTmpStore } from '@aztec/kv-store/lmdb-v2'; import { Checkpoint, type PublishedCheckpoint } from '@aztec/stdlib/checkpoint'; +import { updateInboxRollingHash } from '@aztec/stdlib/messaging'; import '@aztec/stdlib/testing/jest'; import { L1ToL2MessagesNotReadyError } from '../errors.js'; -import type { InboxMessage } from '../structs/inbox_message.js'; +import { type InboxMessage, updateRollingHash } from '../structs/inbox_message.js'; import { makeInboxMessage, makeInboxMessages, @@ -79,7 +80,7 @@ describe('MessageStore', () => { const l1BlockHash = Buffer32.random(); const l1BlockNumber = 10n; await messageStore.setMessageSyncState({ l1BlockNumber, l1BlockHash }, 1n); - await messageStore.addL1ToL2Messages([ + await messageStore.addL1ToL2MessageBuckets([ makeInboxMessage(Buffer16.ZERO, { l1BlockNumber: 5n, l1BlockHash: Buffer32.random() }), ]); await expect(getSynchPoint(blockStore, messageStore)).resolves.toEqual({ @@ -100,7 +101,7 @@ describe('MessageStore', () => { it('stores first message ever', async () => { const msg = makeInboxMessage(Buffer16.ZERO, { index: 0n, checkpointNumber: CheckpointNumber(1) }); - await messageStore.addL1ToL2Messages([msg]); + await messageStore.addL1ToL2MessageBuckets([msg]); await checkMessages([msg]); expect(await messageStore.getL1ToL2Messages(CheckpointNumber(1))).toEqual([msg.leaf]); @@ -108,7 +109,7 @@ describe('MessageStore', () => { it('stores single message', async () => { const msg = makeInboxMessage(Buffer16.ZERO, { checkpointNumber: CheckpointNumber(2) }); - await messageStore.addL1ToL2Messages([msg]); + await messageStore.addL1ToL2MessageBuckets([msg]); await checkMessages([msg]); expect(await messageStore.getL1ToL2Messages(CheckpointNumber(2))).toEqual([msg.leaf]); @@ -116,7 +117,7 @@ describe('MessageStore', () => { it('stores and returns messages across different blocks', async () => { const msgs = makeInboxMessages(5, { initialCheckpointNumber }); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); await checkMessages(msgs); expect(await messageStore.getL1ToL2Messages(CheckpointNumber(initialCheckpointNumber + 2))).toEqual( @@ -126,8 +127,8 @@ describe('MessageStore', () => { it('stores the same messages again', async () => { const msgs = makeInboxMessages(5, { initialCheckpointNumber }); - await messageStore.addL1ToL2Messages(msgs); - await messageStore.addL1ToL2Messages(msgs.slice(2)); + await messageStore.addL1ToL2MessageBuckets(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs.slice(2)); await checkMessages(msgs); }); @@ -140,8 +141,8 @@ describe('MessageStore', () => { initialInboxHash: msgs1.at(-1)!.inboxRollingHash, }); - await messageStore.addL1ToL2Messages(msgs1); - await messageStore.addL1ToL2Messages(msgs2); + await messageStore.addL1ToL2MessageBuckets(msgs1); + await messageStore.addL1ToL2MessageBuckets(msgs2); await checkMessages([...msgs1, ...msgs2]); @@ -153,7 +154,7 @@ describe('MessageStore', () => { it('stores and returns messages with block numbers larger than a byte', async () => { const msgs = makeInboxMessages(5, { initialCheckpointNumber: CheckpointNumber(1000) }); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); await checkMessages(msgs); expect(await messageStore.getL1ToL2Messages(CheckpointNumber(1002))).toEqual([msgs[2]].map(m => m.leaf)); @@ -161,7 +162,7 @@ describe('MessageStore', () => { it('stores and returns multiple messages per block', async () => { const msgs = makeInboxMessagesWithFullBlocks(4); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); await checkMessages(msgs); const blockMessages = await messageStore.getL1ToL2Messages(CheckpointNumber(initialCheckpointNumber + 1)); @@ -173,8 +174,8 @@ describe('MessageStore', () => { it('stores messages in multiple operations', async () => { const msgs = makeInboxMessages(20, { initialCheckpointNumber }); - await messageStore.addL1ToL2Messages(msgs.slice(0, 10)); - await messageStore.addL1ToL2Messages(msgs.slice(10, 20)); + await messageStore.addL1ToL2MessageBuckets(msgs.slice(0, 10)); + await messageStore.addL1ToL2MessageBuckets(msgs.slice(10, 20)); expect(await messageStore.getL1ToL2Messages(CheckpointNumber(initialCheckpointNumber + 2))).toEqual( [msgs[2]].map(m => m.leaf), @@ -187,7 +188,7 @@ describe('MessageStore', () => { it('iterates over messages from start index', async () => { const msgs = makeInboxMessages(10, { initialCheckpointNumber }); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); const iterated = await toArray(messageStore.iterateL1ToL2Messages({ start: msgs[3].index })); expect(iterated).toEqual(msgs.slice(3)); @@ -195,7 +196,7 @@ describe('MessageStore', () => { it('iterates over messages in reverse', async () => { const msgs = makeInboxMessages(10, { initialCheckpointNumber }); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); const iterated = await toArray(messageStore.iterateL1ToL2Messages({ reverse: true, end: msgs[3].index })); expect(iterated).toEqual(msgs.slice(0, 4).reverse()); @@ -203,39 +204,39 @@ describe('MessageStore', () => { it('throws if messages are added out of order', async () => { const msgs = makeInboxMessages(5, { overrideFn: (msg, i) => ({ ...msg, index: BigInt(10 - i) }) }); - await expect(messageStore.addL1ToL2Messages(msgs)).rejects.toThrow(MessageStoreError); + await expect(messageStore.addL1ToL2MessageBuckets(msgs)).rejects.toThrow(MessageStoreError); }); it('throws if block number for the first message is out of order', async () => { const msgs = makeInboxMessages(4, { initialCheckpointNumber }); msgs[2].checkpointNumber = CheckpointNumber(initialCheckpointNumber - 1); - await messageStore.addL1ToL2Messages(msgs.slice(0, 2)); - await expect(messageStore.addL1ToL2Messages(msgs.slice(2, 4))).rejects.toThrow(MessageStoreError); + await messageStore.addL1ToL2MessageBuckets(msgs.slice(0, 2)); + await expect(messageStore.addL1ToL2MessageBuckets(msgs.slice(2, 4))).rejects.toThrow(MessageStoreError); }); it('throws if rolling hash is not correct', async () => { const msgs = makeInboxMessages(5); msgs[1].rollingHash = Buffer16.random(); - await expect(messageStore.addL1ToL2Messages(msgs)).rejects.toThrow(MessageStoreError); + await expect(messageStore.addL1ToL2MessageBuckets(msgs)).rejects.toThrow(MessageStoreError); }); it('throws if rolling hash for first message is not correct', async () => { const msgs = makeInboxMessages(4); msgs[2].rollingHash = Buffer16.random(); - await messageStore.addL1ToL2Messages(msgs.slice(0, CheckpointNumber(2))); - await expect(messageStore.addL1ToL2Messages(msgs.slice(2, 4))).rejects.toThrow(MessageStoreError); + await messageStore.addL1ToL2MessageBuckets(msgs.slice(0, CheckpointNumber(2))); + await expect(messageStore.addL1ToL2MessageBuckets(msgs.slice(2, 4))).rejects.toThrow(MessageStoreError); }); it('throws if index is not in the correct range', async () => { const msgs = makeInboxMessages(5, { initialCheckpointNumber }); msgs.at(-1)!.index += 100n; - await expect(messageStore.addL1ToL2Messages(msgs)).rejects.toThrow(MessageStoreError); + await expect(messageStore.addL1ToL2MessageBuckets(msgs)).rejects.toThrow(MessageStoreError); }); it('throws if first index in block has gaps', async () => { const msgs = makeInboxMessages(4, { initialCheckpointNumber }); msgs[2].index++; - await expect(messageStore.addL1ToL2Messages(msgs)).rejects.toThrow(MessageStoreError); + await expect(messageStore.addL1ToL2MessageBuckets(msgs)).rejects.toThrow(MessageStoreError); }); it('throws if index does not follow previous one', async () => { @@ -248,13 +249,13 @@ describe('MessageStore', () => { }), }); msgs[1].index++; - await expect(messageStore.addL1ToL2Messages(msgs)).rejects.toThrow(MessageStoreError); + await expect(messageStore.addL1ToL2MessageBuckets(msgs)).rejects.toThrow(MessageStoreError); }); it('removes messages up to the given block number', async () => { const msgs = makeInboxMessagesWithFullBlocks(4, { initialCheckpointNumber: CheckpointNumber(1) }); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); await checkMessages(msgs); expect(await messageStore.getL1ToL2Messages(CheckpointNumber(1))).toHaveLength( @@ -286,7 +287,7 @@ describe('MessageStore', () => { it('removes messages starting with the given index', async () => { const msgs = makeInboxMessagesWithFullBlocks(4, { initialCheckpointNumber: CheckpointNumber(1) }); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); await messageStore.removeL1ToL2Messages(msgs[13].index); await checkMessages(msgs.slice(0, 13)); @@ -295,7 +296,7 @@ describe('MessageStore', () => { describe('inbox tree in progress guard', () => { it('throws when checkpointNumber >= treeInProgress', async () => { const msgs = makeInboxMessages(3, { initialCheckpointNumber: CheckpointNumber(5) }); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); // Set treeInProgress to 7, meaning checkpoints 5 and 6 are sealed, 7+ are not await messageStore.setMessageSyncState({ l1BlockNumber: 1n, l1BlockHash: Buffer32.random() }, 7n); @@ -312,7 +313,7 @@ describe('MessageStore', () => { it('returns messages when checkpointNumber < treeInProgress', async () => { const msgs = makeInboxMessages(3, { initialCheckpointNumber: CheckpointNumber(10) }); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); await messageStore.setMessageSyncState({ l1BlockNumber: 1n, l1BlockHash: Buffer32.random() }, 13n); @@ -322,7 +323,7 @@ describe('MessageStore', () => { it('skips guard when treeInProgress is not set', async () => { const msgs = makeInboxMessages(2, { initialCheckpointNumber: CheckpointNumber(1) }); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); // No setMessageSyncState call — guard should be permissive await expect(messageStore.getL1ToL2Messages(CheckpointNumber(1))).resolves.toEqual([msgs[0].leaf]); @@ -345,6 +346,20 @@ describe('MessageStore', () => { return msgs; }; + // Builds a valid message continuing the chain after `previous`, absorbed into the given bucket. + const makeNextMessage = (previous: InboxMessage, bucket: { seq: bigint; timestamp: bigint }): InboxMessage => { + const leaf = Fr.random(); + return { + ...previous, + leaf, + index: previous.index + 1n, + rollingHash: updateRollingHash(previous.rollingHash, leaf), + inboxRollingHash: updateInboxRollingHash(previous.inboxRollingHash, leaf), + bucketSeq: bucket.seq, + bucketTimestamp: bucket.timestamp, + }; + }; + // Three buckets over six messages: bucket 1 = [0,1,2], bucket 2 = [3,4], bucket 3 = [5]. const threeBucketSpec = [ { seq: 1n, timestamp: 100n }, @@ -357,7 +372,7 @@ describe('MessageStore', () => { it('snapshots buckets as messages are inserted', async () => { const msgs = makeBucketedMessages(threeBucketSpec); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); expect(await messageStore.getInboxBucket(1n)).toEqual({ seq: 1n, @@ -366,7 +381,6 @@ describe('MessageStore', () => { timestamp: 100n, msgCount: 3, lastMessageIndex: msgs[2].index, - isOpen: false, }); expect(await messageStore.getInboxBucket(2n)).toEqual({ seq: 2n, @@ -375,7 +389,6 @@ describe('MessageStore', () => { timestamp: 200n, msgCount: 2, lastMessageIndex: msgs[4].index, - isOpen: false, }); expect(await messageStore.getInboxBucket(3n)).toEqual({ seq: 3n, @@ -384,29 +397,57 @@ describe('MessageStore', () => { timestamp: 300n, msgCount: 1, lastMessageIndex: msgs[5].index, - isOpen: true, }); expect(await messageStore.getInboxBucket(4n)).toBeUndefined(); }); - it('continues a bucket that spans two insertion batches', async () => { + it('rejects a bucket delivered without the messages already stored for it', async () => { const msgs = makeBucketedMessages(threeBucketSpec); - await messageStore.addL1ToL2Messages(msgs.slice(0, 2)); - await messageStore.addL1ToL2Messages(msgs.slice(2)); + await messageStore.addL1ToL2MessageBuckets(msgs.slice(0, 2)); - // Bucket 1 keeps accumulating across the batch boundary rather than restarting its message count. - expect(await messageStore.getInboxBucket(1n)).toMatchObject({ msgCount: 3, totalMsgCount: 3n }); - expect(await messageStore.getInboxBucket(3n)).toMatchObject({ msgCount: 1, totalMsgCount: 6n }); + // The second batch continues bucket 1 from its third message, so its snapshot would undercount the bucket. + await expect(messageStore.addL1ToL2MessageBuckets(msgs.slice(2))).rejects.toThrow(/Incomplete Inbox bucket 1/); + }); + + it('rebuilds a stored bucket re-delivered in full with a replaced tail', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2MessageBuckets(msgs.slice(0, 3)); + // An L1 reorg drops bucket 1's last message; the re-sync replays the whole L1 block it lives in. + await messageStore.removeL1ToL2Messages(msgs[2].index); + const replacement = makeNextMessage(msgs[1], { seq: 1n, timestamp: 100n }); + await messageStore.addL1ToL2MessageBuckets([msgs[0], msgs[1], replacement]); + + expect(await messageStore.getInboxBucket(1n)).toEqual({ + seq: 1n, + inboxRollingHash: replacement.inboxRollingHash, + totalMsgCount: 3n, + timestamp: 100n, + msgCount: 3, + lastMessageIndex: replacement.index, + }); + expect(await messageStore.getTotalL1ToL2MessageCount()).toEqual(3n); + }); + + it('rejects a message opening a bucket older than the newest stored one', async () => { + const msgs = makeBucketedMessages([ + { seq: 1n, timestamp: 100n }, + { seq: 2n, timestamp: 200n }, + { seq: 4n, timestamp: 400n }, + ]); + await messageStore.addL1ToL2MessageBuckets(msgs); + + const stale = makeNextMessage(msgs[2], { seq: 3n, timestamp: 300n }); + await expect(messageStore.addL1ToL2MessageBuckets([stale])).rejects.toThrow(/Cannot open Inbox bucket 3/); }); it('throws if the consensus rolling hash is not correct', async () => { const msgs = makeInboxMessages(5); msgs[1].inboxRollingHash = Fr.random(); - await expect(messageStore.addL1ToL2Messages(msgs)).rejects.toThrow(MessageStoreError); + await expect(messageStore.addL1ToL2MessageBuckets(msgs)).rejects.toThrow(MessageStoreError); }); it('resolves the latest bucket at or before a timestamp', async () => { - await messageStore.addL1ToL2Messages(makeBucketedMessages(threeBucketSpec)); + await messageStore.addL1ToL2MessageBuckets(makeBucketedMessages(threeBucketSpec)); expect((await messageStore.getLatestInboxBucketAtOrBefore(100n))!.seq).toEqual(1n); expect((await messageStore.getLatestInboxBucketAtOrBefore(150n))!.seq).toEqual(1n); @@ -422,14 +463,14 @@ describe('MessageStore', () => { { seq: 2n, timestamp: 200n }, { seq: 3n, timestamp: 200n }, ]); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); expect((await messageStore.getLatestInboxBucketAtOrBefore(200n))!.seq).toEqual(3n); }); it('returns messages between buckets in insertion order', async () => { const msgs = makeBucketedMessages(threeBucketSpec); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); const leaves = msgs.map(m => m.leaf); expect(await messageStore.getL1ToL2MessagesBetweenBuckets(0n, 3n)).toEqual(leaves); @@ -445,7 +486,7 @@ describe('MessageStore', () => { it('rewinds buckets when messages are removed', async () => { const msgs = makeBucketedMessages(threeBucketSpec); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); // Remove the last two messages (msgs[4] in bucket 2, msgs[5] in bucket 3), splitting bucket 2. await messageStore.removeL1ToL2Messages(msgs[4].index); @@ -458,9 +499,8 @@ describe('MessageStore', () => { timestamp: 200n, msgCount: 1, lastMessageIndex: msgs[3].index, - isOpen: true, }); - expect(await messageStore.getInboxBucket(1n)).toMatchObject({ msgCount: 3, totalMsgCount: 3n, isOpen: false }); + expect(await messageStore.getInboxBucket(1n)).toMatchObject({ msgCount: 3, totalMsgCount: 3n }); // Bucket 3's timestamp index entry is gone, so an at-or-before lookup falls back to bucket 2. expect((await messageStore.getLatestInboxBucketAtOrBefore(300n))!.seq).toEqual(2n); @@ -472,7 +512,7 @@ describe('MessageStore', () => { { seq: 2n, timestamp: 200n }, { seq: 3n, timestamp: 200n }, ]); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); // Removing the last message deletes bucket 3, whose timestamp (200) is shared with the surviving bucket 2. await messageStore.removeL1ToL2Messages(msgs[2].index); @@ -483,7 +523,7 @@ describe('MessageStore', () => { it('clears all buckets when every message is removed', async () => { const msgs = makeBucketedMessages(threeBucketSpec); - await messageStore.addL1ToL2Messages(msgs); + await messageStore.addL1ToL2MessageBuckets(msgs); await messageStore.removeL1ToL2Messages(msgs[0].index); diff --git a/yarn-project/archiver/src/store/message_store.ts b/yarn-project/archiver/src/store/message_store.ts index 543aebd984e2..133e2a974685 100644 --- a/yarn-project/archiver/src/store/message_store.ts +++ b/yarn-project/archiver/src/store/message_store.ts @@ -23,14 +23,17 @@ import { } from '../structs/inbox_message.js'; /** - * Persisted snapshot of an Inbox rolling-hash bucket. Mirrors the fields the on-chain Inbox tracks per bucket, - * plus the last absorbed message index so the between-buckets query can range-scan messages directly. + * Persisted snapshot of an Inbox rolling-hash bucket. Mirrors the fields the on-chain Inbox tracks per bucket, plus + * the L1 block the bucket was opened in and the index span of its messages, so rollbacks and range queries can work + * off bucket records alone without scanning messages. */ type BucketSnapshot = { inboxRollingHash: Fr; totalMsgCount: bigint; timestamp: bigint; + l1BlockNumber: bigint; msgCount: number; + firstMessageIndex: bigint; lastMessageIndex: bigint; }; @@ -39,7 +42,9 @@ function serializeBucketSnapshot(snapshot: BucketSnapshot): Buffer { snapshot.inboxRollingHash, bigintToUInt64BE(snapshot.totalMsgCount), bigintToUInt64BE(snapshot.timestamp), + bigintToUInt64BE(snapshot.l1BlockNumber), numToUInt32BE(snapshot.msgCount), + bigintToUInt64BE(snapshot.firstMessageIndex), bigintToUInt64BE(snapshot.lastMessageIndex), ]); } @@ -49,9 +54,34 @@ function deserializeBucketSnapshot(buffer: Buffer): BucketSnapshot { const inboxRollingHash = reader.readObject(Fr); const totalMsgCount = reader.readUInt64(); const timestamp = reader.readUInt64(); + const l1BlockNumber = reader.readUInt64(); const msgCount = reader.readNumber(); + const firstMessageIndex = reader.readUInt64(); const lastMessageIndex = reader.readUInt64(); - return { inboxRollingHash, totalMsgCount, timestamp, msgCount, lastMessageIndex }; + return { inboxRollingHash, totalMsgCount, timestamp, l1BlockNumber, msgCount, firstMessageIndex, lastMessageIndex }; +} + +/** The messages of a single Inbox bucket within an incoming batch, in insertion order. */ +type IncomingBucket = { + seq: bigint; + messages: InboxMessage[]; +}; + +/** + * Splits an incoming batch of messages into per-bucket groups, in delivery order. Messages arrive ordered by index + * and a bucket's messages are contiguous within that order, so a group ends as soon as the bucket sequence changes. + */ +function groupMessagesByBucket(messages: InboxMessage[]): IncomingBucket[] { + const buckets: IncomingBucket[] = []; + for (const message of messages) { + const current = buckets.at(-1); + if (current !== undefined && current.seq === message.bucketSeq) { + current.messages.push(message); + } else { + buckets.push({ seq: message.bucketSeq, messages: [message] }); + } + } + return buckets; } export class MessageStoreError extends Error { @@ -137,11 +167,20 @@ export class MessageStore { } /** - * Append L1 to L2 messages to the store. - * Requires new messages to be in order and strictly after the last message added. - * Throws if out of order messages are added or if the rolling hash is invalid. + * Appends L1 to L2 messages to the store, one whole Inbox bucket at a time. + * + * A bucket is opened and closed within a single L1 block, and callers retrieve Inbox logs in whole-L1-block ranges, + * so every message of a bucket reaches this method in the same call — including the rollover buckets a full block + * spills into. The bucket snapshots are derived from that: the batch is split per bucket sequence and each bucket + * gets a single snapshot built from its complete message set. Delivering only part of a bucket already held in the + * store is rejected, since the snapshot would then undercount the bucket. Delivering a stored bucket again from its + * first message is allowed: an L1 reorg can replace a bucket's tail, and the re-sync that follows replays the whole + * L1 block it lives in. + * + * Requires messages to be ordered by index and to continue the stored chain. Throws a `MessageStoreError` if + * messages arrive out of order, if the rolling hash chain breaks, or if a bucket arrives incomplete. */ - public addL1ToL2Messages(messages: InboxMessage[]): Promise { + public addL1ToL2MessageBuckets(messages: InboxMessage[]): Promise { if (messages.length === 0) { return Promise.resolve(); } @@ -150,13 +189,8 @@ export class MessageStore { let lastMessage = await this.getLastMessage(); let messageCount = 0; - // Running cumulative message count and in-progress bucket state, threaded across the batch so we can snapshot - // each Inbox bucket as its messages are inserted. Seeded from the last stored message so a bucket that spans - // two batches keeps accumulating. - let cumulativeTotal = await this.getTotalL1ToL2MessageCount(); - let currentBucketSeq: bigint | undefined = lastMessage?.bucketSeq; - let currentBucketMsgCount = - currentBucketSeq !== undefined ? ((await this.getBucketSnapshotBySeq(currentBucketSeq))?.msgCount ?? 0) : 0; + const incomingBuckets = groupMessagesByBucket(messages); + await this.assertIncomingBucketsAreComplete(incomingBuckets); for (const message of messages) { // Check messages are inserted in increasing order, but allow reinserting messages. @@ -241,31 +275,74 @@ export class MessageStore { await this.#l1ToL2MessageIndices.set(this.leafToIndexKey(message.leaf), message.index); messageCount++; - // Snapshot the bucket this message was absorbed into. A message opens a new bucket whenever its bucket - // sequence differs from the one currently being accumulated; otherwise it extends the current bucket. - cumulativeTotal += 1n; - if (currentBucketSeq === undefined || message.bucketSeq !== currentBucketSeq) { - currentBucketSeq = message.bucketSeq; - currentBucketMsgCount = 0; - } - currentBucketMsgCount += 1; - await this.writeBucketSnapshot(message.bucketSeq, { - inboxRollingHash: message.inboxRollingHash, - totalMsgCount: cumulativeTotal, - timestamp: message.bucketTimestamp, - msgCount: currentBucketMsgCount, - lastMessageIndex: message.index, - }); - this.#log.trace(`Inserted L1 to L2 message ${message.leaf} with index ${message.index} into the store`); lastMessage = message; } + await this.writeIncomingBucketSnapshots(incomingBuckets); + // Update total message count with the number of inserted messages. await this.increaseTotalMessageCount(messageCount); }); } + /** + * Rejects a batch that delivers an Inbox bucket the store already holds without replaying it from its first message, + * or that opens a bucket older than the newest one stored. Either would produce a snapshot that disagrees with the + * messages it covers, since each snapshot is derived from the batch's messages for that bucket alone. + */ + private async assertIncomingBucketsAreComplete(incomingBuckets: IncomingBucket[]): Promise { + const newestStoredSeq = await this.getNewestBucketSeq(); + let previousSeq: bigint | undefined; + for (const bucket of incomingBuckets) { + if (previousSeq !== undefined && bucket.seq <= previousSeq) { + throw new MessageStoreError( + `Inbox bucket ${bucket.seq} arrives after bucket ${previousSeq} in the same batch`, + bucket.messages[0], + ); + } + previousSeq = bucket.seq; + + const stored = await this.getBucketSnapshotBySeq(bucket.seq); + if (stored === undefined) { + if (newestStoredSeq !== undefined && bucket.seq <= newestStoredSeq) { + throw new MessageStoreError( + `Cannot open Inbox bucket ${bucket.seq} after bucket ${newestStoredSeq} has been stored`, + bucket.messages[0], + ); + } + } else if (stored.firstMessageIndex !== bucket.messages[0].index) { + throw new MessageStoreError( + `Incomplete Inbox bucket ${bucket.seq}: stored messages start at index ${stored.firstMessageIndex} ` + + `but the batch starts at index ${bucket.messages[0].index}`, + bucket.messages[0], + ); + } + } + } + + /** + * Writes one snapshot per bucket in the batch, each derived from the bucket's complete message set. Cumulative + * totals thread forward from the bucket preceding the batch, so a bucket re-delivered with extra messages shifts + * the totals of the buckets after it within the same batch. + */ + private async writeIncomingBucketSnapshots(incomingBuckets: IncomingBucket[]): Promise { + let cumulativeTotal = await this.getTotalMsgCountBeforeBucket(incomingBuckets[0].seq); + for (const { seq, messages } of incomingBuckets) { + const lastInBucket = messages.at(-1)!; + cumulativeTotal += BigInt(messages.length); + await this.writeBucketSnapshot(seq, { + inboxRollingHash: lastInBucket.inboxRollingHash, + totalMsgCount: cumulativeTotal, + timestamp: lastInBucket.bucketTimestamp, + l1BlockNumber: lastInBucket.l1BlockNumber, + msgCount: messages.length, + firstMessageIndex: messages[0].index, + lastMessageIndex: lastInBucket.index, + }); + } + } + /** * Gets the L1 to L2 message index in the L1 to L2 message tree. * @param l1ToL2Message - The L1 to L2 message. @@ -391,11 +468,14 @@ export class MessageStore { } msgCount += 1; } + const stored = await this.getBucketSnapshotBySeq(boundarySeq); await this.writeBucketSnapshot(boundarySeq, { inboxRollingHash: lastRemaining.inboxRollingHash, totalMsgCount: await this.getTotalL1ToL2MessageCount(), timestamp: lastRemaining.bucketTimestamp, + l1BlockNumber: stored?.l1BlockNumber ?? lastRemaining.l1BlockNumber, msgCount, + firstMessageIndex: stored?.firstMessageIndex ?? lastRemaining.index - BigInt(msgCount) + 1n, lastMessageIndex: lastRemaining.index, }); } @@ -466,8 +546,21 @@ export class MessageStore { await this.#bucketTimestampToSeq.set(this.timestampToKey(snapshot.timestamp), this.bucketSeqToKey(seq)); } - private async toInboxBucket(seq: bigint, snapshot: BucketSnapshot): Promise { - const lastMessage = await this.getLastMessage(); + /** Returns the sequence number of the newest stored bucket, or undefined if none has been stored yet. */ + private async getNewestBucketSeq(): Promise { + const [seqKey] = await toArray(this.#inboxBuckets.keysAsync({ reverse: true, limit: 1 })); + return seqKey === undefined ? undefined : BigInt(seqKey); + } + + /** Returns the cumulative Inbox message count through the newest stored bucket before the given sequence number. */ + private async getTotalMsgCountBeforeBucket(seq: bigint): Promise { + const [snapBuffer] = await toArray( + this.#inboxBuckets.valuesAsync({ end: this.bucketSeqToKey(seq) - 1, reverse: true, limit: 1 }), + ); + return snapBuffer === undefined ? 0n : deserializeBucketSnapshot(snapBuffer).totalMsgCount; + } + + private toInboxBucket(seq: bigint, snapshot: BucketSnapshot): InboxBucket { return { seq, inboxRollingHash: snapshot.inboxRollingHash, @@ -475,7 +568,6 @@ export class MessageStore { timestamp: snapshot.timestamp, msgCount: snapshot.msgCount, lastMessageIndex: snapshot.lastMessageIndex, - isOpen: lastMessage?.bucketSeq === seq, }; } diff --git a/yarn-project/stdlib/src/interfaces/archiver.test.ts b/yarn-project/stdlib/src/interfaces/archiver.test.ts index 2b005e726b23..11658128468e 100644 --- a/yarn-project/stdlib/src/interfaces/archiver.test.ts +++ b/yarn-project/stdlib/src/interfaces/archiver.test.ts @@ -216,12 +216,12 @@ describe('ArchiverApiSchema', () => { it('getLatestInboxBucketAtOrBefore', async () => { const result = await context.client.getLatestInboxBucketAtOrBefore(123n); - expect(result).toMatchObject({ seq: 1n, msgCount: 3, totalMsgCount: 3n, isOpen: false }); + expect(result).toMatchObject({ seq: 1n, msgCount: 3, totalMsgCount: 3n }); }); it('getInboxBucket', async () => { const result = await context.client.getInboxBucket(2n); - expect(result).toMatchObject({ seq: 2n, msgCount: 3, isOpen: true }); + expect(result).toMatchObject({ seq: 2n, msgCount: 3 }); }); it('getL1ToL2MessagesBetweenBuckets', async () => { @@ -576,7 +576,6 @@ class MockArchiver implements ArchiverApi { timestamp: 100n, msgCount: 3, lastMessageIndex: 2n, - isOpen: false, }); } getInboxBucket(seq: bigint): Promise { @@ -588,7 +587,6 @@ class MockArchiver implements ArchiverApi { timestamp: 100n, msgCount: 3, lastMessageIndex: 2n, - isOpen: true, }); } getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { diff --git a/yarn-project/stdlib/src/messaging/inbox_bucket.ts b/yarn-project/stdlib/src/messaging/inbox_bucket.ts index 5dd08e09d2b9..e515ec6e66b4 100644 --- a/yarn-project/stdlib/src/messaging/inbox_bucket.ts +++ b/yarn-project/stdlib/src/messaging/inbox_bucket.ts @@ -25,11 +25,6 @@ export type InboxBucket = { msgCount: number; /** Global leaf index of the last message absorbed into this bucket. */ lastMessageIndex: bigint; - /** - * Whether this is the latest bucket the archiver has synced. A latest bucket may still grow as more messages - * arrive on L1; earlier buckets are complete. Consumers that need a settled bucket apply a lag on the timestamp. - */ - isOpen: boolean; }; export const InboxBucketSchema = z.object({ @@ -39,5 +34,4 @@ export const InboxBucketSchema = z.object({ timestamp: schemas.BigInt, msgCount: schemas.Integer, lastMessageIndex: schemas.BigInt, - isOpen: z.boolean(), }) satisfies z.ZodType; diff --git a/yarn-project/validator-client/src/validator.integration.test.ts b/yarn-project/validator-client/src/validator.integration.test.ts index 4dbb9302ce25..287705a17ce3 100644 --- a/yarn-project/validator-client/src/validator.integration.test.ts +++ b/yarn-project/validator-client/src/validator.integration.test.ts @@ -460,8 +460,8 @@ describe('ValidatorClient Integration', () => { it('validates and attests with txs anchored to proposed blocks and non-empty l1-to-l2 messages', async () => { // Create l1 to l2 messages and seed them into the archivers const l1ToL2Messages = makeInboxMessages(4, { messagesPerCheckpoint: 4 }); - await proposer.archiver.dataStores.messages.addL1ToL2Messages(l1ToL2Messages); - await attestor.archiver.dataStores.messages.addL1ToL2Messages(l1ToL2Messages); + await proposer.archiver.dataStores.messages.addL1ToL2MessageBuckets(l1ToL2Messages); + await attestor.archiver.dataStores.messages.addL1ToL2MessageBuckets(l1ToL2Messages); // Build txs anchored to the previously proposed block const { blocks, proposal } = await buildCheckpoint( @@ -660,10 +660,10 @@ describe('ValidatorClient Integration', () => { it('refuses block proposal with mismatching l1 to l2 messages', async () => { const l1ToL2Messages = makeInboxMessages(4, { messagesPerCheckpoint: 4 }); - await proposer.archiver.dataStores.messages.addL1ToL2Messages(l1ToL2Messages); + await proposer.archiver.dataStores.messages.addL1ToL2MessageBuckets(l1ToL2Messages); const otherL1ToL2Messages = makeInboxMessages(4, { messagesPerCheckpoint: 4 }); - await attestor.archiver.dataStores.messages.addL1ToL2Messages(otherL1ToL2Messages); + await attestor.archiver.dataStores.messages.addL1ToL2MessageBuckets(otherL1ToL2Messages); const { blocks } = await buildCheckpoint( CheckpointNumber(1), From 4b322d9914292ec0b03f4867db8b9fa1fedb8313 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Mon, 27 Jul 2026 15:11:25 -0300 Subject: [PATCH 5/6] fix(fast-inbox): fail loudly when an inbox bucket range names an unsynced bucket (A-1379) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `getL1ToL2MessagesBetweenBuckets` returned an empty array when either bound was missing from the store, which callers could not tell apart from a range that genuinely holds no messages — and every caller asks because it wants the range's messages. Throw `InboxBucketNotSyncedError` instead, so a node that is behind on L1 sync surfaces that rather than deriving an empty message bundle, and reject an inverted range outright. --- yarn-project/archiver/src/errors.ts | 11 +++++ .../archiver/src/store/message_store.test.ts | 18 +++++--- .../archiver/src/store/message_store.ts | 43 +++++++++++-------- .../src/messaging/l1_to_l2_message_source.ts | 6 ++- 4 files changed, 54 insertions(+), 24 deletions(-) diff --git a/yarn-project/archiver/src/errors.ts b/yarn-project/archiver/src/errors.ts index db3be133c0b0..565a2716e9ba 100644 --- a/yarn-project/archiver/src/errors.ts +++ b/yarn-project/archiver/src/errors.ts @@ -116,6 +116,17 @@ export class L1ToL2MessagesNotReadyError extends Error { } } +/** + * Thrown when a query names an Inbox bucket this archiver has not synced. Distinguishes "not synced yet, retry once + * L1 sync catches up" from a genuinely empty result. + */ +export class InboxBucketNotSyncedError extends Error { + constructor(public readonly bucketSeq: bigint) { + super(`Inbox bucket ${bucketSeq} has not been synced`); + this.name = 'InboxBucketNotSyncedError'; + } +} + /** Thrown when a proposed checkpoint number is stale (already processed). */ export class ProposedCheckpointStaleError extends Error { constructor( diff --git a/yarn-project/archiver/src/store/message_store.test.ts b/yarn-project/archiver/src/store/message_store.test.ts index bfebd4121177..1823bdee8672 100644 --- a/yarn-project/archiver/src/store/message_store.test.ts +++ b/yarn-project/archiver/src/store/message_store.test.ts @@ -8,7 +8,7 @@ import { Checkpoint, type PublishedCheckpoint } from '@aztec/stdlib/checkpoint'; import { updateInboxRollingHash } from '@aztec/stdlib/messaging'; import '@aztec/stdlib/testing/jest'; -import { L1ToL2MessagesNotReadyError } from '../errors.js'; +import { InboxBucketNotSyncedError, L1ToL2MessagesNotReadyError } from '../errors.js'; import { type InboxMessage, updateRollingHash } from '../structs/inbox_message.js'; import { makeInboxMessage, @@ -478,10 +478,18 @@ describe('MessageStore', () => { expect(await messageStore.getL1ToL2MessagesBetweenBuckets(2n, 3n)).toEqual(leaves.slice(5)); // An empty (fromExclusive, toInclusive] range yields no messages. expect(await messageStore.getL1ToL2MessagesBetweenBuckets(3n, 3n)).toEqual([]); - // Unknown upper bucket yields no messages. - expect(await messageStore.getL1ToL2MessagesBetweenBuckets(0n, 9n)).toEqual([]); - // An unknown nonzero lower bucket is treated as unavailable, not as genesis. - expect(await messageStore.getL1ToL2MessagesBetweenBuckets(4n, 3n)).toEqual([]); + }); + + it('throws when a bucket bound has not been synced', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2MessageBuckets(msgs); + + // An unsynced bound is reported rather than collapsing into an empty range. + await expect(messageStore.getL1ToL2MessagesBetweenBuckets(0n, 9n)).rejects.toThrow(InboxBucketNotSyncedError); + await expect(messageStore.getL1ToL2MessagesBetweenBuckets(9n, 12n)).rejects.toThrow(InboxBucketNotSyncedError); + // A nonzero lower bound is never treated as genesis. + await expect(messageStore.getL1ToL2MessagesBetweenBuckets(4n, 5n)).rejects.toThrow(InboxBucketNotSyncedError); + await expect(messageStore.getL1ToL2MessagesBetweenBuckets(4n, 3n)).rejects.toThrow(/Invalid Inbox bucket range/); }); it('rewinds buckets when messages are removed', async () => { diff --git a/yarn-project/archiver/src/store/message_store.ts b/yarn-project/archiver/src/store/message_store.ts index 133e2a974685..d19c43c65f06 100644 --- a/yarn-project/archiver/src/store/message_store.ts +++ b/yarn-project/archiver/src/store/message_store.ts @@ -14,7 +14,7 @@ import { } from '@aztec/kv-store'; import { type InboxBucket, InboxLeaf, updateInboxRollingHash } from '@aztec/stdlib/messaging'; -import { L1ToL2MessagesNotReadyError } from '../errors.js'; +import { InboxBucketNotSyncedError, L1ToL2MessagesNotReadyError } from '../errors.js'; import { type InboxMessage, deserializeInboxMessage, @@ -505,27 +505,27 @@ export class MessageStore { } /** - * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion - * order (AZIP-22 Fast Inbox). Returns an empty array if the upper bucket has not been synced. + * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion order + * (AZIP-22 Fast Inbox). Both bounds must name buckets this archiver has synced, so that an empty result means the + * range holds no messages rather than hiding an unsynced bound; callers route the + * `InboxBucketNotSyncedError` to their own catch-up handling. Sequence 0 is the genesis base case and always + * resolves: the range then starts at the first message of the Inbox. */ public async getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { - const toBucket = await this.getBucketSnapshotBySeq(toInclusive); - if (toBucket === undefined) { - return []; + if (fromExclusive > toInclusive) { + throw new Error(`Invalid Inbox bucket range (${fromExclusive}, ${toInclusive}]`); } - // A nonzero lower bound must reference a synced bucket; otherwise the range is unavailable (rather than - // defaulting to genesis, which would silently over-return). Sequence 0 is the genesis base case: start from - // the beginning of the Inbox. - let startIndex = 0n; - if (fromExclusive > 0n) { - const fromBucket = await this.getBucketSnapshotBySeq(fromExclusive); - if (fromBucket === undefined) { - return []; - } - startIndex = fromBucket.lastMessageIndex + 1n; + if (toInclusive === 0n) { + return []; } - const endIndexExclusive = toBucket.lastMessageIndex + 1n; + const endIndexExclusive = (await this.getSyncedBucketSnapshot(toInclusive)).lastMessageIndex + 1n; + const startIndex = + fromExclusive === 0n ? 0n : (await this.getSyncedBucketSnapshot(fromExclusive)).lastMessageIndex + 1n; + return this.getMessageLeavesInIndexRange(startIndex, endIndexExclusive); + } + /** Collects the message leaves in the global index range `[startIndex, endIndexExclusive)`, in insertion order. */ + private async getMessageLeavesInIndexRange(startIndex: bigint, endIndexExclusive: bigint): Promise { const leaves: Fr[] = []; for await (const msgBuffer of this.#l1ToL2Messages.valuesAsync({ start: this.indexToKey(startIndex), @@ -541,6 +541,15 @@ export class MessageStore { return buffer && deserializeBucketSnapshot(buffer); } + /** Reads a bucket snapshot, failing loudly if the archiver has not synced that bucket. */ + private async getSyncedBucketSnapshot(seq: bigint): Promise { + const snapshot = await this.getBucketSnapshotBySeq(seq); + if (snapshot === undefined) { + throw new InboxBucketNotSyncedError(seq); + } + return snapshot; + } + private async writeBucketSnapshot(seq: bigint, snapshot: BucketSnapshot): Promise { await this.#inboxBuckets.set(this.bucketSeqToKey(seq), serializeBucketSnapshot(snapshot)); await this.#bucketTimestampToSeq.set(this.timestampToKey(snapshot.timestamp), this.bucketSeqToKey(seq)); diff --git a/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts b/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts index 4e3a01172288..e30b78ecfdd8 100644 --- a/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts +++ b/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts @@ -41,8 +41,10 @@ export interface L1ToL2MessageSource { /** * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion - * order, for streaming message-bundle derivation (AZIP-22 Fast Inbox). Returns an empty array if the upper - * bucket has not been synced. + * order, for streaming message-bundle derivation (AZIP-22 Fast Inbox). Both bounds must name buckets the source + * has synced; it throws otherwise, so that an empty result means the range holds no messages instead of hiding an + * unsynced bound. Callers that can tolerate an unsynced source resolve both bounds first, or map the failure to + * their own catch-up handling. * @param fromExclusive - The lower bucket sequence bound, exclusive (0 means from the start of the Inbox). * @param toInclusive - The upper bucket sequence bound, inclusive. */ From 562c5aa9f7c3a5b322ac5ec26d6c38895708c62b Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Mon, 27 Jul 2026 15:13:52 -0300 Subject: [PATCH 6/6] refactor(fast-inbox): index inbox buckets by timestamp through a multimap (A-1379) The bucket rewind after a message removal deleted the timestamp index entry of every bucket it dropped, which also unindexed the rollover siblings sharing that timestamp, and then relied on rewriting the surviving boundary bucket to put the entry back. Give each bucket its own entry by making the index a multimap, so deletions touch only the bucket being removed and nothing has to be restored. Recency lookups now take the highest sequence among the entries at the newest timestamp at-or-before the requested one, and a bucket re-delivered from an L1 block re-mined at a different timestamp drops its stale entry. Also documents why the boundary bucket still has to be rebuilt: a removal can cut inside a bucket, since the synchronizer rolls back to the last message it still shares with L1 and that message need not be a bucket's last. --- .../archiver/src/store/message_store.test.ts | 33 +++++++ .../archiver/src/store/message_store.ts | 89 +++++++++++-------- 2 files changed, 87 insertions(+), 35 deletions(-) diff --git a/yarn-project/archiver/src/store/message_store.test.ts b/yarn-project/archiver/src/store/message_store.test.ts index 1823bdee8672..983624f51b8e 100644 --- a/yarn-project/archiver/src/store/message_store.test.ts +++ b/yarn-project/archiver/src/store/message_store.test.ts @@ -529,6 +529,39 @@ describe('MessageStore', () => { expect((await messageStore.getLatestInboxBucketAtOrBefore(200n))!.seq).toEqual(2n); }); + it('keeps rollover siblings indexed when a bucket sharing their timestamp is removed', async () => { + // Three buckets rolling over within one L1 block, so all three share timestamp 100. + const msgs = makeBucketedMessages([ + { seq: 1n, timestamp: 100n }, + { seq: 2n, timestamp: 100n }, + { seq: 3n, timestamp: 100n }, + ]); + await messageStore.addL1ToL2MessageBuckets(msgs); + + await messageStore.removeL1ToL2Messages(msgs[2].index); + + expect(await messageStore.getInboxBucket(3n)).toBeUndefined(); + expect(await messageStore.getInboxBucket(1n)).toMatchObject({ msgCount: 1, totalMsgCount: 1n }); + expect((await messageStore.getLatestInboxBucketAtOrBefore(100n))!.seq).toEqual(2n); + }); + + it('reindexes a bucket re-delivered from an L1 block with a different timestamp', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2MessageBuckets(msgs.slice(0, 3)); + await messageStore.removeL1ToL2Messages(msgs[2].index); + + // The reorged L1 block holding bucket 1 was re-mined at a later timestamp. + const replayed = [msgs[0], msgs[1], makeNextMessage(msgs[1], { seq: 1n, timestamp: 100n })].map(msg => ({ + ...msg, + bucketTimestamp: 150n, + })); + await messageStore.addL1ToL2MessageBuckets(replayed); + + expect(await messageStore.getInboxBucket(1n)).toMatchObject({ timestamp: 150n, msgCount: 3 }); + expect(await messageStore.getLatestInboxBucketAtOrBefore(100n)).toBeUndefined(); + expect((await messageStore.getLatestInboxBucketAtOrBefore(150n))!.seq).toEqual(1n); + }); + it('clears all buckets when every message is removed', async () => { const msgs = makeBucketedMessages(threeBucketSpec); await messageStore.addL1ToL2MessageBuckets(msgs); diff --git a/yarn-project/archiver/src/store/message_store.ts b/yarn-project/archiver/src/store/message_store.ts index d19c43c65f06..f23135da5e72 100644 --- a/yarn-project/archiver/src/store/message_store.ts +++ b/yarn-project/archiver/src/store/message_store.ts @@ -8,6 +8,7 @@ import { BufferReader, bigintToUInt64BE, numToUInt32BE, serializeToBuffer } from import { type AztecAsyncKVStore, type AztecAsyncMap, + type AztecAsyncMultiMap, type AztecAsyncSingleton, type CustomRange, mapRange, @@ -109,8 +110,12 @@ export class MessageStore { #messagesFinalizedL1Block: AztecAsyncSingleton; /** Maps from Inbox bucket sequence number to its serialized snapshot. */ #inboxBuckets: AztecAsyncMap; - /** Maps from a bucket's L1 timestamp (key) to the highest bucket sequence number opened at that timestamp. */ - #bucketTimestampToSeq: AztecAsyncMap; + /** + * Maps from a bucket's L1 timestamp (key) to the sequence numbers of the buckets opened at that timestamp. Holds + * several sequences per timestamp when a full bucket rolls over within one L1 block, so that deleting one bucket + * leaves its rollover siblings indexed. + */ + #bucketTimestampToSeq: AztecAsyncMultiMap; #log = createLogger('archiver:message_store'); @@ -122,7 +127,7 @@ export class MessageStore { this.#inboxTreeInProgress = db.openSingleton('archiver_inbox_tree_in_progress'); this.#messagesFinalizedL1Block = db.openSingleton('archiver_messages_finalized_l1_block'); this.#inboxBuckets = db.openMap('archiver_inbox_buckets'); - this.#bucketTimestampToSeq = db.openMap('archiver_inbox_bucket_timestamps'); + this.#bucketTimestampToSeq = db.openMultiMap('archiver_inbox_bucket_timestamps'); } public async getTotalL1ToL2MessageCount(): Promise { @@ -441,44 +446,44 @@ export class MessageStore { } /** - * Rewinds the Inbox bucket snapshots to match the messages remaining after a removal. Buckets whose messages - * were all removed are deleted, and the boundary bucket (the one holding the last surviving message) is - * recomputed from its remaining messages, since a checkpoint-aligned removal can split a bucket. Must run - * inside the removal transaction, after the total message count has been updated. + * Rewinds the Inbox bucket snapshots to match the messages left after a removal. Each bucket past the surviving tip + * is deleted together with its own timestamp index entry, leaving the entries of the rollover siblings it shares a + * timestamp with in place. + * + * The bucket holding the last surviving message is rewritten from the messages it has left, because a removal can + * cut inside a bucket: the synchronizer rolls back to the last message it still shares with L1 after a reorg, and + * that message need not be the last of its bucket. Must run inside the removal transaction, after the total message + * count has been updated. */ private async rewindBucketsAfterRemoval(): Promise { const lastRemaining = await this.getLastMessage(); const boundarySeq = lastRemaining?.bucketSeq; - // Delete snapshots (and their timestamp index entries) for buckets entirely past the surviving tip. const deleteFromKey = boundarySeq === undefined ? 0 : this.bucketSeqToKey(boundarySeq) + 1; for await (const [seqKey, snapBuffer] of this.#inboxBuckets.entriesAsync({ start: deleteFromKey })) { const snapshot = deserializeBucketSnapshot(snapBuffer); - await this.#bucketTimestampToSeq.delete(this.timestampToKey(snapshot.timestamp)); + await this.#bucketTimestampToSeq.deleteValue(this.timestampToKey(snapshot.timestamp), seqKey); await this.#inboxBuckets.delete(seqKey); } - // Recompute the boundary bucket from its surviving messages. This also restores its timestamp index entry if a - // just-deleted rollover bucket shared the timestamp. - if (lastRemaining !== undefined && boundarySeq !== undefined) { - let msgCount = 0; - for await (const msg of this.iterateL1ToL2Messages({ reverse: true })) { - if (msg.bucketSeq !== boundarySeq) { - break; - } - msgCount += 1; + if (lastRemaining === undefined || boundarySeq === undefined) { + return; + } + let msgCount = 0; + for await (const msg of this.iterateL1ToL2Messages({ reverse: true })) { + if (msg.bucketSeq !== boundarySeq) { + break; } - const stored = await this.getBucketSnapshotBySeq(boundarySeq); - await this.writeBucketSnapshot(boundarySeq, { - inboxRollingHash: lastRemaining.inboxRollingHash, - totalMsgCount: await this.getTotalL1ToL2MessageCount(), - timestamp: lastRemaining.bucketTimestamp, - l1BlockNumber: stored?.l1BlockNumber ?? lastRemaining.l1BlockNumber, - msgCount, - firstMessageIndex: stored?.firstMessageIndex ?? lastRemaining.index - BigInt(msgCount) + 1n, - lastMessageIndex: lastRemaining.index, - }); + msgCount += 1; } + const stored = await this.getSyncedBucketSnapshot(boundarySeq); + await this.writeBucketSnapshot(boundarySeq, { + ...stored, + inboxRollingHash: lastRemaining.inboxRollingHash, + totalMsgCount: await this.getTotalL1ToL2MessageCount(), + msgCount, + lastMessageIndex: lastRemaining.index, + }); } /** @@ -495,13 +500,22 @@ export class MessageStore { * was opened strictly after it (AZIP-22 Fast Inbox). */ public async getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise { - // Bucket timestamps are non-decreasing in sequence number, and the index holds the highest sequence per - // timestamp. A reverse scan bounded above (inclusively) by the requested timestamp yields, first, the value - // at the largest timestamp at-or-before it — the bucket sequence we want. - const [seq] = await toArray( - this.#bucketTimestampToSeq.valuesAsync({ end: this.timestampToKey(timestamp), reverse: true, limit: 1 }), - ); - return seq === undefined ? undefined : this.getInboxBucket(BigInt(seq)); + // Bucket timestamps are non-decreasing in sequence number, so the bucket we want is the highest sequence indexed + // at the largest timestamp at-or-before the requested one. A reverse scan bounded above (inclusively) by that + // timestamp visits it first; rollover buckets share a timestamp, so keep the highest of its sequences. + let latestTimestampKey: number | undefined; + let latestSeqKey: number | undefined; + for await (const [timestampKey, seqKey] of this.#bucketTimestampToSeq.entriesAsync({ + end: this.timestampToKey(timestamp), + reverse: true, + })) { + if (latestTimestampKey !== undefined && timestampKey !== latestTimestampKey) { + break; + } + latestTimestampKey = timestampKey; + latestSeqKey = latestSeqKey === undefined ? seqKey : Math.max(latestSeqKey, seqKey); + } + return latestSeqKey === undefined ? undefined : this.getInboxBucket(BigInt(latestSeqKey)); } /** @@ -551,6 +565,11 @@ export class MessageStore { } private async writeBucketSnapshot(seq: bigint, snapshot: BucketSnapshot): Promise { + // A reorg can re-deliver a bucket from an L1 block mined at a different timestamp, so drop the stale index entry. + const stored = await this.getBucketSnapshotBySeq(seq); + if (stored !== undefined && stored.timestamp !== snapshot.timestamp) { + await this.#bucketTimestampToSeq.deleteValue(this.timestampToKey(stored.timestamp), this.bucketSeqToKey(seq)); + } await this.#inboxBuckets.set(this.bucketSeqToKey(seq), serializeBucketSnapshot(snapshot)); await this.#bucketTimestampToSeq.set(this.timestampToKey(snapshot.timestamp), this.bucketSeqToKey(seq)); }