From c08b2d121f0bf67430b96508d207f90e3a7ee026 Mon Sep 17 00:00:00 2001 From: Alex Gherghisan Date: Thu, 15 Oct 2026 16:28:08 +0000 Subject: [PATCH] feat: stop history pruning after certain checkpoint --- yarn-project/aztec-node/src/factory.ts | 2 +- yarn-project/foundation/src/config/env_var.ts | 1 + .../src/actions/snapshot-sync.test.ts | 221 ++++++++++++++ .../node-lib/src/actions/snapshot-sync.ts | 275 ++++++++++++++---- yarn-project/stdlib/src/snapshots/download.ts | 46 +-- yarn-project/stdlib/src/snapshots/index.ts | 8 +- yarn-project/world-state/README.md | 20 ++ .../src/native/native_world_state.ts | 1 + .../world-state/src/synchronizer/config.ts | 10 + .../src/synchronizer/history.test.ts | 118 ++++++++ .../world-state/src/synchronizer/history.ts | 55 ++++ .../world-state/src/synchronizer/index.ts | 1 + .../server_world_state_synchronizer.ts | 31 +- .../world-state/src/test/integration.test.ts | 91 +++++- 14 files changed, 796 insertions(+), 84 deletions(-) create mode 100644 yarn-project/node-lib/src/actions/snapshot-sync.test.ts create mode 100644 yarn-project/world-state/src/synchronizer/history.test.ts create mode 100644 yarn-project/world-state/src/synchronizer/history.ts diff --git a/yarn-project/aztec-node/src/factory.ts b/yarn-project/aztec-node/src/factory.ts index 1830d6bf15e..7c168bcf721 100644 --- a/yarn-project/aztec-node/src/factory.ts +++ b/yarn-project/aztec-node/src/factory.ts @@ -187,7 +187,7 @@ export async function createAztecNodeService( const blobClient = await createBlobClientWithFileStores(config, log.createChild('blob-client')); // attempt snapshot sync if possible - await trySnapshotSync(config, log); + await trySnapshotSync(config, log, options.genesis); const epochCache = await EpochCache.create(config.rollupAddress, config, { dateProvider }); diff --git a/yarn-project/foundation/src/config/env_var.ts b/yarn-project/foundation/src/config/env_var.ts index fdf5f99b720..767c600b717 100644 --- a/yarn-project/foundation/src/config/env_var.ts +++ b/yarn-project/foundation/src/config/env_var.ts @@ -328,6 +328,7 @@ export type EnvVar = | 'WS_DATA_DIRECTORY' | 'WS_NUM_HISTORIC_CHECKPOINTS' | 'WS_NUM_HISTORIC_BLOCKS' + | 'WS_PRUNE_BEFORE_CHECKPOINT' | 'ETHEREUM_SLOT_DURATION' | 'AZTEC_SLOT_DURATION' | 'AZTEC_EPOCH_DURATION' diff --git a/yarn-project/node-lib/src/actions/snapshot-sync.test.ts b/yarn-project/node-lib/src/actions/snapshot-sync.test.ts new file mode 100644 index 00000000000..ffcb229dcc9 --- /dev/null +++ b/yarn-project/node-lib/src/actions/snapshot-sync.test.ts @@ -0,0 +1,221 @@ +import { ARCHIVER_DB_VERSION, createArchiverStore } from '@aztec/archiver'; +import { makePublishedCheckpoint } from '@aztec/archiver/test'; +import { type Anvil, EthCheatCodes, startAnvil } from '@aztec/ethereum/test'; +import { BlockNumber, CheckpointNumber } from '@aztec/foundation/branded-types'; +import { fromEntries } from '@aztec/foundation/collection'; +import { EthAddress } from '@aztec/foundation/eth-address'; +import { createLogger } from '@aztec/foundation/log'; +import { DateProvider } from '@aztec/foundation/timer'; +import { GENESIS_BLOCK_HEADER_HASH } from '@aztec/stdlib/block'; +import { createReadOnlyFileStore } from '@aztec/stdlib/file-store'; +import { type SnapshotMetadata, getBasePath } from '@aztec/stdlib/snapshots'; +import { NATIVE_WORLD_STATE_DBS, WORLD_STATE_DB_VERSION, createWorldState } from '@aztec/world-state'; +import { mockCheckpoint } from '@aztec/world-state/test'; + +import { mkdir, mkdtemp, readFile, rm, writeFile } from 'fs/promises'; +import { tmpdir } from 'os'; +import { join } from 'path'; + +import { snapshotSync, trySnapshotSync } from './snapshot-sync.js'; + +describe('snapshot bootstrap history', () => { + let directory: string; + let config: Parameters[0]; + let anvil: Anvil | undefined; + let rpcUrl: string; + const log = createLogger('node-lib:test:snapshot-history'); + + const summary = (oldest: number, latest: number) => ({ + oldestHistoricalBlock: oldest, + unfinalizedBlockNumber: latest, + finalizedBlockNumber: latest, + treesAreSynched: true, + }); + + beforeAll(async () => { + ({ anvil, rpcUrl } = await startAnvil({ port: 0, chainId: 1 })); + await new EthCheatCodes([rpcUrl], new DateProvider()).mine(10000); + }); + + afterAll(() => anvil?.stop()); + + beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), 'snapshot-history-')); + config = { + dataDirectory: join(directory, 'node'), + dataStoreMapSizeKb: 1024 * 1024, + rollupAddress: EthAddress.random(), + rollupVersion: 1, + l1ChainId: 1, + l1RpcUrls: [rpcUrl], + aztecEpochDuration: 32, + syncMode: 'snapshot', + snapshotsUrls: [`file://${directory}/snapshots`], + worldStateBlockCheckIntervalMS: 100, + worldStateCheckpointHistory: 64, + worldStatePruneBeforeCheckpoint: CheckpointNumber(3), + }; + }); + + afterEach(() => rm(directory, { recursive: true, force: true })); + + const writeSnapshot = async (block: number, oldest = 1): Promise => { + const sourceConfig = { + ...config, + dataDirectory: join(directory, `source-${block}-${oldest}`), + worldStateDataDirectory: undefined, + }; + const archive = await createArchiverStore(sourceConfig, GENESIS_BLOCK_HEADER_HASH); + const worldState = await createWorldState(sourceConfig); + try { + const fork = await worldState.fork(BlockNumber.ZERO); + const checkpoints = []; + try { + for (let start = 1; start <= block; start += 3) { + const { checkpoint, messages } = await mockCheckpoint(CheckpointNumber(Math.ceil(start / 3)), fork, { + startBlockNumber: BlockNumber(start), + numBlocks: Math.min(3, block - start + 1), + numTxsPerBlock: 1, + maxEffects: 1, + }); + checkpoints.push({ checkpoint, messages }); + } + } finally { + await fork.close(); + } + for (const { checkpoint, messages } of checkpoints) { + await archive.blocks.addCheckpoints([makePublishedCheckpoint(checkpoint, 1000 + checkpoint.blocks[0].number)]); + for (const [index, entry] of checkpoint.blocks.entries()) { + await worldState.handleL2BlockAndMessages(entry, index === 0 ? messages : []); + } + } + await worldState.setFinalized(BlockNumber(block)); + if (oldest > 1) { + await worldState.removeHistoricalBlocks(BlockNumber(oldest)); + } + return { + l1BlockNumber: 1000 + block, + l2BlockNumber: block, + l2BlockHash: (await archive.blocks.getBlockData({ number: BlockNumber(block) }))!.blockHash.toString(), + timestamp: block, + schemaVersions: { archiver: ARCHIVER_DB_VERSION, worldState: WORLD_STATE_DB_VERSION }, + dataUrls: { + archiver: `file://${join(sourceConfig.dataDirectory, 'archiver', 'data.mdb')}`, + ...fromEntries( + NATIVE_WORLD_STATE_DBS.map( + ([key, tree]) => + [key, `file://${join(sourceConfig.dataDirectory, 'world_state', tree, 'data.mdb')}`] as const, + ), + ), + }, + }; + } finally { + await archive.db.close(); + await worldState.close(); + } + }; + + const writeIndex = async (snapshots: SnapshotMetadata[]) => { + const metadata = { l1ChainId: 1, rollupVersion: 1, rollupAddress: config.rollupAddress }; + const path = join(directory, 'snapshots', getBasePath(metadata)); + await mkdir(path, { recursive: true }); + await writeFile(join(path, 'index.json'), JSON.stringify({ ...metadata, snapshots })); + }; + + const installSnapshot = async (snapshot: SnapshotMetadata) => { + const fileStore = await createReadOnlyFileStore(`file://${directory}`); + await snapshotSync(snapshot, log, { + dataDirectory: config.dataDirectory!, + rollupAddress: config.rollupAddress, + fileStore, + }); + }; + + const readInstalledState = async () => { + const worldState = await createWorldState(config); + try { + return await worldState.getStatusSummary(); + } finally { + await worldState.close(); + } + }; + + it('resolves legacy snapshot block numbers to the newest snapshot in the pinned checkpoint', async () => { + await writeIndex([await writeSnapshot(5), await writeSnapshot(12), await writeSnapshot(8)]); + await expect(trySnapshotSync(config, log)).resolves.toBe(true); + expect(await readInstalledState()).toEqual(summary(1, 8)); + }); + + it('restores world state into its configured data directory', async () => { + config.worldStateDataDirectory = join(directory, 'state-volume'); + await writeIndex([await writeSnapshot(8)]); + await expect(trySnapshotSync(config, log)).resolves.toBe(true); + expect(await readInstalledState()).toEqual(summary(1, 8)); + }); + + it('falls back through missing files and incompatible retained history in the same index', async () => { + const missing = await writeSnapshot(9); + await rm(new URL(missing.dataUrls['public-data-tree'])); + await writeIndex([missing, await writeSnapshot(8, 8), await writeSnapshot(5)]); + await expect(trySnapshotSync(config, log)).resolves.toBe(true); + expect(await readInstalledState()).toEqual(summary(1, 5)); + }); + + it('leaves genesis replay to the synchronizer when no snapshot is old enough', async () => { + await writeIndex([await writeSnapshot(12)]); + await expect(trySnapshotSync(config, log)).resolves.toBe(false); + }); + + it.each([0, 3])('rejects missing history with checkpoint %i pinned even in force-snapshot mode', async checkpoint => { + config.worldStatePruneBeforeCheckpoint = CheckpointNumber(checkpoint); + await installSnapshot(await writeSnapshot(12, 11)); + await writeIndex([await writeSnapshot(5)]); + await expect(trySnapshotSync({ ...config, syncMode: 'force-snapshot' }, log)).rejects.toThrow( + checkpoint === 0 ? 'all checkpoint history is required' : 'history is required from the start of checkpoint 3', + ); + expect(await readInstalledState()).toEqual(summary(11, 12)); + }); + + it('preserves compatible existing history without replacing it with an older snapshot', async () => { + await installSnapshot(await writeSnapshot(12, 5)); + await writeIndex([await writeSnapshot(8)]); + await expect(trySnapshotSync({ ...config, syncMode: 'force-snapshot' }, log)).resolves.toBe(false); + expect(await readInstalledState()).toEqual(summary(5, 12)); + }); + + it.each([128, 0])('accepts a pruned snapshot without a pin when history is %i', async history => { + config.worldStatePruneBeforeCheckpoint = undefined; + config.worldStateCheckpointHistory = history; + await writeIndex([await writeSnapshot(12, 11)]); + await expect(trySnapshotSync(config, log)).resolves.toBe(true); + expect(await readInstalledState()).toEqual(summary(11, 12)); + }); + + it.each([128, 0])('repairs a missing archiver without a pin when history is %i', async history => { + await installSnapshot(await writeSnapshot(12, 11)); + await rm(join(config.dataDirectory!, 'archiver'), { recursive: true }); + config.worldStatePruneBeforeCheckpoint = undefined; + config.worldStateCheckpointHistory = history; + await writeIndex([await writeSnapshot(9, 8)]); + await expect(trySnapshotSync({ ...config, syncMode: 'full' }, log)).resolves.toBe(false); + await expect(trySnapshotSync(config, log)).resolves.toBe(true); + expect(await readInstalledState()).toEqual(summary(8, 9)); + }); + + it('validates a staged snapshot before replacing installed databases', async () => { + const path = join(config.dataDirectory!, 'archiver', 'data.mdb'); + await mkdir(join(config.dataDirectory!, 'archiver'), { recursive: true }); + await writeFile(path, 'existing database'); + const snapshot = await writeSnapshot(8); + const fileStore = await createReadOnlyFileStore(`file://${directory}`); + await expect( + snapshotSync(snapshot, log, { + dataDirectory: config.dataDirectory!, + rollupAddress: config.rollupAddress, + fileStore, + validate: () => Promise.reject(new Error('missing required history')), + }), + ).rejects.toThrow('missing required history'); + expect(await readFile(path, 'utf8')).toBe('existing database'); + }); +}); diff --git a/yarn-project/node-lib/src/actions/snapshot-sync.ts b/yarn-project/node-lib/src/actions/snapshot-sync.ts index 043ec102dcb..9ecaf4d3cd0 100644 --- a/yarn-project/node-lib/src/actions/snapshot-sync.ts +++ b/yarn-project/node-lib/src/actions/snapshot-sync.ts @@ -8,6 +8,7 @@ import { import { INITIAL_L2_BLOCK_NUM } from '@aztec/constants'; import { type EthereumClientConfig, getPublicClient } from '@aztec/ethereum/client'; import type { L1ContractsConfig } from '@aztec/ethereum/config'; +import { BlockNumber } from '@aztec/foundation/branded-types'; import type { EthAddress } from '@aztec/foundation/eth-address'; import { tryRmDir } from '@aztec/foundation/fs'; import type { Logger } from '@aztec/foundation/log'; @@ -18,15 +19,25 @@ import { DatabaseVersionManager } from '@aztec/stdlib/database-version/manager'; import { type ReadOnlyFileStore, createReadOnlyFileStore } from '@aztec/stdlib/file-store'; import type { DataStoreConfig } from '@aztec/stdlib/kv-store'; import { + SnapshotDataKeys, type SnapshotMetadata, type SnapshotsIndexMetadata, downloadSnapshot, - getLatestSnapshotMetadata, + getSnapshotIndex, makeSnapshotPaths, } from '@aztec/stdlib/snapshots'; -import { NATIVE_WORLD_STATE_DBS, WORLD_STATE_DB_VERSION, WORLD_STATE_DIR } from '@aztec/world-state'; - -import { mkdir, mkdtemp, rename } from 'fs/promises'; +import type { GenesisData } from '@aztec/stdlib/world-state'; +import { + NATIVE_WORLD_STATE_DBS, + WORLD_STATE_DB_VERSION, + WORLD_STATE_DIR, + type WorldStateConfig, + createWorldState, + getWorldStateHistoryStartCheckpoint, + validateWorldStateHistory, +} from '@aztec/world-state'; + +import { copyFile, mkdir, mkdtemp, rename, stat, unlink } from 'fs/promises'; import { join } from 'path'; import type { SharedNodeConfig } from '../config/index.js'; @@ -38,6 +49,7 @@ type SnapshotSyncConfig = Pick & Pick & Pick & Pick & + WorldStateConfig & DataStoreConfig & Required> & EthereumClientConfig & { @@ -46,11 +58,23 @@ type SnapshotSyncConfig = Pick & }; /** - * Connects to a remote snapshot index and downloads the latest snapshot if the local archiver is behind. - * Behaviour depends on syncing mode. + * Validates local history and restores the newest compatible snapshot when bootstrap is needed. + * Fixed history boundaries are resolved against archived block and checkpoint metadata. */ -export async function trySnapshotSync(config: SnapshotSyncConfig, log: Logger) { +export async function trySnapshotSync(config: SnapshotSyncConfig, log: Logger, genesis?: GenesisData) { const { syncMode, snapshotsUrls, dataDirectory, l1ChainId, rollupVersion, rollupAddress } = config; + const historyStartCheckpoint = config.worldStatePruneBeforeCheckpoint; + let maxSnapshotBlock = historyStartCheckpoint === 0 ? BlockNumber.ZERO : undefined; + const worldStateDirectory = config.worldStateDataDirectory ?? dataDirectory; + if ( + historyStartCheckpoint !== undefined && + worldStateDirectory && + (await hasWorldStateDatabase(worldStateDirectory)) + ) { + await validateLocalWorldStateHistory(config, log, genesis); + log.info('Using existing world-state database with compatible retained history', { historyStartCheckpoint }); + return false; + } if (syncMode === 'full') { log.debug('Snapshot sync is disabled. Running full sync.', { syncMode: syncMode }); return false; @@ -84,7 +108,12 @@ export async function trySnapshotSync(config: SnapshotSyncConfig, log: Logger) { } const minL1BlocksToTriggerReplace = config.minL1BlocksToTriggerReplace ?? MIN_L1_BLOCKS_TO_TRIGGER_REPLACE; - if (syncMode === 'snapshot' && archiverL2BlockNumber !== undefined && archiverL2BlockNumber >= INITIAL_L2_BLOCK_NUM) { + if ( + historyStartCheckpoint === undefined && + syncMode === 'snapshot' && + archiverL2BlockNumber !== undefined && + archiverL2BlockNumber >= INITIAL_L2_BLOCK_NUM + ) { log.verbose( `Skipping non-forced snapshot sync as archiver is already synced to L2 block ${archiverL2BlockNumber}.`, ); @@ -93,6 +122,7 @@ export async function trySnapshotSync(config: SnapshotSyncConfig, log: Logger) { const currentL1BlockNumber = await getPublicClient(config).getBlockNumber(); if ( + historyStartCheckpoint === undefined && archiverL1BlockNumber && currentL1BlockNumber >= archiverL1BlockNumber && currentL1BlockNumber - archiverL1BlockNumber < minL1BlocksToTriggerReplace @@ -112,7 +142,7 @@ export async function trySnapshotSync(config: SnapshotSyncConfig, log: Logger) { rollupAddress, }; - // Fetch latest snapshot from each URL + // Index entries may outlive their database files, so retain older candidates for download fallback. type SnapshotCandidate = { snapshot: SnapshotMetadata; url: string; fileStore: ReadOnlyFileStore }; const snapshotCandidates: SnapshotCandidate[] = []; @@ -125,57 +155,67 @@ export async function trySnapshotSync(config: SnapshotSyncConfig, log: Logger) { continue; } - let snapshot: SnapshotMetadata | undefined; + let snapshots: SnapshotMetadata[]; try { - snapshot = await getLatestSnapshotMetadata(indexMetadata, fileStore); + snapshots = (await getSnapshotIndex(indexMetadata, fileStore))?.snapshots ?? []; } catch (err) { - log.error(`Failed to get latest snapshot metadata from ${snapshotsUrl}. Skipping this URL.`, err, { + log.error(`Failed to read snapshot index from ${snapshotsUrl}. Skipping this URL.`, err, { ...indexMetadata, snapshotsUrl, }); continue; } - if (!snapshot) { + if (snapshots.length === 0) { log.verbose(`No snapshot found at ${snapshotsUrl}. Skipping this URL.`, { ...indexMetadata, snapshotsUrl }); continue; } - if (snapshot.schemaVersions.archiver !== ARCHIVER_DB_VERSION) { - log.warn( - `Skipping snapshot from ${snapshotsUrl} as it has schema version ${snapshot.schemaVersions.archiver} but expected ${ARCHIVER_DB_VERSION}.`, - snapshot, - ); - continue; - } - - if (snapshot.schemaVersions.worldState !== WORLD_STATE_DB_VERSION) { - log.warn( - `Skipping snapshot from ${snapshotsUrl} as it has world state schema version ${snapshot.schemaVersions.worldState} but we expected ${WORLD_STATE_DB_VERSION}.`, - snapshot, - ); - continue; - } - - if (archiverL1BlockNumber && snapshot.l1BlockNumber < archiverL1BlockNumber) { - log.verbose( - `Skipping snapshot from ${snapshotsUrl} since local archiver is at L1 block ${archiverL1BlockNumber} which is further than snapshot at ${snapshot.l1BlockNumber}`, - { snapshot, archiverL1BlockNumber, snapshotsUrl }, - ); - continue; - } - - if (archiverL1BlockNumber && snapshot.l1BlockNumber - Number(archiverL1BlockNumber) < minL1BlocksToTriggerReplace) { - log.verbose( - `Skipping snapshot from ${snapshotsUrl} as archiver is less than ${ - snapshot.l1BlockNumber - Number(archiverL1BlockNumber) - } L1 blocks behind this snapshot.`, - { snapshot, archiverL1BlockNumber, snapshotsUrl }, - ); - continue; + for (const snapshot of snapshots) { + if (snapshot.schemaVersions.archiver !== ARCHIVER_DB_VERSION) { + log.warn( + `Skipping snapshot from ${snapshotsUrl} as it has schema version ${snapshot.schemaVersions.archiver} but expected ${ARCHIVER_DB_VERSION}.`, + snapshot, + ); + continue; + } + + if (snapshot.schemaVersions.worldState !== WORLD_STATE_DB_VERSION) { + log.warn( + `Skipping snapshot from ${snapshotsUrl} as it has world state schema version ${snapshot.schemaVersions.worldState} but we expected ${WORLD_STATE_DB_VERSION}.`, + snapshot, + ); + continue; + } + + if ( + historyStartCheckpoint === undefined && + archiverL1BlockNumber && + snapshot.l1BlockNumber < archiverL1BlockNumber + ) { + log.verbose( + `Skipping snapshot from ${snapshotsUrl} since local archiver is at L1 block ${archiverL1BlockNumber} which is further than snapshot at ${snapshot.l1BlockNumber}`, + { snapshot, archiverL1BlockNumber, snapshotsUrl }, + ); + continue; + } + + if ( + historyStartCheckpoint === undefined && + archiverL1BlockNumber && + snapshot.l1BlockNumber - Number(archiverL1BlockNumber) < minL1BlocksToTriggerReplace + ) { + log.verbose( + `Skipping snapshot from ${snapshotsUrl} as archiver is less than ${ + snapshot.l1BlockNumber - Number(archiverL1BlockNumber) + } L1 blocks behind this snapshot.`, + { snapshot, archiverL1BlockNumber, snapshotsUrl }, + ); + continue; + } + + snapshotCandidates.push({ snapshot, url: snapshotsUrl, fileStore }); } - - snapshotCandidates.push({ snapshot, url: snapshotsUrl, fileStore }); } if (snapshotCandidates.length === 0) { @@ -183,11 +223,17 @@ export async function trySnapshotSync(config: SnapshotSyncConfig, log: Logger) { return false; } - // Sort candidates by L1 block number (highest first) - snapshotCandidates.sort((a, b) => b.snapshot.l1BlockNumber - a.snapshot.l1BlockNumber); + snapshotCandidates.sort((a, b) => + historyStartCheckpoint === undefined + ? b.snapshot.l1BlockNumber - a.snapshot.l1BlockNumber + : b.snapshot.l2BlockNumber - a.snapshot.l2BlockNumber || b.snapshot.l1BlockNumber - a.snapshot.l1BlockNumber, + ); // Try each candidate in order until one succeeds for (const { snapshot, url, fileStore } of snapshotCandidates) { + if (maxSnapshotBlock !== undefined && snapshot.l2BlockNumber > maxSnapshotBlock) { + continue; + } const { l1BlockNumber, l2BlockNumber } = snapshot; log.info(`Attempting to sync from snapshot at L1 block ${l1BlockNumber} L2 block ${l2BlockNumber}`, { snapshot, @@ -195,10 +241,55 @@ export async function trySnapshotSync(config: SnapshotSyncConfig, log: Logger) { }); try { + const available = await Promise.all(Object.values(snapshot.dataUrls).map(path => fileStore.exists(path))); + if (!available.every(Boolean)) { + log.warn('Skipping snapshot with missing database files', { l1BlockNumber, l2BlockNumber, snapshotsUrl: url }); + continue; + } await snapshotSync(snapshot, log, { dataDirectory: config.dataDirectory!, rollupAddress: config.rollupAddress, fileStore, + worldStateDataDirectory: config.worldStateDataDirectory, + validateArchiver: + historyStartCheckpoint === undefined || historyStartCheckpoint === 0 + ? undefined + : async stagingDirectory => { + const stores = await createArchiverStore( + { ...config, dataDirectory: stagingDirectory }, + GENESIS_BLOCK_HEADER_HASH, + ); + try { + const latestBlockNumber = await stores.blocks.getLatestL2BlockNumber(); + if (latestBlockNumber === undefined) { + throw new Error('Snapshot archiver has no blocks'); + } + const latestBlock = await stores.blocks.getBlockData({ number: latestBlockNumber }); + if (!latestBlock) { + throw new Error(`Snapshot archiver is missing block ${latestBlockNumber}`); + } + if (latestBlock.checkpointNumber > historyStartCheckpoint) { + const checkpoint = await stores.blocks.getCheckpointData(historyStartCheckpoint); + if (checkpoint) { + maxSnapshotBlock = BlockNumber(checkpoint.startBlock + checkpoint.blockCount - 1); + } + throw new Error( + `Snapshot is at checkpoint ${latestBlock.checkpointNumber}, after requested checkpoint ${historyStartCheckpoint}`, + ); + } + } finally { + await stores.db.close(); + } + }, + validate: + historyStartCheckpoint === undefined + ? undefined + : stagingDirectory => + validateLocalWorldStateHistory( + { ...config, dataDirectory: stagingDirectory, worldStateDataDirectory: stagingDirectory }, + log, + genesis, + ), }); log.info(`Snapshot synced to L1 block ${l1BlockNumber} L2 block ${l2BlockNumber}`, { snapshot, @@ -224,7 +315,14 @@ export async function trySnapshotSync(config: SnapshotSyncConfig, log: Logger) { export async function snapshotSync( snapshot: Pick, log: Logger, - config: { dataDirectory: string; rollupAddress: EthAddress; fileStore: ReadOnlyFileStore }, + config: { + dataDirectory: string; + rollupAddress: EthAddress; + fileStore: ReadOnlyFileStore; + worldStateDataDirectory?: string; + validateArchiver?: (stagingDirectory: string) => Promise; + validate?: (stagingDirectory: string) => Promise; + }, ) { const { dataDirectory, rollupAddress, fileStore } = config; if (!dataDirectory) { @@ -238,8 +336,26 @@ export async function snapshotSync( await mkdir(dataDirectory, { recursive: true }); downloadDir = await mkdtemp(join(dataDirectory, 'download-')); const downloadPaths = makeSnapshotPaths(downloadDir); + await prepareTarget(join(downloadDir, ARCHIVER_STORE_NAME), ARCHIVER_DB_VERSION, rollupAddress); + await prepareTarget(join(downloadDir, WORLD_STATE_DIR), WORLD_STATE_DB_VERSION, rollupAddress); + downloadPaths.archiver = join(downloadDir, ARCHIVER_STORE_NAME, 'data.mdb'); + for (const [name, dir] of NATIVE_WORLD_STATE_DBS) { + downloadPaths[name] = join(downloadDir, WORLD_STATE_DIR, dir, 'data.mdb'); + } log.info(`Downloading snapshot to ${downloadDir}`, { snapshot, downloadPaths }); - await downloadSnapshot(snapshot, downloadPaths, fileStore); + if (config.validateArchiver) { + await downloadSnapshot(snapshot, downloadPaths, fileStore, ['archiver']); + await config.validateArchiver(downloadDir); + await downloadSnapshot( + snapshot, + downloadPaths, + fileStore, + SnapshotDataKeys.filter(key => key !== 'archiver'), + ); + } else { + await downloadSnapshot(snapshot, downloadPaths, fileStore); + } + await config.validate?.(downloadDir); log.info(`Snapshot downloaded at ${downloadDir}`, { snapshot, downloadPaths }); // If download was successful, clear lock and version, and move download there @@ -253,12 +369,12 @@ export async function snapshotSync( }); // Same for the world state dbs, only that we do not close them, since we assume they are not yet in use - const worldStateBasePath = join(dataDirectory, WORLD_STATE_DIR); + const worldStateBasePath = join(config.worldStateDataDirectory ?? dataDirectory, WORLD_STATE_DIR); await prepareTarget(worldStateBasePath, WORLD_STATE_DB_VERSION, rollupAddress); for (const [name, dir] of NATIVE_WORLD_STATE_DBS) { const path = join(worldStateBasePath, dir); await mkdir(path, { recursive: true }); - await rename(downloadPaths[name], join(path, 'data.mdb')); + await moveSnapshotFile(downloadPaths[name], join(path, 'data.mdb')); log.info(`World state database ${name} set up from snapshot`, { path, dbVersion: WORLD_STATE_DB_VERSION, @@ -277,6 +393,59 @@ export async function snapshotSync( } } +async function moveSnapshotFile(source: string, destination: string) { + try { + await rename(source, destination); + } catch (err) { + if (!err || typeof err !== 'object' || !('code' in err) || err.code !== 'EXDEV') { + throw err; + } + await copyFile(source, destination); + await unlink(source); + } +} + +async function hasWorldStateDatabase(dataDirectory: string): Promise { + const sizes = await Promise.all( + NATIVE_WORLD_STATE_DBS.map(async ([, directory]) => { + try { + return (await stat(join(dataDirectory, WORLD_STATE_DIR, directory, 'data.mdb'))).size; + } catch (err) { + if (err && typeof err === 'object' && 'code' in err && err.code === 'ENOENT') { + return 0; + } + throw err; + } + }), + ); + return sizes.some(size => size > 0); +} + +async function validateLocalWorldStateHistory(config: SnapshotSyncConfig, log: Logger, genesis?: GenesisData) { + const worldState = await createWorldState(config, genesis); + try { + const summary = await worldState.getStatusSummary(); + if (getWorldStateHistoryStartCheckpoint(config) !== 0 && summary.oldestHistoricalBlock > 1) { + const archiverStore = await createArchiverStore(config, GENESIS_BLOCK_HEADER_HASH); + try { + await validateWorldStateHistory(config, summary, archiverStore.blocks); + } finally { + await archiverStore.db.close(); + } + } else { + await validateWorldStateHistory(config, summary); + } + log.info('Validated retained world-state history', { + oldestHistoricBlock: summary.oldestHistoricalBlock, + latestBlock: summary.unfinalizedBlockNumber, + historyStartCheckpoint: getWorldStateHistoryStartCheckpoint(config), + checkpointHistory: config.worldStateCheckpointHistory, + }); + } finally { + await worldState.close(); + } +} + /** Deletes target dir and writes the new version file. */ async function prepareTarget(target: string, schemaVersion: number, rollupAddress: EthAddress) { const noOpen = () => Promise.resolve(undefined); diff --git a/yarn-project/stdlib/src/snapshots/download.ts b/yarn-project/stdlib/src/snapshots/download.ts index 89d94b36041..4f04232b503 100644 --- a/yarn-project/stdlib/src/snapshots/download.ts +++ b/yarn-project/stdlib/src/snapshots/download.ts @@ -95,29 +95,37 @@ export async function downloadSnapshot( snapshot: Pick, localPaths: Record, store: ReadOnlyFileStore, + keys: readonly SnapshotDataKeys[] = SnapshotDataKeys, ): Promise { - await Promise.all( - getEntries(localPaths).map(async ([key, path]) => { - await fs.mkdir(pathMod.dirname(path), { recursive: true }); + const results = await Promise.allSettled( + getEntries(localPaths) + .filter(([key]) => keys.includes(key)) + .map(async ([key, path]) => { + await fs.mkdir(pathMod.dirname(path), { recursive: true }); - const tmpPath = `${path}.download`; - try { - const url = snapshot.dataUrls[key]; - await store.download(url, tmpPath); + const tmpPath = `${path}.download`; + try { + const url = snapshot.dataUrls[key]; + await store.download(url, tmpPath); - const isGzip = await detectGzip(tmpPath); + const isGzip = await detectGzip(tmpPath); - const read = createReadStream(tmpPath); - const write = createWriteStream(path); - if (isGzip) { - const gunzip = createGunzip(); - await pipeline(read, gunzip, write); - } else { - await pipeline(read, write); + const read = createReadStream(tmpPath); + const write = createWriteStream(path); + if (isGzip) { + const gunzip = createGunzip(); + await pipeline(read, gunzip, write); + } else { + await pipeline(read, write); + } + } finally { + await fs.unlink(tmpPath).catch(() => undefined); } - } finally { - await fs.unlink(tmpPath).catch(() => undefined); - } - }), + }), ); + for (const result of results) { + if (result.status === 'rejected') { + throw result.reason; + } + } } diff --git a/yarn-project/stdlib/src/snapshots/index.ts b/yarn-project/stdlib/src/snapshots/index.ts index 72d5831c1be..8f7cf120183 100644 --- a/yarn-project/stdlib/src/snapshots/index.ts +++ b/yarn-project/stdlib/src/snapshots/index.ts @@ -1,3 +1,9 @@ export * from './types.js'; export * from './upload.js'; -export { getBasePath, downloadSnapshot, getLatestSnapshotMetadata, makeSnapshotPaths } from './download.js'; +export { + getBasePath, + downloadSnapshot, + getSnapshotIndex, + getLatestSnapshotMetadata, + makeSnapshotPaths, +} from './download.js'; diff --git a/yarn-project/world-state/README.md b/yarn-project/world-state/README.md index 4a5cbe330ba..85ec98176a8 100644 --- a/yarn-project/world-state/README.md +++ b/yarn-project/world-state/README.md @@ -33,6 +33,26 @@ Once a new block is received, the synchronizer checks the uncommitted root value The interface to the Merkle Tree DB offers a unified asynchronous API to the set of trees available. Reads from the Merkle Trees need to be marked as to whether they should include uncommitted state. For this reason, the MerkleTreeReadOperationsFacade exists to abstract this detail away from the end consumer. +### History retention + +By default, `WS_NUM_HISTORIC_CHECKPOINTS=64` retains a rolling window of finalized checkpoints, along with all newer state. +Setting this value to zero disables history pruning. Increasing the window accumulates more history as the chain advances; +it does not require an existing database or snapshot to contain that history already. + +Set `WS_PRUNE_BEFORE_CHECKPOINT=N` to retain every block in checkpoint N and every later checkpoint permanently. This +overrides the rolling checkpoint setting. Set N to zero to retain all history from genesis. Pruning always stops at the +first block of a checkpoint; the checkpoint containing the finalized tip is always retained. Future checkpoint numbers +are supported. + +When `WS_PRUNE_BEFORE_CHECKPOINT` is explicitly set, startup accepts a database with more history than requested and +rejects one missing required history, including with `SYNC_MODE=force-snapshot`. Rebuild using a fresh world-state +database to request older history. + +When bootstrapping a fresh world-state database with a pin and snapshot sync enabled, the node tries compatible snapshots in +checkpoint N or earlier, newest first, then syncs forward while preserving complete checkpoints. Existing snapshot +indexes are mapped from block numbers using the archived checkpoint metadata. Missing or incompatible snapshots are +skipped. If none can be used, the node rebuilds from genesis. `SYNC_MODE=full` always skips snapshot downloads. + # Building/Testing Building the package is as simple as calling `yarn build` from the package root. diff --git a/yarn-project/world-state/src/native/native_world_state.ts b/yarn-project/world-state/src/native/native_world_state.ts index 4253a04df6b..38ad1b94690 100644 --- a/yarn-project/world-state/src/native/native_world_state.ts +++ b/yarn-project/world-state/src/native/native_world_state.ts @@ -120,6 +120,7 @@ export class NativeWorldStateService implements MerkleTreeDatabase { await worldState.init(); } catch (e) { log.error(`Error initializing world state: ${e}`); + await instance.close(); throw e; } diff --git a/yarn-project/world-state/src/synchronizer/config.ts b/yarn-project/world-state/src/synchronizer/config.ts index c4b75f07484..5bddc797ff7 100644 --- a/yarn-project/world-state/src/synchronizer/config.ts +++ b/yarn-project/world-state/src/synchronizer/config.ts @@ -1,3 +1,4 @@ +import { type CheckpointNumber, CheckpointNumberSchema } from '@aztec/foundation/branded-types'; import { type ConfigMappingsType, getConfigFromMappings, @@ -36,6 +37,9 @@ export interface WorldStateConfig { /** The number of historic checkpoints worth of blocks to maintain */ worldStateCheckpointHistory: number; + + /** Retain this checkpoint and every later checkpoint, overriding the rolling checkpoint window. */ + worldStatePruneBeforeCheckpoint?: CheckpointNumber; } export const worldStateConfigMappings: ConfigMappingsType = { @@ -95,6 +99,12 @@ export const worldStateConfigMappings: ConfigMappingsType = { fallback: ['WS_NUM_HISTORIC_BLOCKS'], ...numberConfigHelper(64), }, + worldStatePruneBeforeCheckpoint: { + env: 'WS_PRUNE_BEFORE_CHECKPOINT', + description: + 'Retain world state from the start of this checkpoint onwards, overriding WS_NUM_HISTORIC_CHECKPOINTS. Set to 0 to retain all history.', + parseEnv: value => CheckpointNumberSchema.parse(value), + }, }; /** diff --git a/yarn-project/world-state/src/synchronizer/history.test.ts b/yarn-project/world-state/src/synchronizer/history.test.ts new file mode 100644 index 00000000000..72dd822e52e --- /dev/null +++ b/yarn-project/world-state/src/synchronizer/history.test.ts @@ -0,0 +1,118 @@ +import { BlockNumber, CheckpointNumber, IndexWithinCheckpoint } from '@aztec/foundation/branded-types'; +import type { BlockData } from '@aztec/stdlib/block'; + +import { mock } from 'jest-mock-extended'; + +import type { WorldStateStatusSummary } from '../native/message.js'; +import { getWorldStateHistoryStartCheckpoint, validateWorldStateHistory } from './history.js'; + +describe('world-state history policy', () => { + const summary: WorldStateStatusSummary = { + oldestHistoricalBlock: BlockNumber(80), + unfinalizedBlockNumber: BlockNumber(120), + finalizedBlockNumber: BlockNumber(100), + treesAreSynched: true, + }; + + const blockSource = { + getBlockData: () => + Promise.resolve( + mock({ + checkpointNumber: CheckpointNumber(8), + indexWithinCheckpoint: IndexWithinCheckpoint(0), + }), + ), + }; + + it('uses a fixed boundary in preference to the checkpoint count', () => { + expect( + getWorldStateHistoryStartCheckpoint({ + worldStateCheckpointHistory: 0, + worldStatePruneBeforeCheckpoint: CheckpointNumber(90), + }), + ).toBe(90); + expect(getWorldStateHistoryStartCheckpoint({ worldStateCheckpointHistory: 64 })).toBeUndefined(); + expect(getWorldStateHistoryStartCheckpoint({ worldStateCheckpointHistory: 0 })).toBe(0); + }); + + it.each([8, 9, 15])('accepts extra history when retaining from checkpoint %i', async start => { + await expect( + validateWorldStateHistory( + { worldStateCheckpointHistory: 64, worldStatePruneBeforeCheckpoint: CheckpointNumber(start) }, + summary, + blockSource, + ), + ).resolves.toBeUndefined(); + }); + + it('rejects missing history before the pinned checkpoint', async () => { + await expect( + validateWorldStateHistory( + { worldStateCheckpointHistory: 64, worldStatePruneBeforeCheckpoint: CheckpointNumber(7) }, + summary, + blockSource, + ), + ).rejects.toThrow('history is required from the start of checkpoint 7'); + }); + + it('rejects a pruned database when all history is requested', async () => { + await expect(validateWorldStateHistory({ worldStateCheckpointHistory: 0 }, summary)).rejects.toThrow( + 'all checkpoint history is required', + ); + }); + + it('rejects pruned history when checkpoint zero is pinned', async () => { + await expect( + validateWorldStateHistory( + { worldStateCheckpointHistory: 64, worldStatePruneBeforeCheckpoint: CheckpointNumber.ZERO }, + summary, + ), + ).rejects.toThrow('all checkpoint history is required'); + }); + + it('accepts a fresh database and an unpruned database for genesis retention', async () => { + for (const oldestHistoricalBlock of [BlockNumber.ZERO, BlockNumber(1)]) { + await expect( + validateWorldStateHistory({ worldStateCheckpointHistory: 0 }, { ...summary, oldestHistoricalBlock }), + ).resolves.toBeUndefined(); + } + }); + + describe.each(['rolling', 'pinned'])('%s checkpoint retention', mode => { + it.each([ + { checkpoint: 8, index: 2, compatible: true }, + { checkpoint: 9, index: 0, compatible: true }, + { checkpoint: 9, index: 1, compatible: false }, + { checkpoint: 10, index: 0, compatible: false }, + ])('requires complete checkpoints: %j', async ({ checkpoint, index, compatible }) => { + const blocks = new Map([ + [ + 100, + mock({ checkpointNumber: CheckpointNumber(10), indexWithinCheckpoint: IndexWithinCheckpoint(3) }), + ], + [ + 80, + mock({ + checkpointNumber: CheckpointNumber(checkpoint), + indexWithinCheckpoint: IndexWithinCheckpoint(index), + }), + ], + ]); + const result = validateWorldStateHistory( + { + worldStateCheckpointHistory: 2, + worldStatePruneBeforeCheckpoint: mode === 'pinned' ? CheckpointNumber(9) : undefined, + }, + summary, + { + getBlockData: query => Promise.resolve('number' in query ? blocks.get(query.number) : undefined), + }, + ); + if (compatible) { + await expect(result).resolves.toBeUndefined(); + } else { + await expect(result).rejects.toThrow('history is required from the start of checkpoint 9'); + } + }); + }); +}); diff --git a/yarn-project/world-state/src/synchronizer/history.ts b/yarn-project/world-state/src/synchronizer/history.ts new file mode 100644 index 00000000000..8ce86367d58 --- /dev/null +++ b/yarn-project/world-state/src/synchronizer/history.ts @@ -0,0 +1,55 @@ +import { CheckpointNumber, CheckpointNumberSchema } from '@aztec/foundation/branded-types'; +import type { L2BlockSource } from '@aztec/stdlib/block'; + +import type { WorldStateStatusSummary } from '../native/message.js'; +import type { WorldStateConfig } from './config.js'; + +/** Returns the fixed checkpoint boundary, or undefined for a rolling checkpoint window. */ +export function getWorldStateHistoryStartCheckpoint( + config: Pick, +): CheckpointNumber | undefined { + if (config.worldStatePruneBeforeCheckpoint !== undefined) { + return CheckpointNumberSchema.parse(config.worldStatePruneBeforeCheckpoint); + } + return config.worldStateCheckpointHistory < 1 ? CheckpointNumber.ZERO : undefined; +} + +/** Rejects a database whose retained history cannot satisfy the configured pruning policy. */ +export async function validateWorldStateHistory( + config: Pick, + summary: WorldStateStatusSummary, + blockSource?: Pick, +): Promise { + const startCheckpoint = getWorldStateHistoryStartCheckpoint(config); + const { oldestHistoricalBlock, unfinalizedBlockNumber, finalizedBlockNumber } = summary; + if (oldestHistoricalBlock <= 1) { + return; + } + const available = `World state retains blocks ${oldestHistoricalBlock} to ${unfinalizedBlockNumber}`; + const recovery = 'Use a fresh world-state database to rebuild the required history from a snapshot or genesis.'; + if (startCheckpoint === CheckpointNumber.ZERO) { + throw new Error(`${available}, but all checkpoint history is required. ${recovery}`); + } + + const oldest = await blockSource?.getBlockData({ number: oldestHistoricalBlock }); + let requiredCheckpoint = startCheckpoint; + if (requiredCheckpoint === undefined) { + const finalized = await blockSource?.getBlockData({ number: finalizedBlockNumber }); + if (finalized) { + requiredCheckpoint = CheckpointNumber( + Math.max(1, finalized.checkpointNumber - config.worldStateCheckpointHistory + 1), + ); + } + } + if (!oldest || requiredCheckpoint === undefined) { + throw new Error(`${available}, but its checkpoint history could not be verified against the archiver.`); + } + if ( + oldest.checkpointNumber > requiredCheckpoint || + (oldest.checkpointNumber === requiredCheckpoint && oldest.indexWithinCheckpoint !== 0) + ) { + throw new Error( + `${available}, but history is required from the start of checkpoint ${requiredCheckpoint}. ${recovery}`, + ); + } +} diff --git a/yarn-project/world-state/src/synchronizer/index.ts b/yarn-project/world-state/src/synchronizer/index.ts index eb6e92ee213..e2eda0e7c8e 100644 --- a/yarn-project/world-state/src/synchronizer/index.ts +++ b/yarn-project/world-state/src/synchronizer/index.ts @@ -1,3 +1,4 @@ export * from './server_world_state_synchronizer.js'; export * from './factory.js'; export * from './errors.js'; +export * from './history.js'; diff --git a/yarn-project/world-state/src/synchronizer/server_world_state_synchronizer.ts b/yarn-project/world-state/src/synchronizer/server_world_state_synchronizer.ts index fb14ea56c8d..56c0a3c0d6e 100644 --- a/yarn-project/world-state/src/synchronizer/server_world_state_synchronizer.ts +++ b/yarn-project/world-state/src/synchronizer/server_world_state_synchronizer.ts @@ -30,6 +30,7 @@ import type { WorldStateStatusFull } from '../native/message.js'; import type { MerkleTreeAdminDatabase } from '../world-state-db/merkle_tree_db.js'; import type { WorldStateConfig } from './config.js'; import { WorldStateSynchronizerError } from './errors.js'; +import { getWorldStateHistoryStartCheckpoint, validateWorldStateHistory } from './history.js'; export type { SnapshotDataKeys }; @@ -44,7 +45,6 @@ export class ServerWorldStateSynchronizer private readonly merkleTreeCommitted: MerkleTreeReadOperations; private latestBlockNumberAtStart = BlockNumber.ZERO; - private historyToKeep: number | undefined; private currentState: WorldStateRunningState = WorldStateRunningState.IDLE; private syncPromise = promiseWithResolvers(); @@ -62,12 +62,10 @@ export class ServerWorldStateSynchronizer private readonly log: Logger = createLogger('world_state'), ) { this.merkleTreeCommitted = this.merkleTreeDb.getCommitted(); - this.historyToKeep = config.worldStateCheckpointHistory < 1 ? undefined : config.worldStateCheckpointHistory; - this.log.info( - `Created world state synchroniser with block history of ${ - this.historyToKeep === undefined ? 'infinity' : this.historyToKeep - }`, - ); + this.log.info('Created world state synchronizer', { + historyStartCheckpoint: getWorldStateHistoryStartCheckpoint(config), + checkpointHistory: config.worldStateCheckpointHistory, + }); } public getCommitted(): MerkleTreeReadOperations { @@ -138,6 +136,17 @@ export class ServerWorldStateSynchronizer return this.syncPromise; } + const summary = await this.merkleTreeDb.getStatusSummary(); + if (this.config.worldStatePruneBeforeCheckpoint !== undefined) { + await validateWorldStateHistory(this.config, summary, this.l2BlockSource); + } + this.log.info('World state history available at startup', { + oldestHistoricBlock: summary.oldestHistoricalBlock, + latestBlock: summary.unfinalizedBlockNumber, + historyStartCheckpoint: getWorldStateHistoryStartCheckpoint(this.config), + checkpointHistory: this.config.worldStateCheckpointHistory, + }); + // Get the current latest block number this.latestBlockNumberAtStart = BlockNumber(await this.l2BlockSource.getBlockNumber()); @@ -439,7 +448,8 @@ export class ServerWorldStateSynchronizer return; } const summary = await this.merkleTreeDb.setFinalized(blockNumber); - if (this.historyToKeep === undefined) { + const historyStartCheckpoint = getWorldStateHistoryStartCheckpoint(this.config); + if (historyStartCheckpoint === CheckpointNumber.ZERO) { return; } const finalisedBlockData = await this.l2BlockSource.getBlockData({ number: summary.finalizedBlockNumber }); @@ -450,7 +460,10 @@ export class ServerWorldStateSynchronizer return; } // Compute the required historic checkpoint number - const newHistoricCheckpointNumber = finalisedBlockData.checkpointNumber - this.historyToKeep + 1; + const newHistoricCheckpointNumber = + historyStartCheckpoint === undefined + ? finalisedBlockData.checkpointNumber - this.config.worldStateCheckpointHistory + 1 + : Math.min(historyStartCheckpoint, finalisedBlockData.checkpointNumber); if (newHistoricCheckpointNumber <= 1) { return; } diff --git a/yarn-project/world-state/src/test/integration.test.ts b/yarn-project/world-state/src/test/integration.test.ts index 3748bd51c82..b58143dc88b 100644 --- a/yarn-project/world-state/src/test/integration.test.ts +++ b/yarn-project/world-state/src/test/integration.test.ts @@ -1,4 +1,4 @@ -import { MockPrefilledArchiver } from '@aztec/archiver/test'; +import { MockArchiver, MockPrefilledArchiver } from '@aztec/archiver/test'; import { GENESIS_ARCHIVE_ROOT } from '@aztec/constants'; import { BlockNumber, CheckpointNumber } from '@aztec/foundation/branded-types'; import { timesAsync } from '@aztec/foundation/collection'; @@ -237,6 +237,87 @@ describe('world-state integration', () => { }); describe('finalized chain', () => { + it('retains complete checkpoints before and after a future pin is reached', async () => { + const source = new MultiBlockArchiver(); + await source.setInitialHeader(db.getInitialHeader()); + source.setGenesisArchiveRoot(new Fr(GENESIS_ARCHIVE_ROOT)); + const fork = await db.fork(BlockNumber.ZERO); + for (let i = 0; i < 7; i++) { + const { checkpoint, messages } = await mockCheckpoint(CheckpointNumber(i + 1), fork, { + startBlockNumber: BlockNumber(3 * i + 1), + numBlocks: 3, + }); + source.addCheckpoint(checkpoint, messages); + } + await fork.close(); + synchronizer = new TestWorldStateSynchronizer(db, source, { + ...config, + worldStateCheckpointHistory: 2, + worldStatePruneBeforeCheckpoint: CheckpointNumber(3), + }); + await synchronizer.start(); + + for (const [finalized, checkpoint, oldest] of [ + [5, 2, 4], + [10, 4, 7], + [17, 6, 7], + ]) { + await synchronizer.handleBlockStreamEvent({ + type: 'chain-finalized', + block: { number: BlockNumber(finalized), hash: '' }, + checkpoint: { number: CheckpointNumber(checkpoint), hash: '' }, + }); + expect((await db.getStatusSummary()).oldestHistoricalBlock).toBe(oldest); + await expect(db.getSnapshot(BlockNumber(oldest - 1)).getTreeInfo(MerkleTreeId.ARCHIVE)).rejects.toThrow(); + for (let number = oldest; number < oldest + 3; number++) { + const snapshot = await db.getSnapshot(BlockNumber(number)).getTreeInfo(MerkleTreeId.ARCHIVE); + expect(new Fr(snapshot.root)).toEqual((await source.getBlock({ number: BlockNumber(number) }))!.archive.root); + } + } + + const nextFork = await db.fork(BlockNumber(21)); + const next = await mockCheckpoint(CheckpointNumber(8), nextFork, { + startBlockNumber: BlockNumber(22), + numBlocks: 3, + }); + await nextFork.close(); + source.addCheckpoint(next.checkpoint, next.messages); + await synchronizer.syncImmediate(BlockNumber(24)); + expect((await db.getStatusSummary()).unfinalizedBlockNumber).toBe(24); + expect((await db.getStatusSummary()).oldestHistoricalBlock).toBe(7); + }); + + it.each([128, 0])('accepts a restart with an increased history window of %i', async history => { + await archiver.createBlocks(20); + await synchronizer.start(); + await synchronizer.stopBlockStream(); + await db.setFinalized(BlockNumber(15)); + await db.removeHistoricalBlocks(BlockNumber(10)); + + synchronizer = new TestWorldStateSynchronizer(db, archiver, { + ...config, + worldStateCheckpointHistory: history, + }); + await synchronizer.start(); + await expectSynchedToBlock(20, 15); + expect((await db.getStatusSummary()).oldestHistoricalBlock).toBe(10); + }); + + it('rejects a restart that requests history already pruned from the database', async () => { + await archiver.createBlocks(20); + await synchronizer.start(); + await synchronizer.stopBlockStream(); + await db.setFinalized(BlockNumber(15)); + await db.removeHistoricalBlocks(BlockNumber(10)); + + synchronizer = new TestWorldStateSynchronizer(db, archiver, { + ...config, + worldStatePruneBeforeCheckpoint: CheckpointNumber(5), + }); + await expect(synchronizer.start()).rejects.toThrow('history is required from the start of checkpoint 5'); + expect((await db.getStatusSummary()).oldestHistoricalBlock).toBe(10); + }); + it('syncs finalized chain tip', async () => { await archiver.createBlocks(5); archiver.setFinalizedBlockNumber(3); @@ -297,3 +378,11 @@ class TestWorldStateSynchronizer extends ServerWorldStateSynchronizer { await this.blockStream?.stop(); } } + +class MultiBlockArchiver extends MockArchiver { + public addCheckpoint(checkpoint: Checkpoint, messages: Fr[]) { + this.checkpointList.push(checkpoint); + this.setL1ToL2Messages(checkpoint.number, messages); + this.addProposedBlocks(checkpoint.blocks); + } +}