From 2204f5b77edff470b212b6c4ff2a2e58c41e0e44 Mon Sep 17 00:00:00 2001 From: Andreas Berger Date: Wed, 3 Sep 2025 14:53:40 +0200 Subject: [PATCH 1/3] Prevent deadlock in catalog deferred stitching resolves #30843 Signed-off-by: Andreas Berger --- .changeset/late-swans-press.md | 5 + .../stitcher/markForStitching.test.ts | 139 ++++++++++++++++++ .../operations/stitcher/markForStitching.ts | 106 +++++++++---- 3 files changed, 219 insertions(+), 31 deletions(-) create mode 100644 .changeset/late-swans-press.md diff --git a/.changeset/late-swans-press.md b/.changeset/late-swans-press.md new file mode 100644 index 0000000000..987672bd77 --- /dev/null +++ b/.changeset/late-swans-press.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Prevent deadlock in catalog deferred stitching 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 1afa0b9c30..33712b99f4 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts @@ -435,4 +435,143 @@ describe('markForStitching', () => { } }, ); + + const deadlockTestDatabases = TestDatabases.create({ + ids: ['POSTGRES_17', 'POSTGRES_16', 'SQLITE_3'], + disableDocker: false, + }); + it.each(deadlockTestDatabases.eachSupportedId())( + 'reproduces deadlock scenario when concurrent transactions update overlapping entity sets %p', + async databaseId => { + const knex = await deadlockTestDatabases.init(databaseId); + await applyDatabaseMigrations(knex); + + // Setup test data with multiple entities + const entityRefs = [ + 'k:ns/entity-a', + 'k:ns/entity-b', + 'k:ns/entity-c', + 'k:ns/entity-d', + 'k:ns/entity-e', + 'k:ns/entity-f', + ]; + + await knex('refresh_state').insert( + entityRefs.map((ref, i) => ({ + entity_id: `${i + 1}`, + entity_ref: ref, + unprocessed_entity: '{}', + processed_entity: '{}', + errors: '[]', + next_update_at: knex.fn.now(), + last_discovery_at: knex.fn.now(), + next_stitch_at: null, + next_stitch_ticket: null, + })), + ); + + // This test attempts to reproduce the deadlock by running concurrent transactions + // that update overlapping sets of entities in different orders + const errorResults = []; + + for (let attempt = 0; attempt < 10; attempt++) { + // Transaction 1: Update entities A, B, C, D, E + const transaction1 = knex.transaction(async trx => { + await markForStitching({ + knex: trx, + strategy: { + mode: 'deferred', + pollingInterval: { seconds: 1 }, + stitchTimeout: { seconds: 1 }, + }, + entityRefs: [ + 'k:ns/entity-a', + 'k:ns/entity-b', + 'k:ns/entity-c', + 'k:ns/entity-d', + 'k:ns/entity-e', + ], + }); + + // Add a small delay to increase chance of collision + await new Promise(resolve => setTimeout(resolve, 10)); + + await markForStitching({ + knex: trx, + strategy: { + mode: 'deferred', + pollingInterval: { seconds: 1 }, + stitchTimeout: { seconds: 1 }, + }, + entityRefs: ['k:ns/entity-f'], + }); + }); + + // Transaction 2: Update entities F, E, D, C, B (reverse order) + const transaction2 = knex.transaction(async trx => { + await markForStitching({ + knex: trx, + strategy: { + mode: 'deferred', + pollingInterval: { seconds: 1 }, + stitchTimeout: { seconds: 1 }, + }, + entityRefs: [ + 'k:ns/entity-f', + 'k:ns/entity-e', + 'k:ns/entity-d', + 'k:ns/entity-c', + 'k:ns/entity-b', + ], + }); + + // Add a small delay to increase chance of collision + await new Promise(resolve => setTimeout(resolve, 10)); + + await markForStitching({ + knex: trx, + strategy: { + mode: 'deferred', + pollingInterval: { seconds: 1 }, + stitchTimeout: { seconds: 1 }, + }, + entityRefs: ['k:ns/entity-a'], + }); + }); + + // Run both transactions concurrently to create potential deadlock + errorResults.push( + Promise.allSettled([transaction1, transaction2]).then(results => + results + .filter(r => r.status === 'rejected') + .map(r => (r as PromiseRejectedResult).reason), + ), + ); + } + + const allResults = await Promise.all(errorResults); + + const deadlockErrors = allResults + .flat() + .filter( + error => + error?.code === '40P01' || + error?.message?.includes('deadlock detected') || + error?.message?.includes('deadlock'), + ); + expect(deadlockErrors.length).toEqual(0); + + // Verify final state - all entities should have been marked for stitching + const finalState = await knex('refresh_state') + .select('entity_ref', 'next_stitch_at', 'next_stitch_ticket') + .whereNotNull('next_stitch_at') + .orderBy('entity_ref'); + + expect(finalState.length).toBeGreaterThan(0); + finalState.forEach(row => { + expect(row.next_stitch_at).not.toBeNull(); + expect(row.next_stitch_ticket).not.toBeNull(); + }); + }, + ); }); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts index ecc364a9cc..a3d63778c3 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts @@ -20,6 +20,11 @@ import { v4 as uuid } from 'uuid'; 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; + /** * Marks a number of entities for stitching some time in the near * future. @@ -32,9 +37,9 @@ export async function markForStitching(options: { entityRefs?: Iterable; entityIds?: Iterable; }): Promise { - // Splitting to chunks just to cover pathological cases that upset the db - const entityRefs = split(options.entityRefs); - const entityIds = split(options.entityIds); + // Sort inputs to ensure consistent lock order across concurrent writers + const entityRefs = split(options.entityRefs, true); + const entityIds = split(options.entityIds, true); const knex = options.knex; const mode = options.strategy.mode; @@ -51,13 +56,15 @@ export async function markForStitching(options: { .select('entity_id') .whereIn('entity_ref', chunk), ); - await knex - .table('refresh_state') - .update({ - result_hash: 'force-stitching', - next_update_at: knex.fn.now(), - }) - .whereIn('entity_ref', chunk); + await retryOnDeadlock(async () => { + await knex + .table('refresh_state') + .update({ + result_hash: 'force-stitching', + next_update_at: knex.fn.now(), + }) + .whereIn('entity_ref', chunk); + }); } for (const chunk of entityIds) { @@ -67,44 +74,81 @@ export async function markForStitching(options: { hash: 'force-stitching', }) .whereIn('entity_id', chunk); - await knex - .table('refresh_state') - .update({ - result_hash: 'force-stitching', - next_update_at: knex.fn.now(), - }) - .whereIn('entity_id', chunk); + await retryOnDeadlock(async () => { + await knex + .table('refresh_state') + .update({ + result_hash: 'force-stitching', + next_update_at: knex.fn.now(), + }) + .whereIn('entity_id', chunk); + }); } } else if (mode === 'deferred') { // It's OK that this is shared across refresh state rows; it just needs to // be uniquely generated for every new stitch request. const ticket = uuid(); + // Update by primary key in deterministic order to avoid deadlocks for (const chunk of entityRefs) { - await knex('refresh_state') - .update({ - next_stitch_at: knex.fn.now(), - next_stitch_ticket: ticket, - }) - .whereIn('entity_ref', chunk); + await retryOnDeadlock(async () => { + await knex('refresh_state') + .update({ + next_stitch_at: knex.fn.now(), + next_stitch_ticket: ticket, + }) + .whereIn('entity_ref', chunk); + }); } for (const chunk of entityIds) { - await knex('refresh_state') - .update({ - next_stitch_at: knex.fn.now(), - next_stitch_ticket: ticket, - }) - .whereIn('entity_id', chunk); + // Ensure ids are sorted for deterministic lock order + + await retryOnDeadlock(async () => { + await knex('refresh_state') + .update({ + next_stitch_at: knex.fn.now(), + next_stitch_ticket: ticket, + }) + .whereIn('entity_id', chunk); + }); } } else { throw new Error(`Unknown stitching strategy mode ${mode}`); } } -function split(input: Iterable | undefined): string[][] { +function split(input: Iterable | undefined, sort = false): string[][] { if (!input) { return []; } - return splitToChunks(Array.isArray(input) ? input : [...input], 200); + const array = Array.isArray(input) ? input.slice() : [...input]; + if (sort) { + array.sort(); + } + return splitToChunks(array, UPDATE_CHUNK_SIZE); +} + +async function retryOnDeadlock( + fn: () => Promise, + retries = DEADLOCK_RETRY_ATTEMPTS, + baseMs = DEADLOCK_BASE_DELAY_MS, +): Promise { + let attempt = 0; + for (;;) { + try { + return await fn(); + } catch (e: any) { + if (e?.code === DEADLOCK_SQLSTATE && attempt < retries) { + await sleep(baseMs * Math.pow(2, attempt)); + attempt++; + continue; + } + throw e; + } + } +} + +function sleep(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)); } From 2b208f1963ef4e135233a2b2f0d5a6dfd096b6a0 Mon Sep 17 00:00:00 2001 From: Andreas Berger Date: Thu, 4 Sep 2025 10:40:22 +0200 Subject: [PATCH 2/3] Adjustments after review Signed-off-by: Andreas Berger --- .../stitcher/markForStitching.test.ts | 10 ++--- .../operations/stitcher/markForStitching.ts | 41 ++++++++++++------- 2 files changed, 29 insertions(+), 22 deletions(-) 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; From a79d7cbe36e701d88e024b698ac0b0fe11fcd0f5 Mon Sep 17 00:00:00 2001 From: benjdlambert Date: Tue, 9 Sep 2025 10:32:05 +0200 Subject: [PATCH 3/3] chore: cleanup a little bit Signed-off-by: benjdlambert --- .../operations/stitcher/markForStitching.ts | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts index 913936202f..484d6e4ed6 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts @@ -17,7 +17,9 @@ import { Knex } from 'knex'; import splitToChunks from 'lodash/chunk'; import { v4 as uuid } from 'uuid'; +import { ErrorLike, isError } from '@backstage/errors'; import { StitchingStrategy } from '../../../stitching/types'; +import { setTimeout as sleep } from 'timers/promises'; import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables'; const UPDATE_CHUNK_SIZE = 100; // Smaller chunks reduce contention @@ -30,10 +32,13 @@ 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 { +function isDeadlockError( + knex: Knex | Knex.Transaction, + e: unknown, +): e is ErrorLike { if (knex.client.config.client.includes('pg')) { // PostgreSQL deadlock detection - return (e as any)?.code === POSTGRES_DEADLOCK_SQLSTATE; + return isError(e) && e.code === POSTGRES_DEADLOCK_SQLSTATE; } // Add more database engine checks here as needed @@ -149,7 +154,7 @@ async function retryOnDeadlock( for (;;) { try { return await fn(); - } catch (e: any) { + } catch (e: unknown) { if (isDeadlockError(knex, e) && attempt < retries) { await sleep(baseMs * Math.pow(2, attempt)); attempt++; @@ -159,7 +164,3 @@ async function retryOnDeadlock( } } } - -function sleep(ms: number): Promise { - return new Promise(resolve => setTimeout(resolve, ms)); -}