From 0b9f259827e1df05b2e2cd6baca722a523272bf6 Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Thu, 5 Nov 2026 11:49:25 -0600 Subject: [PATCH 1/4] fix(search): queue a durable retry when startup search-index backfill fails A failed startup lexical or semantic backfill only logged a warning. The index then stayed behind until the next ingest or restart marked the scope dirty, so an idle source could stay unsearchable indefinitely. On failure, startup now marks every extant stream of the affected connector instances dirty in search_index_dirty, in one transaction per instance and under the instance deletion fence. The existing reconcile loop retries them with its backoff. A scope that is already dirty or parked keeps its attempt count and backoff; a clean scope starts fresh. Tombstoned instances and instances owned by another connector get no ticket. One instance that cannot be queued no longer blocks its siblings. Also move the semantic "rebuild completed" log after the final transaction commits, and reword the non-participating-stream cleanup log so it no longer claims stale data was dropped. Assisted-by: AI Signed-off-by: Tim Nunamaker --- .../scripts/postgres-template-eligibility.ts | 2 + .../scripts/run-tests-resource-policy.test.ts | 5 +- reference-implementation/server/index.ts | 22 +- .../server/queries/index.ts | 3 + .../index-dirty/mark-startup-failure.sql | 12 + .../index-dirty/startup-index-instances.sql | 8 + .../search/index-dirty/startup-streams.sql | 9 + .../server/search-semantic.ts | 12 +- reference-implementation/server/search.ts | 2 +- .../server/startup-search-index-retry.ts | 46 ++ .../server/stores/search-index-dirty-store.ts | 102 ++++- .../test/helpers/close-test-server.ts | 2 +- ...ill-catches-crash-abandoned-record.test.ts | 125 +++++ .../startup-backfill-durable-retry.test.ts | 427 ++++++++++++++++++ 14 files changed, 764 insertions(+), 13 deletions(-) create mode 100644 reference-implementation/server/queries/search/index-dirty/mark-startup-failure.sql create mode 100644 reference-implementation/server/queries/search/index-dirty/startup-index-instances.sql create mode 100644 reference-implementation/server/queries/search/index-dirty/startup-streams.sql create mode 100644 reference-implementation/server/startup-search-index-retry.ts create mode 100644 reference-implementation/test/startup-backfill-durable-retry.test.ts diff --git a/reference-implementation/scripts/postgres-template-eligibility.ts b/reference-implementation/scripts/postgres-template-eligibility.ts index 3e82b9010..1ea2a68f8 100644 --- a/reference-implementation/scripts/postgres-template-eligibility.ts +++ b/reference-implementation/scripts/postgres-template-eligibility.ts @@ -245,6 +245,8 @@ export const POSTGRES_TEMPLATE_COLD_REQUIRED_FILES: readonly string[] = [ "test/semantic-index-skip-unchanged-postgres.test.ts", // Exercises additive semantic input-identity migration and real publication. "test/semantic-startup-backfill-catches-crash-abandoned-record.test.ts", + // Starts the actual server over retained state. + "test/startup-backfill-durable-retry.test.ts", "test/spine-events-connector-instance-id-backfill.test.ts", "test/spine-source-boot-backfill.test.ts", ]; diff --git a/reference-implementation/scripts/run-tests-resource-policy.test.ts b/reference-implementation/scripts/run-tests-resource-policy.test.ts index 766f1428c..9fb1ee25d 100644 --- a/reference-implementation/scripts/run-tests-resource-policy.test.ts +++ b/reference-implementation/scripts/run-tests-resource-policy.test.ts @@ -33,8 +33,9 @@ test("only reviewed files bypass outer allocation; restore remains separate", () assert.equal(isDedicatedPostgresTestDatabaseName(restoreName), true); }); -test("all 37 cold-required entries remain cold, including helper-owned bootstrap", () => { - assert.equal(POSTGRES_TEMPLATE_COLD_REQUIRED_FILES.length, 37); +test("all 38 cold-required entries remain cold, including helper-owned bootstrap and durable startup retry", () => { + assert.equal(POSTGRES_TEMPLATE_COLD_REQUIRED_FILES.length, 38); + assert.ok(POSTGRES_TEMPLATE_COLD_REQUIRED_FILES.includes("test/startup-backfill-durable-retry.test.ts")); for (const file of POSTGRES_TEMPLATE_COLD_REQUIRED_FILES) { assert.equal(isPostgresTemplateEligibleFilePath(file), false, file); } diff --git a/reference-implementation/server/index.ts b/reference-implementation/server/index.ts index 13cb81d52..d8804ed1a 100644 --- a/reference-implementation/server/index.ts +++ b/reference-implementation/server/index.ts @@ -7,6 +7,8 @@ * Combined AS + RS implementing PDPP v0.1.0 core spec. * Starts on port 7662 (AS/introspection) and 7663 (RS query API). */ +import { queueStartupSearchIndexRetry } from "./startup-search-index-retry.ts"; +import { currentStorageGeneration } from "./storage-generation.ts"; import { createHash, randomBytes } from "node:crypto"; import { existsSync, readFileSync, renameSync } from "node:fs"; import os from "node:os"; @@ -912,6 +914,8 @@ interface ServerOpts { } | null; trustedProxies?: string | null; ownerToken?: string | null; + /** Single-use capability for synthetic retained-state restart tests. */ + testOnlyAlreadyAdmittedPostgresChildAttachment?: string; postgresBootstrapLockTimeoutMs?: number; postgresSemanticHnswMaintenanceImpl?: typeof schedulePostgresSemanticHnswMaintenance; preRegisteredPublicClients?: unknown; @@ -1286,10 +1290,12 @@ async function collectRetrievalStartupBackfillManifests({ } async function runRetrievalStartupBackfill({ + generation = currentStorageGeneration(), manifests, logger, signal = null, }: { + generation?: number; manifests: ConnectorManifest[]; logger: LoggerLike; signal?: AbortSignal | null; @@ -1311,6 +1317,7 @@ async function runRetrievalStartupBackfill({ logger.info({ connectorId }, "retrieval startup backfill connector started"); // biome-ignore lint/performance/noAwaitInLoops: Work is intentionally sequential to preserve ordering and state transitions. await lexicalIndexBackfillForManifest({ + generation, log: (msg) => logger.info(msg), manifest: manifest as never, signal, @@ -1318,6 +1325,7 @@ async function runRetrievalStartupBackfill({ const semanticBackend = getSemanticBackend(); if (semanticBackend?.available()) { await semanticIndexBackfillForManifest({ + generation, log: (msg) => logger.info(msg), manifest: manifest as never, signal, @@ -1332,6 +1340,11 @@ async function runRetrievalStartupBackfill({ return; } logger.warn({ connectorId, err }, "retrieval startup backfill failed for connector"); + try { + await queueStartupSearchIndexRetry(manifest as never, generation); + } catch (queueError) { + logger.warn({ connectorId, err, queueError }, "retrieval startup backfill retry queue write failed"); + } } } @@ -1342,10 +1355,12 @@ async function runRetrievalStartupBackfill({ } function scheduleRetrievalStartupBackfill({ + generation = currentStorageGeneration(), manifests, logger, signal = null, }: { + generation?: number; manifests: ConnectorManifest[]; logger: LoggerLike; signal?: AbortSignal | null; @@ -1357,7 +1372,7 @@ function scheduleRetrievalStartupBackfill({ logger.info({ connectorCount: manifests.length }, "retrieval startup backfill scheduled after AS/RS listen"); return new Promise((resolve) => setImmediate(resolve)) - .then(() => runRetrievalStartupBackfill({ logger, manifests, signal })) + .then(() => runRetrievalStartupBackfill({ generation, logger, manifests, signal })) .catch((err) => { // Abort-driven exits travel through this catch when the loop // re-throws an AbortError-like value before reaching the inner @@ -8744,6 +8759,9 @@ export async function startServer(opts: ServerOpts = {}) { ...(opts.postgresBootstrapLockTimeoutMs === undefined ? {} : { bootstrapLockTimeoutMs: opts.postgresBootstrapLockTimeoutMs }), + ...(opts.testOnlyAlreadyAdmittedPostgresChildAttachment === undefined + ? {} + : { testOnlyAlreadyAdmittedChildAttachment: opts.testOnlyAlreadyAdmittedPostgresChildAttachment }), log: (msg: string) => logger.info(msg), }); const uninstallRecoveryStore = createConnectorInstallStore(); @@ -9685,6 +9703,7 @@ export async function startServer(opts: ServerOpts = {}) { }); await schedulerManager.start(); const startupBackfillAbortController = new AbortController(); + const startupBackfillGeneration = currentStorageGeneration(); const startupBackfillDone = manifestReconciliationDone .then(() => collectRetrievalStartupBackfillManifests({ @@ -9694,6 +9713,7 @@ export async function startServer(opts: ServerOpts = {}) { ) .then((manifests) => scheduleRetrievalStartupBackfill({ + generation: startupBackfillGeneration, logger, manifests, signal: startupBackfillAbortController.signal, diff --git a/reference-implementation/server/queries/index.ts b/reference-implementation/server/queries/index.ts index dbb5fe543..9ca30019d 100644 --- a/reference-implementation/server/queries/index.ts +++ b/reference-implementation/server/queries/index.ts @@ -476,10 +476,13 @@ export interface ReferenceQueryRegistry extends Readonly ?) + AND NOT EXISTS (SELECT 1 FROM connector_instance_tombstones WHERE connector_instance_id = ?) +ON CONFLICT(connector_instance_id, stream) DO UPDATE SET + connector_id = excluded.connector_id, + marked_at = CASE WHEN search_index_dirty.dirty <> 0 THEN search_index_dirty.marked_at ELSE excluded.marked_at END, + attempts = CASE WHEN search_index_dirty.dirty <> 0 OR search_index_dirty.operator_parked <> 0 THEN search_index_dirty.attempts ELSE 0 END, + next_attempt_at = CASE WHEN search_index_dirty.dirty <> 0 OR search_index_dirty.operator_parked <> 0 THEN search_index_dirty.next_attempt_at ELSE NULL END, + dirty = 1, + revision = search_index_dirty.revision + 1 diff --git a/reference-implementation/server/queries/search/index-dirty/startup-index-instances.sql b/reference-implementation/server/queries/search/index-dirty/startup-index-instances.sql new file mode 100644 index 000000000..69d26882a --- /dev/null +++ b/reference-implementation/server/queries/search/index-dirty/startup-index-instances.sql @@ -0,0 +1,8 @@ +-- @terminator: iterate +-- @cursor_field: connector_instance_id +SELECT connector_instance_id FROM lexical_search_meta WHERE connector_id = ? +UNION SELECT connector_instance_id FROM lexical_search_index WHERE connector_id = ? +UNION SELECT connector_instance_id FROM semantic_search_meta WHERE connector_id = ? +UNION SELECT connector_instance_id FROM semantic_search_backfill_progress WHERE connector_id = ? +UNION SELECT connector_instance_id FROM semantic_search_blob WHERE connector_id = ? +UNION SELECT connector_instance_id FROM semantic_search_rowid WHERE connector_id = ? diff --git a/reference-implementation/server/queries/search/index-dirty/startup-streams.sql b/reference-implementation/server/queries/search/index-dirty/startup-streams.sql new file mode 100644 index 000000000..68b627953 --- /dev/null +++ b/reference-implementation/server/queries/search/index-dirty/startup-streams.sql @@ -0,0 +1,9 @@ +-- @terminator: iterate +-- @cursor_field: stream +SELECT stream FROM records WHERE connector_instance_id = ? AND connector_id = ? +UNION SELECT stream FROM lexical_search_meta WHERE connector_instance_id = ? AND connector_id = ? +UNION SELECT stream FROM semantic_search_meta WHERE connector_instance_id = ? AND connector_id = ? +UNION SELECT stream FROM semantic_search_backfill_progress WHERE connector_instance_id = ? AND connector_id = ? +UNION SELECT stream FROM lexical_search_index WHERE connector_instance_id = ? AND connector_id = ? +UNION SELECT json_extract(scope_key, '$[0]') AS stream FROM semantic_search_blob WHERE connector_instance_id = ? AND connector_id = ? AND json_valid(scope_key) AND json_type(scope_key, '$[0]') = 'text' +UNION SELECT json_extract(scope_key, '$[0]') AS stream FROM semantic_search_rowid WHERE connector_instance_id = ? AND connector_id = ? AND json_valid(scope_key) AND json_type(scope_key, '$[0]') = 'text' diff --git a/reference-implementation/server/search-semantic.ts b/reference-implementation/server/search-semantic.ts index 404f5272a..4ecf8b64c 100644 --- a/reference-implementation/server/search-semantic.ts +++ b/reference-implementation/server/search-semantic.ts @@ -2921,7 +2921,7 @@ function deleteBackfillProgress({ // can be current, removed from the manifest, or represented only by an // interrupted backfill row; all three need the same per-instance critical // section. The caller then holds one fence per instance, never two at once. -async function resolveSemanticBackfillConnectorInstanceIds({ +export async function resolveSemanticBackfillConnectorInstanceIds({ connectorId, manifest, }: { @@ -4008,10 +4008,6 @@ async function rebuildSemanticBackfillInstance({ await semanticRepairPhase("after-verification", scope); await assertSemanticBackfillManifestCurrent(connectorId, context.manifest); assertSemanticRepairBackend(scope); - log( - `[PDPP] Semantic index rebuild completed for ${connectorId} stream='${stream}' ` + - `(records=${recordsToScan}, indexed=${indexed})` - ); await withConnectorInstanceWrite(connectorInstanceId, async () => { if (usePostgres) { await withPostgresBulkTransaction( @@ -4053,6 +4049,10 @@ async function rebuildSemanticBackfillInstance({ }); } }); + log( + `[PDPP] Semantic index rebuild completed for ${connectorId} stream='${stream}' ` + + `(records=${recordsToScan}, indexed=${indexed})` + ); return rebuildingJob; } @@ -4136,7 +4136,7 @@ async function backfillSemanticManifestStream({ if (ids.length > 0) { log( `[PDPP] Semantic index: stream='${stream}' connector='${connectorId}' ` + - "no longer declares semantic_fields — dropping stale index + meta/progress" + "checking cleanup for stream without semantic_fields" ); await deleteSemanticNonParticipatingStream({ connectorInstanceIds: ids.filter((id) => id === fencedConnectorInstanceId), diff --git a/reference-implementation/server/search.ts b/reference-implementation/server/search.ts index 33d480b1c..5489998c6 100644 --- a/reference-implementation/server/search.ts +++ b/reference-implementation/server/search.ts @@ -1412,7 +1412,7 @@ async function lexicalRecordsCountNonDeleted({ return await getSearchIndexStore().recordsCountNonDeleted({ connectorInstanceId, stream }); } -async function resolveLexicalBackfillConnectorInstanceIds({ +export async function resolveLexicalBackfillConnectorInstanceIds({ connectorId, manifest, }: { diff --git a/reference-implementation/server/startup-search-index-retry.ts b/reference-implementation/server/startup-search-index-retry.ts new file mode 100644 index 000000000..5113deb8f --- /dev/null +++ b/reference-implementation/server/startup-search-index-retry.ts @@ -0,0 +1,46 @@ +// Copyright The PDP-Connect Contributors +// SPDX-License-Identifier: Apache-2.0 +import { resolveLexicalBackfillConnectorInstanceIds } from "./search.ts"; +import { resolveSemanticBackfillConnectorInstanceIds } from "./search-semantic.ts"; +import { assertCurrentStorageGeneration } from "./storage-generation.ts"; +import { listStartupSearchIndexInstances, markStartupSearchIndexFailure } from "./stores/search-index-dirty-store.ts"; + +type Manifest = Parameters[0]["manifest"] & + Parameters[0]["manifest"]; + +/** Startup-only caller effect: reconcile failures own their revision/backoff. */ +export async function queueStartupSearchIndexRetry(manifest: Manifest, generation: number): Promise { + assertCurrentStorageGeneration(generation); + const connectorId = manifest.connector_id; + const lexical = await resolveLexicalBackfillConnectorInstanceIds({ + connectorId, + manifest, + }); + assertCurrentStorageGeneration(generation); + const semantic = await resolveSemanticBackfillConnectorInstanceIds({ + connectorId, + manifest, + }); + assertCurrentStorageGeneration(generation); + const pin = manifest.storage_binding?.connector_instance_id || manifest.connector_instance_id; + const orphanInstances = pin ? [] : await listStartupSearchIndexInstances(connectorId); + assertCurrentStorageGeneration(generation); + // One instance's failed mark must not leave its siblings without a durable + // retry, so every instance is attempted before the failures are reported. + const failures: unknown[] = []; + for (const connectorInstanceId of new Set([...lexical, ...semantic, ...orphanInstances])) { + try { + // biome-ignore lint/performance/noAwaitInLoops: Never hold more than one instance deletion fence. + await markStartupSearchIndexFailure({ + connectorId, + connectorInstanceId, + generation, + }); + } catch (error) { + failures.push(error); + } + } + if (failures.length > 0) { + throw new AggregateError(failures, `startup search-index retry could not be queued for ${connectorId}`); + } +} diff --git a/reference-implementation/server/stores/search-index-dirty-store.ts b/reference-implementation/server/stores/search-index-dirty-store.ts index f33bfa5a5..96f62ef32 100644 --- a/reference-implementation/server/stores/search-index-dirty-store.ts +++ b/reference-implementation/server/stores/search-index-dirty-store.ts @@ -22,9 +22,11 @@ * for that scope. */ -import { exec, getOne, iterate, referenceQueries } from "../../lib/db.ts"; +import { exec, getOne, iterate, referenceQueries, writeTransaction } from "../../lib/db.ts"; +import { withConnectorInstanceControlPlaneWrite } from "../connector-instance-write-coordinator.ts"; import type { PostgresTransactionClient } from "../postgres-storage.ts"; -import { isPostgresStorageBackend, postgresQuery } from "../postgres-storage.ts"; +import { isPostgresStorageBackend, postgresQuery, withPostgresTransaction } from "../postgres-storage.ts"; +import { assertCurrentStorageGeneration } from "../storage-generation.ts"; interface SqlClient { query: (sql: string, params?: unknown[]) => Promise<{ rows: unknown[] }>; @@ -345,3 +347,99 @@ export async function countDirtySearchIndexScopes(): Promise { const row = getOne<{ n: number }>(referenceQueries.searchIndexDirtyCountDirty, []); return row?.n ?? 0; } + +/** Startup failures preserve retry policy; ingest alone resets changed-input history. + * Discover extant streams inside the same instance fence as deletion. No instance + * row is required for legacy direct ingest, but retained/index state is required. + */ +export async function markStartupSearchIndexFailure({ + connectorId, + connectorInstanceId, + generation, +}: { + connectorId: string; + connectorInstanceId: string; + generation: number; +}): Promise { + await withConnectorInstanceControlPlaneWrite(connectorInstanceId, async () => { + assertCurrentStorageGeneration(generation); + const now = new Date().toISOString(); + if (isPostgresStorageBackend()) { + await withPostgresTransaction( + async (client) => { + assertCurrentStorageGeneration(generation); + const extant = await client.query<{ stream: string }>( + `SELECT stream FROM records WHERE connector_instance_id = $1 AND connector_id = $2 + UNION SELECT stream FROM lexical_search_meta WHERE connector_instance_id = $1 AND connector_id = $2 + UNION SELECT stream FROM semantic_search_meta WHERE connector_instance_id = $1 AND connector_id = $2 + UNION SELECT stream FROM semantic_search_backfill_progress WHERE connector_instance_id = $1 AND connector_id = $2 + UNION SELECT stream FROM lexical_search_index WHERE connector_instance_id = $1 AND connector_id = $2 + UNION SELECT (scope_key::jsonb)->>0 AS stream FROM semantic_search_blob WHERE connector_instance_id = $1 AND connector_id = $2`, + [connectorInstanceId, connectorId] + ); + for (const { stream } of extant.rows) { + // Extant orphan streams must also be repaired even after manifest removal. + assertCurrentStorageGeneration(generation); + // biome-ignore lint/performance/noAwaitInLoops: One connection's stream batch shares its deletion fence and transaction. + await client.query( + `INSERT INTO search_index_dirty(connector_instance_id, connector_id, stream, dirty, marked_at, revision, operator_parked) + SELECT $1, $2, $3, 1, $4, 1, 0 + WHERE NOT EXISTS (SELECT 1 FROM connector_instances WHERE connector_instance_id = $1 AND connector_id <> $2) + AND NOT EXISTS (SELECT 1 FROM connector_instance_tombstones WHERE connector_instance_id = $1) + ON CONFLICT(connector_instance_id, stream) DO UPDATE SET + connector_id = excluded.connector_id, + marked_at = CASE WHEN search_index_dirty.dirty <> 0 THEN search_index_dirty.marked_at ELSE excluded.marked_at END, + attempts = CASE WHEN search_index_dirty.dirty <> 0 OR search_index_dirty.operator_parked <> 0 THEN search_index_dirty.attempts ELSE 0 END, + next_attempt_at = CASE WHEN search_index_dirty.dirty <> 0 OR search_index_dirty.operator_parked <> 0 THEN search_index_dirty.next_attempt_at ELSE NULL END, + dirty = 1, revision = search_index_dirty.revision + 1`, + [connectorInstanceId, connectorId, stream, now] + ); + } + assertCurrentStorageGeneration(generation); + }, + { lockConnectorInstanceId: connectorInstanceId } + ); + return; + } + writeTransaction(() => { + assertCurrentStorageGeneration(generation); + for (const { stream } of [ + ...iterate<{ stream: string }>( + referenceQueries.searchIndexDirtyStartupStreams, + Array.from({ length: 7 }, () => [connectorInstanceId, connectorId]).flat() + ), + ]) { + exec(referenceQueries.searchIndexDirtyMarkStartupFailure, [ + connectorInstanceId, + connectorId, + stream, + now, + connectorInstanceId, + connectorId, + connectorInstanceId, + ]); + } + }); + }); +} + +/** Supplement backfill discovery with orphan index instances, still subject to manifest pins. */ +export async function listStartupSearchIndexInstances(connectorId: string): Promise { + if (isPostgresStorageBackend()) { + const result = await postgresQuery<{ connector_instance_id: string }>( + `SELECT connector_instance_id FROM lexical_search_meta WHERE connector_id = $1 + UNION SELECT connector_instance_id FROM lexical_search_index WHERE connector_id = $1 + UNION SELECT connector_instance_id FROM semantic_search_meta WHERE connector_id = $1 + UNION SELECT connector_instance_id FROM semantic_search_backfill_progress WHERE connector_id = $1 + UNION SELECT connector_instance_id FROM semantic_search_blob WHERE connector_id = $1`, + [connectorId] + ); + return result.rows.map((row) => row.connector_instance_id); + } + return [ + ...iterate<{ connector_instance_id: string }>( + referenceQueries.searchIndexDirtyStartupIndexInstances, + Array.from({ length: 6 }, () => connectorId) + ), + ].map((row) => row.connector_instance_id); +} diff --git a/reference-implementation/test/helpers/close-test-server.ts b/reference-implementation/test/helpers/close-test-server.ts index 12f5bba5f..bf1c0984c 100644 --- a/reference-implementation/test/helpers/close-test-server.ts +++ b/reference-implementation/test/helpers/close-test-server.ts @@ -78,7 +78,7 @@ function flattenErrors(error: unknown): unknown[] { return error instanceof AggregateError ? error.errors.flatMap(flattenErrors) : [error]; } -async function closeHttpServer(server: HttpServer): Promise { +export async function closeHttpServer(server: HttpServer): Promise { const errors: unknown[] = []; for (const closeConnections of [server.closeIdleConnections, server.closeAllConnections]) { if (!closeConnections) continue; diff --git a/reference-implementation/test/semantic-startup-backfill-catches-crash-abandoned-record.test.ts b/reference-implementation/test/semantic-startup-backfill-catches-crash-abandoned-record.test.ts index 9467cde37..de78af577 100644 --- a/reference-implementation/test/semantic-startup-backfill-catches-crash-abandoned-record.test.ts +++ b/reference-implementation/test/semantic-startup-backfill-catches-crash-abandoned-record.test.ts @@ -448,6 +448,13 @@ for (const kind of ["blob-flat", "sqlite-vec", "postgres"] as const) { orphanStreamCleanup: false, }); } + async function repairWithLog(stream: string, log: (message: string) => void) { + await semanticIndexBackfillForManifest({ + log, + manifest: pinned(stream), + orphanStreamCleanup: false, + }); + } async function rows(stream = "items") { return sql( `SELECT scope_key, record_key, input_identity FROM ${table} WHERE connector_instance_id = $1 AND scope_key = $2 ORDER BY record_key`, @@ -505,6 +512,7 @@ for (const kind of ["blob-flat", "sqlite-vec", "postgres"] as const) { vector, writes, pinned, + repairWithLog, childAttachment, }; } @@ -661,6 +669,123 @@ for (const kind of ["blob-flat", "sqlite-vec", "postgres"] as const) { }) ); + await t.test( + "nonparticipating zero-state repeated repairs report cleanup attempts without claiming stale data was dropped", + async () => + fixture(async (f) => { + await f.ingest("k1", "ordinary record without semantic fields"); + const noFields = { + ...f.manifest, + streams: f.manifest.streams + .filter((stream) => stream.name === "items") + .map((stream) => ({ ...stream, query: { search: {} } })), + }; + await registerConnector(noFields, { backfillRetrievalIndexes: false }); + const pinnedNoFields = { ...noFields, connector_instance_id: f.instance }; + const logs: string[] = []; + await semanticIndexBackfillForManifest({ + log: (message) => logs.push(message), + manifest: pinnedNoFields, + orphanStreamCleanup: false, + }); + await semanticIndexBackfillForManifest({ + log: (message) => logs.push(message), + manifest: pinnedNoFields, + orphanStreamCleanup: false, + }); + assert.equal(logs.length, 2, "each repair says it checked cleanup"); + assert.ok(logs.every((message) => message.includes("checking cleanup"))); + assert.ok(logs.every((message) => !message.includes("dropping stale"))); + assert.deepEqual(await f.rows(), [], "no semantic vectors are created for a nonparticipating stream"); + assert.equal( + ( + await f.sql("SELECT 1 FROM semantic_search_meta WHERE connector_instance_id = $1 AND stream = 'items'", [ + f.instance, + ]) + ).length, + 0, + "zero-state cleanup does not create completion metadata" + ); + assert.equal( + ( + await f.sql( + "SELECT 1 FROM semantic_search_backfill_progress WHERE connector_instance_id = $1 AND stream = 'items'", + [f.instance] + ) + ).length, + 0, + "zero-state cleanup does not create progress" + ); + }) + ); + + await t.test("completion log follows the final metadata and progress transaction", async () => + fixture(async (f) => { + await f.ingest("k1", "completion transaction fault"); + const completionLogs: string[] = []; + const failTrigger = "fail_semantic_completion_meta"; + try { + if (kind === "postgres") { + await f.sql( + "CREATE OR REPLACE FUNCTION semantic_test_fail_completion() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'completion metadata failure'; END $$" + ); + await f.sql( + `CREATE TRIGGER ${failTrigger} BEFORE INSERT OR UPDATE ON semantic_search_meta FOR EACH ROW EXECUTE FUNCTION semantic_test_fail_completion()` + ); + } else { + getDb().exec( + `CREATE TRIGGER ${failTrigger} BEFORE INSERT ON semantic_search_meta BEGIN SELECT RAISE(ABORT, 'completion metadata failure'); END;` + ); + } + await assert.rejects( + f.repairWithLog("items", (message) => { + if (message.includes("Semantic index rebuild completed")) { + completionLogs.push(message); + } + }), + (error) => error instanceof Error && error.message.includes("completion metadata failure") + ); + assert.equal(completionLogs.length, 0, "a failed final transaction cannot announce completion"); + assert.equal( + ( + await f.sql( + "SELECT 1 FROM semantic_search_backfill_progress WHERE connector_instance_id = $1 AND stream = 'items'", + [f.instance] + ) + ).length, + 1, + "the failed final transaction retains progress for retry" + ); + } finally { + if (kind === "postgres") { + await f.sql(`DROP TRIGGER IF EXISTS ${failTrigger} ON semantic_search_meta`); + await f.sql("DROP FUNCTION IF EXISTS semantic_test_fail_completion()"); + } else { + getDb().exec(`DROP TRIGGER IF EXISTS ${failTrigger}`); + } + } + await f.repairWithLog("items", (message) => { + if (message.includes("Semantic index rebuild completed")) { + completionLogs.push(message); + } + }); + assert.equal(completionLogs.length, 1, "successful retry emits one completion event"); + assert.equal( + ( + await f.sql( + "SELECT 1 FROM semantic_search_backfill_progress WHERE connector_instance_id = $1 AND stream = 'items'", + [f.instance] + ) + ).length, + 0, + "successful retry commits progress deletion" + ); + const calls = f.inputs.length; + await f.repair("items"); + assert.equal(f.inputs.length, calls, "unchanged second pass reuses committed input identities"); + }) + ); + await t.test( "legacy NULL trailing orphan/removed field/empty canonical stream are deleted, with wildcard and other-instance siblings intact", async () => diff --git a/reference-implementation/test/startup-backfill-durable-retry.test.ts b/reference-implementation/test/startup-backfill-durable-retry.test.ts new file mode 100644 index 000000000..72074de86 --- /dev/null +++ b/reference-implementation/test/startup-backfill-durable-retry.test.ts @@ -0,0 +1,427 @@ +// Copyright The PDP-Connect Contributors +// SPDX-License-Identifier: Apache-2.0 +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { registerConnector } from "../server/auth.ts"; +import { withConnectorInstanceControlPlaneWrite } from "../server/connector-instance-write-coordinator.ts"; +import { closeDb, getDb, initDb } from "../server/db.ts"; +import { startServer } from "../server/index.ts"; +import { LocalTransformerExecutorError } from "../server/local-transformer-executor.ts"; +import { closePostgresStorage, initPostgresStorage, postgresQuery } from "../server/postgres-storage.ts"; +import { createAlreadyAdmittedTestDatabaseChildAttachment } from "../server/postgres-test-database-guard.ts"; +import { ingestRecord } from "../server/records.ts"; +import { runSearchIndexDirtyReconcileRound } from "../server/search-index-reconcile.ts"; +import { computeSemanticEntries, makeStubBackend } from "../server/search-semantic.ts"; +import { queueStartupSearchIndexRetry } from "../server/startup-search-index-retry.ts"; +import { bumpStorageGeneration, currentStorageGeneration } from "../server/storage-generation.ts"; +import { + clearSearchIndexDirty, + listDirtySearchIndexScopes, + markStartupSearchIndexFailure, +} from "../server/stores/search-index-dirty-store.ts"; +import { closeTestServer } from "./helpers/close-test-server.ts"; + +const STALE_GENERATION = /storage generation changed/; + +function required(value: T | undefined): T { + assert.ok(value); + return value; +} + +const manifest = { + connector_id: "startup-retry", + display_name: "Startup retry", + protocol_version: "0.1.0", + streams: [ + { + name: "items", + primary_key: ["id"], + query: { search: { semantic_fields: ["subject"] } }, + schema: { + properties: { id: { type: "string" }, subject: { type: "string" } }, + required: ["id", "subject"], + type: "object", + }, + selection: { fields: true, resources: true }, + semantics: "mutable_state", + }, + ], + version: "1.0.0", + manifest_uri: "https://example.test/startup-retry", +}; + +test("actual startup timeout queues retained work and maintenance converges without ingest", async () => { + const dir = mkdtempSync(join(tmpdir(), "startup-retry-")); + const dbPath = join(dir, "db.sqlite"); + let healthy = false; + let failedCalls = 0; + const stub = makeStubBackend(); + const backend = { + ...stub, + embedDocument: async (text: string) => { + if (!healthy) { + failedCalls += 1; + throw new LocalTransformerExecutorError("transformer_deadline"); + } + return await stub.embedDocument(text); + }, + }; + const pg = Boolean(process.env.PDPP_TEST_POSTGRES_URL); + if (pg) { + await initPostgresStorage({ + backend: "postgres", + databaseUrl: process.env.PDPP_TEST_POSTGRES_URL || "", + }); + } else { + initDb(dbPath); + } + const attachment = pg + ? await createAlreadyAdmittedTestDatabaseChildAttachment(process.env.PDPP_TEST_POSTGRES_URL || "") + : undefined; + await registerConnector(manifest, { backfillRetrievalIndexes: false }); + await ingestRecord( + { + connector_id: manifest.connector_id, + connector_instance_id: "cin_startup_retry", + }, + { + data: { id: "k1", subject: "retained synthetic subject" }, + emitted_at: "2026-07-16T00:00:00.000Z", + key: "k1", + stream: "items", + }, + { deferIndexes: true } + ); + if (pg) { + await postgresQuery("DELETE FROM search_index_dirty"); + } else { + getDb().prepare("DELETE FROM search_index_dirty").run(); + closeDb(); + } + const server = await startServer({ + asPort: 0, + dbPath, + rsPort: 0, + storageBackend: pg ? "postgres" : "sqlite", + ...(pg + ? { + databaseUrl: process.env.PDPP_TEST_POSTGRES_URL || "", + testOnlyAlreadyAdmittedPostgresChildAttachment: attachment || "", + } + : {}), + quiet: true, + reconcilePolyfillManifests: false, + semanticRetrievalBackend: backend, + }); + try { + await server.startupBackfillDone; + assert.ok(failedCalls > 0, "real startup attempted embedding and failed"); + const tickets = await listDirtySearchIndexScopes(25); + assert.ok( + tickets.some((t) => t.connectorInstanceId === "cin_startup_retry" && t.stream === "items"), + "startup failure must create durable retry ticket" + ); + healthy = true; + const round = await runSearchIndexDirtyReconcileRound(); + assert.equal(round.succeeded, 1); + assert.equal((await listDirtySearchIndexScopes(25)).length, 0); + const computed = await computeSemanticEntries({ + connectorId: manifest.connector_id, + connectorInstanceId: "cin_startup_retry", + data: { id: "k1", subject: "retained synthetic subject" }, + recordKey: "k1", + stream: "items", + }); + assert.ok(computed); + const rows = pg + ? ( + await postgresQuery( + "SELECT record_key, scope_key, input_identity FROM semantic_search_blob WHERE connector_instance_id = $1", + ["cin_startup_retry"] + ) + ).rows + : getDb() + .prepare( + "SELECT record_key, scope_key, input_identity FROM semantic_search_rowid WHERE connector_instance_id = ? UNION ALL SELECT record_key, scope_key, input_identity FROM semantic_search_blob WHERE connector_instance_id = ?" + ) + .all("cin_startup_retry", "cin_startup_retry"); + assert.deepEqual( + rows.map((r) => ({ ...r })), + [...computed.entries, ...computed.retained].map((e) => ({ + input_identity: e.inputIdentity, + record_key: "k1", + scope_key: e.scopeKey, + })) + ); + } finally { + await closeTestServer(server); + if (pg) { + await postgresQuery("DELETE FROM records"); + await postgresQuery("DELETE FROM search_index_dirty"); + await closePostgresStorage(); + } else { + closeDb(); + } + rmSync(dir, { force: true, recursive: true }); + } +}); + +test("startup retry preserves history and parks, fences clears, respects pins and deleted state", async () => { + const pg = Boolean(process.env.PDPP_TEST_POSTGRES_URL); + if (pg) { + await initPostgresStorage({ + backend: "postgres", + databaseUrl: process.env.PDPP_TEST_POSTGRES_URL || "", + }); + } else { + initDb(":memory:"); + } + const sql = async (query: string, args: unknown[] = []) => { + if (pg) { + let n = 0; + return ( + await postgresQuery( + query.replaceAll("?", () => { + n += 1; + return `$${n}`; + }), + args + ) + ).rows; + } + const stmt = getDb().prepare(query); + if (query.startsWith("SELECT")) { + return stmt.all(...args); + } + stmt.run(...args); + return []; + }; + try { + await registerConnector(manifest, { backfillRetrievalIndexes: false }); + for await (const instance of ["cin_retry_control", "cin_retry_other"]) { + await ingestRecord( + { + connector_id: manifest.connector_id, + connector_instance_id: instance, + }, + { + data: { id: "k1", subject: "synthetic control" }, + emitted_at: "2026-07-16T00:00:00.000Z", + key: "k1", + stream: "items", + }, + { deferIndexes: true } + ); + } + const key = { connectorInstanceId: "cin_retry_control", stream: "items" }; + await sql( + "UPDATE search_index_dirty SET attempts = 7, next_attempt_at = ?, marked_at = ? WHERE connector_instance_id = ?", + ["2099-01-01T00:00:00.000Z", "2020-01-01T00:00:00.000Z", key.connectorInstanceId] + ); + const read = async () => + required( + (await sql("SELECT * FROM search_index_dirty WHERE connector_instance_id = ?", [key.connectorInstanceId]))[0] + ); + const otherBefore = await sql("SELECT * FROM search_index_dirty WHERE connector_instance_id = ?", [ + "cin_retry_other", + ]); + const before = await read(); + const generation = currentStorageGeneration(); + await queueStartupSearchIndexRetry( + { + ...manifest, + storage_binding: { connector_instance_id: key.connectorInstanceId }, + } as never, + generation + ); + const after = await read(); + assert.equal(Number(after.revision), Number(before.revision) + 1); + for (const field of ["attempts", "next_attempt_at", "marked_at", "operator_parked"]) { + assert.equal(after[field], before[field]); + } + assert.deepEqual( + await sql("SELECT * FROM search_index_dirty WHERE connector_instance_id = ?", ["cin_retry_other"]), + otherBefore + ); + assert.equal(await clearSearchIndexDirty(key, Number(before.revision), new Date().toISOString()), false); + await sql("UPDATE search_index_dirty SET operator_parked = 1 WHERE connector_instance_id = ?", [ + key.connectorInstanceId, + ]); + await queueStartupSearchIndexRetry( + { ...manifest, connector_instance_id: key.connectorInstanceId } as never, + generation + ); + assert.equal(Number((await read()).operator_parked), 1); + assert.equal(Number((await read()).attempts), 7); + await sql("UPDATE search_index_dirty SET dirty = 0, operator_parked = 0 WHERE connector_instance_id = ?", [ + key.connectorInstanceId, + ]); + await queueStartupSearchIndexRetry( + { ...manifest, connector_instance_id: key.connectorInstanceId } as never, + generation + ); + assert.equal(Number((await read()).dirty), 1); + assert.equal(Number((await read()).attempts), 0); + assert.equal((await read()).next_attempt_at, null); + await sql("DELETE FROM records WHERE connector_instance_id = ?", [key.connectorInstanceId]); + await sql("DELETE FROM search_index_dirty WHERE connector_instance_id = ?", [key.connectorInstanceId]); + await markStartupSearchIndexFailure({ + connectorId: manifest.connector_id, + connectorInstanceId: key.connectorInstanceId, + generation, + }); + assert.equal( + (await sql("SELECT * FROM search_index_dirty WHERE connector_instance_id = ?", [key.connectorInstanceId])).length, + 0, + "deleted scope must not resurrect" + ); + // Deletion commits its tombstone before asynchronous projection teardown. + // Another PostgreSQL process can acquire the transaction fence in between. + await sql( + "INSERT INTO lexical_search_meta(connector_id, connector_instance_id, stream, fields_fingerprint, updated_at) VALUES (?, ?, ?, ?, ?)", + [manifest.connector_id, key.connectorInstanceId, key.stream, "residual-projection", "2026-11-03T00:00:00.000Z"] + ); + await markStartupSearchIndexFailure({ + connectorId: manifest.connector_id, + connectorInstanceId: key.connectorInstanceId, + generation, + }); + assert.equal( + (await sql("SELECT * FROM search_index_dirty WHERE connector_instance_id = ?", [key.connectorInstanceId])).length, + 1, + "legacy instance-less index state still needs a retry when not deleted" + ); + await sql("DELETE FROM search_index_dirty WHERE connector_instance_id = ?", [key.connectorInstanceId]); + await sql( + "INSERT INTO connector_instance_tombstones(connector_instance_id, owner_subject_id, connector_id, source_kind, source_binding_key, deleted_at) VALUES (?, ?, ?, ?, ?, ?)", + [ + key.connectorInstanceId, + "synthetic-owner", + manifest.connector_id, + "synthetic", + "deleted", + "2026-11-03T00:00:00.000Z", + ] + ); + await markStartupSearchIndexFailure({ + connectorId: manifest.connector_id, + connectorInstanceId: key.connectorInstanceId, + generation, + }); + assert.equal( + (await sql("SELECT * FROM search_index_dirty WHERE connector_instance_id = ?", [key.connectorInstanceId])).length, + 0, + "deleted connection must not regain a retry ticket from residual index state" + ); + let unlock: () => void = () => { + /* assigned synchronously below */ + }; + let entered: () => void = () => { + /* assigned synchronously below */ + }; + const held = new Promise((resolve) => { + entered = resolve; + }); + const release = new Promise((resolve) => { + unlock = resolve; + }); + const fence = withConnectorInstanceControlPlaneWrite("cin_retry_other", async () => { + entered(); + await release; + }); + await held; + const delayed = markStartupSearchIndexFailure({ + connectorId: manifest.connector_id, + connectorInstanceId: "cin_retry_other", + generation, + }); + bumpStorageGeneration(); + unlock(); + await fence; + await assert.rejects(delayed, STALE_GENERATION); + await assert.rejects(queueStartupSearchIndexRetry(manifest as never, generation), STALE_GENERATION); + assert.deepEqual( + await sql("SELECT * FROM search_index_dirty WHERE connector_instance_id = ?", ["cin_retry_other"]), + otherBefore + ); + } finally { + if (pg) { + for (const table of ["records", "lexical_search_meta", "search_index_dirty", "connector_instance_tombstones"]) { + // biome-ignore lint/performance/noAwaitInLoops: Cleanup is sequential and order-independent. + await postgresQuery(`DELETE FROM ${table}`); + } + await closePostgresStorage(); + } else { + closeDb(); + } + } +}); + +test("one instance that cannot be queued does not leave its siblings without a retry ticket", async () => { + const pg = Boolean(process.env.PDPP_TEST_POSTGRES_URL); + if (pg) { + await initPostgresStorage({ + backend: "postgres", + databaseUrl: process.env.PDPP_TEST_POSTGRES_URL || "", + }); + } else { + initDb(":memory:"); + } + const poisoned = "cin_a_poisoned"; + const healthy = "cin_b_healthy"; + try { + await registerConnector(manifest, { backfillRetrievalIndexes: false }); + for await (const instance of [poisoned, healthy]) { + await ingestRecord( + { + connector_id: manifest.connector_id, + connector_instance_id: instance, + }, + { + data: { id: "k1", subject: "synthetic sibling" }, + emitted_at: "2026-07-16T00:00:00.000Z", + key: "k1", + stream: "items", + }, + { deferIndexes: true } + ); + } + if (pg) { + await postgresQuery("DELETE FROM search_index_dirty"); + await postgresQuery( + "CREATE OR REPLACE FUNCTION startup_retry_test_reject() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'poisoned instance'; END $$" + ); + await postgresQuery( + `CREATE TRIGGER startup_retry_test_reject BEFORE INSERT OR UPDATE ON search_index_dirty FOR EACH ROW WHEN (NEW.connector_instance_id = '${poisoned}') EXECUTE FUNCTION startup_retry_test_reject()` + ); + } else { + getDb().prepare("DELETE FROM search_index_dirty").run(); + getDb().exec( + `CREATE TRIGGER startup_retry_test_reject BEFORE INSERT ON search_index_dirty WHEN NEW.connector_instance_id = '${poisoned}' BEGIN SELECT RAISE(ABORT, 'poisoned instance'); END` + ); + } + await assert.rejects( + queueStartupSearchIndexRetry(manifest as never, currentStorageGeneration()), + (error) => error instanceof AggregateError && error.errors.length === 1 + ); + assert.deepEqual( + (await listDirtySearchIndexScopes(25)) + .map((scope) => scope.connectorInstanceId) + .filter((id) => id === poisoned || id === healthy), + [healthy] + ); + } finally { + if (pg) { + await postgresQuery("DROP TRIGGER IF EXISTS startup_retry_test_reject ON search_index_dirty"); + await postgresQuery("DROP FUNCTION IF EXISTS startup_retry_test_reject()"); + await postgresQuery("DELETE FROM records"); + await postgresQuery("DELETE FROM search_index_dirty"); + await closePostgresStorage(); + } else { + closeDb(); + } + } +}); From ef4f30ad6193fcc27d0856580609f5f3f2bfe100 Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Thu, 5 Nov 2026 14:12:00 -0600 Subject: [PATCH 2/4] fix(search): retry a failed startup ticket write and queue when semantic backend is unavailable A failed startup backfill could still leave no durable ticket: if the queue write itself threw (for example a transient read or write fault while finding the scopes to mark), the error was only logged. Manifests whose queue write failed now get one more attempt after the startup loop. The list is local to the run, so no new state is stored. When a semantic backend is configured but not available at startup, the semantic pass was skipped and startup logged "completed" with no ticket. It now queues a ticket for connectors that declare semantic fields. With no backend configured, or no semantic fields declared, nothing is queued. The tests run the real startup over retained state on SQLite and PostgreSQL. Both new behavior tests fail without the change. Assisted-by: AI Signed-off-by: Tim Nunamaker --- reference-implementation/server/index.ts | 47 ++++- .../server/startup-search-index-retry.ts | 12 ++ .../startup-backfill-durable-retry.test.ts | 198 +++++++++++++++++- 3 files changed, 249 insertions(+), 8 deletions(-) diff --git a/reference-implementation/server/index.ts b/reference-implementation/server/index.ts index d8804ed1a..05a2771d6 100644 --- a/reference-implementation/server/index.ts +++ b/reference-implementation/server/index.ts @@ -7,7 +7,7 @@ * Combined AS + RS implementing PDPP v0.1.0 core spec. * Starts on port 7662 (AS/introspection) and 7663 (RS query API). */ -import { queueStartupSearchIndexRetry } from "./startup-search-index-retry.ts"; +import { manifestDeclaresSemanticFields, queueStartupSearchIndexRetry } from "./startup-search-index-retry.ts"; import { currentStorageGeneration } from "./storage-generation.ts"; import { createHash, randomBytes } from "node:crypto"; import { existsSync, readFileSync, renameSync } from "node:fs"; @@ -1307,6 +1307,23 @@ async function runRetrievalStartupBackfill({ const startedAt = Date.now(); logger.info({ connectorCount: manifests.length }, "retrieval startup backfill started"); + // A failed queue write (for example a transient read or write fault while + // finding the scopes to mark) must not leave a failed backfill without a + // ticket. Those manifests get one more attempt after the loop, so the + // retry costs no new state: only a local list for this run. + const queueWriteFailed: ConnectorManifest[] = []; + const queueRetry = async (manifest: ConnectorManifest, err: unknown) => { + try { + await queueStartupSearchIndexRetry(manifest as never, generation); + } catch (queueError) { + logger.warn( + { connectorId: manifest.connector_id, err, queueError }, + "retrieval startup backfill retry queue write failed" + ); + queueWriteFailed.push(manifest); + } + }; + for (const manifest of manifests) { if (signal?.aborted) { logger.info({ reason: "shutdown" }, "retrieval startup backfill aborted between connectors"); @@ -1330,6 +1347,13 @@ async function runRetrievalStartupBackfill({ manifest: manifest as never, signal, }); + } else if (semanticBackend && manifestDeclaresSemanticFields(manifest)) { + // Configured but not available (for example still warming): the semantic + // half is unproven, so leave a ticket. With no backend configured at all + // nothing is queued, or every boot would dirty every semantic connector. + logger.warn({ connectorId }, "retrieval startup backfill skipped semantic pass: backend unavailable"); + await queueRetry(manifest, new Error("semantic backend unavailable at startup")); + continue; } logger.info({ connectorId }, "retrieval startup backfill connector completed"); } catch (err) { @@ -1340,11 +1364,22 @@ async function runRetrievalStartupBackfill({ return; } logger.warn({ connectorId, err }, "retrieval startup backfill failed for connector"); - try { - await queueStartupSearchIndexRetry(manifest as never, generation); - } catch (queueError) { - logger.warn({ connectorId, err, queueError }, "retrieval startup backfill retry queue write failed"); - } + await queueRetry(manifest, err); + } + } + + for (const manifest of queueWriteFailed) { + if (signal?.aborted) { + return; + } + try { + // biome-ignore lint/performance/noAwaitInLoops: Never hold more than one instance deletion fence. + await queueStartupSearchIndexRetry(manifest as never, generation); + } catch (queueError) { + logger.warn( + { connectorId: manifest.connector_id, queueError }, + "retrieval startup backfill ticket write failed again after retry; the next startup repeats the backfill" + ); } } diff --git a/reference-implementation/server/startup-search-index-retry.ts b/reference-implementation/server/startup-search-index-retry.ts index 5113deb8f..3b6d67eb9 100644 --- a/reference-implementation/server/startup-search-index-retry.ts +++ b/reference-implementation/server/startup-search-index-retry.ts @@ -8,6 +8,18 @@ import { listStartupSearchIndexInstances, markStartupSearchIndexFailure } from " type Manifest = Parameters[0]["manifest"] & Parameters[0]["manifest"]; +/** True when any stream declares semantic fields, so a semantic pass is owed. */ +export function manifestDeclaresSemanticFields(manifest: { streams?: unknown }): boolean { + if (!Array.isArray(manifest.streams)) { + return false; + } + return manifest.streams.some((stream) => { + const fields = (stream as { query?: { search?: { semantic_fields?: unknown } } } | null)?.query?.search + ?.semantic_fields; + return Array.isArray(fields) && fields.length > 0; + }); +} + /** Startup-only caller effect: reconcile failures own their revision/backoff. */ export async function queueStartupSearchIndexRetry(manifest: Manifest, generation: number): Promise { assertCurrentStorageGeneration(generation); diff --git a/reference-implementation/test/startup-backfill-durable-retry.test.ts b/reference-implementation/test/startup-backfill-durable-retry.test.ts index 72074de86..86a25bbd7 100644 --- a/reference-implementation/test/startup-backfill-durable-retry.test.ts +++ b/reference-implementation/test/startup-backfill-durable-retry.test.ts @@ -5,6 +5,7 @@ import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import test from "node:test"; +import pino from "pino"; import { registerConnector } from "../server/auth.ts"; import { withConnectorInstanceControlPlaneWrite } from "../server/connector-instance-write-coordinator.ts"; import { closeDb, getDb, initDb } from "../server/db.ts"; @@ -24,6 +25,24 @@ import { } from "../server/stores/search-index-dirty-store.ts"; import { closeTestServer } from "./helpers/close-test-server.ts"; +const RETAINED_PG_TABLES = [ + "semantic_search_blob", + "semantic_search_meta", + "semantic_search_backfill_progress", + "lexical_search_index", + "lexical_search_meta", + "search_index_dirty", + "records", +]; + +/** Rows a finished test leaves in the shared database would make the next restart a no-op. */ +async function clearRetainedPgState() { + for (const table of RETAINED_PG_TABLES) { + // biome-ignore lint/performance/noAwaitInLoops: Cleanup order is fixed and tiny. + await postgresQuery(`DELETE FROM ${table}`); + } +} + const STALE_GENERATION = /storage generation changed/; function required(value: T | undefined): T { @@ -159,8 +178,7 @@ test("actual startup timeout queues retained work and maintenance converges with } finally { await closeTestServer(server); if (pg) { - await postgresQuery("DELETE FROM records"); - await postgresQuery("DELETE FROM search_index_dirty"); + await clearRetainedPgState(); await closePostgresStorage(); } else { closeDb(); @@ -425,3 +443,179 @@ test("one instance that cannot be queued does not leave its siblings without a r } } }); + +const FAULT = "startup_retry_test_fault"; + +/** + * Restart over one retained record and no tickets, then let the real startup + * backfill run. `fault` makes the first ticket write fail on either backend. + * SQLite cannot keep a counter across a rolled-back statement, so there the + * fault trigger is dropped when the server logs the failed queue write. + */ +async function startOverRetainedRecord({ + backend, + fault = false, + semanticRetrievalSupported, + source = manifest, +}: { + backend: unknown; + fault?: boolean; + semanticRetrievalSupported?: boolean; + source?: typeof manifest; +}) { + const dir = mkdtempSync(join(tmpdir(), "startup-retry-")); + const dbPath = join(dir, "db.sqlite"); + const pg = Boolean(process.env.PDPP_TEST_POSTGRES_URL); + if (pg) { + await initPostgresStorage({ backend: "postgres", databaseUrl: process.env.PDPP_TEST_POSTGRES_URL || "" }); + } else { + initDb(dbPath); + } + const attachment = pg + ? await createAlreadyAdmittedTestDatabaseChildAttachment(process.env.PDPP_TEST_POSTGRES_URL || "") + : undefined; + await registerConnector(source, { backfillRetrievalIndexes: false }); + await ingestRecord( + { connector_id: source.connector_id, connector_instance_id: "cin_startup_retry" }, + { + data: { id: "k1", subject: "retained synthetic subject" }, + emitted_at: "2026-07-16T00:00:00.000Z", + key: "k1", + stream: "items", + }, + { deferIndexes: true } + ); + if (pg) { + await postgresQuery("DELETE FROM search_index_dirty"); + if (fault) { + await postgresQuery(`CREATE SEQUENCE IF NOT EXISTS ${FAULT}_seq`); + await postgresQuery(`SELECT setval('${FAULT}_seq', 1, false)`); + await postgresQuery( + `CREATE OR REPLACE FUNCTION ${FAULT}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF nextval('${FAULT}_seq') = 1 THEN RAISE EXCEPTION 'transient fault'; END IF; RETURN NEW; END $$` + ); + await postgresQuery( + `CREATE TRIGGER ${FAULT} BEFORE INSERT OR UPDATE ON search_index_dirty FOR EACH ROW EXECUTE FUNCTION ${FAULT}()` + ); + } + } else { + getDb().prepare("DELETE FROM search_index_dirty").run(); + if (fault) { + getDb().exec( + `CREATE TRIGGER ${FAULT} BEFORE INSERT ON search_index_dirty BEGIN SELECT RAISE(ABORT, 'transient fault'); END` + ); + } + closeDb(); + } + const warnings: string[] = []; + const logger = pino( + { level: "warn" }, + { + write(line: string) { + const message = String(JSON.parse(line).msg); + warnings.push(message); + if (!pg && fault && message.includes("retry queue write failed")) { + getDb().exec(`DROP TRIGGER IF EXISTS ${FAULT}`); + } + }, + } + ); + const server = await startServer({ + asPort: 0, + dbPath, + logger, + rsPort: 0, + storageBackend: pg ? "postgres" : "sqlite", + ...(pg + ? { + databaseUrl: process.env.PDPP_TEST_POSTGRES_URL || "", + testOnlyAlreadyAdmittedPostgresChildAttachment: attachment || "", + } + : {}), + quiet: true, + reconcilePolyfillManifests: false, + ...(semanticRetrievalSupported === undefined ? {} : { semanticRetrievalSupported }), + semanticRetrievalBackend: backend, + }); + return { + server, + warnings, + async finish() { + await closeTestServer(server); + if (pg) { + await postgresQuery(`DROP TRIGGER IF EXISTS ${FAULT} ON search_index_dirty`); + await postgresQuery(`DROP FUNCTION IF EXISTS ${FAULT}()`); + await postgresQuery(`DROP SEQUENCE IF EXISTS ${FAULT}_seq`); + await clearRetainedPgState(); + await closePostgresStorage(); + } else { + closeDb(); + } + rmSync(dir, { force: true, recursive: true }); + }, + }; +} + +function failingEmbeddingBackend() { + const stub = makeStubBackend(); + return { + ...stub, + embedDocument: async () => { + throw new LocalTransformerExecutorError("transformer_deadline"); + }, + }; +} + +async function retainedTickets(connectorId: string = manifest.connector_id) { + return await listDirtySearchIndexScopes(25, { connectorId }); +} + +test("a failed ticket write at startup is retried once so the failed backfill is not left without a ticket", async () => { + const run = await startOverRetainedRecord({ backend: failingEmbeddingBackend(), fault: true }); + try { + await run.server.startupBackfillDone; + assert.equal( + run.warnings.filter((m) => m.includes("retry queue write failed")).length, + 1, + `the first ticket write failed once: ${JSON.stringify(run.warnings)}` + ); + assert.equal((await retainedTickets()).length, 1, "the retry after the loop wrote the ticket"); + } finally { + await run.finish(); + } +}); + +test("a semantic backend that is configured but unavailable at startup queues a ticket", async () => { + const run = await startOverRetainedRecord({ backend: { ...makeStubBackend(), available: () => false } }); + try { + await run.server.startupBackfillDone; + assert.equal((await retainedTickets()).length, 1); + } finally { + await run.finish(); + } +}); + +test("no ticket is queued when no semantic backend is configured or the connector declares no semantic fields", async () => { + const unconfigured = await startOverRetainedRecord({ backend: undefined, semanticRetrievalSupported: false }); + try { + await unconfigured.server.startupBackfillDone; + assert.equal((await retainedTickets()).length, 0, "no backend configured"); + } finally { + await unconfigured.finish(); + } + const lexicalOnly = { + ...manifest, + connector_id: "startup-retry-lexical-only", + manifest_uri: "https://example.test/startup-retry-lexical-only", + streams: manifest.streams.map(({ query: _query, ...stream }) => ({ ...stream, query: { search: {} } })), + }; + const noFields = await startOverRetainedRecord({ + backend: { ...makeStubBackend(), available: () => false }, + source: lexicalOnly as never, + }); + try { + await noFields.server.startupBackfillDone; + assert.equal((await retainedTickets(lexicalOnly.connector_id)).length, 0, "no semantic_fields declared"); + } finally { + await noFields.finish(); + } +}); From e55f92636159fb5efe73bb96a423159fba179c2a Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Thu, 5 Nov 2026 16:30:22 -0600 Subject: [PATCH 3/4] fix(search): name the instance in each failed startup ticket write When a startup ticket write failed for one instance, the AggregateError carried the raw error only, so the log could not say which instance was never queued. Each failure is now wrapped with its connector instance id and keeps the original error as its cause. Also document why the orphan-instance lookup does not read `records`: the semantic resolver already lists every instance with records for a manifest stream. A test pins that, and another pins that startup continues when the ticket write keeps failing. Assisted-by: AI Signed-off-by: Tim Nunamaker --- .../index-dirty/startup-index-instances.sql | 5 + .../server/startup-search-index-retry.ts | 6 +- .../startup-backfill-durable-retry.test.ts | 96 +++++++++++++++++-- 3 files changed, 99 insertions(+), 8 deletions(-) diff --git a/reference-implementation/server/queries/search/index-dirty/startup-index-instances.sql b/reference-implementation/server/queries/search/index-dirty/startup-index-instances.sql index 69d26882a..5890e0bd2 100644 --- a/reference-implementation/server/queries/search/index-dirty/startup-index-instances.sql +++ b/reference-implementation/server/queries/search/index-dirty/startup-index-instances.sql @@ -1,5 +1,10 @@ -- @terminator: iterate -- @cursor_field: connector_instance_id +-- Instances that hold search-index state for a connector, so state the backfill +-- cleans up is still queued when no active binding names the instance. `records` +-- is left out on purpose: the semantic resolver already lists every instance +-- that has records for a manifest stream (test/startup-backfill-durable-retry), +-- so a second `records` read here would only repeat that lookup. SELECT connector_instance_id FROM lexical_search_meta WHERE connector_id = ? UNION SELECT connector_instance_id FROM lexical_search_index WHERE connector_id = ? UNION SELECT connector_instance_id FROM semantic_search_meta WHERE connector_id = ? diff --git a/reference-implementation/server/startup-search-index-retry.ts b/reference-implementation/server/startup-search-index-retry.ts index 3b6d67eb9..7d7df5b1c 100644 --- a/reference-implementation/server/startup-search-index-retry.ts +++ b/reference-implementation/server/startup-search-index-retry.ts @@ -49,7 +49,11 @@ export async function queueStartupSearchIndexRetry(manifest: Manifest, generatio generation, }); } catch (error) { - failures.push(error); + failures.push( + new Error(`startup search-index retry could not be queued for instance ${connectorInstanceId}`, { + cause: error, + }) + ); } } if (failures.length > 0) { diff --git a/reference-implementation/test/startup-backfill-durable-retry.test.ts b/reference-implementation/test/startup-backfill-durable-retry.test.ts index 86a25bbd7..64f435a20 100644 --- a/reference-implementation/test/startup-backfill-durable-retry.test.ts +++ b/reference-implementation/test/startup-backfill-durable-retry.test.ts @@ -15,12 +15,18 @@ import { closePostgresStorage, initPostgresStorage, postgresQuery } from "../ser import { createAlreadyAdmittedTestDatabaseChildAttachment } from "../server/postgres-test-database-guard.ts"; import { ingestRecord } from "../server/records.ts"; import { runSearchIndexDirtyReconcileRound } from "../server/search-index-reconcile.ts"; -import { computeSemanticEntries, makeStubBackend } from "../server/search-semantic.ts"; +import { resolveLexicalBackfillConnectorInstanceIds } from "../server/search.ts"; +import { + computeSemanticEntries, + makeStubBackend, + resolveSemanticBackfillConnectorInstanceIds, +} from "../server/search-semantic.ts"; import { queueStartupSearchIndexRetry } from "../server/startup-search-index-retry.ts"; import { bumpStorageGeneration, currentStorageGeneration } from "../server/storage-generation.ts"; import { clearSearchIndexDirty, listDirtySearchIndexScopes, + listStartupSearchIndexInstances, markStartupSearchIndexFailure, } from "../server/stores/search-index-dirty-store.ts"; import { closeTestServer } from "./helpers/close-test-server.ts"; @@ -423,7 +429,12 @@ test("one instance that cannot be queued does not leave its siblings without a r } await assert.rejects( queueStartupSearchIndexRetry(manifest as never, currentStorageGeneration()), - (error) => error instanceof AggregateError && error.errors.length === 1 + (error) => + error instanceof AggregateError && + error.errors.length === 1 && + error.errors[0] instanceof Error && + error.errors[0].message.includes(poisoned) && + error.errors[0].cause instanceof Error ); assert.deepEqual( (await listDirtySearchIndexScopes(25)) @@ -448,7 +459,7 @@ const FAULT = "startup_retry_test_fault"; /** * Restart over one retained record and no tickets, then let the real startup - * backfill run. `fault` makes the first ticket write fail on either backend. + * backfill run. `fault` makes the first ("once") or every ("always") ticket write fail on either backend. * SQLite cannot keep a counter across a rolled-back statement, so there the * fault trigger is dropped when the server logs the failed queue write. */ @@ -459,7 +470,7 @@ async function startOverRetainedRecord({ source = manifest, }: { backend: unknown; - fault?: boolean; + fault?: false | "once" | "always"; semanticRetrievalSupported?: boolean; source?: typeof manifest; }) { @@ -491,7 +502,7 @@ async function startOverRetainedRecord({ await postgresQuery(`CREATE SEQUENCE IF NOT EXISTS ${FAULT}_seq`); await postgresQuery(`SELECT setval('${FAULT}_seq', 1, false)`); await postgresQuery( - `CREATE OR REPLACE FUNCTION ${FAULT}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF nextval('${FAULT}_seq') = 1 THEN RAISE EXCEPTION 'transient fault'; END IF; RETURN NEW; END $$` + `CREATE OR REPLACE FUNCTION ${FAULT}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF ${fault === "always" ? "true" : `nextval('${FAULT}_seq') = 1`} THEN RAISE EXCEPTION 'transient fault'; END IF; RETURN NEW; END $$` ); await postgresQuery( `CREATE TRIGGER ${FAULT} BEFORE INSERT OR UPDATE ON search_index_dirty FOR EACH ROW EXECUTE FUNCTION ${FAULT}()` @@ -513,7 +524,7 @@ async function startOverRetainedRecord({ write(line: string) { const message = String(JSON.parse(line).msg); warnings.push(message); - if (!pg && fault && message.includes("retry queue write failed")) { + if (!pg && fault === "once" && message.includes("retry queue write failed")) { getDb().exec(`DROP TRIGGER IF EXISTS ${FAULT}`); } }, @@ -570,7 +581,7 @@ async function retainedTickets(connectorId: string = manifest.connector_id) { } test("a failed ticket write at startup is retried once so the failed backfill is not left without a ticket", async () => { - const run = await startOverRetainedRecord({ backend: failingEmbeddingBackend(), fault: true }); + const run = await startOverRetainedRecord({ backend: failingEmbeddingBackend(), fault: "once" }); try { await run.server.startupBackfillDone; assert.equal( @@ -619,3 +630,74 @@ test("no ticket is queued when no semantic backend is configured or the connecto await noFields.finish(); } }); + +test("an instance that has only records is queued through the semantic resolver, not the orphan lookup", async () => { + const pg = Boolean(process.env.PDPP_TEST_POSTGRES_URL); + if (pg) { + await initPostgresStorage({ backend: "postgres", databaseUrl: process.env.PDPP_TEST_POSTGRES_URL || "" }); + } else { + initDb(":memory:"); + } + const unbound = "cin_records_only"; + try { + await registerConnector(manifest, { backfillRetrievalIndexes: false }); + await ingestRecord( + { connector_id: manifest.connector_id, connector_instance_id: unbound }, + { + data: { id: "k1", subject: "synthetic records only" }, + emitted_at: "2026-07-16T00:00:00.000Z", + key: "k1", + stream: "items", + }, + { deferIndexes: true } + ); + if (pg) { + await postgresQuery("DELETE FROM search_index_dirty"); + } else { + getDb().prepare("DELETE FROM search_index_dirty").run(); + } + const backfilled = await resolveLexicalBackfillConnectorInstanceIds({ + connectorId: manifest.connector_id, + manifest: manifest as never, + }); + assert.ok(!backfilled.includes(unbound), "the lexical resolver does not return it"); + assert.deepEqual(await listStartupSearchIndexInstances(manifest.connector_id), [], "no index state exists for it"); + assert.ok( + ( + await resolveSemanticBackfillConnectorInstanceIds({ + connectorId: manifest.connector_id, + manifest: manifest as never, + }) + ).includes(unbound), + "the semantic resolver lists every instance with records for a manifest stream" + ); + await queueStartupSearchIndexRetry(manifest as never, currentStorageGeneration()); + assert.ok( + (await listDirtySearchIndexScopes(25)).some((t) => t.connectorInstanceId === unbound), + "so a failed startup backfill still queues it, and the orphan lookup needs no `records` branch" + ); + } finally { + if (pg) { + await clearRetainedPgState(); + await closePostgresStorage(); + } else { + closeDb(); + } + } +}); + +test("a ticket write that keeps failing does not fail startup", async () => { + const run = await startOverRetainedRecord({ backend: failingEmbeddingBackend(), fault: "always" }); + try { + await run.server.startupBackfillDone; + assert.ok( + run.warnings.some((m) => m.includes("ticket write failed again after retry")), + `the retry also failed and was logged: ${JSON.stringify(run.warnings)}` + ); + assert.equal((await retainedTickets()).length, 0, "no ticket could be written"); + const response = await fetch(`http://localhost:${run.server.asPort}/.well-known/oauth-authorization-server`); + assert.ok(response.status < 500, "the server still answers"); + } finally { + await run.finish(); + } +}); From 01f7a3e010b198f48b5a6554a31bbb926027bf23 Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Fri, 6 Nov 2026 14:01:02 -0600 Subject: [PATCH 4/4] fix(search): keep retrying a failed startup ticket write and cover retained semantic state Review of the startup retry found four gaps. - A ticket write that failed twice was dropped, so idle sources stayed unqueued until the next restart. Failed manifests now stay in the startup task and are retried with capped exponential backoff (250 ms to 30 s) until a mark commits or shutdown aborts the sleep. - On PostgreSQL the mark cast every retained semantic scope key to jsonb. One malformed key failed the whole mark and left the instance without a ticket. Keys are now decoded in code; a key that names no stream is skipped and left for the repair that fails closed on it. - A manifest that drops its last semantic field skipped the ticket when the backend was unavailable, though retained semantic state still owed removed-field cleanup. Retained meta, progress or vector rows now count like declared fields. - Instance discovery for a ticket no longer calls the lexical and semantic backfill resolvers. One store query lists every instance with records or index state, so those two resolvers are private again. The tests assert outcomes (a ticket exists), not which lookup found it. Assisted-by: AI Signed-off-by: Tim Nunamaker --- reference-implementation/server/index.ts | 76 +++++++-- .../server/queries/index.ts | 1 + .../index-dirty/startup-index-instances.sql | 12 +- .../index-dirty/startup-semantic-state.sql | 10 ++ .../server/search-semantic.ts | 2 +- reference-implementation/server/search.ts | 2 +- .../server/startup-search-index-retry.ts | 24 +-- .../server/stores/search-index-dirty-store.ts | 56 ++++++- .../startup-backfill-durable-retry.test.ts | 146 +++++++++++++----- 9 files changed, 244 insertions(+), 85 deletions(-) create mode 100644 reference-implementation/server/queries/search/index-dirty/startup-semantic-state.sql diff --git a/reference-implementation/server/index.ts b/reference-implementation/server/index.ts index 05a2771d6..9e2a5bc5b 100644 --- a/reference-implementation/server/index.ts +++ b/reference-implementation/server/index.ts @@ -640,6 +640,7 @@ import { } from "./stores/run-history-backfill-stage.ts"; import { getDefaultSchedulerStore, type SchedulerStore } from "./stores/scheduler-store.ts"; import { + connectorHasRetainedSemanticState, countDirtySearchIndexScopes, parkSearchIndexDirtyScope, rearmSearchIndexDirtyScope, @@ -1289,6 +1290,32 @@ async function collectRetrievalStartupBackfillManifests({ return registered.map(({ manifest }) => manifest); } +const STARTUP_TICKET_RETRY_BASE_MS = 250; +const STARTUP_TICKET_RETRY_MAX_MS = 30_000; + +/** Exponential backoff from 250 ms, capped at 30 s. */ +function startupTicketRetryDelayMs(attempt: number): number { + return Math.min(STARTUP_TICKET_RETRY_MAX_MS, STARTUP_TICKET_RETRY_BASE_MS * 2 ** attempt); +} + +/** Waits `ms`; resolves false at once if `signal` aborts first. */ +function sleepUnlessAborted(ms: number, signal: AbortSignal | null): Promise { + if (signal?.aborted) { + return Promise.resolve(false); + } + return new Promise((resolve) => { + const onAbort = () => { + clearTimeout(timer); + resolve(false); + }; + const timer = setTimeout(() => { + signal?.removeEventListener("abort", onAbort); + resolve(true); + }, ms); + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} + async function runRetrievalStartupBackfill({ generation = currentStorageGeneration(), manifests, @@ -1309,8 +1336,11 @@ async function runRetrievalStartupBackfill({ // A failed queue write (for example a transient read or write fault while // finding the scopes to mark) must not leave a failed backfill without a - // ticket. Those manifests get one more attempt after the loop, so the - // retry costs no new state: only a local list for this run. + // ticket. Those manifests stay in this local list and are retried with + // bounded backoff until a mark commits or shutdown aborts the run, so a fault + // that outlasts a fixed number of attempts still ends in a ticket. No new + // durable state: if the process stops first, the next startup repeats the + // backfill. const queueWriteFailed: ConnectorManifest[] = []; const queueRetry = async (manifest: ConnectorManifest, err: unknown) => { try { @@ -1347,10 +1377,15 @@ async function runRetrievalStartupBackfill({ manifest: manifest as never, signal, }); - } else if (semanticBackend && manifestDeclaresSemanticFields(manifest)) { + } else if ( + semanticBackend && + (manifestDeclaresSemanticFields(manifest) || (await connectorHasRetainedSemanticState(String(connectorId)))) + ) { // Configured but not available (for example still warming): the semantic - // half is unproven, so leave a ticket. With no backend configured at all - // nothing is queued, or every boot would dirty every semantic connector. + // half is unproven, so leave a ticket. Retained semantic state counts like + // declared fields, because removed-field cleanup also needs the backend. + // With no backend configured at all nothing is queued, or every boot would + // dirty every semantic connector. logger.warn({ connectorId }, "retrieval startup backfill skipped semantic pass: backend unavailable"); await queueRetry(manifest, new Error("semantic backend unavailable at startup")); continue; @@ -1368,19 +1403,30 @@ async function runRetrievalStartupBackfill({ } } - for (const manifest of queueWriteFailed) { - if (signal?.aborted) { + let pending = queueWriteFailed; + for (let attempt = 0; pending.length > 0; attempt++) { + // biome-ignore lint/performance/noAwaitInLoops: Backoff is sequential by design. + if (!(await sleepUnlessAborted(startupTicketRetryDelayMs(attempt), signal))) { + logger.info({ reason: "shutdown" }, "retrieval startup ticket retry aborted"); return; } - try { - // biome-ignore lint/performance/noAwaitInLoops: Never hold more than one instance deletion fence. - await queueStartupSearchIndexRetry(manifest as never, generation); - } catch (queueError) { - logger.warn( - { connectorId: manifest.connector_id, queueError }, - "retrieval startup backfill ticket write failed again after retry; the next startup repeats the backfill" - ); + const stillFailing: ConnectorManifest[] = []; + for (const manifest of pending) { + try { + // biome-ignore lint/performance/noAwaitInLoops: Never hold more than one instance deletion fence. + await queueStartupSearchIndexRetry(manifest as never, generation); + } catch (queueError) { + if (signal?.aborted) { + return; + } + logger.warn( + { attempt: attempt + 1, connectorId: manifest.connector_id, queueError }, + "retrieval startup backfill ticket write failed again; retrying" + ); + stillFailing.push(manifest); + } } + pending = stillFailing; } logger.info( diff --git a/reference-implementation/server/queries/index.ts b/reference-implementation/server/queries/index.ts index 9ca30019d..040a1cb9d 100644 --- a/reference-implementation/server/queries/index.ts +++ b/reference-implementation/server/queries/index.ts @@ -482,6 +482,7 @@ export interface ReferenceQueryRegistry extends Readonly[0]["manifest"] & - Parameters[0]["manifest"]; +type Manifest = { + connector_id: string; + connector_instance_id?: string | null; + storage_binding?: { connector_instance_id?: string | null } | null; + streams?: unknown; +}; /** True when any stream declares semantic fields, so a semantic pass is owed. */ export function manifestDeclaresSemanticFields(manifest: { streams?: unknown }): boolean { @@ -24,23 +26,13 @@ export function manifestDeclaresSemanticFields(manifest: { streams?: unknown }): export async function queueStartupSearchIndexRetry(manifest: Manifest, generation: number): Promise { assertCurrentStorageGeneration(generation); const connectorId = manifest.connector_id; - const lexical = await resolveLexicalBackfillConnectorInstanceIds({ - connectorId, - manifest, - }); - assertCurrentStorageGeneration(generation); - const semantic = await resolveSemanticBackfillConnectorInstanceIds({ - connectorId, - manifest, - }); - assertCurrentStorageGeneration(generation); const pin = manifest.storage_binding?.connector_instance_id || manifest.connector_instance_id; - const orphanInstances = pin ? [] : await listStartupSearchIndexInstances(connectorId); + const instances = pin ? [pin] : await listStartupSearchIndexInstances(connectorId); assertCurrentStorageGeneration(generation); // One instance's failed mark must not leave its siblings without a durable // retry, so every instance is attempted before the failures are reported. const failures: unknown[] = []; - for (const connectorInstanceId of new Set([...lexical, ...semantic, ...orphanInstances])) { + for (const connectorInstanceId of new Set(instances)) { try { // biome-ignore lint/performance/noAwaitInLoops: Never hold more than one instance deletion fence. await markStartupSearchIndexFailure({ diff --git a/reference-implementation/server/stores/search-index-dirty-store.ts b/reference-implementation/server/stores/search-index-dirty-store.ts index 96f62ef32..450b93a5b 100644 --- a/reference-implementation/server/stores/search-index-dirty-store.ts +++ b/reference-implementation/server/stores/search-index-dirty-store.ts @@ -348,6 +348,16 @@ export async function countDirtySearchIndexScopes(): Promise { return row?.n ?? 0; } +/** The stream a canonical semantic scope key (`[stream, field]` JSON) names, or null for anything else. */ +function streamOfSemanticScopeKey(scopeKey: string): string | null { + try { + const parsed: unknown = JSON.parse(scopeKey); + return Array.isArray(parsed) && typeof parsed[0] === "string" ? parsed[0] : null; + } catch { + return null; + } +} + /** Startup failures preserve retry policy; ingest alone resets changed-input history. * Discover extant streams inside the same instance fence as deletion. No instance * row is required for legacy direct ingest, but retained/index state is required. @@ -373,11 +383,24 @@ export async function markStartupSearchIndexFailure({ UNION SELECT stream FROM lexical_search_meta WHERE connector_instance_id = $1 AND connector_id = $2 UNION SELECT stream FROM semantic_search_meta WHERE connector_instance_id = $1 AND connector_id = $2 UNION SELECT stream FROM semantic_search_backfill_progress WHERE connector_instance_id = $1 AND connector_id = $2 - UNION SELECT stream FROM lexical_search_index WHERE connector_instance_id = $1 AND connector_id = $2 - UNION SELECT (scope_key::jsonb)->>0 AS stream FROM semantic_search_blob WHERE connector_instance_id = $1 AND connector_id = $2`, + UNION SELECT stream FROM lexical_search_index WHERE connector_instance_id = $1 AND connector_id = $2`, [connectorInstanceId, connectorId] ); - for (const { stream } of extant.rows) { + // Decode semantic keys here, not with a SQL jsonb cast: one malformed key + // must not fail the whole mark. Untrusted keys name no stream and are left + // for the repair that fails closed on them (same rule as the SQLite query). + const blobKeys = await client.query<{ scope_key: string }>( + "SELECT DISTINCT scope_key FROM semantic_search_blob WHERE connector_instance_id = $1 AND connector_id = $2", + [connectorInstanceId, connectorId] + ); + const streams = new Set(extant.rows.map((row) => row.stream)); + for (const { scope_key: scopeKey } of blobKeys.rows) { + const stream = streamOfSemanticScopeKey(scopeKey); + if (stream !== null) { + streams.add(stream); + } + } + for (const stream of streams) { // Extant orphan streams must also be repaired even after manifest removal. assertCurrentStorageGeneration(generation); // biome-ignore lint/performance/noAwaitInLoops: One connection's stream batch shares its deletion fence and transaction. @@ -423,11 +446,12 @@ export async function markStartupSearchIndexFailure({ }); } -/** Supplement backfill discovery with orphan index instances, still subject to manifest pins. */ +/** Every instance that holds records or search-index state for a connector. Instances with neither cannot need a ticket. */ export async function listStartupSearchIndexInstances(connectorId: string): Promise { if (isPostgresStorageBackend()) { const result = await postgresQuery<{ connector_instance_id: string }>( - `SELECT connector_instance_id FROM lexical_search_meta WHERE connector_id = $1 + `SELECT connector_instance_id FROM records WHERE connector_id = $1 + UNION SELECT connector_instance_id FROM lexical_search_meta WHERE connector_id = $1 UNION SELECT connector_instance_id FROM lexical_search_index WHERE connector_id = $1 UNION SELECT connector_instance_id FROM semantic_search_meta WHERE connector_id = $1 UNION SELECT connector_instance_id FROM semantic_search_backfill_progress WHERE connector_id = $1 @@ -439,7 +463,27 @@ export async function listStartupSearchIndexInstances(connectorId: string): Prom return [ ...iterate<{ connector_instance_id: string }>( referenceQueries.searchIndexDirtyStartupIndexInstances, - Array.from({ length: 6 }, () => connectorId) + Array.from({ length: 7 }, () => connectorId) ), ].map((row) => row.connector_instance_id); } + +/** True when any instance of the connector still holds semantic index state (meta, progress or vectors). */ +export async function connectorHasRetainedSemanticState(connectorId: string): Promise { + if (isPostgresStorageBackend()) { + const result = await postgresQuery<{ present: boolean }>( + `SELECT EXISTS ( + SELECT 1 FROM semantic_search_meta WHERE connector_id = $1 + UNION ALL SELECT 1 FROM semantic_search_backfill_progress WHERE connector_id = $1 + UNION ALL SELECT 1 FROM semantic_search_blob WHERE connector_id = $1 + ) AS present`, + [connectorId] + ); + return result.rows[0]?.present === true; + } + const row = getOne<{ present: number }>( + referenceQueries.searchIndexDirtyStartupSemanticState, + Array.from({ length: 4 }, () => connectorId) + ); + return row?.present === 1; +} diff --git a/reference-implementation/test/startup-backfill-durable-retry.test.ts b/reference-implementation/test/startup-backfill-durable-retry.test.ts index 64f435a20..cfe0db791 100644 --- a/reference-implementation/test/startup-backfill-durable-retry.test.ts +++ b/reference-implementation/test/startup-backfill-durable-retry.test.ts @@ -15,18 +15,12 @@ import { closePostgresStorage, initPostgresStorage, postgresQuery } from "../ser import { createAlreadyAdmittedTestDatabaseChildAttachment } from "../server/postgres-test-database-guard.ts"; import { ingestRecord } from "../server/records.ts"; import { runSearchIndexDirtyReconcileRound } from "../server/search-index-reconcile.ts"; -import { resolveLexicalBackfillConnectorInstanceIds } from "../server/search.ts"; -import { - computeSemanticEntries, - makeStubBackend, - resolveSemanticBackfillConnectorInstanceIds, -} from "../server/search-semantic.ts"; +import { computeSemanticEntries, makeStubBackend } from "../server/search-semantic.ts"; import { queueStartupSearchIndexRetry } from "../server/startup-search-index-retry.ts"; import { bumpStorageGeneration, currentStorageGeneration } from "../server/storage-generation.ts"; import { clearSearchIndexDirty, listDirtySearchIndexScopes, - listStartupSearchIndexInstances, markStartupSearchIndexFailure, } from "../server/stores/search-index-dirty-store.ts"; import { closeTestServer } from "./helpers/close-test-server.ts"; @@ -459,18 +453,21 @@ const FAULT = "startup_retry_test_fault"; /** * Restart over one retained record and no tickets, then let the real startup - * backfill run. `fault` makes the first ("once") or every ("always") ticket write fail on either backend. + * backfill run. `fault` makes the first ("once"), the first two ("twice") or every ("always") ticket write fail on either backend. * SQLite cannot keep a counter across a rolled-back statement, so there the * fault trigger is dropped when the server logs the failed queue write. */ async function startOverRetainedRecord({ backend, fault = false, + retainedScopeKeys = [], semanticRetrievalSupported, source = manifest, }: { backend: unknown; - fault?: false | "once" | "always"; + fault?: false | "once" | "twice" | "always"; + /** Semantic blob rows retained from an earlier run, keyed by raw `scope_key` (may be malformed). */ + retainedScopeKeys?: string[]; semanticRetrievalSupported?: boolean; source?: typeof manifest; }) { @@ -496,13 +493,17 @@ async function startOverRetainedRecord({ }, { deferIndexes: true } ); + for (const scopeKey of retainedScopeKeys) { + // biome-ignore lint/performance/noAwaitInLoops: A handful of fixed rows. + await insertRetainedBlob(pg, source.connector_id, scopeKey); + } if (pg) { await postgresQuery("DELETE FROM search_index_dirty"); if (fault) { await postgresQuery(`CREATE SEQUENCE IF NOT EXISTS ${FAULT}_seq`); await postgresQuery(`SELECT setval('${FAULT}_seq', 1, false)`); await postgresQuery( - `CREATE OR REPLACE FUNCTION ${FAULT}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF ${fault === "always" ? "true" : `nextval('${FAULT}_seq') = 1`} THEN RAISE EXCEPTION 'transient fault'; END IF; RETURN NEW; END $$` + `CREATE OR REPLACE FUNCTION ${FAULT}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF ${{ always: "true", once: `nextval('${FAULT}_seq') = 1`, twice: `nextval('${FAULT}_seq') <= 2` }[fault]} THEN RAISE EXCEPTION 'transient fault'; END IF; RETURN NEW; END $$` ); await postgresQuery( `CREATE TRIGGER ${FAULT} BEFORE INSERT OR UPDATE ON search_index_dirty FOR EACH ROW EXECUTE FUNCTION ${FAULT}()` @@ -518,14 +519,23 @@ async function startOverRetainedRecord({ closeDb(); } const warnings: string[] = []; + const waiters: Array<{ done: () => void; text: string }> = []; + let failedWrites = 0; const logger = pino( { level: "warn" }, { write(line: string) { const message = String(JSON.parse(line).msg); warnings.push(message); - if (!pg && fault === "once" && message.includes("retry queue write failed")) { - getDb().exec(`DROP TRIGGER IF EXISTS ${FAULT}`); + if (message.includes("queue write failed") || message.includes("ticket write failed again")) { + failedWrites += 1; + const failuresToInject = fault === "once" ? 1 : fault === "twice" ? 2 : Number.POSITIVE_INFINITY; + if (!pg && failedWrites >= failuresToInject) { + getDb().exec(`DROP TRIGGER IF EXISTS ${FAULT}`); + } + } + for (const waiter of waiters.filter((w) => message.includes(w.text))) { + waiter.done(); } }, } @@ -550,6 +560,12 @@ async function startOverRetainedRecord({ return { server, warnings, + /** Resolves once a warning containing `text` has been logged (immediately if it already was). */ + warned(text: string) { + return warnings.some((m) => m.includes(text)) + ? Promise.resolve() + : new Promise((done) => waiters.push({ done, text })); + }, async finish() { await closeTestServer(server); if (pg) { @@ -576,6 +592,21 @@ function failingEmbeddingBackend() { }; } +async function insertRetainedBlob(pg: boolean, connectorId: string, scopeKey: string) { + if (pg) { + await postgresQuery( + "INSERT INTO semantic_search_blob(connector_id, connector_instance_id, scope_key, record_key, embedding, input_identity) VALUES ($1, 'cin_startup_retry', $2, 'retained', '[0,0,0,0,0,0,0,0]'::vector, 'retained')", + [connectorId, scopeKey] + ); + } else { + getDb() + .prepare( + "INSERT INTO semantic_search_blob(connector_id, connector_instance_id, scope_key, record_key, embedding, input_identity) VALUES (?, 'cin_startup_retry', ?, 'retained', x'00', 'retained')" + ) + .run(connectorId, scopeKey); + } +} + async function retainedTickets(connectorId: string = manifest.connector_id) { return await listDirtySearchIndexScopes(25, { connectorId }); } @@ -631,18 +662,18 @@ test("no ticket is queued when no semantic backend is configured or the connecto } }); -test("an instance that has only records is queued through the semantic resolver, not the orphan lookup", async () => { +test("an instance that has only records, with no binding or index state, still gets a ticket", async () => { const pg = Boolean(process.env.PDPP_TEST_POSTGRES_URL); if (pg) { await initPostgresStorage({ backend: "postgres", databaseUrl: process.env.PDPP_TEST_POSTGRES_URL || "" }); } else { initDb(":memory:"); } - const unbound = "cin_records_only"; + const recordsOnly = "cin_records_only"; try { await registerConnector(manifest, { backfillRetrievalIndexes: false }); await ingestRecord( - { connector_id: manifest.connector_id, connector_instance_id: unbound }, + { connector_id: manifest.connector_id, connector_instance_id: recordsOnly }, { data: { id: "k1", subject: "synthetic records only" }, emitted_at: "2026-07-16T00:00:00.000Z", @@ -656,26 +687,8 @@ test("an instance that has only records is queued through the semantic resolver, } else { getDb().prepare("DELETE FROM search_index_dirty").run(); } - const backfilled = await resolveLexicalBackfillConnectorInstanceIds({ - connectorId: manifest.connector_id, - manifest: manifest as never, - }); - assert.ok(!backfilled.includes(unbound), "the lexical resolver does not return it"); - assert.deepEqual(await listStartupSearchIndexInstances(manifest.connector_id), [], "no index state exists for it"); - assert.ok( - ( - await resolveSemanticBackfillConnectorInstanceIds({ - connectorId: manifest.connector_id, - manifest: manifest as never, - }) - ).includes(unbound), - "the semantic resolver lists every instance with records for a manifest stream" - ); await queueStartupSearchIndexRetry(manifest as never, currentStorageGeneration()); - assert.ok( - (await listDirtySearchIndexScopes(25)).some((t) => t.connectorInstanceId === unbound), - "so a failed startup backfill still queues it, and the orphan lookup needs no `records` branch" - ); + assert.ok((await listDirtySearchIndexScopes(25)).some((t) => t.connectorInstanceId === recordsOnly)); } finally { if (pg) { await clearRetainedPgState(); @@ -686,17 +699,70 @@ test("an instance that has only records is queued through the semantic resolver, } }); -test("a ticket write that keeps failing does not fail startup", async () => { - const run = await startOverRetainedRecord({ backend: failingEmbeddingBackend(), fault: "always" }); +test("a ticket write that fails twice is retried until it commits, with no restart", async () => { + const run = await startOverRetainedRecord({ backend: failingEmbeddingBackend(), fault: "twice" }); try { await run.server.startupBackfillDone; - assert.ok( - run.warnings.some((m) => m.includes("ticket write failed again after retry")), - `the retry also failed and was logged: ${JSON.stringify(run.warnings)}` + assert.deepEqual( + (await retainedTickets()).map((t) => t.stream), + ["items"], + `a fault that outlasts one retry still ends in a ticket: ${JSON.stringify(run.warnings)}` ); - assert.equal((await retainedTickets()).length, 0, "no ticket could be written"); + } finally { + await run.finish(); + } +}); + +test("a ticket write that keeps failing does not fail startup, and shutdown stops the retry", async () => { + const run = await startOverRetainedRecord({ backend: failingEmbeddingBackend(), fault: "always" }); + try { + await run.warned("ticket write failed again; retrying"); + assert.equal((await retainedTickets()).length, 0, "no ticket could be written yet"); const response = await fetch(`http://localhost:${run.server.asPort}/.well-known/oauth-authorization-server`); assert.ok(response.status < 500, "the server still answers"); + } finally { + // finish() closes the server, which aborts the backoff sleep; a retry loop + // that ignored the abort would hang here until the test timeout. + await run.finish(); + } +}); + +test("a malformed retained semantic scope key does not stop the startup ticket", async () => { + const run = await startOverRetainedRecord({ + backend: failingEmbeddingBackend(), + retainedScopeKeys: ["not json", "null", '{"a":1}', "[1]"], + }); + try { + await run.server.startupBackfillDone; + assert.deepEqual( + (await retainedTickets()).map((t) => t.stream), + ["items"], + `canonical streams are still queued: ${JSON.stringify(run.warnings)}` + ); + } finally { + await run.finish(); + } +}); + +test("retained semantic state owes a ticket even after the manifest drops its semantic fields", async () => { + const lexicalOnly = { + ...manifest, + connector_id: "startup-retry-lexical-only", + manifest_uri: "https://example.test/startup-retry-lexical-only", + streams: manifest.streams.map(({ query: _query, ...stream }) => ({ ...stream, query: { search: {} } })), + }; + const run = await startOverRetainedRecord({ + backend: { ...makeStubBackend(), available: () => false }, + retainedScopeKeys: [JSON.stringify(["items", "subject"])], + source: lexicalOnly as never, + }); + try { + await run.server.startupBackfillDone; + assert.equal( + (await retainedTickets(lexicalOnly.connector_id)).length, + 1, + "the removed-field cleanup is owed once a backend is available" + ); } finally { await run.finish(); }