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..9e2a5bc5b 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 { 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"; import os from "node:os"; @@ -638,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, @@ -912,6 +915,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; @@ -1285,11 +1290,39 @@ 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, logger, signal = null, }: { + generation?: number; manifests: ConnectorManifest[]; logger: LoggerLike; signal?: AbortSignal | null; @@ -1301,6 +1334,26 @@ 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 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 { + 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"); @@ -1311,6 +1364,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,10 +1372,23 @@ async function runRetrievalStartupBackfill({ const semanticBackend = getSemanticBackend(); if (semanticBackend?.available()) { await semanticIndexBackfillForManifest({ + generation, log: (msg) => logger.info(msg), manifest: manifest as never, signal, }); + } 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. 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; } logger.info({ connectorId }, "retrieval startup backfill connector completed"); } catch (err) { @@ -1332,9 +1399,36 @@ async function runRetrievalStartupBackfill({ return; } logger.warn({ connectorId, err }, "retrieval startup backfill failed for connector"); + await queueRetry(manifest, err); } } + 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; + } + 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( { connectorCount: manifests.length, duration_ms: Date.now() - startedAt }, "retrieval startup backfill completed" @@ -1342,10 +1436,12 @@ async function runRetrievalStartupBackfill({ } function scheduleRetrievalStartupBackfill({ + generation = currentStorageGeneration(), manifests, logger, signal = null, }: { + generation?: number; manifests: ConnectorManifest[]; logger: LoggerLike; signal?: AbortSignal | null; @@ -1357,7 +1453,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 +8840,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 +9784,7 @@ export async function startServer(opts: ServerOpts = {}) { }); await schedulerManager.start(); const startupBackfillAbortController = new AbortController(); + const startupBackfillGeneration = currentStorageGeneration(); const startupBackfillDone = manifestReconciliationDone .then(() => collectRetrievalStartupBackfillManifests({ @@ -9694,6 +9794,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..040a1cb9d 100644 --- a/reference-implementation/server/queries/index.ts +++ b/reference-implementation/server/queries/index.ts @@ -476,10 +476,14 @@ 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..b93c40be0 --- /dev/null +++ b/reference-implementation/server/queries/search/index-dirty/startup-index-instances.sql @@ -0,0 +1,13 @@ +-- @terminator: iterate +-- @cursor_field: connector_instance_id +-- Every instance that holds records or search-index state for a connector, so +-- a failed startup backfill queues state it would have repaired or cleaned up +-- even when no active binding names the instance. An instance with neither +-- records nor index state has nothing to repair and needs no ticket. +SELECT connector_instance_id FROM records WHERE connector_id = ? +UNION 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-semantic-state.sql b/reference-implementation/server/queries/search/index-dirty/startup-semantic-state.sql new file mode 100644 index 000000000..eac7dda7d --- /dev/null +++ b/reference-implementation/server/queries/search/index-dirty/startup-semantic-state.sql @@ -0,0 +1,10 @@ +-- @terminator: one +-- Whether the connector still holds semantic index state (meta, progress or +-- vectors) for any instance. Startup uses it to owe a cleanup ticket when the +-- manifest no longer declares semantic fields but the backend is unavailable. +SELECT EXISTS ( + SELECT 1 FROM semantic_search_meta WHERE connector_id = ? + UNION ALL SELECT 1 FROM semantic_search_backfill_progress WHERE connector_id = ? + UNION ALL SELECT 1 FROM semantic_search_blob WHERE connector_id = ? + UNION ALL SELECT 1 FROM semantic_search_rowid WHERE connector_id = ? +) AS present 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..019da5f3f 100644 --- a/reference-implementation/server/search-semantic.ts +++ b/reference-implementation/server/search-semantic.ts @@ -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/startup-search-index-retry.ts b/reference-implementation/server/startup-search-index-retry.ts new file mode 100644 index 000000000..4f3a4559b --- /dev/null +++ b/reference-implementation/server/startup-search-index-retry.ts @@ -0,0 +1,54 @@ +// Copyright The PDP-Connect Contributors +// SPDX-License-Identifier: Apache-2.0 +import { assertCurrentStorageGeneration } from "./storage-generation.ts"; +import { listStartupSearchIndexInstances, markStartupSearchIndexFailure } from "./stores/search-index-dirty-store.ts"; + +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 { + 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); + const connectorId = manifest.connector_id; + const pin = manifest.storage_binding?.connector_instance_id || manifest.connector_instance_id; + 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(instances)) { + try { + // biome-ignore lint/performance/noAwaitInLoops: Never hold more than one instance deletion fence. + await markStartupSearchIndexFailure({ + connectorId, + connectorInstanceId, + generation, + }); + } catch (error) { + failures.push( + new Error(`startup search-index retry could not be queued for instance ${connectorInstanceId}`, { + cause: 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..450b93a5b 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,143 @@ export async function countDirtySearchIndexScopes(): Promise { const row = getOne<{ n: number }>(referenceQueries.searchIndexDirtyCountDirty, []); 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. + */ +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`, + [connectorInstanceId, connectorId] + ); + // 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. + 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, + ]); + } + }); + }); +} + +/** 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 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 + 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: 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/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..cfe0db791 --- /dev/null +++ b/reference-implementation/test/startup-backfill-durable-retry.test.ts @@ -0,0 +1,769 @@ +// 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 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"; +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 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 { + 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 clearRetainedPgState(); + 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 && + error.errors[0] instanceof Error && + error.errors[0].message.includes(poisoned) && + error.errors[0].cause instanceof Error + ); + 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(); + } + } +}); + +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"), 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" | "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; +}) { + 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 } + ); + 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 ${{ 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}()` + ); + } + } 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 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 (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(); + } + }, + } + ); + 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, + /** 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) { + 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 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 }); +} + +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: "once" }); + 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(); + } +}); + +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 recordsOnly = "cin_records_only"; + try { + await registerConnector(manifest, { backfillRetrievalIndexes: false }); + await ingestRecord( + { connector_id: manifest.connector_id, connector_instance_id: recordsOnly }, + { + 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(); + } + await queueStartupSearchIndexRetry(manifest as never, currentStorageGeneration()); + assert.ok((await listDirtySearchIndexScopes(25)).some((t) => t.connectorInstanceId === recordsOnly)); + } finally { + if (pg) { + await clearRetainedPgState(); + await closePostgresStorage(); + } else { + closeDb(); + } + } +}); + +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.deepEqual( + (await retainedTickets()).map((t) => t.stream), + ["items"], + `a fault that outlasts one retry still ends in a ticket: ${JSON.stringify(run.warnings)}` + ); + } 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(); + } +});