From d294f3db4bc1ec22f777f86b073218215953d19f Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Tue, 25 Aug 2026 16:38:50 -0300 Subject: [PATCH] fix(archiver): serve log-by-tag queries from read-only snapshots The log store's read paths now run inside `db.readOnlyTransaction` instead of `db.transactionAsync`. They still see one consistent snapshot for the `referenceBlock` reorg check and every per-tag scan, but no longer go through the store's single serial writer queue, so a query does not wait for queued writes (notably block ingestion) to commit. Only the read paths move; `addLogs` and `deleteLogs` stay on `transactionAsync`. Cursors bound to one snapshot serialize on that snapshot's mutex natively, so `TAG_SCAN_CONCURRENCY` drops from 8 to 2: on a seeded store a 100-tag query averages 7.4ms at 1, 4.9ms at 2, and 5.0ms at 4 and 8, and with five queries in flight the wider pools are marginally slower. Each open snapshot holds one LMDB reader slot, so the reader table size is now configurable via `DATA_STORE_MAX_READERS` / `DataStoreConfig.dataStoreMaxReaders` instead of being hardcoded, keeping the previous 16 as the default. --- yarn-project/archiver/src/store/log_store.ts | 35 +++++++++++-------- .../store/log_store_write_contention.test.ts | 11 ++++-- yarn-project/foundation/src/config/env_var.ts | 1 + yarn-project/kv-store/src/lmdb-v2/factory.ts | 19 +++++----- .../stdlib/src/ha-signing/local_config.ts | 1 + yarn-project/stdlib/src/kv-store/config.ts | 12 +++++++ 6 files changed, 51 insertions(+), 28 deletions(-) diff --git a/yarn-project/archiver/src/store/log_store.ts b/yarn-project/archiver/src/store/log_store.ts index b9dde9f02c6..cff030ccd98 100644 --- a/yarn-project/archiver/src/store/log_store.ts +++ b/yarn-project/archiver/src/store/log_store.ts @@ -30,8 +30,12 @@ import { tagHexForLog, } from './log_store_codec.js'; -/** How many per-tag range scans a single tag query runs concurrently. */ -const TAG_SCAN_CONCURRENCY = 8; +/** + * How many per-tag range scans a single tag query runs concurrently. Cursors bound to one read-only snapshot serialize + * on that snapshot's mutex natively, so overlapping them only hides the per-cursor round trip: measured on a + * 100-tag query, going from 1 to 2 cuts latency by a third and anything above 2 is flat or marginally worse. + */ +const TAG_SCAN_CONCURRENCY = 2; /** * Indexes every emitted private and public log under a composite hex-string key @@ -48,11 +52,11 @@ const TAG_SCAN_CONCURRENCY = 8; * * Contract-class logs are no longer stored or served by the log store. * - * Tag queries run inside `db.transactionAsync` so the `referenceBlock` reorg check and every per-tag scan see the same - * snapshot of the store and cannot return a torn result. On lmdb-v2 that helper routes its callback through the store's - * single serial writer queue, so a query waits for any queued write (notably block ingestion) to commit first; that - * cost is accepted in exchange for the consistency. The per-tag scans within one query still run concurrently - * ({@link TAG_SCAN_CONCURRENCY} at a time), each on its own cursor over the transaction's snapshot. + * Every read path runs inside `db.readOnlyTransaction`, so the `referenceBlock` reorg check and all the per-tag scans + * of one query observe the same committed snapshot and cannot return a torn result. Unlike `transactionAsync`, a + * read-only snapshot does not go through the store's single serial writer queue, so a query never waits for queued + * writes (notably block ingestion) to commit. The per-tag scans within one query run concurrently + * ({@link TAG_SCAN_CONCURRENCY} at a time), each on its own cursor over the snapshot. */ export class LogStore { /** Primary map: composite private key (tag + tail = 96 hex chars + separators) -> serialized {@link StoredLogValue}. */ @@ -199,13 +203,13 @@ export class LogStore { /** Returns one inner array per element of `query.tags`, in input order. */ getPrivateLogsByTags(query: PrivateLogsQuery): Promise { LogStore.#validateQuery(query); - return this.db.transactionAsync(() => this.#runQuery(query, /* contractHex */ undefined)); + return this.db.readOnlyTransaction(() => this.#runQuery(query, /* contractHex */ undefined)); } /** Returns one inner array per element of `query.tags`, in input order. */ getPublicLogsByTags(query: PublicLogsQuery): Promise { LogStore.#validateQuery(query); - return this.db.transactionAsync(() => this.#runQuery(query, fieldHex(query.contractAddress))); + return this.db.readOnlyTransaction(() => this.#runQuery(query, fieldHex(query.contractAddress))); } static #validateQuery(query: { txHash?: TxHash; fromBlock?: unknown; toBlock?: unknown }): void { @@ -219,7 +223,7 @@ export class LogStore { const tags = (query.tags as ReadonlyArray>) ?? []; const primaryMap = isPublic ? this.#publicLogs : this.#privateLogs; - // referenceBlock reorg check, in-transaction, against the same db the log primary maps live on. The + // referenceBlock reorg check, on the query's snapshot, against the same db the log primary maps live on. The // genesis block is a valid anchor during early sync but is synthetic and never indexed in the block // store, so resolve it directly to the genesis block number rather than mistaking it for a reorg. let referenceBlockNumber: number | undefined; @@ -266,9 +270,8 @@ export class LogStore { const limit = query.limitPerTag ?? MAX_LOGS_PER_TAG; - // Each scan opens its own native cursor over the transaction's snapshot, so they overlap instead of queueing. The - // pool bound stays well under the store's concurrent-cursor budget (maxReaders - 1) so other components can still - // read while a query is in flight. + // Each scan opens its own native cursor over the query's snapshot. The pool bound stays well under the store's + // concurrent-cursor budget (maxReaders - 1) so other components can still read while a query is in flight. const perTagResults = await asyncPool(TAG_SCAN_CONCURRENCY, [...tags], async tagEntry => { const { tagHex, afterLog } = normalizeTagEntry(tagEntry); const prefix = contractHex !== undefined ? encodePublicPrefix(contractHex, tagHex) : tagHex; @@ -338,14 +341,16 @@ export class LogStore { * counts without depending on the removed `getPublicLogs(LogFilter)` API. */ getPrivateLogsForBlock(blockNumber: number): Promise { - return this.db.transactionAsync(() => + return this.db.readOnlyTransaction(() => this.#readBlockLogs(this.#privateKeysByBlock, this.#privateLogs, blockNumber), ); } /** {@inheritDoc LogStore.getPrivateLogsForBlock} */ getPublicLogsForBlock(blockNumber: number): Promise { - return this.db.transactionAsync(() => this.#readBlockLogs(this.#publicKeysByBlock, this.#publicLogs, blockNumber)); + return this.db.readOnlyTransaction(() => + this.#readBlockLogs(this.#publicKeysByBlock, this.#publicLogs, blockNumber), + ); } async #readBlockLogs( diff --git a/yarn-project/archiver/src/store/log_store_write_contention.test.ts b/yarn-project/archiver/src/store/log_store_write_contention.test.ts index b39d57c636b..c55fa75fc5a 100644 --- a/yarn-project/archiver/src/store/log_store_write_contention.test.ts +++ b/yarn-project/archiver/src/store/log_store_write_contention.test.ts @@ -75,24 +75,29 @@ describe('LogStore write contention', () => { const baseline = await logStore.getPrivateLogsByTags({ tags }); const baselineMs = baselineTimer.ms(); - // Fill the store's serial writer queue without awaiting it. Tag queries are routed through that same queue, so - // the contended query below only resolves once these commit. + // Fill the store's serial writer queue without awaiting it. const writes = Array.from({ length: QUEUED_WRITES }, () => db.transactionAsync(() => sleep(WRITE_DURATION_MS))); const contendedTimer = new Timer(); const contended = await logStore.getPrivateLogsByTags({ tags }); const contendedMs = contendedTimer.ms(); + const queuedWriteMs = QUEUED_WRITES * WRITE_DURATION_MS; logger.info(`Baseline read ${baselineMs.toFixed(2)}ms, contended read ${contendedMs.toFixed(2)}ms`, { baselineMs, contendedMs, - queuedWriteMs: QUEUED_WRITES * WRITE_DURATION_MS, + queuedWriteMs, }); expect(contended.length).toBe(TAGS_PER_QUERY); expect(contended.every(logs => logs.length > 0)).toBe(true); expect(contended).toEqual(baseline); + // The query runs on a read-only snapshot instead of the writer queue, so it must not wait for the queued writes. + // Serializing behind them would cost the full `queuedWriteMs` on top of the baseline; half of that is left as + // jitter headroom so the bound only trips when the read really is queued. + expect(contendedMs).toBeLessThan(baselineMs + queuedWriteMs / 2); + await Promise.all(writes); }); }); diff --git a/yarn-project/foundation/src/config/env_var.ts b/yarn-project/foundation/src/config/env_var.ts index fdf5f99b720..97c849b3213 100644 --- a/yarn-project/foundation/src/config/env_var.ts +++ b/yarn-project/foundation/src/config/env_var.ts @@ -63,6 +63,7 @@ export type EnvVar = | 'CRS_PATH' | 'DATA_DIRECTORY' | 'DATA_STORE_MAP_SIZE_KB' + | 'DATA_STORE_MAX_READERS' | 'ARCHIVER_STORE_MAP_SIZE_KB' | 'BLOB_SINK_MAP_SIZE_KB' | 'P2P_STORE_MAP_SIZE_KB' diff --git a/yarn-project/kv-store/src/lmdb-v2/factory.ts b/yarn-project/kv-store/src/lmdb-v2/factory.ts index d4396ef7dd4..82caff51e37 100644 --- a/yarn-project/kv-store/src/lmdb-v2/factory.ts +++ b/yarn-project/kv-store/src/lmdb-v2/factory.ts @@ -5,7 +5,7 @@ import { type SchemaVersionMismatchPolicy, type VersionFileReadFailurePolicy, } from '@aztec/stdlib/database-version/manager'; -import type { DataStoreConfig } from '@aztec/stdlib/kv-store'; +import { DEFAULT_DATA_STORE_MAX_READERS, type DataStoreConfig } from '@aztec/stdlib/kv-store'; import { copyFile, mkdir, mkdtemp, rm } from 'fs/promises'; import { tmpdir } from 'os'; @@ -13,8 +13,6 @@ import { join } from 'path'; import { AztecLMDBStoreV2 } from './store.js'; -const MAX_READERS = 16; - /** Optional versioning hooks for persistent LMDB stores. */ export type CreateStoreOptions = { onUpgrade?: (dataDir: string, currentVersion: number, latestVersion: number) => Promise; @@ -31,6 +29,7 @@ export async function createStore( ): Promise { const log = createLogger('kv-store:lmdb-v2:' + name, bindings); const { dataDirectory, rollupAddress: rollupFromConfig } = config; + const maxReaders = config.dataStoreMaxReaders ?? DEFAULT_DATA_STORE_MAX_READERS; let store: AztecLMDBStoreV2; if (typeof dataDirectory !== 'undefined') { @@ -46,7 +45,7 @@ export async function createStore( rollupAddress, dataDirectory: subDir, onOpen: dbDirectory => - AztecLMDBStoreV2.new(dbDirectory, config.dataStoreMapSizeKb, MAX_READERS, () => Promise.resolve(), bindings), + AztecLMDBStoreV2.new(dbDirectory, config.dataStoreMapSizeKb, maxReaders, () => Promise.resolve(), bindings), onUpgrade: options.onUpgrade, schemaVersionMismatchPolicy: options.schemaVersionMismatchPolicy, versionFileReadFailurePolicy: options.versionFileReadFailurePolicy, @@ -57,7 +56,7 @@ export async function createStore( ); [store] = await versionManager.open(); } else { - store = await openTmpStore(name, true, config.dataStoreMapSizeKb, MAX_READERS, bindings); + store = await openTmpStore(name, true, config.dataStoreMapSizeKb, maxReaders, bindings); } return store; @@ -72,7 +71,7 @@ export async function openTmpStore( name: string, cleanupTmpDir: boolean = true, dbMapSizeKb = 10 * 1_024 * 1_024, // 10GB - maxReaders = MAX_READERS, + maxReaders = DEFAULT_DATA_STORE_MAX_READERS, bindings?: LoggerBindings, ): Promise { const log = createLogger('kv-store:lmdb-v2:' + name, bindings); @@ -105,7 +104,7 @@ export async function openTmpStore( export async function openEphemeralStore( name: string, dbMapSizeKb = 10 * 1_024 * 1_024, // 10GB - maxReaders = MAX_READERS, + maxReaders = DEFAULT_DATA_STORE_MAX_READERS, bindings?: LoggerBindings, ): Promise { const log = createLogger('kv-store:lmdb-v2:' + name, bindings); @@ -127,7 +126,7 @@ export async function openEphemeralStore( export async function openStoreAt( dataDir: string, dbMapSizeKb = 10 * 1_024 * 1_024, // 10GB - maxReaders = MAX_READERS, + maxReaders = DEFAULT_DATA_STORE_MAX_READERS, bindings?: LoggerBindings, ): Promise { const log = createLogger('kv-store:lmdb-v2', bindings); @@ -145,7 +144,7 @@ export async function cloneEphemeralStoreFrom( srcDataMdbPath: string, name: string, dbMapSizeKb = 10 * 1_024 * 1_024, // 10GB - maxReaders = MAX_READERS, + maxReaders = DEFAULT_DATA_STORE_MAX_READERS, bindings?: LoggerBindings, ): Promise { const log = createLogger('kv-store:lmdb-v2:' + name, bindings); @@ -170,7 +169,7 @@ export async function openVersionedStoreAt( schemaVersion: number, rollupAddress: EthAddress, dbMapSizeKb = 10 * 1_024 * 1_024, // 10GB - maxReaders = MAX_READERS, + maxReaders = DEFAULT_DATA_STORE_MAX_READERS, bindings?: LoggerBindings, ): Promise { const log = createLogger('kv-store:lmdb-v2', bindings); diff --git a/yarn-project/stdlib/src/ha-signing/local_config.ts b/yarn-project/stdlib/src/ha-signing/local_config.ts index c16f3e0c4b9..a18a827fcc2 100644 --- a/yarn-project/stdlib/src/ha-signing/local_config.ts +++ b/yarn-project/stdlib/src/ha-signing/local_config.ts @@ -52,6 +52,7 @@ export const LocalSignerConfigSchema = zodFor()( BaseSignerConfigSchema.extend({ dataDirectory: z.string().optional(), dataStoreMapSizeKb: z.number(), + dataStoreMaxReaders: z.number().optional(), signingProtectionMapSizeKb: z.number().optional(), allowEphemeralSigningProtection: z.boolean().optional(), }), diff --git a/yarn-project/stdlib/src/kv-store/config.ts b/yarn-project/stdlib/src/kv-store/config.ts index c0892f96020..2206cec7c65 100644 --- a/yarn-project/stdlib/src/kv-store/config.ts +++ b/yarn-project/stdlib/src/kv-store/config.ts @@ -1,9 +1,16 @@ import { type L1ContractAddresses, pickL1ContractAddressMappings } from '@aztec/ethereum/l1-contract-addresses'; import { type ConfigMappingsType, getConfigFromMappings, numberConfigHelper } from '@aztec/foundation/config'; +/** + * Slots in the LMDB reader table. Every concurrent cursor and every open read-only snapshot holds one, so this bounds + * how many reads a store can have in flight at once. + */ +export const DEFAULT_DATA_STORE_MAX_READERS = 16; + export type DataStoreConfig = { dataDirectory?: string; dataStoreMapSizeKb: number; + dataStoreMaxReaders?: number; } & Partial>; export const dataConfigMappings: ConfigMappingsType = { @@ -16,6 +23,11 @@ export const dataConfigMappings: ConfigMappingsType = { description: 'The maximum possible size of a data store DB in KB. Can be overridden by component-specific options.', ...numberConfigHelper(128 * 1_024 * 1_024), // Defaulted to 128 GB }, + dataStoreMaxReaders: { + env: 'DATA_STORE_MAX_READERS', + description: 'Maximum number of concurrent readers (cursors and read-only snapshots) on a data store DB.', + ...numberConfigHelper(DEFAULT_DATA_STORE_MAX_READERS), + }, ...pickL1ContractAddressMappings('rollupAddress'), };