🏗️ fix: Wait Out Concurrent Index Builds on DocumentDB (#15563)

* 🏗️ fix: Wait Out Concurrent Index Builds on DocumentDB

Mongoose starts every compiled model's automatic index build in the
background as soon as the connection opens. The durable trigger delivery
boot step then issues an explicit createIndexes() for the same collection
while that build is still running. Amazon DocumentDB allows one index build
per collection and rejects the second with code 40333, which the retry
helper did not treat as transient, so the trigger service failed to
initialize and the post-listen catch exited the process on every boot.

createIndexesWithRetry now lets the model's automatic build settle before
building explicitly, and treats the single-build rejection as retryable
with a longer backoff so a peer replica's build of the same collection is
waited out instead of crash-looping the process.

Fixes #15556

* 🏗️ fix: Poll Instead of Backing Off While a Peer Holds the Index Build

A peer replica's index build time is data-dependent, so a fixed retry
budget either expires into the same process exit this change removes or
grows arbitrarily. The single-build conflict is now handled by polling at
a fixed interval with progress logging, unbounded unless a deadline is
supplied; other errors keep the existing backoff and fail-fast behaviour.

packages/data-schemas/src/utils/retry.spec.ts

@@ -0,0 +1,156 @@
+import mongoose from 'mongoose';
+import { MongoMemoryServer } from 'mongodb-memory-server';
+import type { Model } from 'mongoose';
+import type { IAgentTriggerDeliveryDocument } from '~/types/triggerDelivery';
+import triggerDeliverySchema from '~/schema/triggerDelivery';
+import { createIndexesWithRetry } from './retry';
+
+const DB_SETUP_TIMEOUT_MS = 60_000;
+/** Amazon DocumentDB: "Existing index build in progress on the same collection." */
+const INDEX_BUILD_ALREADY_IN_PROGRESS = 40333;
+const BUILD_LATENCY_MS = 5;
+
+let mongoServer: MongoMemoryServer;
+let modelCounter = 0;
+
+function indexBuildInProgressError(): Error {
+  return new mongoose.mongo.MongoServerError({
+    ok: 0,
+    code: INDEX_BUILD_ALREADY_IN_PROGRESS,
+    errmsg:
+      'Existing index build in progress on the same collection. Collection is limited to a single index build at a time.',
+  });
+}
+
+/**
+ * Turns the in-memory MongoDB into a single-index-build engine: a createIndex that
+ * arrives while another build on the same collection is in flight is rejected the
+ * way DocumentDB rejects it, instead of being serialized the way MongoDB does.
+ */
+function enforceSingleIndexBuild(model: Model<IAgentTriggerDeliveryDocument>): {
+  rejected: () => number;
+} {
+  const collection = model.collection;
+  const createIndex = collection.createIndex.bind(collection);
+  let inFlight = 0;
+  let rejected = 0;
+  collection.createIndex = async (fields, options) => {
+    if (inFlight > 0) {
+      rejected += 1;
+      throw indexBuildInProgressError();
+    }
+    inFlight += 1;
+    try {
+      await new Promise((resolve) => setTimeout(resolve, BUILD_LATENCY_MS));
+      return await createIndex(fields, options);
+    } finally {
+      inFlight -= 1;
+    }
+  };
+  return { rejected: () => rejected };
+}
+
+function compileDeliveryModel(): Model<IAgentTriggerDeliveryDocument> {
+  modelCounter += 1;
+  return mongoose.model<IAgentTriggerDeliveryDocument>(
+    `RetrySpecDelivery${modelCounter}`,
+    triggerDeliverySchema,
+    `retry_spec_deliveries_${modelCounter}`,
+  );
+}
+
+async function indexNames(model: Model<IAgentTriggerDeliveryDocument>): Promise<Set<string>> {
+  const indexes = await model.listIndexes();
+  return new Set(indexes.map((index) => index.name));
+}
+
+beforeAll(async () => {
+  mongoServer = await MongoMemoryServer.create();
+  await mongoose.connect(mongoServer.getUri());
+}, DB_SETUP_TIMEOUT_MS);
+
+afterAll(async () => {
+  await mongoose.disconnect();
+  await mongoServer.stop();
+}, DB_SETUP_TIMEOUT_MS);
+
+describe('createIndexesWithRetry on a single-index-build engine', () => {
+  test('waits for the automatic build Mongoose starts at compile before building explicitly', async () => {
+    const model = compileDeliveryModel();
+    const automaticBuildStarted = new Promise<void>((resolve) => {
+      model.once('index-single-start', () => resolve());
+    });
+    const engine = enforceSingleIndexBuild(model);
+    await automaticBuildStarted;
+
+    await createIndexesWithRetry(model, { baseDelayMs: 1, jitter: false });
+
+    await expect(model.init()).resolves.toBeUndefined();
+    const names = await indexNames(model);
+    expect(names.size).toBe(triggerDeliverySchema.indexes().length + 1);
+    expect(names.has('deliveryKey_1')).toBe(true);
+    expect(engine.rejected()).toBe(0);
+  });
+
+  test('keeps polling while another replica holds the collection, then builds', async () => {
+    const model = compileDeliveryModel();
+    await model.init();
+    const collection = model.collection;
+    const createIndex = collection.createIndex.bind(collection);
+    let remainingRejections = 6;
+    let attempts = 0;
+    collection.createIndex = async (fields, options) => {
+      attempts += 1;
+      if (remainingRejections > 0) {
+        remainingRejections -= 1;
+        throw indexBuildInProgressError();
+      }
+      return createIndex(fields, options);
+    };
+
+    await createIndexesWithRetry(model, { maxAttempts: 1, peerBuildPollMs: 1 });
+
+    expect(remainingRejections).toBe(0);
+    expect(attempts).toBeGreaterThan(6);
+    expect((await indexNames(model)).has('deliveryKey_1')).toBe(true);
+  });
+
+  test('surfaces the conflict once the peer-build deadline passes', async () => {
+    const model = compileDeliveryModel();
+    await model.init();
+    let attempts = 0;
+    model.collection.createIndex = async () => {
+      attempts += 1;
+      throw indexBuildInProgressError();
+    };
+
+    await expect(
+      createIndexesWithRetry(model, {
+        maxAttempts: 1,
+        peerBuildPollMs: 1,
+        peerBuildDeadlineMs: 20,
+      }),
+    ).rejects.toMatchObject({ code: INDEX_BUILD_ALREADY_IN_PROGRESS });
+    expect(attempts).toBeGreaterThan(triggerDeliverySchema.indexes().length);
+  });
+
+  test('still fails fast on errors that are not transient', async () => {
+    const model = compileDeliveryModel();
+    await model.init();
+    const collection = model.collection;
+    let attempts = 0;
+    collection.createIndex = async () => {
+      attempts += 1;
+      throw new mongoose.mongo.MongoServerError({
+        ok: 0,
+        code: 67,
+        errmsg: 'CannotCreateIndex: bad index spec',
+      });
+    };
+
+    await expect(createIndexesWithRetry(model, { baseDelayMs: 1, jitter: false })).rejects.toThrow(
+      'bad index spec',
+    );
+    expect(attempts).toBe(triggerDeliverySchema.indexes().length);
+  });
+});

packages/data-schemas/src/utils/retry.ts

@@ -77,17 +77,93 @@ export async function retryWithBackoff<T>(
   }
 }
 
