fix(search): queue a durable retry when startup search-index backfill fails

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",
 ];

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

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<boolean> {
+  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,17 +1364,31 @@ 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,
       });
       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,20 +1399,49 @@ 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"
   );
 }
 
 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,

reference-implementation/server/queries/index.ts

@@ -476,10 +476,14 @@ export interface ReferenceQueryRegistry extends Readonly<Record<string, Register
   readonly searchIndexDirtyGetByScope: ReadOneQuery;
   readonly searchIndexDirtyListDirty: IterateQuery;
   readonly searchIndexDirtyMarkDirty: MutationQuery;
+  readonly searchIndexDirtyMarkStartupFailure: MutationQuery;
   readonly searchIndexDirtyPark: MutationQuery;
   readonly searchIndexDirtyRearm: MutationQuery;
   readonly searchIndexDirtyRecordEvidence: MutationQuery;
   readonly searchIndexDirtyRecordFailure: MutationQuery;
+  readonly searchIndexDirtyStartupIndexInstances: IterateQuery;
+  readonly searchIndexDirtyStartupSemanticState: ReadOneQuery;
+  readonly searchIndexDirtyStartupStreams: IterateQuery;
   readonly searchIndexInsertRow: MutationQuery;
   readonly searchIndexPageAfterRowid: ReadManyQuery;
   readonly searchIndexPageByRecordKeysAfterRowid: ReadManyQuery;

reference-implementation/server/queries/search/index-dirty/mark-startup-failure.sql

@@ -0,0 +1,12 @@
+-- @terminator: exec
+INSERT INTO search_index_dirty(connector_instance_id, connector_id, stream, dirty, marked_at, revision, operator_parked)
+SELECT ?, ?, ?, 1, ?, 1, 0
+WHERE NOT EXISTS (SELECT 1 FROM connector_instances WHERE connector_instance_id = ? AND connector_id <> ?)
+  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

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 = ?

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

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'

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),

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<void> {
+  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}`);
+  }
+}