paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 48 Packages Projects Releases Wiki Activity

Merge pull request #31050 from Andy2003/bugfix/gh-30843-stiching-deadlock

Browse Source 
Prevent deadlock in catalog deferred stitching
This commit is contained in:
Fredrik Adelöw
2025-09-09 10:45:56 +02:00
committed by GitHub
parent 309941a451 a79d7cbe36
commit 16cfaa54d4
3 changed files with 227 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/late-swans-press.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/plugin-catalog-backend': patch
---
Prevent deadlock in catalog deferred stitching
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts
+135
View File
@@ -435,4 +435,139 @@ describe('markForStitching', () => {
}
},
);
it.each(databases.eachSupportedId())(
'reproduces deadlock scenario when concurrent transactions update overlapping entity sets %p',
async databaseId => {
const knex = await databases.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<DbRefreshStateRow>('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).toEqual([]);
// Verify final state - all entities should have been marked for stitching
const finalState = await knex<DbRefreshStateRow>('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();
});
},
);
});
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts
+87 -31
View File
@@ -17,9 +17,34 @@
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
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,
): e is ErrorLike {
if (knex.client.config.client.includes('pg')) {
// PostgreSQL deadlock detection
return isError(e) && e.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.
@@ -32,9 +57,8 @@ export async function markForStitching(options: {
entityRefs?: Iterable<string>;
entityIds?: Iterable<string>;
}): Promise<void> {
// Splitting to chunks just to cover pathological cases that upset the db
const entityRefs = split(options.entityRefs);
const entityIds = split(options.entityIds);
const entityRefs = sortSplit(options.entityRefs);
const entityIds = sortSplit(options.entityIds);
const knex = options.knex;
const mode = options.strategy.mode;
@@ -51,13 +75,15 @@ export async function markForStitching(options: {
.select('entity_id')
.whereIn('entity_ref', chunk),
);
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_ref', chunk);
await retryOnDeadlock(async () => {
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_ref', chunk);
}, knex);
}
for (const chunk of entityIds) {
@@ -67,44 +93,74 @@ export async function markForStitching(options: {
hash: 'force-stitching',
})
.whereIn('entity_id', chunk);
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_id', chunk);
await retryOnDeadlock(async () => {
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
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
// 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<DbRefreshStateRow>('refresh_state')
.update({
next_stitch_at: knex.fn.now(),
next_stitch_ticket: ticket,
})
.whereIn('entity_ref', chunk);
await retryOnDeadlock(async () => {
await knex<DbRefreshStateRow>('refresh_state')
.update({
next_stitch_at: knex.fn.now(),
next_stitch_ticket: ticket,
})
.whereIn('entity_ref', chunk);
}, knex);
}
for (const chunk of entityIds) {
await knex<DbRefreshStateRow>('refresh_state')
.update({
next_stitch_at: knex.fn.now(),
next_stitch_ticket: ticket,
})
.whereIn('entity_id', chunk);
await retryOnDeadlock(async () => {
await knex<DbRefreshStateRow>('refresh_state')
.update({
next_stitch_at: knex.fn.now(),
next_stitch_ticket: ticket,
})
.whereIn('entity_id', chunk);
}, knex);
}
} else {
throw new Error(`Unknown stitching strategy mode ${mode}`);
}
}
function split(input: Iterable<string> | undefined): string[][] {
function sortSplit(input: Iterable<string> | undefined): string[][] {
if (!input) {
return [];
}
return splitToChunks(Array.isArray(input) ? input : [...input], 200);
const array = Array.isArray(input) ? input.slice() : [...input];
array.sort();
return splitToChunks(array, UPDATE_CHUNK_SIZE);
}
async function retryOnDeadlock<T>(
fn: () => Promise<T>,
knex: Knex | Knex.Transaction,
retries = DEADLOCK_RETRY_ATTEMPTS,
baseMs = DEADLOCK_BASE_DELAY_MS,
): Promise<T> {
let attempt = 0;
for (;;) {
try {
return await fn();
} catch (e: unknown) {
if (isDeadlockError(knex, e) && attempt < retries) {
await sleep(baseMs * Math.pow(2, attempt));
attempt++;
continue;
}
throw e;
}
}
}