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'), };