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

Merge pull request #34193 from backstage/freben/remove-immediate-stitching

Browse Source 
catalog-backend: remove immediate mode stitching
This commit is contained in:
Fredrik Adelöw
2026-05-26 14:46:15 +02:00
committed by GitHub
parent 0f5286f23c a3d55d7a0e
commit f1918e1b64
21 changed files with 923 additions and 1507 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/remove-immediate-stitching.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/plugin-catalog-backend': minor
---
Removed the immediate mode stitching strategy. All stitching now uses the deferred mode, which processes entities asynchronously via a worker queue. If your configuration includes `catalog.stitchingStrategy.mode: 'immediate'`, it will be ignored with a deprecation warning. The `pollingInterval` and `stitchTimeout` settings continue to work as before.
plugins/catalog-backend/config.d.ts
Vendored
+9 -18
View File
@@ -175,25 +175,16 @@ export interface Config {
orphanProviderStrategy?: 'keep' | 'delete';
/**
* The strategy to use when stitching together the final entities. The default mode is "deferred".
* The strategy to use when stitching together the final entities.
*/
stitchingStrategy?:
| {
/**
* Perform stitching in-band immediately when needed.
*
* @deprecated Immediate mode stitching has been deprecated and will be removed in a future release. Migrate to deferred mode (the default).
*/
mode: 'immediate';
}
| {
/** Defer stitching to be performed asynchronously */
mode: 'deferred';
/** Polling interval for tasks in seconds */
pollingInterval?: HumanDuration | string;
/** How long to wait for a stitch to complete before giving up in seconds */
stitchTimeout?: HumanDuration | string;
};
stitchingStrategy?: {
/** @deprecated Immediate mode has been removed. This field is ignored. */
mode?: string;
/** Polling interval for tasks in seconds */
pollingInterval?: HumanDuration | string;
/** How long to wait for a stitch to complete before giving up in seconds */
stitchTimeout?: HumanDuration | string;
};
/**
* The strategy to use when there is a conflict with a location being registered.
plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.test.ts
+2 -3
View File
@@ -85,10 +85,9 @@ describe.each(databases.eachSupportedId())(
}
async function entitiesMarkedForStitching(knex: Knex) {
const rows = await knex<DbRefreshStateRow>('refresh_state')
const rows = await knex('stitch_queue')
.orderBy('entity_ref')
.select('entity_ref')
.where('result_hash', '=', 'force-stitching');
.select('entity_ref');
return rows.map(r => r.entity_ref);
}
plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts
+11 -21
View File
@@ -16,11 +16,8 @@
import { Knex } from 'knex';
import lodash from 'lodash';
import {
DbFinalEntitiesRow,
DbRefreshStateReferencesRow,
DbRefreshStateRow,
} from '../../tables';
import { DbRefreshStateReferencesRow } from '../../tables';
import { markForStitching } from '../stitcher/markForStitching';
/**
* Given a number of entity refs originally created by a given entity provider
@@ -57,6 +54,10 @@ export async function deleteWithEagerPruningOfChildren(options: {
.delete()
.from('refresh_state')
.whereIn('entity_ref', refsToDelete);
await knex
.delete()
.from('stitch_queue')
.whereIn('entity_ref', refsToDelete);
}
// Delete the references that originate only from this entity provider. Note
@@ -239,7 +240,7 @@ async function markEntitiesAffectedByDeletionForStitching(options: {
// change, but not here - this code by its very definition is meant to not
// leave any orphans behind, so we can simplify away that.
const affectedIds = await knex
.select('refresh_state.entity_id AS entity_id')
.distinct('refresh_state.entity_id AS entity_id')
.from('relations')
.join(
'refresh_state',
@@ -249,19 +250,8 @@ async function markEntitiesAffectedByDeletionForStitching(options: {
.whereIn('relations.target_entity_ref', entityRefs)
.then(rows => rows.map(row => row.entity_id));
for (const ids of lodash.chunk(affectedIds, 1000)) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
.whereIn('entity_id', ids);
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_id', ids);
}
await markForStitching({
knex,
entityIds: affectedIds,
});
}
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts
+262 -534
View File
@@ -17,550 +17,278 @@
import { TestDatabases } from '@backstage/backend-test-utils';
import { applyDatabaseMigrations } from '../../migrations';
import { markForStitching } from './markForStitching';
import {
DbFinalEntitiesRow,
DbRefreshStateRow,
DbStitchQueueRow,
} from '../../tables';
import { DbRefreshStateRow, DbStitchQueueRow } from '../../tables';
jest.setTimeout(60_000);
const databases = TestDatabases.create();
describe.each(databases.eachSupportedId())(
'markForStitching, %p',
databaseId => {
it('marks the right rows in deferred mode', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
it.each(databases.eachSupportedId())(
'marks the right rows %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: '1',
entity_ref: 'k:ns/one',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '2',
entity_ref: 'k:ns/two',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '3',
entity_ref: 'k:ns/three',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '4',
entity_ref: 'k:ns/four',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
// Entity 4 has an existing stitch_queue row with old stitch data
await knex<DbStitchQueueRow>('stitch_queue').insert([
{
entity_ref: 'k:ns/four',
stitch_ticket: 'old',
next_stitch_at: '1971-01-01T00:00:00.000',
},
]);
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: '1',
entity_ref: 'k:ns/one',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '2',
entity_ref: 'k:ns/two',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '3',
entity_ref: 'k:ns/three',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '4',
entity_ref: 'k:ns/four',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
// Entity 4 has an existing stitch_queue row with old stitch data
await knex<DbStitchQueueRow>('stitch_queue').insert([
{
entity_ref: 'k:ns/four',
stitch_ticket: 'old',
next_stitch_at: '1971-01-01T00:00:00.000',
},
]);
async function result() {
return knex<DbStitchQueueRow>('stitch_queue')
.select('entity_ref', 'next_stitch_at', 'stitch_ticket')
.orderBy('entity_ref', 'asc');
}
// Initially only entity 4 has a stitch_queue row
const original = await result();
expect(original).toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
]);
// Calling with empty set should not create any new rows
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: new Set(),
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
]);
// Mark entity 1 - should create a new stitch_queue row
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: new Set(['k:ns/one']),
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Mark entity 2 - should create another new stitch_queue row
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: ['k:ns/two'],
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/two',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Mark entities 3 and 4 by ID - entity 3 creates new row, entity 4 updates existing
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityIds: ['3', '4'],
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/three',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/two',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Entity 4's ticket should have been updated (was 'old', now something else)
const final = await result();
const entity4Final = final.find(r => r.entity_ref === 'k:ns/four');
expect(entity4Final?.stitch_ticket).not.toEqual('old');
});
it('marks the right rows in immediate mode', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: '1',
entity_ref: 'k:ns/one',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '2',
entity_ref: 'k:ns/two',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '3',
entity_ref: 'k:ns/three',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '4',
entity_ref: 'k:ns/four',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
await knex<DbFinalEntitiesRow>('final_entities').insert([
{
entity_id: '1',
final_entity: '{}',
entity_ref: 'k:ns/one',
hash: 'old',
},
{
entity_id: '2',
final_entity: '{}',
entity_ref: 'k:ns/two',
hash: 'old',
},
{
entity_id: '3',
final_entity: '{}',
entity_ref: 'k:ns/three',
hash: 'old',
},
{
entity_id: '4',
final_entity: '{}',
entity_ref: 'k:ns/four',
hash: 'old',
},
]);
async function result() {
return knex<DbRefreshStateRow>('refresh_state')
.leftJoin(
'final_entities',
'final_entities.entity_id',
'refresh_state.entity_id',
)
.select({
entity_id: 'refresh_state.entity_id',
next_update_at: 'refresh_state.next_update_at',
refresh_state_hash: 'refresh_state.result_hash',
final_entities_hash: 'final_entities.hash',
})
.orderBy('entity_id', 'asc');
}
// Ensure that now() isn't evaluating to the same thing
await new Promise(resolve => setTimeout(resolve, 1100));
const original = await result();
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityRefs: new Set(),
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
]);
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityRefs: new Set(['k:ns/one']),
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
]);
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityRefs: ['k:ns/two'],
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
]);
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityIds: ['3', '4'],
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
]);
// It overwrites timers
const final = await result();
for (let i = 0; i < final.length; ++i) {
expect(original[i].next_update_at).not.toEqual(final[i].next_update_at);
}
});
it('reproduces deadlock scenario when concurrent transactions update overlapping entity sets', async () => {
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(),
})),
);
// 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<DbStitchQueueRow>('stitch_queue')
async function result() {
return knex<DbStitchQueueRow>('stitch_queue')
.select('entity_ref', 'next_stitch_at', 'stitch_ticket')
.orderBy('entity_ref');
.orderBy('entity_ref', 'asc');
}
expect(finalState.length).toBeGreaterThan(0);
finalState.forEach(row => {
expect(row.next_stitch_at).not.toBeNull();
expect(row.stitch_ticket).not.toBeNull();
// Initially only entity 4 has a stitch_queue row
const original = await result();
expect(original).toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
]);
// Calling with empty set should not create any new rows
await markForStitching({
knex,
entityRefs: new Set(),
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
]);
// Mark entity 1 - should create a new stitch_queue row
await markForStitching({
knex,
entityRefs: new Set(['k:ns/one']),
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Mark entity 2 - should create another new stitch_queue row
await markForStitching({
knex,
entityRefs: ['k:ns/two'],
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/two',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Mark entities 3 and 4 by ID - entity 3 creates new row, entity 4 updates existing
await markForStitching({
knex,
entityIds: ['3', '4'],
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/three',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/two',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Entity 4's ticket should have been updated (was 'old', now something else)
const final = await result();
const entity4Final = final.find(r => r.entity_ref === 'k:ns/four');
expect(entity4Final?.stitch_ticket).not.toEqual('old');
},
);
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(),
})),
);
// 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,
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,
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,
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,
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<DbStitchQueueRow>('stitch_queue')
.select('entity_ref', 'next_stitch_at', 'stitch_ticket')
.orderBy('entity_ref');
expect(finalState.length).toBeGreaterThan(0);
finalState.forEach(row => {
expect(row.next_stitch_at).not.toBeNull();
expect(row.stitch_ticket).not.toBeNull();
});
},
);
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts
+39 -86
View File
@@ -17,12 +17,7 @@
import { Knex } from 'knex';
import splitToChunks from 'lodash/chunk';
import { randomUUID as uuid } from 'node:crypto';
import { StitchingStrategy } from '../../../stitching/types';
import {
DbFinalEntitiesRow,
DbRefreshStateRow,
DbStitchQueueRow,
} from '../../tables';
import { DbRefreshStateRow, DbStitchQueueRow } from '../../tables';
import { retryOnDeadlock } from '../../util';
const UPDATE_CHUNK_SIZE = 100; // Smaller chunks reduce contention
@@ -35,96 +30,54 @@ const UPDATE_CHUNK_SIZE = 100; // Smaller chunks reduce contention
*/
export async function markForStitching(options: {
knex: Knex | Knex.Transaction;
strategy: StitchingStrategy;
entityRefs?: Iterable<string>;
entityIds?: Iterable<string>;
}): Promise<void> {
const entityRefs = sortSplit(options.entityRefs);
const entityIds = sortSplit(options.entityIds);
const knex = options.knex;
const mode = options.strategy.mode;
if (mode === 'immediate') {
for (const chunk of entityRefs) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
.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);
}
// It's OK that this is shared across stitch_queue rows; it just needs to
// be uniquely generated for every new stitch request.
const ticket = uuid();
for (const chunk of entityIds) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
for (const chunk of entityRefs) {
await retryOnDeadlock(async () => {
if (chunk.length > 0) {
await knex<DbStitchQueueRow>('stitch_queue')
.insert(
chunk.map(ref => ({
entity_ref: ref,
stitch_ticket: ticket,
next_stitch_at: knex.fn.now(),
})),
)
.onConflict('entity_ref')
.merge(['next_stitch_at', 'stitch_ticket']);
}
}, knex);
}
for (const chunk of entityIds) {
await retryOnDeadlock(async () => {
// Look up entity_refs from refresh_state by entity_id
const refreshStateRows = await knex<DbRefreshStateRow>('refresh_state')
.select('entity_ref')
.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 stitch_queue rows; it just needs to
// be uniquely generated for every new stitch request.
const ticket = uuid();
for (const chunk of entityRefs) {
await retryOnDeadlock(async () => {
if (chunk.length > 0) {
await knex<DbStitchQueueRow>('stitch_queue')
.insert(
chunk.map(ref => ({
entity_ref: ref,
stitch_ticket: ticket,
next_stitch_at: knex.fn.now(),
})),
)
.onConflict('entity_ref')
.merge(['next_stitch_at', 'stitch_ticket']);
}
}, knex);
}
for (const chunk of entityIds) {
await retryOnDeadlock(async () => {
// Look up entity_refs from refresh_state by entity_id
const refreshStateRows = await knex<DbRefreshStateRow>('refresh_state')
.select('entity_ref')
.whereIn('entity_id', chunk);
if (refreshStateRows.length > 0) {
await knex<DbStitchQueueRow>('stitch_queue')
.insert(
refreshStateRows.map(row => ({
entity_ref: row.entity_ref,
stitch_ticket: ticket,
next_stitch_at: knex.fn.now(),
})),
)
.onConflict('entity_ref')
.merge(['next_stitch_at', 'stitch_ticket']);
}
}, knex);
}
} else {
throw new Error(`Unknown stitching strategy mode ${mode}`);
if (refreshStateRows.length > 0) {
await knex<DbStitchQueueRow>('stitch_queue')
.insert(
refreshStateRows.map(row => ({
entity_ref: row.entity_ref,
stitch_ticket: ticket,
next_stitch_at: knex.fn.now(),
})),
)
.onConflict('entity_ref')
.merge(['next_stitch_at', 'stitch_ticket']);
}
}, knex);
}
}
plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts
+360 -352
View File
@@ -31,366 +31,275 @@ jest.setTimeout(60_000);
const databases = TestDatabases.create();
describe.each(databases.eachSupportedId())(
'performStitching, %p',
databaseId => {
it.each(databases.eachSupportedId())(
'runs the happy path for %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
const logger = mockServices.logger.mock();
// NOTE(freben): Testing the deferred path since it's a superset of the immediate one
it('runs the happy path in deferred mode', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
let entities: DbFinalEntitiesRow[];
let entity: Entity;
let entities: DbFinalEntitiesRow[];
let entity: Entity;
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
unprocessed_entity: JSON.stringify({}),
processed_entity: JSON.stringify({
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
},
spec: {
k: 'v',
},
}),
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
await knex<DbRefreshStateReferencesRow>(
'refresh_state_references',
).insert([{ source_key: 'a', target_entity_ref: 'k:ns/n' }]);
await knex<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
// handles and ignores duplicates
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
]);
const deferredStrategy = {
mode: 'deferred' as const,
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
};
await markForStitching({
knex,
strategy: deferredStrategy,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
strategy: deferredStrategy,
entityRef: 'k:ns/n',
stitchTicket: (
await knex('stitch_queue')
.where('entity_ref', 'k:ns/n')
.select('stitch_ticket')
.first()
)?.stitch_ticket,
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: [
{
type: 'looksAt',
targetRef: 'k:ns/other',
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
unprocessed_entity: JSON.stringify({}),
processed_entity: JSON.stringify({
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
},
],
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entity.metadata.etag).toEqual(entities[0].hash);
const last_updated_at = entities[0].last_updated_at;
expect(last_updated_at).not.toBeNull();
const firstHash = entities[0].hash;
const search = await knex<DbSearchRow>('search');
expect(search).toEqual(
expect.arrayContaining([
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/other',
value: 'k:ns/other',
spec: {
k: 'v',
},
{
entity_id: 'my-id',
key: 'apiversion',
original_value: 'a',
value: 'a',
},
{
entity_id: 'my-id',
key: 'kind',
original_value: 'k',
value: 'k',
},
{
entity_id: 'my-id',
key: 'metadata.name',
original_value: 'n',
value: 'n',
},
{
entity_id: 'my-id',
key: 'metadata.namespace',
original_value: 'ns',
value: 'ns',
},
{
entity_id: 'my-id',
key: 'metadata.uid',
original_value: 'my-id',
value: 'my-id',
},
{
entity_id: 'my-id',
key: 'spec.k',
original_value: 'v',
value: 'v',
},
]),
);
// Re-stitch without any changes
await markForStitching({
knex,
strategy: deferredStrategy,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
strategy: deferredStrategy,
entityRef: 'k:ns/n',
stitchTicket: (
await knex('stitch_queue')
.where('entity_ref', 'k:ns/n')
.select('stitch_ticket')
.first()
)?.stitch_ticket,
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entities[0].hash).toEqual(firstHash);
expect(entity.metadata.etag).toEqual(firstHash);
// Now add one more relation and re-stitch
await knex<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/third',
},
]);
await markForStitching({
knex,
strategy: deferredStrategy,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
strategy: deferredStrategy,
entityRef: 'k:ns/n',
stitchTicket: (
await knex('stitch_queue')
.where('entity_ref', 'k:ns/n')
.select('stitch_ticket')
.first()
)?.stitch_ticket,
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: expect.arrayContaining([
{
type: 'looksAt',
targetRef: 'k:ns/other',
},
{
type: 'looksAt',
targetRef: 'k:ns/third',
},
]),
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entities[0].hash).not.toEqual(firstHash);
expect(entities[0].hash).toEqual(entity.metadata.etag);
expect(await knex<DbSearchRow>('search')).toEqual(
expect.arrayContaining([
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/other',
value: 'k:ns/other',
},
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/third',
value: 'k:ns/third',
},
{
entity_id: 'my-id',
key: 'apiversion',
original_value: 'a',
value: 'a',
},
{
entity_id: 'my-id',
key: 'kind',
original_value: 'k',
value: 'k',
},
{
entity_id: 'my-id',
key: 'metadata.name',
original_value: 'n',
value: 'n',
},
{
entity_id: 'my-id',
key: 'metadata.namespace',
original_value: 'ns',
value: 'ns',
},
{
entity_id: 'my-id',
key: 'metadata.uid',
original_value: 'my-id',
value: 'my-id',
},
{
entity_id: 'my-id',
key: 'spec.k',
original_value: 'v',
value: 'v',
},
]),
);
});
it('handles conflicts with past stitches', async () => {
if (databaseId === 'MYSQL_8') {
// MySQL doesn't handle conflicts in the same way as the other two, most
// likely due to the conflict probably being handled with a merged even
// if it's for the entity_ref. This probably means that MySQL will have
// inconsistencies in the final_entities table in some cases, but they
// should heal fairly quickly.
return;
}
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
// Drop the foreign key constraint so we can insert a conflicting entity
// where the ID is inconsistent with the entity ref across refresh_state
// and final_entities
await knex.schema.alterTable('final_entities', table => {
table.dropForeign('entity_id');
});
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: 'other-id',
entity_ref: 'k:ns/n',
unprocessed_entity: JSON.stringify({}),
processed_entity: JSON.stringify({
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
},
spec: {
k: 'v',
},
}),
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
await knex<DbFinalEntitiesRow>('final_entities').insert([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
hash: '',
final_entity: JSON.stringify({}),
},
]);
const stitchLogger = mockServices.logger.mock();
await expect(
performStitching({
knex,
logger: stitchLogger,
strategy: { mode: 'immediate' },
entityRef: 'k:ns/n',
}),
).resolves.toBe('abandoned');
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
await knex<DbRefreshStateReferencesRow>('refresh_state_references').insert([
{ source_key: 'a', target_entity_ref: 'k:ns/n' },
]);
await knex<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
// handles and ignores duplicates
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
]);
expect(stitchLogger.debug).toHaveBeenCalledWith(
'Skipping stitching of k:ns/n, conflict',
expect.anything(),
);
await markForStitching({
knex,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
entityRef: 'k:ns/n',
stitchTicket: await getStitchTicket(knex, 'k:ns/n'),
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: [
{
type: 'looksAt',
targetRef: 'k:ns/other',
},
],
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entity.metadata.etag).toEqual(entities[0].hash);
const last_updated_at = entities[0].last_updated_at;
expect(last_updated_at).not.toBeNull();
const firstHash = entities[0].hash;
const search = await knex<DbSearchRow>('search');
expect(search).toEqual(
expect.arrayContaining([
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/other',
value: 'k:ns/other',
},
{
entity_id: 'my-id',
key: 'apiversion',
original_value: 'a',
value: 'a',
},
{
entity_id: 'my-id',
key: 'kind',
original_value: 'k',
value: 'k',
},
{
entity_id: 'my-id',
key: 'metadata.name',
original_value: 'n',
value: 'n',
},
{
entity_id: 'my-id',
key: 'metadata.namespace',
original_value: 'ns',
value: 'ns',
},
{
entity_id: 'my-id',
key: 'metadata.uid',
original_value: 'my-id',
value: 'my-id',
},
{
entity_id: 'my-id',
key: 'spec.k',
original_value: 'v',
value: 'v',
},
]),
);
// Re-stitch without any changes
await markForStitching({
knex,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
entityRef: 'k:ns/n',
stitchTicket: await getStitchTicket(knex, 'k:ns/n'),
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entities[0].hash).toEqual(firstHash);
expect(entity.metadata.etag).toEqual(firstHash);
// Now add one more relation and re-stitch
await knex<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/third',
},
]);
await markForStitching({
knex,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
entityRef: 'k:ns/n',
stitchTicket: await getStitchTicket(knex, 'k:ns/n'),
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: expect.arrayContaining([
{
type: 'looksAt',
targetRef: 'k:ns/other',
},
{
type: 'looksAt',
targetRef: 'k:ns/third',
},
]),
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entities[0].hash).not.toEqual(firstHash);
expect(entities[0].hash).toEqual(entity.metadata.etag);
expect(await knex<DbSearchRow>('search')).toEqual(
expect.arrayContaining([
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/other',
value: 'k:ns/other',
},
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/third',
value: 'k:ns/third',
},
{
entity_id: 'my-id',
key: 'apiversion',
original_value: 'a',
value: 'a',
},
{
entity_id: 'my-id',
key: 'kind',
original_value: 'k',
value: 'k',
},
{
entity_id: 'my-id',
key: 'metadata.name',
original_value: 'n',
value: 'n',
},
{
entity_id: 'my-id',
key: 'metadata.namespace',
original_value: 'ns',
value: 'ns',
},
{
entity_id: 'my-id',
key: 'metadata.uid',
original_value: 'my-id',
value: 'my-id',
},
{
entity_id: 'my-id',
key: 'spec.k',
original_value: 'v',
value: 'v',
},
]),
);
},
);
describe.each(databases.eachSupportedId())(
'performStitching edge cases, %p',
databaseId => {
it('stitches when final_entities row already exists', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
@@ -425,13 +334,15 @@ describe.each(databases.eachSupportedId())(
},
]);
await markForStitching({ knex, entityRefs: ['k:ns/n'] });
const stitchLogger = mockServices.logger.mock();
await expect(
performStitching({
knex,
logger: stitchLogger,
strategy: { mode: 'immediate' },
entityRef: 'k:ns/n',
stitchTicket: await getStitchTicket(knex, 'k:ns/n'),
}),
).resolves.toBe('changed');
@@ -440,5 +351,102 @@ describe.each(databases.eachSupportedId())(
expect(entities[0].hash).not.toBe('');
expect(entities[0].final_entity).toBeDefined();
});
it('rejects a stale stitch ticket without overwriting', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
unprocessed_entity: JSON.stringify({}),
processed_entity: JSON.stringify({
apiVersion: 'a',
kind: 'k',
metadata: { name: 'n', namespace: 'ns' },
spec: { original: true },
}),
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
// First stitch: create the final_entities row with a valid ticket
await markForStitching({ knex, entityRefs: ['k:ns/n'] });
const validTicket = await getStitchTicket(knex, 'k:ns/n');
const result1 = await performStitching({
knex,
logger: mockServices.logger.mock(),
entityRef: 'k:ns/n',
stitchTicket: validTicket,
});
expect(result1).toBe('changed');
const afterFirst = await knex<DbFinalEntitiesRow>('final_entities');
expect(afterFirst).toHaveLength(1);
const firstHash = afterFirst[0].hash;
// Now change the processed entity and mark for stitching again
await knex<DbRefreshStateRow>('refresh_state')
.where('entity_id', 'my-id')
.update({
processed_entity: JSON.stringify({
apiVersion: 'a',
kind: 'k',
metadata: { name: 'n', namespace: 'ns' },
spec: { original: false, stale: true },
}),
});
await markForStitching({ knex, entityRefs: ['k:ns/n'] });
const freshTicket = await getStitchTicket(knex, 'k:ns/n');
// Attempt to stitch with a WRONG ticket (simulating a stale worker)
const result2 = await performStitching({
knex,
logger: mockServices.logger.mock(),
entityRef: 'k:ns/n',
stitchTicket: 'stale-ticket-that-does-not-match',
});
expect(result2).toBe('abandoned');
// The final_entities row should still have the FIRST hash,
// not be overwritten by the stale worker
const afterStale = await knex<DbFinalEntitiesRow>('final_entities');
expect(afterStale).toHaveLength(1);
expect(afterStale[0].hash).toBe(firstHash);
// Now stitch with the correct fresh ticket — should succeed
const result3 = await performStitching({
knex,
logger: mockServices.logger.mock(),
entityRef: 'k:ns/n',
stitchTicket: freshTicket,
});
expect(result3).toBe('changed');
const afterFresh = await knex<DbFinalEntitiesRow>('final_entities');
expect(afterFresh).toHaveLength(1);
expect(afterFresh[0].hash).not.toBe(firstHash);
const freshEntity = JSON.parse(afterFresh[0].final_entity!);
expect(freshEntity.spec).toEqual({ original: false, stale: true });
});
},
);
async function getStitchTicket(
knex: import('knex').Knex,
entityRef: string,
): Promise<string> {
const row = await knex('stitch_queue')
.where('entity_ref', entityRef)
.select('stitch_ticket')
.first();
if (!row) {
throw new Error(`No stitch_queue entry for ${entityRef}`);
}
return row.stitch_ticket;
}
plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts
+64 -71
View File
@@ -26,19 +26,11 @@ import { SerializedError } from '@backstage/errors';
import { Knex } from 'knex';
import { createHash } from 'node:crypto';
import stableStringify from 'fast-json-stable-stringify';
import { StitchingStrategy } from '../../../stitching/types';
import {
DbFinalEntitiesRow,
DbRefreshStateRow,
DbStitchQueueRow,
} from '../../tables';
import { DbFinalEntitiesRow, DbStitchQueueRow } from '../../tables';
import { buildEntitySearch } from './buildEntitySearch';
import { markDeferredStitchCompleted } from './markDeferredStitchCompleted';
import { syncSearchRows } from './syncSearchRows';
import {
LoggerService,
isDatabaseConflictError,
} from '@backstage/backend-plugin-api';
import { LoggerService } from '@backstage/backend-plugin-api';
function generateStableHash(entity: Entity) {
return createHash('sha1')
@@ -59,56 +51,21 @@ const scriptProtocolPattern =
export async function performStitching(options: {
knex: Knex | Knex.Transaction;
logger: LoggerService;
strategy: StitchingStrategy;
entityRef: string;
stitchTicket?: string;
stitchTicket: string;
}): Promise<'changed' | 'unchanged' | 'abandoned'> {
const { knex, logger, entityRef } = options;
const stitchTicket = options.stitchTicket;
const { knex, logger, entityRef, stitchTicket } = options;
// In deferred mode, the entity is removed from the stitch queue on ANY
// completion, except when an exception is thrown. In the latter case, the
// entity will be retried at a later time.
let removeFromStitchQueueOnCompletion = options.strategy.mode === 'deferred';
// The entity is removed from the stitch queue on ANY completion, except when
// an exception is thrown. In the latter case, the entity will be retried at a
// later time.
let removeFromStitchQueueOnCompletion = true;
try {
const entityResult = await knex<DbRefreshStateRow>('refresh_state')
.where({ entity_ref: entityRef })
.limit(1)
.select('entity_id');
if (!entityResult.length) {
// Entity does no exist in refresh state table, no stitching required.
return 'abandoned';
}
// Ensure that a final_entities row exists for this entity.
try {
await knex<DbFinalEntitiesRow>('final_entities')
.insert({
entity_id: entityResult[0].entity_id,
hash: '',
entity_ref: entityRef,
})
.onConflict('entity_id')
.ignore();
} catch (error) {
// It's possible to hit a race where a refresh_state table delete + insert
// is done just after we read the entity_id from it. This conflict is safe
// to ignore because the current stitching operation will be triggered by
// the old entry, and the new entry will trigger it's own stitching that
// will update the entity.
if (isDatabaseConflictError(error)) {
logger.debug(`Skipping stitching of ${entityRef}, conflict`, error);
return 'abandoned';
}
throw error;
}
// Selecting from refresh_state and final_entities should yield exactly
// one row (except in abnormal cases where the stitch was invoked for
// something that didn't exist at all, in which case it's zero rows).
// The join with the temporary incoming_references still gives one row.
// Selecting from refresh_state (with an optional left join to
// final_entities for the previous hash) should yield exactly one row,
// except in abnormal cases where the entity was deleted between the
// stitch request and now.
const [processedResult, relationsResult] = await Promise.all([
knex
.with('incoming_references', function incomingReferences(builder) {
@@ -238,30 +195,66 @@ export async function performStitching(options: {
// to write the search index.
const searchEntries = buildEntitySearch(entityId, entity);
let updateQuery = knex<DbFinalEntitiesRow>('final_entities')
.update({
// Guard against concurrent stitchers: if our stitch_ticket no longer
// matches stitch_queue, another worker has newer data and we should
// not overwrite it. On PostgreSQL and SQLite, this is done atomically
// via a WHERE on the upsert merge path. MySQL does not support
// ON CONFLICT ... DO UPDATE ... WHERE, so it uses a separate check
// with a negligible TOCTOU window (self-corrects on the next ~1s
// stitch cycle).
const isMySQL = String(knex.client.config.client).includes('mysql');
if (isMySQL) {
const ticketValid = await knex<DbStitchQueueRow>('stitch_queue')
.where('entity_ref', entityRef)
.where('stitch_ticket', stitchTicket)
.first();
if (!ticketValid) {
logger.debug(
`Entity ${entityRef} is already stitched, skipping write.`,
);
return 'abandoned';
}
}
let upsert = knex<DbFinalEntitiesRow>('final_entities')
.insert({
entity_id: entityId,
entity_ref: entityRef,
final_entity: JSON.stringify(entity),
hash,
last_updated_at: knex.fn.now(),
})
.where('entity_id', entityId);
.onConflict('entity_id')
.merge(['final_entity', 'hash', 'last_updated_at']);
// In deferred mode, guard against concurrent stitchers by checking that
// the stitch_ticket in stitch_queue still matches what we were given.
if (options.strategy.mode === 'deferred' && stitchTicket) {
updateQuery = updateQuery.whereExists(
knex<DbStitchQueueRow>('stitch_queue')
.where('stitch_queue.entity_ref', entityRef)
.where('stitch_queue.stitch_ticket', stitchTicket)
.select(knex.raw('1')),
if (!isMySQL) {
upsert = upsert.where(
knex.raw(
'exists (select 1 from stitch_queue where entity_ref = ? and stitch_ticket = ?)',
[entityRef, stitchTicket],
),
);
}
const amountOfRowsChanged = await updateQuery;
await upsert;
if (amountOfRowsChanged === 0) {
logger.debug(`Entity ${entityRef} is already stitched, skipping write.`);
return 'abandoned';
// Verify the write took effect. INSERT return values vary across
// database engines (row IDs vs row counts vs empty arrays), so we
// check the hash directly — we already know hash !== previousHash
// from the check above, so a mismatch means the write was blocked.
if (!isMySQL) {
const written = await knex<DbFinalEntitiesRow>('final_entities')
.where('entity_id', entityId)
.where('hash', hash)
.select(knex.raw('1'))
.first();
if (!written) {
logger.debug(
`Entity ${entityRef} is already stitched, skipping write.`,
);
return 'abandoned';
}
}
await syncSearchRows(knex, entityId, searchEntries);
@@ -271,7 +264,7 @@ export async function performStitching(options: {
removeFromStitchQueueOnCompletion = false;
throw error;
} finally {
if (removeFromStitchQueueOnCompletion && stitchTicket) {
if (removeFromStitchQueueOnCompletion) {
await markDeferredStitchCompleted({
knex: knex,
entityRef,
plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts
+4 -96
View File
@@ -16,7 +16,6 @@
import { TestDatabases } from '@backstage/backend-test-utils';
import { Knex } from 'knex';
import { StitchingStrategy } from '../../../stitching/types';
import { applyDatabaseMigrations } from '../../migrations';
import {
DbFinalEntitiesRow,
@@ -39,17 +38,14 @@ describe.each(databases.eachSupportedId())(
return knex;
}
async function run(
knex: Knex,
strategy: StitchingStrategy,
): Promise<number> {
async function run(knex: Knex): Promise<number> {
let result: number;
await knex.transaction(
async tx => {
// We can't return here, as knex swallows the return type in case the
// transaction is rolled back:
// https://github.com/knex/knex/blob/e37aeaa31c8ef9c1b07d2e4d3ec6607e557d800d/lib/transaction.js#L136
result = await deleteOrphanedEntities({ knex: tx, strategy });
result = await deleteOrphanedEntities({ knex: tx });
},
{
// If we explicitly trigger a rollback, don't fail.
@@ -133,7 +129,7 @@ describe.each(databases.eachSupportedId())(
});
}
it('works for some mixed paths in immediate mode', async () => {
it('works for some mixed paths', async () => {
/*
In this graph, edges represent refresh state references, not entity relations:
@@ -189,95 +185,7 @@ describe.each(databases.eachSupportedId())(
await insertRelation(knex, 'E2', 'E3');
await insertRelation(knex, 'E10', 'E6');
await insertRelation(knex, 'E7', 'E6');
await expect(run(knex, { mode: 'immediate' })).resolves.toEqual(5);
await expect(refreshState(knex)).resolves.toEqual([
{ entity_ref: 'E1', result_hash: 'original' },
{ entity_ref: 'E2', result_hash: 'force-stitching' },
{ entity_ref: 'E7', result_hash: 'force-stitching' },
{ entity_ref: 'E8', result_hash: 'original' },
{ entity_ref: 'E9', result_hash: 'original' },
]);
await expect(stitchQueue(knex)).resolves.toEqual([]);
await expect(finalEntities(knex)).resolves.toEqual([
{ entity_ref: 'E1', hash: 'original', next_stitch_at: null },
{
entity_ref: 'E2',
hash: 'force-stitching',
next_stitch_at: null,
},
{
entity_ref: 'E7',
hash: 'force-stitching',
next_stitch_at: null,
},
{ entity_ref: 'E8', hash: 'original', next_stitch_at: null },
{ entity_ref: 'E9', hash: 'original', next_stitch_at: null },
]);
});
it('works for some mixed paths in deferred mode', async () => {
/*
In this graph, edges represent refresh state references, not entity relations:
P1 - E1 -- E2
/
E3
/
E4
\
E5
/
E6
\
E7
/
P2 - E8
P3 - E9
E10
Result: E3, E4, E5, E6, and E10 deleted; others remain
Entities that had relations pointing at orphans are marked for reprocessing
*/
const knex = await createDatabase();
await insertEntity(
knex,
'E1',
'E2',
'E3',
'E4',
'E5',
'E6',
'E7',
'E8',
'E9',
'E10',
);
await insertReference(
knex,
{ source_key: 'P1', target_entity_ref: 'E1' },
{ source_entity_ref: 'E1', target_entity_ref: 'E2' },
{ source_entity_ref: 'E3', target_entity_ref: 'E2' },
{ source_entity_ref: 'E4', target_entity_ref: 'E3' },
{ source_entity_ref: 'E4', target_entity_ref: 'E5' },
{ source_entity_ref: 'E6', target_entity_ref: 'E5' },
{ source_entity_ref: 'E6', target_entity_ref: 'E7' },
{ source_key: 'P2', target_entity_ref: 'E8' },
{ source_entity_ref: 'E8', target_entity_ref: 'E7' },
{ source_key: 'P3', target_entity_ref: 'E9' },
);
await insertRelation(knex, 'E1', 'E2');
await insertRelation(knex, 'E2', 'E3');
await insertRelation(knex, 'E10', 'E6');
await insertRelation(knex, 'E7', 'E6');
await expect(
run(knex, {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
}),
).resolves.toEqual(5);
await expect(run(knex)).resolves.toEqual(5);
await expect(refreshState(knex)).resolves.toEqual([
{ entity_ref: 'E1', result_hash: 'original' },
{ entity_ref: 'E2', result_hash: 'original' },
plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts
+1 -4
View File
@@ -16,7 +16,6 @@
import { Knex } from 'knex';
import uniq from 'lodash/uniq';
import { StitchingStrategy } from '../../../stitching/types';
import { DbRefreshStateRow } from '../../tables';
import { markForStitching } from '../stitcher/markForStitching';
@@ -27,9 +26,8 @@ import { markForStitching } from '../stitcher/markForStitching';
*/
export async function deleteOrphanedEntities(options: {
knex: Knex.Transaction | Knex;
strategy: StitchingStrategy;
}): Promise<number> {
const { knex, strategy } = options;
const { knex } = options;
let total = 0;
@@ -83,7 +81,6 @@ export async function deleteOrphanedEntities(options: {
// Mark all of the things that the orphans had relations to for stitching
await markForStitching({
knex,
strategy,
entityIds: orphanRelationIds,
});
}
plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts
+19 -25
View File
@@ -20,11 +20,17 @@ import waitForExpect from 'wait-for-expect';
import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase';
import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
import { CatalogProcessingOrchestrator } from './types';
import { Stitcher } from '../stitching/types';
import { ConfigReader } from '@backstage/config';
import { mockServices } from '@backstage/backend-test-utils';
import { metricsServiceMock } from '@backstage/backend-test-utils/alpha';
jest.mock('../database/operations/stitcher/markForStitching', () => ({
markForStitching: jest.fn().mockResolvedValue(undefined),
}));
const { markForStitching } = jest.requireMock(
'../database/operations/stitcher/markForStitching',
) as { markForStitching: jest.Mock };
describe('DefaultCatalogProcessingEngine', () => {
const db = {
transaction: jest.fn(),
@@ -37,9 +43,6 @@ describe('DefaultCatalogProcessingEngine', () => {
const orchestrator: jest.Mocked<CatalogProcessingOrchestrator> = {
process: jest.fn(),
};
const stitcher = {
stitch: jest.fn(),
} as unknown as jest.Mocked<Stitcher>;
const hash = {
update: () => hash,
digest: jest.fn(),
@@ -69,7 +72,6 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
createHash: () => hash,
scheduler: mockServices.scheduler(),
events: mockServices.events.mock(),
@@ -139,7 +141,6 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
events: mockServices.events.mock(),
@@ -225,7 +226,6 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
events: mockServices.events.mock(),
@@ -305,7 +305,6 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
events: mockServices.events.mock(),
@@ -367,7 +366,6 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
pollingIntervalMs: 100,
@@ -467,12 +465,12 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(2);
expect(markForStitching).toHaveBeenCalledTimes(2);
});
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other2']),
);
expect([...stitcher.stitch.mock.calls[1][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[1][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other3']),
);
await engine.stop();
@@ -485,7 +483,6 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
pollingIntervalMs: 100,
@@ -572,15 +569,15 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(2);
expect(markForStitching).toHaveBeenCalledTimes(2);
});
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1']),
);
// As a result of switching the relationship for source other1 to
// a new target entity, the other1 relationship source must be
// restitched.
expect([...stitcher.stitch.mock.calls[1][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[1][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1']),
);
await engine.stop();
@@ -593,7 +590,6 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
pollingIntervalMs: 100,
@@ -664,9 +660,9 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(1);
expect(markForStitching).toHaveBeenCalledTimes(1);
});
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me']),
);
await engine.stop();
@@ -679,7 +675,6 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
pollingIntervalMs: 100,
@@ -755,9 +750,9 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(1);
expect(markForStitching).toHaveBeenCalledTimes(1);
});
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other2']),
);
await engine.stop();
@@ -770,7 +765,6 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
pollingIntervalMs: 100,
@@ -836,9 +830,9 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(1);
expect(markForStitching).toHaveBeenCalledTimes(1);
});
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other2']),
);
await engine.stop();
plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts
+5 -11
View File
@@ -27,7 +27,7 @@ import { trace } from '@opentelemetry/api';
import { ProcessingDatabase, RefreshStateItem } from '../database/types';
import { createCounterMetric, createSummaryMetric } from '../util/metrics';
import { CatalogProcessingOrchestrator, EntityProcessingResult } from './types';
import { Stitcher, stitchingStrategyFromConfig } from '../stitching/types';
import { markForStitching } from '../database/operations/stitcher/markForStitching';
import { startTaskPipeline } from './TaskPipeline';
import { Config } from '@backstage/config';
import {
@@ -65,7 +65,6 @@ export class DefaultCatalogProcessingEngine {
private readonly knex: Knex;
private readonly processingDatabase: ProcessingDatabase;
private readonly orchestrator: CatalogProcessingOrchestrator;
private readonly stitcher: Stitcher;
private readonly createHash: () => Hash;
private readonly pollingIntervalMs: number;
private readonly orphanCleanupIntervalMs: number;
@@ -85,7 +84,6 @@ export class DefaultCatalogProcessingEngine {
knex: Knex;
processingDatabase: ProcessingDatabase;
orchestrator: CatalogProcessingOrchestrator;
stitcher: Stitcher;
createHash: () => Hash;
pollingIntervalMs?: number;
orphanCleanupIntervalMs?: number;
@@ -103,7 +101,6 @@ export class DefaultCatalogProcessingEngine {
this.knex = options.knex;
this.processingDatabase = options.processingDatabase;
this.orchestrator = options.orchestrator;
this.stitcher = options.stitcher;
this.createHash = options.createHash;
this.pollingIntervalMs = options.pollingIntervalMs ?? 1_000;
this.orphanCleanupIntervalMs = options.orphanCleanupIntervalMs ?? 30_000;
@@ -283,7 +280,8 @@ export class DefaultCatalogProcessingEngine {
});
});
await this.stitcher.stitch({
await markForStitching({
knex: this.knex,
entityRefs: [stringifyEntityRef(unprocessedEntity)],
});
@@ -338,7 +336,8 @@ export class DefaultCatalogProcessingEngine {
}
});
await this.stitcher.stitch({
await markForStitching({
knex: this.knex,
entityRefs: setOfThingsToStitch,
});
@@ -358,15 +357,10 @@ export class DefaultCatalogProcessingEngine {
return () => {};
}
const stitchingStrategy = stitchingStrategyFromConfig(this.config, {
logger: this.logger,
});
const runOnce = async () => {
try {
const n = await deleteOrphanedEntities({
knex: this.knex,
strategy: stitchingStrategy,
});
if (n > 0) {
this.logger.info(`Deleted ${n} orphaned entities`);
plugins/catalog-backend/src/service/CatalogBuilder.ts
-2
View File
@@ -435,7 +435,6 @@ export class CatalogBuilder {
const unauthorizedEntitiesCatalog = new DefaultEntitiesCatalog({
database: dbClient,
logger,
stitcher,
enableRelationsCompatibility,
});
@@ -511,7 +510,6 @@ export class CatalogBuilder {
knex: dbClient,
processingDatabase,
orchestrator,
stitcher,
createHash: () => createHash('sha1'),
pollingIntervalMs: 1000,
onProcessingError: event => {
plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts
+8 -46
View File
@@ -33,7 +33,6 @@ import {
DbRefreshStateRow,
DbSearchRow,
} from '../database/tables';
import { Stitcher } from '../stitching/types';
import { DefaultEntitiesCatalog } from './DefaultEntitiesCatalog';
import { EntitiesRequest } from '../catalog/types';
import { buildEntitySearch } from '../database/operations/stitcher/buildEntitySearch';
@@ -52,9 +51,6 @@ describe.each(databases.eachSupportedId())(
await knex.destroy();
});
const stitch = jest.fn();
const stitcher: Stitcher = { stitch } as any;
async function createDatabase() {
knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
@@ -165,7 +161,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const result = await catalog.entityAncestry('k:default/root');
expect(result.rootEntityRef).toEqual('k:default/root');
@@ -195,7 +190,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(() =>
catalog.entityAncestry('k:default/root'),
@@ -238,7 +232,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const result = await catalog.entityAncestry('k:default/root');
expect(result.rootEntityRef).toEqual('k:default/root');
@@ -294,7 +287,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const testFilter = {
@@ -331,7 +323,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const testFilter = {
@@ -382,7 +373,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const testFilter1 = {
@@ -439,7 +429,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const testFilter1 = {
@@ -484,7 +473,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const testFilter = {
@@ -530,7 +518,7 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
enableRelationsCompatibility: true,
});
@@ -585,7 +573,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
function f(
@@ -648,7 +635,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
function f(
@@ -726,7 +712,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
async function page(limit: number, offset?: number): Promise<string[]> {
@@ -795,7 +780,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
async function page(
@@ -862,7 +846,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
async function page(order: 'asc' | 'desc'): Promise<string[]> {
@@ -910,7 +893,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const res = await catalog.entitiesBatch({
@@ -963,7 +945,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const res = await catalog.entitiesBatch({
@@ -1009,7 +990,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const filter = {
@@ -1183,7 +1163,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const filter = {
@@ -1358,7 +1337,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const filter = {
@@ -1415,7 +1393,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const request: QueryEntitiesInitialRequest = {
@@ -1466,7 +1443,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const filter = {
@@ -1562,7 +1538,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const filter = {
@@ -1639,7 +1614,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const request: QueryEntitiesInitialRequest = {
@@ -1672,7 +1646,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const request: QueryEntitiesInitialRequest = {
@@ -1720,7 +1693,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const limit = 2;
@@ -1841,7 +1813,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const limit = 2;
@@ -1902,7 +1873,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const limit = 2;
@@ -1994,7 +1964,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
// Entities without the sort field are excluded — sorting by a field
@@ -2034,7 +2003,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2130,7 +2098,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
// Query with orderField
@@ -2161,7 +2128,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
// Use filter to restrict to kind=component, and query to restrict to name=A
@@ -2246,7 +2212,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await catalog.removeEntityByUid(uid);
@@ -2262,9 +2227,13 @@ describe.each(databases.eachSupportedId())(
{ entity_ref: 'k:default/unrelated1', result_hash: 'not-changed' },
{ entity_ref: 'k:default/unrelated2', result_hash: 'not-changed' },
]);
expect(stitch).toHaveBeenCalledWith({
entityRefs: new Set(['k:default/unrelated1', 'k:default/unrelated2']),
});
const stitchQueue = await knex('stitch_queue')
.select('entity_ref')
.orderBy('entity_ref');
expect(stitchQueue.map(r => r.entity_ref)).toEqual([
'k:default/unrelated1',
'k:default/unrelated2',
]);
});
});
@@ -2293,7 +2262,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2364,7 +2332,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2410,7 +2377,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2451,7 +2417,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2500,7 +2465,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2523,7 +2487,6 @@ describe.each(databases.eachSupportedId())(
return new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
}
@@ -2571,7 +2534,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
// With filter: unstitched entity should be excluded because the
plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts
+3 -5
View File
@@ -41,7 +41,7 @@ import {
DbRelationsRow,
DbSearchRow,
} from '../database/tables';
import { Stitcher } from '../stitching/types';
import { markForStitching } from '../database/operations/stitcher/markForStitching';
import {
expandLegacyCompoundRelationsInEntity,
@@ -102,18 +102,15 @@ function stringifyPagination(
export class DefaultEntitiesCatalog implements EntitiesCatalog {
private readonly database: Knex;
private readonly logger: LoggerService;
private readonly stitcher: Stitcher;
private readonly enableRelationsCompatibility: boolean;
constructor(options: {
database: Knex;
logger: LoggerService;
stitcher: Stitcher;
enableRelationsCompatibility?: boolean;
}) {
this.database = options.database;
this.logger = options.logger;
this.stitcher = options.stitcher;
this.enableRelationsCompatibility = Boolean(
options.enableRelationsCompatibility,
);
@@ -759,7 +756,8 @@ export class DefaultEntitiesCatalog implements EntitiesCatalog {
});
if (relationPeerRefs.size > 0) {
await this.stitcher.stitch({
await markForStitching({
knex: this.database,
entityRefs: relationPeerRefs,
});
}
plugins/catalog-backend/src/service/DefaultRefreshService.test.ts
-7
View File
@@ -35,7 +35,6 @@ import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProc
import { EntityProcessingRequest } from '../processing/types';
import { DefaultRefreshService } from './DefaultRefreshService';
import { ConfigReader } from '@backstage/config';
import { DefaultStitcher } from '../stitching/DefaultStitcher';
import { LoggerService } from '@backstage/backend-plugin-api';
import { metricsServiceMock } from '@backstage/backend-test-utils/alpha';
@@ -113,17 +112,11 @@ describe.each(databases.eachSupportedId())(
}
}
const stitcher = DefaultStitcher.fromConfig(new ConfigReader({}), {
knex,
logger: defaultLogger,
metrics: metricsServiceMock.mock(),
});
const engine = new DefaultCatalogProcessingEngine({
config: new ConfigReader({}),
logger: defaultLogger,
processingDatabase: db,
knex: knex,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
orchestrator: {
async process(request: EntityProcessingRequest) {
plugins/catalog-backend/src/stitching/DefaultStitcher.test.ts
+50 -8
View File
@@ -17,6 +17,7 @@
import { TestDatabases, mockServices } from '@backstage/backend-test-utils';
import { Entity } from '@backstage/catalog-model';
import { applyDatabaseMigrations } from '../database/migrations';
import { markForStitching } from '../database/operations/stitcher/markForStitching';
import {
DbFinalEntitiesRow,
DbRefreshStateReferencesRow,
@@ -41,11 +42,12 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => {
const stitcher = new DefaultStitcher({
knex: db,
logger,
strategy: { mode: 'immediate' },
strategy: {
pollingInterval: { milliseconds: 50 },
stitchTimeout: { seconds: 10 },
},
metrics: metricsServiceMock.mock(),
});
let entities: DbFinalEntitiesRow[];
let entity: Entity;
await db<DbRefreshStateRow>('refresh_state').insert([
{
@@ -87,12 +89,21 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => {
},
]);
await stitcher.stitch({ entityRefs: ['k:ns/n'] });
await markForStitching({ knex: db, entityRefs: ['k:ns/n'] });
await stitcher.start();
entities = await db<DbFinalEntitiesRow>('final_entities');
// Wait for stitching to complete
await waitForCondition(async () => {
const entities = await db<DbFinalEntitiesRow>('final_entities');
return entities.length === 1 && entities[0].final_entity !== null;
});
await stitcher.stop();
let entities = await db<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
let entity: Entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: [
{
@@ -167,7 +178,15 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => {
);
// Re-stitch without any changes
await stitcher.stitch({ entityRefs: ['k:ns/n'] });
await markForStitching({ knex: db, entityRefs: ['k:ns/n'] });
await stitcher.start();
await waitForCondition(async () => {
const queue = await db('stitch_queue').select('*');
return queue.length === 0;
});
await stitcher.stop();
entities = await db<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
@@ -185,7 +204,15 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => {
},
]);
await stitcher.stitch({ entityRefs: ['k:ns/n'] });
await markForStitching({ knex: db, entityRefs: ['k:ns/n'] });
await stitcher.start();
await waitForCondition(async () => {
const e = await db<DbFinalEntitiesRow>('final_entities');
return e.length === 1 && e[0].hash !== firstHash;
});
await stitcher.stop();
entities = await db<DbFinalEntitiesRow>('final_entities');
@@ -272,3 +299,18 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => {
);
});
});
async function waitForCondition(
condition: () => Promise<boolean>,
timeoutMs = 10_000,
intervalMs = 50,
) {
const start = Date.now();
while (Date.now() - start < timeoutMs) {
if (await condition()) {
return;
}
await new Promise(resolve => setTimeout(resolve, intervalMs));
}
throw new Error('Timed out waiting for condition');
}
plugins/catalog-backend/src/stitching/DefaultStitcher.ts
+30 -80
View File
@@ -17,19 +17,12 @@
import { Config } from '@backstage/config';
import { durationToMilliseconds, HumanDuration } from '@backstage/types';
import { Knex } from 'knex';
import splitToChunks from 'lodash/chunk';
import { DateTime } from 'luxon';
import { getDeferredStitchableEntities } from '../database/operations/stitcher/getDeferredStitchableEntities';
import { markForStitching } from '../database/operations/stitcher/markForStitching';
import { performStitching } from '../database/operations/stitcher/performStitching';
import { DbRefreshStateRow } from '../database/tables';
import { startTaskPipeline } from '../processing/TaskPipeline';
import { progressTracker } from './progressTracker';
import {
Stitcher,
StitchingStrategy,
stitchingStrategyFromConfig,
} from './types';
import { StitchingStrategy, stitchingStrategyFromConfig } from './types';
import { LoggerService } from '@backstage/backend-plugin-api';
import { MetricsService } from '@backstage/backend-plugin-api/alpha';
@@ -44,7 +37,7 @@ type StitchProgressTracker = ReturnType<typeof progressTracker>;
* ingestion process, and stitching them together into the final entity JSON
* shape.
*/
export class DefaultStitcher implements Stitcher {
export class DefaultStitcher {
private readonly knex: Knex;
private readonly logger: LoggerService;
private readonly strategy: StitchingStrategy;
@@ -85,80 +78,38 @@ export class DefaultStitcher implements Stitcher {
);
}
async stitch(options: {
entityRefs?: Iterable<string>;
entityIds?: Iterable<string>;
}) {
const { entityRefs, entityIds } = options;
if (this.strategy.mode === 'deferred') {
await markForStitching({
knex: this.knex,
strategy: this.strategy,
entityRefs,
entityIds,
});
return;
}
if (entityRefs) {
for (const entityRef of entityRefs) {
await this.#stitchOne({ entityRef });
}
}
if (entityIds) {
const chunks = splitToChunks(
Array.isArray(entityIds) ? entityIds : [...entityIds],
100,
);
for (const chunk of chunks) {
const rows = await this.knex<DbRefreshStateRow>('refresh_state')
.select('entity_ref')
.whereIn('entity_id', chunk);
for (const row of rows) {
await this.#stitchOne({ entityRef: row.entity_ref });
}
}
}
}
async start() {
if (this.strategy.mode === 'deferred') {
if (this.stopFunc) {
throw new Error('Processing engine is already started');
}
const { pollingInterval, stitchTimeout } = this.strategy;
const stopPipeline = startTaskPipeline<DeferredStitchItem>({
lowWatermark: 2,
highWatermark: 5,
pollingIntervalMs: durationToMilliseconds(pollingInterval),
loadTasks: async count => {
return await this.#getStitchableEntities(count, stitchTimeout);
},
processTask: async item => {
return await this.#stitchOne({
entityRef: item.entityRef,
stitchTicket: item.stitchTicket,
stitchRequestedAt: item.stitchRequestedAt,
});
},
});
this.stopFunc = () => {
stopPipeline();
};
if (this.stopFunc) {
throw new Error('Stitcher is already started');
}
const { pollingInterval, stitchTimeout } = this.strategy;
const stopPipeline = startTaskPipeline<DeferredStitchItem>({
lowWatermark: 2,
highWatermark: 5,
pollingIntervalMs: durationToMilliseconds(pollingInterval),
loadTasks: async count => {
return await this.#getStitchableEntities(count, stitchTimeout);
},
processTask: async item => {
return await this.#stitchOne({
entityRef: item.entityRef,
stitchTicket: item.stitchTicket,
stitchRequestedAt: item.stitchRequestedAt,
});
},
});
this.stopFunc = () => {
stopPipeline();
};
}
async stop() {
if (this.strategy.mode === 'deferred') {
if (this.stopFunc) {
this.stopFunc();
this.stopFunc = undefined;
}
if (this.stopFunc) {
this.stopFunc();
this.stopFunc = undefined;
}
}
@@ -177,7 +128,7 @@ export class DefaultStitcher implements Stitcher {
async #stitchOne(options: {
entityRef: string;
stitchTicket?: string;
stitchTicket: string;
stitchRequestedAt?: DateTime;
}) {
const track = this.tracker.stitchStart({
@@ -189,7 +140,6 @@ export class DefaultStitcher implements Stitcher {
const result = await performStitching({
knex: this.knex,
logger: this.logger,
strategy: this.strategy,
entityRef: options.entityRef,
stitchTicket: options.stitchTicket,
});
plugins/catalog-backend/src/stitching/types.ts
+27 -63
View File
@@ -19,51 +19,19 @@ import { Config, readDurationFromConfig } from '@backstage/config';
import { HumanDuration } from '@backstage/types';
/**
* Performs the act of stitching - to take all of the various outputs from the
* ingestion process, and stitching them together into the final entity JSON
* shape.
* Configuration for the stitching process, controlling polling and timeout
* behavior for the deferred stitching worker.
*/
export interface Stitcher {
stitch(options: {
entityRefs?: Iterable<string>;
entityIds?: Iterable<string>;
}): Promise<void>;
}
/**
* The strategies supported by the stitching process, in terms of when to
* perform stitching.
*
* @remarks
*
* In immediate mode, stitching happens "in-band" (blocking) immediately when
* each processing task finishes. When set to `'deferred'`, stitching is instead
* deferred to happen on a separate asynchronous worker queue just like
* processing.
*
* Deferred stitching should make performance smoother when ingesting large
* amounts of entities, and reduce p99 processing times and repeated
* over-stitching of hot spot entities when fan-out/fan-in in terms of relations
* is very large. It does however also come with some performance cost due to
* the queuing with how much wall-clock time some types of task take.
*
* Note: Immediate mode is deprecated and will be removed in a future release.
*/
export type StitchingStrategy =
| {
mode: 'immediate';
}
| {
mode: 'deferred';
pollingInterval: HumanDuration;
stitchTimeout: HumanDuration;
};
export type StitchingStrategy = {
pollingInterval: HumanDuration;
stitchTimeout: HumanDuration;
};
let immediateDeprecationLogged = false;
export function stitchingStrategyFromConfig(
config: Config,
options: { logger: LoggerService },
options?: { logger?: LoggerService },
): StitchingStrategy {
const strategyMode = config.getOptionalString(
'catalog.stitchingStrategy.mode',
@@ -72,32 +40,28 @@ export function stitchingStrategyFromConfig(
if (strategyMode === 'immediate') {
if (!immediateDeprecationLogged) {
immediateDeprecationLogged = true;
options.logger.warn(
'DEPRECATED: Immediate mode stitching has been deprecated, and will be removed in the next Backstage release.',
options?.logger?.warn(
"The 'immediate' stitching strategy mode has been removed and is no longer supported. Falling back to deferred stitching. Please remove the 'catalog.stitchingStrategy.mode' configuration key.",
);
}
return {
mode: 'immediate',
};
} else if (strategyMode === undefined || strategyMode === 'deferred') {
const pollingIntervalKey = 'catalog.stitchingStrategy.pollingInterval';
const stitchTimeoutKey = 'catalog.stitchingStrategy.stitchTimeout';
const pollingInterval = config.has(pollingIntervalKey)
? readDurationFromConfig(config, { key: pollingIntervalKey })
: { seconds: 1 };
const stitchTimeout = config.has(stitchTimeoutKey)
? readDurationFromConfig(config, { key: stitchTimeoutKey })
: { seconds: 60 };
return {
mode: 'deferred',
pollingInterval: pollingInterval,
stitchTimeout: stitchTimeout,
};
} else if (strategyMode !== undefined && strategyMode !== 'deferred') {
options?.logger?.warn(
`Unknown stitching strategy mode '${strategyMode}', falling back to deferred stitching. Please remove or correct the 'catalog.stitchingStrategy.mode' configuration key.`,
);
}
throw new Error(
`Invalid stitching strategy mode '${strategyMode}', expected one of 'immediate' or 'deferred'`,
);
const pollingIntervalKey = 'catalog.stitchingStrategy.pollingInterval';
const stitchTimeoutKey = 'catalog.stitchingStrategy.stitchTimeout';
const pollingInterval = config.has(pollingIntervalKey)
? readDurationFromConfig(config, { key: pollingIntervalKey })
: { seconds: 1 };
const stitchTimeout = config.has(stitchTimeoutKey)
? readDurationFromConfig(config, { key: stitchTimeoutKey })
: { seconds: 60 };
return {
pollingInterval: pollingInterval,
stitchTimeout: stitchTimeout,
};
}
plugins/catalog-backend/src/tests/integration.test.ts
+19 -20
View File
@@ -202,6 +202,7 @@ class WaitingProgressTracker implements ProgressTrackerWithErrorReports {
class TestHarness {
readonly #catalog: EntitiesCatalog;
readonly #engine: CatalogProcessingEngine;
readonly #stitcher: DefaultStitcher;
readonly #refresh: RefreshService;
readonly #provider: TestProvider;
readonly #proxyProgressTracker: ProxyProgressTracker;
@@ -226,11 +227,6 @@ class TestHarness {
connection: ':memory:',
},
},
catalog: {
stitchingStrategy: {
mode: 'immediate',
},
},
});
const logger = options?.logger ?? mockServices.logger.mock();
@@ -287,7 +283,6 @@ class TestHarness {
const catalog = new DefaultEntitiesCatalog({
database: options.db,
logger,
stitcher,
enableRelationsCompatibility: options?.enableRelationsCompatibility,
});
const proxyProgressTracker = new ProxyProgressTracker(
@@ -300,7 +295,6 @@ class TestHarness {
processingDatabase,
knex: options.db,
orchestrator,
stitcher,
scheduler: mockServices.scheduler(),
createHash: () => createHash('sha1'),
pollingIntervalMs: 50,
@@ -335,16 +329,8 @@ class TestHarness {
return new TestHarness(
catalog,
{
async start() {
await engine.start();
await stitcher.start();
},
async stop() {
await engine.stop();
await stitcher.stop();
},
},
engine,
stitcher,
refresh,
provider,
proxyProgressTracker,
@@ -355,6 +341,7 @@ class TestHarness {
constructor(
catalog: EntitiesCatalog,
engine: CatalogProcessingEngine,
stitcher: DefaultStitcher,
refresh: RefreshService,
provider: TestProvider,
proxyProgressTracker: ProxyProgressTracker,
@@ -362,6 +349,7 @@ class TestHarness {
) {
this.#catalog = catalog;
this.#engine = engine;
this.#stitcher = stitcher;
this.#refresh = refresh;
this.#provider = provider;
this.#proxyProgressTracker = proxyProgressTracker;
@@ -372,13 +360,25 @@ class TestHarness {
const tracker = new WaitingProgressTracker(entityRefs);
this.#proxyProgressTracker.setTracker(tracker);
this.#engine.start();
await this.#engine.start();
await this.#stitcher.start();
const errors = await tracker.wait();
this.#engine.stop();
await this.#engine.stop();
await tracker.waitForFinish();
// Wait for the stitch queue to drain while the stitcher is still running
const start = Date.now();
while (Date.now() - start < 10_000) {
const queue = await this.#db('stitch_queue').select('*');
if (queue.length === 0) {
break;
}
await new Promise(resolve => setTimeout(resolve, 50));
}
await this.#stitcher.stop();
this.#proxyProgressTracker.setTracker(new NoopProgressTracker());
return errors;
@@ -411,7 +411,6 @@ class TestHarness {
async removeOrphanedEntities() {
await deleteOrphanedEntities({
knex: this.#db,
strategy: { mode: 'immediate' },
});
}
plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts
+5 -55
View File
@@ -23,7 +23,6 @@ import {
import { catalogProcessingExtensionPoint } from '@backstage/plugin-catalog-node';
import { createDeferred } from '@backstage/types';
import { Knex } from 'knex';
import { default as catalogPlugin } from '../..';
import { applyDatabaseMigrations } from '../../database/migrations';
import {
SyntheticLoadEntitiesProcessor,
@@ -149,58 +148,9 @@ const databases = TestDatabases.create({
});
describePerformanceTest('stitchingPerformance', () => {
describe.each(databases.eachSupportedId())('%p', databaseId => {
it('runs stitching in immediate mode', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
const load: SyntheticLoadOptions = {
baseEntitiesCount: 1000,
baseRelationsCount: 3,
baseRelationsSkew: 0.3,
childrenCount: 3,
};
const config = {
backend: { baseUrl: 'http://localhost:7007' },
catalog: { stitchingStrategy: { mode: 'immediate' } },
};
const tracker = new Tracker(knex, load);
const backend = await startTestBackend({
features: [
catalogPlugin,
mockServices.rootConfig.factory({ data: config }),
mockServices.database.factory({ knex }),
createBackendModule({
pluginId: 'catalog',
moduleId: 'synthetic-load-entities',
register(reg) {
reg.registerInit({
deps: {
catalog: catalogProcessingExtensionPoint,
},
async init({ catalog }) {
catalog.addEntityProvider(
new SyntheticLoadEntitiesProvider(load, tracker.events()),
);
catalog.addProcessor(
new SyntheticLoadEntitiesProcessor(load),
);
},
});
},
}),
],
});
await expect(tracker.completion()).resolves.toBeUndefined();
await backend.stop();
await knex.destroy();
});
it('runs stitching in deferred mode', async () => {
it.each(databases.eachSupportedId())(
'runs stitching in deferred mode, %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
@@ -248,6 +198,6 @@ describePerformanceTest('stitchingPerformance', () => {
await expect(tracker.completion()).resolves.toBeUndefined();
await backend.stop();
await knex.destroy();
});
});
},
);
});