diff --git a/resources/databases.ts b/resources/databases.ts index 8b4b0d842f..cad1bf435e 100644 --- a/resources/databases.ts +++ b/resources/databases.ts @@ -29,6 +29,7 @@ import { OpenDBIObject } from '../utility/lmdb/OpenDBIObject.ts'; import { RocksDatabase, type RocksDatabaseOptions } from '@harperfast/rocksdb-js'; import { replayLogs } from './replayLogs.ts'; import { totalmem } from 'node:os'; +import { setTimeout as delay } from 'node:timers/promises'; import { RocksIndexStore } from './RocksIndexStore.ts'; import { when } from '../utility/when.ts'; import { resolveRocksMemoryConfig } from '../utility/rocksMemoryConfig.ts'; @@ -1406,18 +1407,39 @@ export function table(tableDefinition: TableDefinition): Tabl } const MAX_OUTSTANDING_INDEXING = 1000; const MIN_OUTSTANDING_INDEXING = 10; -async function runIndexing(Table, attributes, indicesToRemove) { +// When a backfill pass fails with only transient RocksDB errors (e.g. the write buffer filling under +// bulk-ingest load), re-run the pass in-process instead of parking the index until the next restart. +// Bounded with exponential backoff; on exhaustion we fall back to the existing park-and-retry-on-restart +// behavior, so this is strictly an improvement. See issue #1356. +const INDEXING_MAX_PASS_RETRIES = 10; +const INDEXING_RETRY_BASE_DELAY_MS = 50; +const INDEXING_RETRY_MAX_DELAY_MS = 1000; + +// ERR_BUSY (optimistic write conflict) and ERR_TRY_AGAIN (RocksDB kTryAgain — the snapshot fell outside +// the memtable conflict-check window, which happens under bulk ingest such as a backfill) are transient +// and retryable. Mirrors the classification in DatabaseTransaction's commit retry. A pass failing only +// with these can be retried in-process; any other error is treated as permanent and parks the index. +function isTransientIndexingError(error: any): boolean { + return error?.code === 'ERR_BUSY' || error?.code === 'ERR_TRY_AGAIN'; +} +async function runIndexing(Table, attributes, indicesToRemove, retryAttempt = 0) { try { logger.info(`Indexing ${Table.tableName} attributes`, attributes); - await signalling.signalSchemaChange( - new SchemaEventMsg(process.pid, 'schema-change', Table.databaseName, Table.tableName) - ); let lastResolution; - for (const index of indicesToRemove) { - lastResolution = index.drop(); + if (retryAttempt === 0) { + // Only signal the schema change and drop removed indices on the first attempt; in-process + // transient retries (see below) re-enter this function with indicesToRemove already emptied. + await signalling.signalSchemaChange( + new SchemaEventMsg(process.pid, 'schema-change', Table.databaseName, Table.tableName) + ); + for (const index of indicesToRemove) { + lastResolution = index.drop(); + } } let interrupted; let hadIndexingErrors = false; + let hadPermanentIndexingError = false; + let passStartKey: any; const attributeErrorReported = {}; let indexed = 0; const attributesLength = attributes.length; @@ -1437,6 +1459,11 @@ async function runIndexing(Table, attributes, indicesToRemove) { } } } + // Capture this pass's starting checkpoint. A transient-only retry resets each attribute's + // lastIndexedKey back to here and re-runs the pass, re-reading any row that errored + // mid-pass — the intra-pass checkpoint below can advance past a put that later rejects + // asynchronously, so it is not a safe resume point. Re-indexing is idempotent. + passStartKey = start; let outstanding = 0; // this means that a new attribute has been introduced that needs to be indexed for (const { key, value: record } of Table.primaryStore.getRange({ @@ -1477,6 +1504,7 @@ async function runIndexing(Table, attributes, indicesToRemove) { } } catch (error) { hadIndexingErrors = true; + if (!isTransientIndexingError(error)) hadPermanentIndexingError = true; if (!attributeErrorReported[property]) { // just report an indexing error once per attribute so we don't spam the logs attributeErrorReported[property] = true; @@ -1490,6 +1518,7 @@ async function runIndexing(Table, attributes, indicesToRemove) { (error) => { outstanding--; hadIndexingErrors = true; + if (!isTransientIndexingError(error)) hadPermanentIndexingError = true; logger.error(error); } ); @@ -1521,12 +1550,44 @@ async function runIndexing(Table, attributes, indicesToRemove) { await lastResolution; } catch (error) { hadIndexingErrors = true; + if (!isTransientIndexingError(error)) hadPermanentIndexingError = true; logger.error(error); } // Yield one more event turn so any queued when() error callbacks (which fire as // microtasks when their tracked promise settles) have a chance to set hadIndexingErrors // before we decide whether to mark indexing as complete. await new Promise((resolve) => setImmediate(resolve)); + // If the pass failed only with transient RocksDB errors (ERR_BUSY/ERR_TRY_AGAIN under + // bulk-ingest load), re-run it in-process from this pass's start rather than parking the + // index until the next restart. Bounded with exponential backoff; on exhaustion (or any + // permanent error) we fall through to the park path below, so this never does worse than the + // prior behavior. Reset and persist the resume point so a crash mid-retry also resumes from + // here, then re-enter with an empty indicesToRemove (the drops already happened). See #1356. + if ( + hadIndexingErrors && + !hadPermanentIndexingError && + attributesLength > 0 && + retryAttempt < INDEXING_MAX_PASS_RETRIES + ) { + const checkpointWrites: Promise[] = []; + for (const attribute of attributes) { + attribute.lastIndexedKey = passStartKey; + checkpointWrites.push(Table.dbisDB.put(attribute.key, attribute)); + } + // Await the descriptor writes before retrying (as the park/complete paths do) so a metadata + // write that rejects under the same RocksDB pressure — or while the DB is closing — is not + // dropped as an unhandled rejection, and we never retry without the persisted checkpoint. A + // failure here propagates to the outer catch, which parks the index: a safe fallback. + await Promise.all(checkpointWrites); + const backoff = Math.min(INDEXING_RETRY_BASE_DELAY_MS * 2 ** retryAttempt, INDEXING_RETRY_MAX_DELAY_MS); + logger.warn( + `Indexing of ${Table.tableName} hit transient errors; retrying pass ` + + `${retryAttempt + 1}/${INDEXING_MAX_PASS_RETRIES} after ${backoff}ms. ` + + `Affected attributes: ${attributes.map((a) => a.name).join(', ')}` + ); + await delay(backoff); + return runIndexing(Table, attributes, [], retryAttempt + 1); + } if (hadIndexingErrors) { // Some records failed to index. Persist the failure marker in the descriptor so // the next call to table() (including after a restart with a fresh PID) re-triggers diff --git a/unitTests/resources/schemaMigrationFragility.test.js b/unitTests/resources/schemaMigrationFragility.test.js index be625f81a7..4c79f880e3 100644 --- a/unitTests/resources/schemaMigrationFragility.test.js +++ b/unitTests/resources/schemaMigrationFragility.test.js @@ -83,7 +83,7 @@ describe('schema-migration fragility: silent gaps when per-record indexing error ); }); - it('reproduces a silent gap when a per-record index write rejects mid-flight', async () => { + it('parks the index (does not silently complete) when a per-record index write fails permanently', async () => { // Force a fresh migration cycle by recreating with a DIFFERENT attribute // name so a NEW index has to be backfilled. const TABLE2 = TABLE + 'WithThrowingIndex'; @@ -100,8 +100,10 @@ describe('schema-migration fragility: silent gaps when per-record indexing error } await last; - // Now add @indexed to tag. During runIndexing, intercept dbi.put calls - // for some records to simulate transient errors (e.g. ERR_BUSY). + // Now add @indexed to tag. During runIndexing, intercept dbi.put calls for some records to + // simulate a permanent (non-retryable) per-record error — which must park the index, not + // silently complete. (Transient ERR_BUSY/ERR_TRY_AGAIN errors are retried instead; see the + // next test.) resetDatabases(); Tbl = table({ table: TABLE2, @@ -117,11 +119,9 @@ describe('schema-migration fragility: silent gaps when per-record indexing error let opCount = 0; tagIndex.put = function (indexedValue, primaryKey, options) { opCount++; - // reject every 10th index put with a simulated transient error + // reject every 10th index put with a permanent (non-retryable) error if (opCount % 10 === 0) { - return Promise.reject( - Object.assign(new Error('simulated transient index put failure'), { code: 'ERR_BUSY' }) - ); + return Promise.reject(new Error('simulated permanent index put failure')); } return origPut(indexedValue, primaryKey, options); }; @@ -178,6 +178,142 @@ describe('schema-migration fragility: silent gaps when per-record indexing error } assert.equal(viaIndex, N, `After restart-triggered retry, all ${N} rows should be indexed. Got ${viaIndex}.`); }); + + it('retries transient ERR_BUSY index write errors instead of parking the index', async () => { + // Same fragility setup, but the injected error is a TRANSIENT ERR_BUSY (RocksDB write + // contention). The backfill should retry it with bounded backoff and complete cleanly in a + // single pass — no park, no 503, no restart needed. + const TABLE3 = TABLE + 'WithTransientIndex'; + resetDatabases(); + let Tbl = table({ + table: TABLE3, + database: DB, + attributes: [{ name: 'id', isPrimaryKey: true }, { name: 'tag' }], + }); + let last; + const VALUES = ['alpha', 'beta', 'gamma']; + for (let i = 0; i < N; i++) { + last = Tbl.put({ id: 't3-' + i, tag: VALUES[i % VALUES.length] }); + } + await last; + + resetDatabases(); + Tbl = table({ + table: TABLE3, + database: DB, + attributes: [ + { name: 'id', isPrimaryKey: true }, + { name: 'tag', indexed: true }, + ], + }); + const tagIndex = Tbl.indices?.tag; + let injected = 0; + const failedKeys = new Set(); + if (tagIndex && typeof tagIndex.put === 'function') { + const origPut = tagIndex.put.bind(tagIndex); + tagIndex.put = function (indexedValue, primaryKey, options) { + // Fail a handful of keys exactly once with a TRANSIENT ERR_BUSY — alternating a + // SYNCHRONOUS throw (the RocksIndexStore putSync path) and an async rejection so both + // retry branches are exercised. On the in-process retry pass the key is no longer in + // the fail set, so the put succeeds and the backfill completes cleanly (no park, no 503). + const targeted = /-(?:4|14|24|34|44)$/.test(String(primaryKey)); + if (targeted && !failedKeys.has(primaryKey)) { + failedKeys.add(primaryKey); + injected++; + const err = Object.assign(new Error('simulated transient index put failure'), { code: 'ERR_BUSY' }); + if (injected % 2 === 1) throw err; + return Promise.reject(err); + } + return origPut(indexedValue, primaryKey, options); + }; + } + if (Tbl.indexingOperation) await Tbl.indexingOperation; + + assert.ok(injected > 0, 'expected the test to inject at least one transient ERR_BUSY'); + + // The retry should have completed the backfill cleanly: search must NOT throw 503 (would throw + // if the index were parked), and every row must be present in the new index. + let total = 0; + for (const v of VALUES) { + const rows = await collect(Tbl.search({ conditions: [{ attribute: 'tag', value: v }] })); + total += rows.length; + } + assert.equal(total, N, `transient errors should be retried and all ${N} rows indexed; got ${total}`); + }); + + it('retried transient errors re-cover rows that the intra-pass checkpoint advanced past (no gap)', async () => { + // Regression guard for the failure mode that sank the per-put retry approach: the intra-pass + // checkpoint advances `lastIndexedKey` every 100 rows, even past a put that later fails. If a + // retry resumed from that checkpoint it would skip the failed early row, leaving a silent gap. + // The pass-level retry instead re-runs from the pass start, re-reading every row. + const TABLE4 = TABLE + 'CheckpointGap'; + const M = 250; // > 100 so the checkpoint fires and advances past the early failing row + resetDatabases(); + let Tbl = table({ + table: TABLE4, + database: DB, + attributes: [{ name: 'id', isPrimaryKey: true }, { name: 'tag' }], + }); + let last; + const VALUES = ['alpha', 'beta', 'gamma']; + for (let i = 0; i < M; i++) { + // zero-padded so lexicographic key order matches numeric order and the targeted row is + // reliably scanned early (before the first 100-row checkpoint). + last = Tbl.put({ id: 't4-' + String(i).padStart(3, '0'), tag: VALUES[i % VALUES.length] }); + } + await last; + + resetDatabases(); + Tbl = table({ + table: TABLE4, + database: DB, + attributes: [ + { name: 'id', isPrimaryKey: true }, + { name: 'tag', indexed: true }, + ], + }); + const tagIndex = Tbl.indices?.tag; + let injected = 0; + const failedKeys = new Set(); + if (tagIndex && typeof tagIndex.put === 'function') { + const origPut = tagIndex.put.bind(tagIndex); + tagIndex.put = function (indexedValue, primaryKey, options) { + // Fail one EARLY row once (transiently). The checkpoint at row 100/200 advances the + // resume point well past it, so only a from-pass-start retry can re-cover it. + if (String(primaryKey).endsWith('t4-005') && !failedKeys.has(primaryKey)) { + failedKeys.add(primaryKey); + injected++; + return Promise.reject( + Object.assign(new Error('simulated transient index put failure'), { code: 'ERR_BUSY' }) + ); + } + return origPut(indexedValue, primaryKey, options); + }; + } + if (Tbl.indexingOperation) await Tbl.indexingOperation; + + assert.ok(injected > 0, 'expected the test to inject a transient ERR_BUSY on the early row'); + + let total = 0; + for (const v of VALUES) { + const rows = await collect(Tbl.search({ conditions: [{ attribute: 'tag', value: v }] })); + total += rows.length; + } + assert.equal( + total, + M, + `all ${M} rows must be indexed after retry — a resume-from-checkpoint retry would skip the ` + + `early failed row, leaving a gap. Got ${total}.` + ); + // And the specific early row must be findable by its indexed value. + const earlyRows = await collect( + Tbl.search({ conditions: [{ attribute: 'tag', value: VALUES[5 % VALUES.length] }] }) + ); + assert.ok( + earlyRows.some((r) => r.id === 't4-005'), + 'the early row that failed transiently must be present in the index after the retry' + ); + }); }); describe('schema-migration fragility: stale index entry from concurrent write during reindex (F3)', () => {