diff --git a/yarn-project/archiver/README.md b/yarn-project/archiver/README.md index bac1af3774cd..5d4a8e92a35a 100644 --- a/yarn-project/archiver/README.md +++ b/yarn-project/archiver/README.md @@ -43,14 +43,18 @@ Two independent syncpoints track progress on L1: ### L1-to-L2 Messages -Messages are synced from the Inbox contract via `handleL1ToL2Messages()`: +Messages are synced from the Inbox contract. The sync compares local state (message count and rolling hash) against the Inbox contract state on L1, downloads any missing messages, and verifies consistency afterwards. On success, the syncpoint advances to the current L1 block. On failure (L1 reorg or inconsistency), the syncpoint rolls back to the last known-good message and the operation retries (up to 3 times within the same sync iteration). 1. Query Inbox state at the current L1 block (message count + rolling hash) -2. Compare local vs remote state -3. If they match, nothing to do -4. If mismatch, validate the local last message still exists on L1 with the same rolling hash - - If not found or hash differs, an L1 reorg occurred: find the last common message, delete everything after, and rollback the syncpoint -5. Fetch `MessageSent` events in batches and store +2. Compare local state against remote +3. If they match, advance syncpoint and return +4. If mismatch, fetch `MessageSent` events in batches and store them + - If storing fails due to a rolling hash mismatch (indicating an L1 reorg changed or removed messages), find the last common message with L1, delete everything after, reset the syncpoint, and retry +5. After storing, verify local state matches the remote state queried in step 1 + - If still mismatched (e.g., messages missed due to a concurrent L1 reorg), rollback and retry +6. On success, advance the syncpoint + +The syncpoint and the `inboxTreeInProgress` marker (which tracks which checkpoint's messages are currently being filled on L1) are updated atomically. The marker is only advanced after messages are stored, so concurrent reads don't see an unsealed checkpoint as readable before its messages are available. ### Checkpoints @@ -81,6 +85,8 @@ The `blocksSynchedTo` syncpoint is updated: Note that the `blocksSynchedTo` pointer is NOT updated during normal sync when there are no new checkpoints. This protects against small L1 reorgs that could add a checkpoint on an L1 block we have flagged as already synced. +The `messagesSynchedTo` pointer is always advanced to the current L1 block on success. If a rolling hash mismatch or post-download inconsistency is detected, the pointer rolls back to the last common message and the operation retries. The rolling hash chain and pre/post-sync consistency checks provide the primary reorg protection. + ### Block Queue The archiver implements `L2BlockSink`, allowing other subsystems to push blocks before they appear on L1: diff --git a/yarn-project/archiver/src/archiver.ts b/yarn-project/archiver/src/archiver.ts index c922de6a683e..ffee1d49a819 100644 --- a/yarn-project/archiver/src/archiver.ts +++ b/yarn-project/archiver/src/archiver.ts @@ -490,7 +490,10 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra await this.store.rollbackL1ToL2MessagesToCheckpoint(targetCheckpointNumber); this.log.info(`Setting L1 syncpoints to ${targetL1BlockNumber}`); await this.store.setCheckpointSynchedL1BlockNumber(targetL1BlockNumber); - await this.store.setMessageSynchedL1Block({ l1BlockNumber: targetL1BlockNumber, l1BlockHash: targetL1BlockHash }); + await this.store.setMessageSyncState( + { l1BlockNumber: targetL1BlockNumber, l1BlockHash: targetL1BlockHash }, + undefined, + ); if (targetL2BlockNumber < currentProvenBlock) { this.log.info(`Rolling back proven L2 checkpoint to ${targetCheckpointNumber}`); await this.updater.setProvenCheckpointNumber(targetCheckpointNumber); diff --git a/yarn-project/archiver/src/l1/data_retrieval.ts b/yarn-project/archiver/src/l1/data_retrieval.ts index 599a33bc1bfb..4b1505565092 100644 --- a/yarn-project/archiver/src/l1/data_retrieval.ts +++ b/yarn-project/archiver/src/l1/data_retrieval.ts @@ -344,14 +344,10 @@ export async function getCheckpointBlobDataFromBlobs( /** Given an L1 to L2 message, retrieves its corresponding event from the Inbox within a specific block range. */ export async function retrieveL1ToL2Message( inbox: InboxContract, - leaf: Fr, - fromBlock: bigint, - toBlock: bigint, + message: InboxMessage, ): Promise { - const logs = await inbox.getMessageSentEventByHash(leaf.toString(), fromBlock, toBlock); - - const messages = mapLogsInboxMessage(logs); - return messages.length > 0 ? messages[0] : undefined; + const log = await inbox.getMessageSentEventByHash(message.leaf.toString(), message.l1BlockHash.toString()); + return log && mapLogInboxMessage(log); } /** @@ -374,22 +370,22 @@ export async function retrieveL1ToL2Messages( break; } - retrievedL1ToL2Messages.push(...mapLogsInboxMessage(messageSentLogs)); + retrievedL1ToL2Messages.push(...messageSentLogs.map(mapLogInboxMessage)); searchStartBlock = messageSentLogs.at(-1)!.l1BlockNumber + 1n; } return retrievedL1ToL2Messages; } -function mapLogsInboxMessage(logs: MessageSentLog[]): InboxMessage[] { - return logs.map(log => ({ +function mapLogInboxMessage(log: MessageSentLog): InboxMessage { + return { index: log.args.index, leaf: log.args.leaf, l1BlockNumber: log.l1BlockNumber, l1BlockHash: log.l1BlockHash, checkpointNumber: log.args.checkpointNumber, rollingHash: log.args.rollingHash, - })); + }; } /** Retrieves L2ProofVerified events from the rollup contract. */ diff --git a/yarn-project/archiver/src/modules/l1_synchronizer.ts b/yarn-project/archiver/src/modules/l1_synchronizer.ts index cef612efe72f..7f6f01f1b8d7 100644 --- a/yarn-project/archiver/src/modules/l1_synchronizer.ts +++ b/yarn-project/archiver/src/modules/l1_synchronizer.ts @@ -1,6 +1,6 @@ import type { BlobClientInterface } from '@aztec/blob-client/client'; import { EpochCache } from '@aztec/epoch-cache'; -import { InboxContract, RollupContract } from '@aztec/ethereum/contracts'; +import { InboxContract, type InboxContractState, RollupContract } from '@aztec/ethereum/contracts'; import type { L1BlockId } from '@aztec/ethereum/l1-types'; import type { ViemPublicClient, ViemPublicDebugClient } from '@aztec/ethereum/types'; import { asyncPool } from '@aztec/foundation/async-pool'; @@ -10,9 +10,10 @@ import { Buffer16, Buffer32 } from '@aztec/foundation/buffer'; import { pick } from '@aztec/foundation/collection'; import { Fr } from '@aztec/foundation/curves/bn254'; import { type Logger, createLogger } from '@aztec/foundation/log'; +import { retryTimes } from '@aztec/foundation/retry'; import { count } from '@aztec/foundation/string'; import { DateProvider, Timer, elapsed } from '@aztec/foundation/timer'; -import { isDefined } from '@aztec/foundation/types'; +import { isDefined, isErrorClass } from '@aztec/foundation/types'; import { type ArchiverEmitter, L2BlockSourceEvents, type ValidateCheckpointResult } from '@aztec/stdlib/block'; import { PublishedCheckpoint } from '@aztec/stdlib/checkpoint'; import { type L1RollupConstants, getEpochAtSlot, getSlotAtNextL1Block } from '@aztec/stdlib/epoch-helpers'; @@ -28,6 +29,7 @@ import { } from '../l1/data_retrieval.js'; import type { KVArchiverDataStore } from '../store/kv_archiver_store.js'; import type { L2TipsCache } from '../store/l2_tips_cache.js'; +import { MessageStoreError } from '../store/message_store.js'; import type { InboxMessage } from '../structs/inbox_message.js'; import { ArchiverDataStoreUpdater } from './data_store_updater.js'; import type { ArchiverInstrumentation } from './instrumentation.js'; @@ -121,10 +123,15 @@ export class ArchiverL1Synchronizer implements Traceable { @trackSpan('Archiver.syncFromL1') public async syncFromL1(initialSyncComplete: boolean): Promise { + // In between the various calls to L1, the block number can move meaning some of the following + // calls will return data for blocks that were not present during earlier calls. To combat this + // we ensure that all data retrieval methods only retrieve data up to the currentBlockNumber + // captured at the top of this function. const currentL1Block = await this.publicClient.getBlock({ includeTransactions: false }); const currentL1BlockNumber = currentL1Block.number; const currentL1BlockHash = Buffer32.fromString(currentL1Block.hash); const currentL1Timestamp = currentL1Block.timestamp; + const currentL1BlockData = { l1BlockNumber: currentL1BlockNumber, l1BlockHash: currentL1BlockHash }; if (this.l1BlockHash && currentL1BlockHash.equals(this.l1BlockHash)) { this.log.trace(`No new L1 blocks since last sync at L1 block ${this.l1BlockNumber}`); @@ -141,45 +148,15 @@ export class ArchiverL1Synchronizer implements Traceable { ); } - // Load sync point for blocks and messages defaulting to start block - const { - blocksSynchedTo = this.l1Constants.l1StartBlock, - messagesSynchedTo = { - l1BlockNumber: this.l1Constants.l1StartBlock, - l1BlockHash: this.l1Constants.l1StartBlockHash, - }, - } = await this.store.getSynchPoint(); + // Load sync point for blocks defaulting to start block + const { blocksSynchedTo = this.l1Constants.l1StartBlock } = await this.store.getSynchPoint(); + this.log.debug(`Starting new archiver sync iteration`, { blocksSynchedTo, currentL1BlockData }); - this.log.debug(`Starting new archiver sync iteration`, { - blocksSynchedTo, - messagesSynchedTo, - currentL1BlockNumber, - currentL1BlockHash, - }); + // Sync L1 to L2 messages. We retry this a few times since there are error conditions that reset the sync point, requiring a new iteration. + // Note that we cannot just wait for the l1 synchronizer to loop again, since the synchronizer would report as synced up to the current L1 + // block, when that wouldn't be the case, since L1 to L2 messages would need another iteration. + await retryTimes(() => this.handleL1ToL2Messages(currentL1BlockData), 'Handling L1 to L2 messages', 3, 0.1); - // ********** Ensuring Consistency of data pulled from L1 ********** - - /** - * There are a number of calls in this sync operation to L1 for retrieving - * events and transaction data. There are a couple of things we need to bear in mind - * to ensure that data is read exactly once. - * - * The first is the problem of eventually consistent ETH service providers like Infura. - * Each L1 read operation will query data from the last L1 block that it saw emit its kind of data. - * (so pending L1 to L2 messages will read from the last L1 block that emitted a message and so on) - * This will mean the archiver will lag behind L1 and will only advance when there's L2-relevant activity on the chain. - * - * The second is that in between the various calls to L1, the block number can move meaning some - * of the following calls will return data for blocks that were not present during earlier calls. - * To combat this for the time being we simply ensure that all data retrieval methods only retrieve - * data up to the currentBlockNumber captured at the top of this function. We might want to improve on this - * in future but for the time being it should give us the guarantees that we need - */ - - // ********** Events that are processed per L1 block ********** - await this.handleL1ToL2Messages(messagesSynchedTo, currentL1BlockNumber); - - // ********** Events that are processed per checkpoint ********** if (currentL1BlockNumber > blocksSynchedTo) { // First we retrieve new checkpoints and L2 blocks and store them in the DB. This will also update the // pending chain validation status, proven checkpoint number, and synched L1 block number. @@ -388,64 +365,87 @@ export class ArchiverL1Synchronizer implements Traceable { } @trackSpan('Archiver.handleL1ToL2Messages') - private async handleL1ToL2Messages(messagesSyncPoint: L1BlockId, currentL1BlockNumber: bigint): Promise { - this.log.trace(`Handling L1 to L2 messages from ${messagesSyncPoint.l1BlockNumber} to ${currentL1BlockNumber}.`); - if (currentL1BlockNumber <= messagesSyncPoint.l1BlockNumber) { - return; + private async handleL1ToL2Messages(currentL1Block: L1BlockId): Promise { + // Load the syncpoint, which may have been updated in a previous iteration + const { + messagesSynchedTo = { + l1BlockNumber: this.l1Constants.l1StartBlock, + l1BlockHash: this.l1Constants.l1StartBlockHash, + }, + } = await this.store.getSynchPoint(); + + // Nothing to do if L1 block number has not moved forward + const currentL1BlockNumber = currentL1Block.l1BlockNumber; + if (currentL1BlockNumber <= messagesSynchedTo.l1BlockNumber) { + return true; } - // Load remote and local inbox states. - const localMessagesInserted = await this.store.getTotalL1ToL2MessageCount(); - const localLastMessage = await this.store.getLastL1ToL2Message(); + // Compare local message store state with the remote. If they match, we just advance the match pointer. const remoteMessagesState = await this.inbox.getState({ blockNumber: currentL1BlockNumber }); - await this.store.setInboxTreeInProgress(remoteMessagesState.treeInProgress); + const localLastMessage = await this.store.getLastL1ToL2Message(); + if (await this.localStateMatches(localLastMessage, remoteMessagesState)) { + this.log.trace(`Local L1 to L2 messages are already in sync with remote at L1 block ${currentL1BlockNumber}`); + await this.store.setMessageSyncState(currentL1Block, remoteMessagesState.treeInProgress); + return true; + } - this.log.trace(`Retrieved remote inbox state at L1 block ${currentL1BlockNumber}.`, { - localMessagesInserted, - localLastMessage, - remoteMessagesState, - }); + // If not, then we are out of sync. Most likely there are new messages on the inbox, so we try retrieving them. + // However, it could also be the case that there was an L1 reorg and our syncpoint is no longer valid. + // If that's the case, we'd get an exception out of the message store since the rolling hash of the first message + // we try to insert would not match the one in the db, in which case we rollback to the last common message with L1. + try { + await this.retrieveAndStoreMessages(messagesSynchedTo.l1BlockNumber, currentL1BlockNumber); + } catch (error) { + if (isErrorClass(error, MessageStoreError)) { + this.log.warn( + `Failed to store L1 to L2 messages retrieved from L1: ${error.message}. Rolling back syncpoint to retry.`, + { inboxMessage: error.inboxMessage }, + ); + await this.rollbackL1ToL2Messages(remoteMessagesState.treeInProgress); + return false; + } + throw error; + } - // Compare message count and rolling hash. If they match, no need to retrieve anything. - if ( - remoteMessagesState.totalMessagesInserted === localMessagesInserted && - remoteMessagesState.messagesRollingHash.equals(localLastMessage?.rollingHash ?? Buffer16.ZERO) - ) { - this.log.trace( - `No L1 to L2 messages to query between L1 blocks ${messagesSyncPoint.l1BlockNumber} and ${currentL1BlockNumber}.`, + // Note that, if there are no new messages to insert, but there was an L1 reorg that pruned out last messages, + // we'd notice by comparing our local state with the remote one again, and seeing they don't match even after + // our sync attempt. In this case, we also rollback our syncpoint, and trigger a retry. + const localLastMessageAfterSync = await this.store.getLastL1ToL2Message(); + if (!(await this.localStateMatches(localLastMessageAfterSync, remoteMessagesState))) { + this.log.warn( + `Local L1 to L2 messages state does not match remote after sync attempt. Rolling back syncpoint to retry.`, + { localLastMessageAfterSync, remoteMessagesState }, ); - return; + await this.rollbackL1ToL2Messages(remoteMessagesState.treeInProgress); + return false; } - // Check if our syncpoint is still valid. If not, there was an L1 reorg and we need to re-retrieve messages. - // Note that we need to fetch it from logs and not from inbox state at the syncpoint l1 block number, since it - // could be older than 128 blocks and non-archive nodes cannot resolve it. - if (localLastMessage) { - const remoteLastMessage = await this.retrieveL1ToL2Message(localLastMessage.leaf); - this.log.trace(`Retrieved remote message for local last`, { remoteLastMessage, localLastMessage }); - if (!remoteLastMessage || !remoteLastMessage.rollingHash.equals(localLastMessage.rollingHash)) { - this.log.warn(`Rolling back L1 to L2 messages due to hash mismatch or msg not found.`, { - remoteLastMessage, - messagesSyncPoint, - localLastMessage, - }); + // Advance the syncpoint after a successful sync + await this.store.setMessageSyncState(currentL1Block, remoteMessagesState.treeInProgress); + return true; + } - messagesSyncPoint = await this.rollbackL1ToL2Messages(localLastMessage, messagesSyncPoint); - this.log.debug(`Rolled back L1 to L2 messages to L1 block ${messagesSyncPoint.l1BlockNumber}.`, { - messagesSyncPoint, - }); - } - } + /** Checks if the local rolling hash and message count matches the remote state */ + private async localStateMatches(localLastMessage: InboxMessage | undefined, remoteState: InboxContractState) { + const localMessageCount = await this.store.getTotalL1ToL2MessageCount(); + this.log.trace(`Comparing local and remote inbox state`, { localMessageCount, localLastMessage, remoteState }); + + return ( + remoteState.totalMessagesInserted === localMessageCount && + remoteState.messagesRollingHash.equals(localLastMessage?.rollingHash ?? Buffer16.ZERO) + ); + } - // Retrieve and save messages in batches. Each batch is estimated to acommodate up to L2 'blockBatchSize' blocks, + /** Retrieves L1 to L2 messages from L1 in batches and stores them. */ + private async retrieveAndStoreMessages(fromL1Block: bigint, toL1Block: bigint): Promise { let searchStartBlock: bigint = 0n; - let searchEndBlock: bigint = messagesSyncPoint.l1BlockNumber; + let searchEndBlock: bigint = fromL1Block; let lastMessage: InboxMessage | undefined; let messageCount = 0; do { - [searchStartBlock, searchEndBlock] = this.nextRange(searchEndBlock, currentL1BlockNumber); + [searchStartBlock, searchEndBlock] = this.nextRange(searchEndBlock, toL1Block); 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(); @@ -457,81 +457,65 @@ export class ArchiverL1Synchronizer implements Traceable { lastMessage = msg; messageCount++; } - } while (searchEndBlock < currentL1BlockNumber); + } while (searchEndBlock < toL1Block); - // Log stats for messages retrieved (if any). if (messageCount > 0) { this.log.info( `Retrieved ${messageCount} new L1 to L2 messages up to message with index ${lastMessage?.index} for checkpoint ${lastMessage?.checkpointNumber}`, { lastMessage, messageCount }, ); } - - // Warn if the resulting rolling hash does not match the remote state we had retrieved. - if (lastMessage && !lastMessage.rollingHash.equals(remoteMessagesState.messagesRollingHash)) { - this.log.warn(`Last message retrieved rolling hash does not match remote state.`, { - lastMessage, - remoteMessagesState, - }); - } } - private async retrieveL1ToL2Message(leaf: Fr): Promise { - const currentL1BlockNumber = await this.publicClient.getBlockNumber(); - let searchStartBlock: bigint = 0n; - let searchEndBlock: bigint = this.l1Constants.l1StartBlock - 1n; - - do { - [searchStartBlock, searchEndBlock] = this.nextRange(searchEndBlock, currentL1BlockNumber); - - const message = await retrieveL1ToL2Message(this.inbox, leaf, searchStartBlock, searchEndBlock); - - if (message) { - return message; - } - } while (searchEndBlock < currentL1BlockNumber); - - return undefined; - } - - private async rollbackL1ToL2Messages( - localLastMessage: InboxMessage, - messagesSyncPoint: L1BlockId, - ): Promise { + /** + * Rolls back local L1 to L2 messages to the last common message with L1, and updates the syncpoint to the L1 block of that message. + * If no common message is found, rolls back all messages and sets the syncpoint to the start block. + */ + private async rollbackL1ToL2Messages(remoteTreeInProgress: bigint): Promise { // Slowly go back through our messages until we find the last common message. // We could query the logs in batch as an optimization, but the depth of the reorg should not be deep, and this // is a very rare case, so it's fine to query one log at a time. let commonMsg: undefined | InboxMessage; - this.log.verbose(`Searching most recent common L1 to L2 message at or before index ${localLastMessage.index}`); - for await (const msg of this.store.iterateL1ToL2Messages({ reverse: true, end: localLastMessage.index })) { - const remoteMsg = await this.retrieveL1ToL2Message(msg.leaf); - const logCtx = { remoteMsg, localMsg: msg }; - if (remoteMsg && remoteMsg.rollingHash.equals(msg.rollingHash)) { + let messagesToDelete = 0; + this.log.verbose(`Searching most recent common L1 to L2 message`); + for await (const localMsg of this.store.iterateL1ToL2Messages({ reverse: true })) { + const remoteMsg = await retrieveL1ToL2Message(this.inbox, localMsg); + const logCtx = { remoteMsg, localMsg: localMsg }; + if (remoteMsg && remoteMsg.rollingHash.equals(localMsg.rollingHash)) { this.log.verbose( - `Found most recent common L1 to L2 message at index ${msg.index} on L1 block ${msg.l1BlockNumber}`, + `Found most recent common L1 to L2 message at index ${localMsg.index} on L1 block ${localMsg.l1BlockNumber}`, logCtx, ); commonMsg = remoteMsg; break; } else if (remoteMsg) { - this.log.debug(`Local L1 to L2 message with index ${msg.index} has different rolling hash`, logCtx); + this.log.debug(`Local L1 to L2 message with index ${localMsg.index} has different rolling hash`, logCtx); + messagesToDelete++; } else { - this.log.debug(`Local L1 to L2 message with index ${msg.index} not found on L1`, logCtx); + this.log.debug(`Local L1 to L2 message with index ${localMsg.index} not found on L1`, logCtx); + messagesToDelete++; } } - // Delete everything after the common message we found. - const lastGoodIndex = commonMsg?.index; - this.log.warn(`Deleting all local L1 to L2 messages after index ${lastGoodIndex ?? 'undefined'}`); - await this.store.removeL1ToL2Messages(lastGoodIndex !== undefined ? lastGoodIndex + 1n : 0n); + // Delete everything after the common message we found, if anything needs to be deleted. + // Do not exit early if there are no messages to delete, we still want to update the syncpoint. + if (messagesToDelete > 0) { + const lastGoodIndex = commonMsg?.index; + this.log.warn(`Rolling back all local L1 to L2 messages after index ${lastGoodIndex ?? 'initial'}`); + await this.store.removeL1ToL2Messages(lastGoodIndex !== undefined ? lastGoodIndex + 1n : 0n); + } // Update the syncpoint so the loop below reprocesses the changed messages. We go to the block before // the last common one, so we force reprocessing it, in case new messages were added on that same L1 block // after the last common message. const syncPointL1BlockNumber = commonMsg ? commonMsg.l1BlockNumber - 1n : this.l1Constants.l1StartBlock; const syncPointL1BlockHash = await this.getL1BlockHash(syncPointL1BlockNumber); - messagesSyncPoint = { l1BlockNumber: syncPointL1BlockNumber, l1BlockHash: syncPointL1BlockHash }; - await this.store.setMessageSynchedL1Block(messagesSyncPoint); + const messagesSyncPoint = { l1BlockNumber: syncPointL1BlockNumber, l1BlockHash: syncPointL1BlockHash }; + await this.store.setMessageSyncState(messagesSyncPoint, remoteTreeInProgress); + this.log.verbose(`Updated messages syncpoint to L1 block ${syncPointL1BlockNumber}`, { + ...messagesSyncPoint, + remoteTreeInProgress, + }); return messagesSyncPoint; } diff --git a/yarn-project/archiver/src/store/kv_archiver_store.test.ts b/yarn-project/archiver/src/store/kv_archiver_store.test.ts index d8f5451a989a..003bed9effe4 100644 --- a/yarn-project/archiver/src/store/kv_archiver_store.test.ts +++ b/yarn-project/archiver/src/store/kv_archiver_store.test.ts @@ -1809,23 +1809,13 @@ describe('KVArchiverDataStore', () => { } satisfies ArchiverL1SynchPoint); }); - it('returns the L1 block number that most recently added messages from inbox', async () => { + it('returns the L1 block set via setMessageSyncState', async () => { const l1BlockHash = Buffer32.random(); const l1BlockNumber = 10n; - await store.setMessageSynchedL1Block({ l1BlockNumber: 5n, l1BlockHash: Buffer32.random() }); - await store.addL1ToL2Messages([makeInboxMessage(Buffer16.ZERO, { l1BlockNumber, l1BlockHash })]); - await expect(store.getSynchPoint()).resolves.toEqual({ - blocksSynchedTo: undefined, - messagesSynchedTo: { l1BlockHash, l1BlockNumber }, - } satisfies ArchiverL1SynchPoint); - }); - - it('returns the latest syncpoint if latest message is behind', async () => { - const l1BlockHash = Buffer32.random(); - const l1BlockNumber = 10n; - await store.setMessageSynchedL1Block({ l1BlockNumber, l1BlockHash }); - const msg = makeInboxMessage(Buffer16.ZERO, { l1BlockNumber: 5n, l1BlockHash: Buffer32.random() }); - await store.addL1ToL2Messages([msg]); + await store.setMessageSyncState({ l1BlockNumber, l1BlockHash }, 1n); + await store.addL1ToL2Messages([ + makeInboxMessage(Buffer16.ZERO, { l1BlockNumber: 5n, l1BlockHash: Buffer32.random() }), + ]); await expect(store.getSynchPoint()).resolves.toEqual({ blocksSynchedTo: undefined, messagesSynchedTo: { l1BlockHash, l1BlockNumber }, @@ -2243,7 +2233,7 @@ describe('KVArchiverDataStore', () => { await store.addL1ToL2Messages(msgs); // Set treeInProgress to 7, meaning checkpoints 5 and 6 are sealed, 7+ are not - await store.setInboxTreeInProgress(7n); + await store.setMessageSyncState({ l1BlockNumber: 1n, l1BlockHash: Buffer32.random() }, 7n); // Sealed checkpoint should succeed await expect(store.getL1ToL2Messages(CheckpointNumber(5))).resolves.toEqual([msgs[0].leaf]); @@ -2259,7 +2249,7 @@ describe('KVArchiverDataStore', () => { const msgs = makeInboxMessages(3, { initialCheckpointNumber: CheckpointNumber(10) }); await store.addL1ToL2Messages(msgs); - await store.setInboxTreeInProgress(13n); + await store.setMessageSyncState({ l1BlockNumber: 1n, l1BlockHash: Buffer32.random() }, 13n); await expect(store.getL1ToL2Messages(CheckpointNumber(10))).resolves.toEqual([msgs[0].leaf]); await expect(store.getL1ToL2Messages(CheckpointNumber(11))).resolves.toEqual([msgs[1].leaf]); @@ -2269,7 +2259,7 @@ describe('KVArchiverDataStore', () => { const msgs = makeInboxMessages(2, { initialCheckpointNumber: CheckpointNumber(1) }); await store.addL1ToL2Messages(msgs); - // No setInboxTreeInProgress call — guard should be permissive + // No setMessageSyncState call — guard should be permissive await expect(store.getL1ToL2Messages(CheckpointNumber(1))).resolves.toEqual([msgs[0].leaf]); }); }); diff --git a/yarn-project/archiver/src/store/kv_archiver_store.ts b/yarn-project/archiver/src/store/kv_archiver_store.ts index 6d6904669dc6..fd07ccc684d0 100644 --- a/yarn-project/archiver/src/store/kv_archiver_store.ts +++ b/yarn-project/archiver/src/store/kv_archiver_store.ts @@ -558,13 +558,6 @@ export class KVArchiverDataStore implements ContractDataSource { await this.#blockStore.setSynchedL1BlockNumber(l1BlockNumber); } - /** - * Stores the l1 block that messages have been synched until - */ - async setMessageSynchedL1Block(l1Block: L1BlockId) { - await this.#messageStore.setSynchedL1Block(l1Block); - } - /** * Returns the number of the most recent proven block * @returns The number of the most recent proven block @@ -597,9 +590,9 @@ export class KVArchiverDataStore implements ContractDataSource { return this.#messageStore.rollbackL1ToL2MessagesToCheckpoint(targetCheckpointNumber); } - /** Persists the inbox tree-in-progress checkpoint number from L1 state. */ - public setInboxTreeInProgress(value: bigint): Promise { - return this.#messageStore.setInboxTreeInProgress(value); + /** Atomically updates the message sync state: the L1 sync point and the inbox tree-in-progress marker. */ + public setMessageSyncState(l1Block: L1BlockId, treeInProgress: bigint | undefined): Promise { + return this.#messageStore.setMessageSyncState(l1Block, treeInProgress); } /** Returns an async iterator to all L1 to L2 messages on the range. */ diff --git a/yarn-project/archiver/src/store/message_store.ts b/yarn-project/archiver/src/store/message_store.ts index ab0029800b1b..12b53e935f55 100644 --- a/yarn-project/archiver/src/store/message_store.ts +++ b/yarn-project/archiver/src/store/message_store.ts @@ -161,15 +161,6 @@ export class MessageStore { lastMessage = message; } - // Update the L1 sync point to that of the last message added. - const currentSyncPoint = await this.getSynchedL1Block(); - if (!currentSyncPoint || currentSyncPoint.l1BlockNumber < lastMessage!.l1BlockNumber) { - await this.setSynchedL1Block({ - l1BlockNumber: lastMessage!.l1BlockNumber, - l1BlockHash: lastMessage!.l1BlockHash, - }); - } - // Update total message count with the number of inserted messages. await this.increaseTotalMessageCount(messageCount); }); @@ -194,9 +185,16 @@ export class MessageStore { return this.#inboxTreeInProgress.getAsync(); } - /** Persists the inbox tree-in-progress checkpoint number from L1 state. */ - public async setInboxTreeInProgress(value: bigint): Promise { - await this.#inboxTreeInProgress.set(value); + /** Atomically updates the message sync state: the L1 sync point and the inbox tree-in-progress marker. */ + public setMessageSyncState(l1Block: L1BlockId, treeInProgress: bigint | undefined): Promise { + return this.db.transactionAsync(async () => { + await this.setSynchedL1Block(l1Block); + if (treeInProgress !== undefined) { + await this.#inboxTreeInProgress.set(treeInProgress); + } else { + await this.#inboxTreeInProgress.delete(); + } + }); } public async getL1ToL2Messages(checkpointNumber: CheckpointNumber): Promise { diff --git a/yarn-project/archiver/src/structs/inbox_message.ts b/yarn-project/archiver/src/structs/inbox_message.ts index 9c1b19b78a16..9f981b505e9b 100644 --- a/yarn-project/archiver/src/structs/inbox_message.ts +++ b/yarn-project/archiver/src/structs/inbox_message.ts @@ -8,7 +8,7 @@ export type InboxMessage = { index: bigint; leaf: Fr; checkpointNumber: CheckpointNumber; - l1BlockNumber: bigint; // L1 block number - NOT Aztec L2 + l1BlockNumber: bigint; l1BlockHash: Buffer32; rollingHash: Buffer16; }; diff --git a/yarn-project/archiver/src/test/fake_l1_state.ts b/yarn-project/archiver/src/test/fake_l1_state.ts index e6cac82fd574..1eaa06b026bb 100644 --- a/yarn-project/archiver/src/test/fake_l1_state.ts +++ b/yarn-project/archiver/src/test/fake_l1_state.ts @@ -451,19 +451,33 @@ export class FakeL1State { createMockInboxContract(_publicClient: MockProxy): MockProxy { const mockInbox = mock(); - mockInbox.getState.mockImplementation(() => { + mockInbox.getState.mockImplementation((opts: { blockTag?: string; blockNumber?: bigint } = {}) => { + // Filter messages visible at the given block number (or all if not specified) + const blockNumber = opts.blockNumber ?? this.l1BlockNumber; + const visibleMessages = this.messages.filter(m => m.l1BlockNumber <= blockNumber); + // treeInProgress must be > any sealed checkpoint. On L1, a checkpoint can only be proposed // after its messages are sealed, so treeInProgress > checkpointNumber for all published checkpoints. const maxFromMessages = - this.messages.length > 0 ? Math.max(...this.messages.map(m => Number(m.checkpointNumber))) + 1 : 0; + visibleMessages.length > 0 ? Math.max(...visibleMessages.map(m => Number(m.checkpointNumber))) + 1 : 0; const maxFromCheckpoints = this.checkpoints.length > 0 - ? Math.max(...this.checkpoints.filter(cp => !cp.pruned).map(cp => Number(cp.checkpointNumber))) + 1 + ? Math.max( + ...this.checkpoints + .filter(cp => !cp.pruned && cp.l1BlockNumber <= blockNumber) + .map(cp => Number(cp.checkpointNumber)), + 0, + ) + 1 : 0; const treeInProgress = Math.max(maxFromMessages, maxFromCheckpoints, INITIAL_CHECKPOINT_NUMBER); + + // Compute rolling hash only for visible messages + const rollingHash = + visibleMessages.length > 0 ? visibleMessages[visibleMessages.length - 1].rollingHash : Buffer16.ZERO; + return Promise.resolve({ - messagesRollingHash: this.messagesRollingHash, - totalMessagesInserted: BigInt(this.messages.length), + messagesRollingHash: rollingHash, + totalMessagesInserted: BigInt(visibleMessages.length), treeInProgress: BigInt(treeInProgress), }); }); @@ -473,8 +487,8 @@ export class FakeL1State { Promise.resolve(this.getMessageSentLogs(fromBlock, toBlock)), ); - mockInbox.getMessageSentEventByHash.mockImplementation((hash: string, fromBlock: bigint, toBlock: bigint) => - Promise.resolve(this.getMessageSentLogs(fromBlock, toBlock, hash)), + mockInbox.getMessageSentEventByHash.mockImplementation((msgHash: string, l1BlockHash: string) => + Promise.resolve(this.getMessageSentLogByHash(msgHash, l1BlockHash) as MessageSentLog), ); return mockInbox; @@ -594,6 +608,26 @@ export class FakeL1State { })); } + private getMessageSentLogByHash(msgHash: string, l1BlockHash: string): MessageSentLog | undefined { + const msg = this.messages.find( + msg => msg.leaf.toString() === msgHash && Buffer32.fromBigInt(msg.l1BlockNumber).toString() === l1BlockHash, + ); + if (!msg) { + return undefined; + } + return { + l1BlockNumber: msg.l1BlockNumber, + l1BlockHash: Buffer32.fromBigInt(msg.l1BlockNumber), + l1TransactionHash: `0x${msg.l1BlockNumber.toString(16)}` as `0x${string}`, + args: { + checkpointNumber: msg.checkpointNumber, + index: msg.index, + leaf: msg.leaf, + rollingHash: msg.rollingHash, + }, + }; + } + private async makeRollupTx( checkpoint: Checkpoint, signers: Secp256k1Signer[], diff --git a/yarn-project/ethereum/src/contracts/inbox.ts b/yarn-project/ethereum/src/contracts/inbox.ts index 614ca96163ff..cae66f0878f2 100644 --- a/yarn-project/ethereum/src/contracts/inbox.ts +++ b/yarn-project/ethereum/src/contracts/inbox.ts @@ -82,10 +82,10 @@ export class InboxContract { .map(log => this.mapMessageSentLog(log)); } - /** Fetches MessageSent events for a specific message hash within the given block range. */ - async getMessageSentEventByHash(hash: Hex, fromBlock: bigint, toBlock: bigint): Promise { - const logs = await this.inbox.getEvents.MessageSent({ hash }, { fromBlock, toBlock }); - return logs.map(log => this.mapMessageSentLog(log)); + /** Fetches MessageSent events for a specific message hash at a specific block. */ + async getMessageSentEventByHash(msgHash: Hex, l1BlockHash: Hex): Promise { + const [log] = await this.inbox.getEvents.MessageSent({ hash: msgHash }, { blockHash: l1BlockHash }); + return log && this.mapMessageSentLog(log); } private mapMessageSentLog(log: { diff --git a/yarn-project/foundation/src/retry/index.ts b/yarn-project/foundation/src/retry/index.ts index 38a8b9348246..cec8b8a8d3d0 100644 --- a/yarn-project/foundation/src/retry/index.ts +++ b/yarn-project/foundation/src/retry/index.ts @@ -104,6 +104,36 @@ export async function retryUntil( } } +/** + * Retry an asynchronous function until it returns a truthy value or the maximum number of retries is exceeded. + * The function is retried periodically with a fixed interval between attempts. + * + * @param fn - The asynchronous function to be retried, which should return a truthy value upon success or undefined otherwise. + * @param name - The optional name of the operation, used for generating error messages. + * @param maxRetries - The maximum number of retry attempts before throwing an error. + * @param retryInterval - The optional interval, in seconds, between retry attempts. Defaults to 1 second. + * @returns A Promise that resolves with the successful (truthy) result of the provided function, or rejects if retries are exhausted. + */ +export async function retryTimes( + fn: () => (T | undefined) | Promise, + name = '', + maxRetries: number, + retryInterval = 1, +) { + for (let attempt = 0; attempt <= maxRetries; attempt++) { + const result = await fn(); + if (result) { + return result; + } + + if (attempt < maxRetries) { + await sleep(retryInterval * 1000); + } + } + + throw new Error(name ? `Retries exhausted awaiting ${name}` : 'Retries exhausted'); +} + /** * Convenience wrapper around retryUntil with fast polling for tests. * Uses 10s timeout and 100ms polling interval by default.