fix(databases): retry transient index-backfill errors in-process instead of parking

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<TableResourceType>(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<unknown>[] = [];
+			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

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)', () => {