diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts index 33712b99f4..ff450e9ccf 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts @@ -436,14 +436,10 @@ describe('markForStitching', () => { }, ); - const deadlockTestDatabases = TestDatabases.create({ - ids: ['POSTGRES_17', 'POSTGRES_16', 'SQLITE_3'], - disableDocker: false, - }); - it.each(deadlockTestDatabases.eachSupportedId())( + it.each(databases.eachSupportedId())( 'reproduces deadlock scenario when concurrent transactions update overlapping entity sets %p', async databaseId => { - const knex = await deadlockTestDatabases.init(databaseId); + const knex = await databases.init(databaseId); await applyDatabaseMigrations(knex); // Setup test data with multiple entities @@ -559,7 +555,7 @@ describe('markForStitching', () => { error?.message?.includes('deadlock detected') || error?.message?.includes('deadlock'), ); - expect(deadlockErrors.length).toEqual(0); + expect(deadlockErrors).toEqual([]); // Verify final state - all entities should have been marked for stitching const finalState = await knex('refresh_state') diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts index a3d63778c3..913936202f 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts @@ -21,10 +21,25 @@ import { StitchingStrategy } from '../../../stitching/types'; import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables'; const UPDATE_CHUNK_SIZE = 100; // Smaller chunks reduce contention -const DEADLOCK_SQLSTATE = '40P01'; const DEADLOCK_RETRY_ATTEMPTS = 3; const DEADLOCK_BASE_DELAY_MS = 25; +// PostgreSQL deadlock error code +const POSTGRES_DEADLOCK_SQLSTATE = '40P01'; + +/** + * Checks if the given error is a deadlock error for the database engine in use. + */ +function isDeadlockError(knex: Knex | Knex.Transaction, e: unknown): boolean { + if (knex.client.config.client.includes('pg')) { + // PostgreSQL deadlock detection + return (e as any)?.code === POSTGRES_DEADLOCK_SQLSTATE; + } + + // Add more database engine checks here as needed + return false; +} + /** * Marks a number of entities for stitching some time in the near * future. @@ -37,9 +52,8 @@ export async function markForStitching(options: { entityRefs?: Iterable; entityIds?: Iterable; }): Promise { - // Sort inputs to ensure consistent lock order across concurrent writers - const entityRefs = split(options.entityRefs, true); - const entityIds = split(options.entityIds, true); + const entityRefs = sortSplit(options.entityRefs); + const entityIds = sortSplit(options.entityIds); const knex = options.knex; const mode = options.strategy.mode; @@ -64,7 +78,7 @@ export async function markForStitching(options: { next_update_at: knex.fn.now(), }) .whereIn('entity_ref', chunk); - }); + }, knex); } for (const chunk of entityIds) { @@ -82,7 +96,7 @@ export async function markForStitching(options: { next_update_at: knex.fn.now(), }) .whereIn('entity_id', chunk); - }); + }, knex); } } else if (mode === 'deferred') { // It's OK that this is shared across refresh state rows; it just needs to @@ -98,12 +112,10 @@ export async function markForStitching(options: { next_stitch_ticket: ticket, }) .whereIn('entity_ref', chunk); - }); + }, knex); } for (const chunk of entityIds) { - // Ensure ids are sorted for deterministic lock order - await retryOnDeadlock(async () => { await knex('refresh_state') .update({ @@ -111,26 +123,25 @@ export async function markForStitching(options: { next_stitch_ticket: ticket, }) .whereIn('entity_id', chunk); - }); + }, knex); } } else { throw new Error(`Unknown stitching strategy mode ${mode}`); } } -function split(input: Iterable | undefined, sort = false): string[][] { +function sortSplit(input: Iterable | undefined): string[][] { if (!input) { return []; } const array = Array.isArray(input) ? input.slice() : [...input]; - if (sort) { - array.sort(); - } + array.sort(); return splitToChunks(array, UPDATE_CHUNK_SIZE); } async function retryOnDeadlock( fn: () => Promise, + knex: Knex | Knex.Transaction, retries = DEADLOCK_RETRY_ATTEMPTS, baseMs = DEADLOCK_BASE_DELAY_MS, ): Promise { @@ -139,7 +150,7 @@ async function retryOnDeadlock( try { return await fn(); } catch (e: any) { - if (e?.code === DEADLOCK_SQLSTATE && attempt < retries) { + if (isDeadlockError(knex, e) && attempt < retries) { await sleep(baseMs * Math.pow(2, attempt)); attempt++; continue;