diff --git a/packages/data-schemas/src/utils/retry.spec.ts b/packages/data-schemas/src/utils/retry.spec.ts new file mode 100644 index 0000000000..a7dbb257c6 --- /dev/null +++ b/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): { + 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 { + modelCounter += 1; + return mongoose.model( + `RetrySpecDelivery${modelCounter}`, + triggerDeliverySchema, + `retry_spec_deliveries_${modelCounter}`, + ); +} + +async function indexNames(model: Model): Promise> { + 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((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); + }); +}); diff --git a/packages/data-schemas/src/utils/retry.ts b/packages/data-schemas/src/utils/retry.ts index 55becf76ac..0b42e7a753 100644 --- a/packages/data-schemas/src/utils/retry.ts +++ b/packages/data-schemas/src/utils/retry.ts @@ -77,17 +77,93 @@ export async function retryWithBackoff( } } +interface IndexedModel { + createIndexes: () => Promise; + init?: () => Promise; + 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 { + 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 { + 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; modelName: string }, - options: RetryOptions = {}, + model: IndexedModel, + options: IndexBuildOptions = {}, ): Promise { + await settleAutomaticIndexBuild(model); + const label = `createIndexes(${model.modelName})`; await retryWithBackoff( - () => model.createIndexes() as Promise, - `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; - createIndexes: () => Promise; - modelName: string; - } - >, - options: RetryOptions = {}, + models: Record Promise }>, + options: IndexBuildOptions = {}, ): Promise<{ totalMs: number; perModel: Array<{ name: string; ms: number }> }> { const perModel: Array<{ name: string; ms: number }> = []; const t0 = Date.now();