+interface IndexedModel {
+  createIndexes: () => Promise<unknown>;
+  init?: () => Promise<unknown>;
+  modelName: string;
+}
+
+export interface IndexBuildOptions extends RetryOptions {
+  /** Interval between attempts while another build holds the collection. */
+  peerBuildPollMs?: number;
+  /** Longest wait for another build to finish; unbounded when omitted. */
+  peerBuildDeadlineMs?: number;
+}
+
+/** Amazon DocumentDB rejects a createIndex that arrives while another build on the
+ * same collection is running (code 40333), where MongoDB would serialize it. */
+const INDEX_BUILD_ALREADY_IN_PROGRESS = 40333;
+const INDEX_BUILD_IN_PROGRESS_MESSAGE = 'index build in progress';
+const DEFAULT_PEER_BUILD_POLL_MS = 5_000;
+
+function isIndexBuildInProgress(error: unknown): boolean {
+  const candidate = error as { code?: number; message?: string } | null;
+  if (candidate?.code === INDEX_BUILD_ALREADY_IN_PROGRESS) {
+    return true;
+  }
+  return (candidate?.message ?? '').toLowerCase().includes(INDEX_BUILD_IN_PROGRESS_MESSAGE);
+}
+
+/**
+ * Mongoose starts every compiled model's automatic index build in the background
+ * as soon as its connection opens. An explicit build issued while that one is
+ * still running is a second concurrent build of the same collection, which
+ * single-build engines reject outright. Letting the automatic build settle first
+ * keeps the two from overlapping; if it failed, the explicit build below is the
+ * retry, so its outcome is deliberately ignored here.
+ */
+async function settleAutomaticIndexBuild(model: IndexedModel): Promise<void> {
+  if (model.init == null) {
+    return;
+  }
+  await model.init().catch(() => undefined);
+}
+
+/**
+ * Builds the model's indexes, waiting while another process (a peer replica
+ * booting the same release) holds the collection's single index build slot.
+ * A peer's build time is data-dependent, so the wait polls rather than backs
+ * off: the process cannot become ready without these indexes, and giving up
+ * only restarts it into the same wait. Every other error propagates at once.
+ */
+async function buildIndexesWhenCollectionFree(
+  model: IndexedModel,
+  label: string,
+  options: IndexBuildOptions,
+): Promise<unknown> {
+  const pollMs = options.peerBuildPollMs ?? DEFAULT_PEER_BUILD_POLL_MS;
+  const startedAt = Date.now();
+  for (;;) {
+    try {
+      return await model.createIndexes();
+    } catch (error: unknown) {
+      const elapsedMs = Date.now() - startedAt;
+      const deadlineReached =
+        options.peerBuildDeadlineMs != null && elapsedMs >= options.peerBuildDeadlineMs;
+      if (!isIndexBuildInProgress(error) || deadlineReached) {
+        throw error;
+      }
+      logger.warn(
+        `[createIndexesWithRetry] ${label} is waiting for another index build on the collection to finish (${Math.round(elapsedMs / 1000)}s elapsed, next attempt in ${pollMs}ms)`,
+      );
+      await new Promise((resolve) => setTimeout(resolve, pollMs));
+    }
+  }
+}
+
 /**
  * Creates all indexes for a Mongoose model with deadlock retry.
- * Use this instead of raw `model.createIndexes()` on FerretDB.
+ * Use this instead of raw `model.createIndexes()` on FerretDB or DocumentDB.
  */
 export async function createIndexesWithRetry(
-  model: { createIndexes: () => Promise<unknown>; modelName: string },
-  options: RetryOptions = {},
+  model: IndexedModel,
+  options: IndexBuildOptions = {},
 ): Promise<void> {
+  await settleAutomaticIndexBuild(model);
+  const label = `createIndexes(${model.modelName})`;
   await retryWithBackoff(
-    () => model.createIndexes() as Promise<unknown>,
-    `createIndexes(${model.modelName})`,
+    () => buildIndexesWhenCollectionFree(model, label, options),
+    label,
     options,
   );
 }
@@ -98,15 +174,8 @@ export async function createIndexesWithRetry(
  * contention on the DocumentDB catalog.
  */
 export async function initializeOrgCollections(
-  models: Record<
-    string,
-    {
-      createCollection: () => Promise<unknown>;
-      createIndexes: () => Promise<unknown>;
-      modelName: string;
-    }
-  >,
-  options: RetryOptions = {},
+  models: Record<string, IndexedModel & { createCollection: () => Promise<unknown> }>,
+  options: IndexBuildOptions = {},
 ): Promise<{ totalMs: number; perModel: Array<{ name: string; ms: number }> }> {
   const perModel: Array<{ name: string; ms: number }> = [];
   const t0 = Date.now();