From dc7678cdcc7fc810c7c4396664ee17426ed29695 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Mon, 11 May 2026 23:20:55 +0200 Subject: [PATCH 01/17] fix(catalog-backend): remove immediate mode stitching MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove the immediate stitching strategy from the catalog-backend plugin. All stitching now uses the deferred mode exclusively, which processes entities asynchronously via a worker queue. This simplifies the stitching subsystem by removing dead-weight code paths and the strategy parameter threading, and fixes a bug where deleteWithEagerPruningOfChildren hardcoded immediate-mode semantics regardless of configured mode. Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .changeset/remove-immediate-stitching.md | 5 + plugins/catalog-backend/config.d.ts | 27 +- .../deleteWithEagerPruningOfChildren.test.ts | 5 +- .../deleteWithEagerPruningOfChildren.ts | 30 +- .../stitcher/markForStitching.test.ts | 796 ++++++------------ .../operations/stitcher/markForStitching.ts | 125 +-- .../stitcher/performStitching.test.ts | 547 ++++++------ .../operations/stitcher/performStitching.ts | Bin 10203 -> 9992 bytes .../util/deleteOrphanedEntities.test.ts | 100 +-- .../operations/util/deleteOrphanedEntities.ts | 5 +- .../DefaultCatalogProcessingEngine.test.ts | 53 +- .../DefaultCatalogProcessingEngine.ts | 16 +- .../src/service/CatalogBuilder.ts | 2 - .../service/DefaultEntitiesCatalog.test.ts | 57 +- .../src/service/DefaultEntitiesCatalog.ts | 8 +- .../src/service/DefaultRefreshService.test.ts | 7 - .../src/stitching/DefaultStitcher.test.ts | 58 +- .../src/stitching/DefaultStitcher.ts | 108 +-- .../catalog-backend/src/stitching/types.ts | 93 +- .../src/tests/integration.test.ts | 35 +- .../performance/stitchingPerformance.test.ts | 60 +- 21 files changed, 772 insertions(+), 1365 deletions(-) create mode 100644 .changeset/remove-immediate-stitching.md diff --git a/.changeset/remove-immediate-stitching.md b/.changeset/remove-immediate-stitching.md new file mode 100644 index 0000000000..df48c8f633 --- /dev/null +++ b/.changeset/remove-immediate-stitching.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +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. diff --git a/plugins/catalog-backend/config.d.ts b/plugins/catalog-backend/config.d.ts index 41be62feb4..38c5ed0435 100644 --- a/plugins/catalog-backend/config.d.ts +++ b/plugins/catalog-backend/config.d.ts @@ -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. diff --git a/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.test.ts b/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.test.ts index b9a458a3ab..d80b2dffda 100644 --- a/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.test.ts +++ b/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.test.ts @@ -85,10 +85,9 @@ describe.each(databases.eachSupportedId())( } async function entitiesMarkedForStitching(knex: Knex) { - const rows = await knex('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); } diff --git a/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts b/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts index 70f7d08ed1..7d2f699ea9 100644 --- a/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts +++ b/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts @@ -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 @@ -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('final_entities') - .update({ - hash: 'force-stitching', - }) - .whereIn('entity_id', ids); - await knex - .table('refresh_state') - .update({ - result_hash: 'force-stitching', - next_update_at: knex.fn.now(), - }) - .whereIn('entity_id', ids); - } + await markForStitching({ + knex, + entityIds: affectedIds, + }); } diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts index c9a89f8892..e3650edf03 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts @@ -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('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('stitch_queue').insert([ - { - entity_ref: 'k:ns/four', - stitch_ticket: 'old', - next_stitch_at: '1971-01-01T00:00:00.000', - }, - ]); + await knex('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('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('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('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('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('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('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('stitch_queue') + async function result() { + return knex('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('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('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(); }); }, ); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts index f0dfc5db71..d3c6bf3a5d 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts @@ -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; entityIds?: Iterable; }): Promise { 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('final_entities') - .update({ - hash: 'force-stitching', - }) - .whereIn('entity_ref', chunk); - await retryOnDeadlock(async () => { - await knex - .table('refresh_state') - .update({ - result_hash: 'force-stitching', - next_update_at: knex.fn.now(), - }) - .whereIn('entity_ref', chunk); - }, 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('final_entities') - .update({ - hash: 'force-stitching', - }) + for (const chunk of entityRefs) { + await retryOnDeadlock(async () => { + if (chunk.length > 0) { + await knex('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('refresh_state') + .select('entity_ref') .whereIn('entity_id', chunk); - await retryOnDeadlock(async () => { - await knex - .table('refresh_state') - .update({ - result_hash: 'force-stitching', - next_update_at: knex.fn.now(), - }) - .whereIn('entity_id', chunk); - }, 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('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('refresh_state') - .select('entity_ref') - .whereIn('entity_id', chunk); - - if (refreshStateRows.length > 0) { - await knex('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('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); } } diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index 6099ee031b..2dde95db48 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -31,301 +31,292 @@ 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('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( + 'refresh_state_references', + ).insert([{ source_key: 'a', target_entity_ref: 'k:ns/n' }]); + await knex('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', + }, + ]); - await knex('refresh_state').insert([ + await markForStitching({ + knex, + entityRefs: ['k:ns/n'], + }); + + await performStitching({ + knex, + logger, + entityRef: 'k:ns/n', + stitchTicket: ( + await knex('stitch_queue') + .where('entity_ref', 'k:ns/n') + .select('stitch_ticket') + .first() + )?.stitch_ticket, + }); + + entities = await knex('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('search'); + expect(search).toEqual( + expect.arrayContaining([ { 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(), + key: 'relations.looksat', + original_value: 'k:ns/other', + value: 'k:ns/other', }, - ]); - await knex( - 'refresh_state_references', - ).insert([{ source_key: 'a', target_entity_ref: 'k:ns/n' }]); - await knex('relations').insert([ { - originating_entity_id: 'my-id', - source_entity_ref: 'k:ns/n', - type: 'looksAt', - target_entity_ref: 'k:ns/other', + entity_id: 'my-id', + key: 'apiversion', + original_value: 'a', + value: 'a', }, - // handles and ignores duplicates { - originating_entity_id: 'my-id', - source_entity_ref: 'k:ns/n', - type: 'looksAt', - target_entity_ref: 'k:ns/other', + entity_id: 'my-id', + key: 'kind', + original_value: 'k', + value: 'k', }, - ]); - - 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('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('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, - 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('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('relations').insert([ { - originating_entity_id: 'my-id', - source_entity_ref: 'k:ns/n', - type: 'looksAt', - target_entity_ref: 'k:ns/third', + entity_id: 'my-id', + key: 'metadata.name', + original_value: 'n', + value: 'n', }, - ]); - - 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('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', + { + entity_id: 'my-id', + key: 'metadata.namespace', + original_value: 'ns', + value: 'ns', }, - spec: { - k: 'v', + { + 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', + }, + ]), + ); - expect(entities[0].hash).not.toEqual(firstHash); - expect(entities[0].hash).toEqual(entity.metadata.etag); - - expect(await knex('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', - }, - ]), - ); + // Re-stitch without any changes + await markForStitching({ + knex, + entityRefs: ['k:ns/n'], }); + await performStitching({ + knex, + logger, + entityRef: 'k:ns/n', + stitchTicket: ( + await knex('stitch_queue') + .where('entity_ref', 'k:ns/n') + .select('stitch_ticket') + .first() + )?.stitch_ticket, + }); + + entities = await knex('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('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 knex('stitch_queue') + .where('entity_ref', 'k:ns/n') + .select('stitch_ticket') + .first() + )?.stitch_ticket, + }); + + entities = await knex('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('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 => { + const logger = mockServices.logger.mock(); + 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 @@ -380,7 +371,6 @@ describe.each(databases.eachSupportedId())( performStitching({ knex, logger: stitchLogger, - strategy: { mode: 'immediate' }, entityRef: 'k:ns/n', }), ).resolves.toBe('abandoned'); @@ -430,7 +420,6 @@ describe.each(databases.eachSupportedId())( performStitching({ knex, logger: stitchLogger, - strategy: { mode: 'immediate' }, entityRef: 'k:ns/n', }), ).resolves.toBe('changed'); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index ff5df8ce9d965cad20317df0e1f0eb13227b277e..c68a277a0fb41083043aa184d20797b732cfb934 100644 GIT binary patch delta 114 zcmccZ-{H5xlx1@<%Pl5m23`5hQS4oelbJaWGlopQ$!R#5pG$(3OF=b& z87QxU%bca8s5Eu+Wv>4cjP8@4DjP{Elw_nT6qjU{Bxl5zWF}{)mH>5v^%|&HPxez6 K+q_c6h#3H-*diAI delta 309 zcmeD1yY0Whl*L#fxFoYAIU_SKJ-DPOu_QIUQlVBMttdZNL0wNzAA*WuO7u%A3sQ?W zYq30K;*BXa00LbQW$3n9h`ozZx)^GL6{ew+XK)-B_RLdANli;FDoRaJ$jwhl)ln#! zoWN { + async function run(knex: Knex): Promise { 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' }, diff --git a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts index 5e2e46cf89..7cb4cc4187 100644 --- a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts +++ b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts @@ -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 { - 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, }); } diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts index 88829fc448..b473891a7f 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts @@ -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 = { process: jest.fn(), }; - const stitcher = { - stitch: jest.fn(), - } as unknown as jest.Mocked; const hash = { update: () => hash, digest: jest.fn(), @@ -69,7 +72,7 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - stitcher: stitcher, + createHash: () => hash, scheduler: mockServices.scheduler(), events: mockServices.events.mock(), @@ -139,7 +142,7 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, events: mockServices.events.mock(), @@ -225,7 +228,7 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, events: mockServices.events.mock(), @@ -305,7 +308,7 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, events: mockServices.events.mock(), @@ -367,7 +370,7 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, @@ -467,12 +470,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 +488,7 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, @@ -572,15 +575,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 +596,7 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, @@ -664,9 +667,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 +682,7 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, @@ -755,9 +758,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 +773,7 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, @@ -836,9 +839,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(); diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index a8cca7fa0d..1212a43eb6 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -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`); diff --git a/plugins/catalog-backend/src/service/CatalogBuilder.ts b/plugins/catalog-backend/src/service/CatalogBuilder.ts index 881c29db3f..7b6ea06e0b 100644 --- a/plugins/catalog-backend/src/service/CatalogBuilder.ts +++ b/plugins/catalog-backend/src/service/CatalogBuilder.ts @@ -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 => { diff --git a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts index 3836a1d30a..da1497bb1e 100644 --- a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts +++ b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts @@ -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,7 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, + }); async function page(limit: number, offset?: number): Promise { @@ -795,7 +781,7 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, + }); async function page( @@ -862,7 +848,7 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, + }); async function page(order: 'asc' | 'desc'): Promise { @@ -910,7 +896,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const res = await catalog.entitiesBatch({ @@ -963,7 +948,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const res = await catalog.entitiesBatch({ @@ -1009,7 +993,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const filter = { @@ -1183,7 +1166,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const filter = { @@ -1358,7 +1340,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const filter = { @@ -1415,7 +1396,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const request: QueryEntitiesInitialRequest = { @@ -1466,7 +1446,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const filter = { @@ -1562,7 +1541,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const filter = { @@ -1639,7 +1617,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const request: QueryEntitiesInitialRequest = { @@ -1672,7 +1649,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const request: QueryEntitiesInitialRequest = { @@ -1720,7 +1696,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const limit = 2; @@ -1841,7 +1816,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const limit = 2; @@ -1902,7 +1876,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); const limit = 2; @@ -1994,7 +1967,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 +2006,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); await expect( @@ -2130,7 +2101,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); // Query with orderField @@ -2161,7 +2131,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 +2215,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); await catalog.removeEntityByUid(uid); @@ -2262,9 +2230,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 +2265,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); await expect( @@ -2364,7 +2335,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); await expect( @@ -2410,7 +2380,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); await expect( @@ -2451,7 +2420,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); await expect( @@ -2500,7 +2468,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); await expect( @@ -2523,7 +2490,6 @@ describe.each(databases.eachSupportedId())( return new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - stitcher, }); } @@ -2571,7 +2537,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 diff --git a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts index 2a3c7aad9d..bafc2d4212 100644 --- a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts +++ b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts @@ -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, }); } diff --git a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts index c3244de20a..ff73a7a927 100644 --- a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts +++ b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts @@ -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) { diff --git a/plugins/catalog-backend/src/stitching/DefaultStitcher.test.ts b/plugins/catalog-backend/src/stitching/DefaultStitcher.test.ts index 5ec9b39c71..8f383c04fc 100644 --- a/plugins/catalog-backend/src/stitching/DefaultStitcher.test.ts +++ b/plugins/catalog-backend/src/stitching/DefaultStitcher.test.ts @@ -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('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('final_entities'); + // Wait for stitching to complete + await waitForCondition(async () => { + const entities = await db('final_entities'); + return entities.length === 1 && entities[0].final_entity !== null; + }); + + await stitcher.stop(); + + let entities = await db('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('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('final_entities'); + return e.length === 1 && e[0].hash !== firstHash; + }); + + await stitcher.stop(); entities = await db('final_entities'); @@ -272,3 +299,18 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => { ); }); }); + +async function waitForCondition( + condition: () => Promise, + 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'); +} diff --git a/plugins/catalog-backend/src/stitching/DefaultStitcher.ts b/plugins/catalog-backend/src/stitching/DefaultStitcher.ts index f67adf7942..335d82f173 100644 --- a/plugins/catalog-backend/src/stitching/DefaultStitcher.ts +++ b/plugins/catalog-backend/src/stitching/DefaultStitcher.ts @@ -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; * 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; - entityIds?: Iterable; - }) { - 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('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({ - 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('Processing engine is already started'); } + + const { pollingInterval, stitchTimeout } = this.strategy; + + const stopPipeline = startTaskPipeline({ + 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; } } @@ -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, }); diff --git a/plugins/catalog-backend/src/stitching/types.ts b/plugins/catalog-backend/src/stitching/types.ts index 7dc8ee1ac0..4ca171de29 100644 --- a/plugins/catalog-backend/src/stitching/types.ts +++ b/plugins/catalog-backend/src/stitching/types.ts @@ -19,85 +19,40 @@ 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; - entityIds?: Iterable; - }): Promise; -} - -/** - * 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; - }; - -let immediateDeprecationLogged = false; +export type StitchingStrategy = { + pollingInterval: HumanDuration; + stitchTimeout: HumanDuration; +}; export function stitchingStrategyFromConfig( config: Config, - options: { logger: LoggerService }, + options?: { logger?: LoggerService }, ): StitchingStrategy { const strategyMode = config.getOptionalString( 'catalog.stitchingStrategy.mode', ); 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.', - ); - } - 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, - }; + 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.", + ); } - 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, + }; } diff --git a/plugins/catalog-backend/src/tests/integration.test.ts b/plugins/catalog-backend/src/tests/integration.test.ts index d03a23f896..d4235c2c61 100644 --- a/plugins/catalog-backend/src/tests/integration.test.ts +++ b/plugins/catalog-backend/src/tests/integration.test.ts @@ -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; @@ -373,12 +361,24 @@ class TestHarness { this.#proxyProgressTracker.setTracker(tracker); this.#engine.start(); + await this.#stitcher.start(); const errors = await tracker.wait(); 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' }, }); } diff --git a/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts b/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts index 3987e5e937..98fd496a90 100644 --- a/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts +++ b/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts @@ -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(); - }); - }); + }, + ); }); From d0d1b9b5b97d05bbf7a5b0806704957c9c777932 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 16:30:20 +0200 Subject: [PATCH 02/17] Run prettier on files with formatting issues MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../database/operations/stitcher/performStitching.test.ts | 6 +++--- .../src/service/DefaultEntitiesCatalog.test.ts | 3 --- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index 2dde95db48..7c1ab0fd01 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -62,9 +62,9 @@ it.each(databases.eachSupportedId())( last_discovery_at: knex.fn.now(), }, ]); - await knex( - 'refresh_state_references', - ).insert([{ source_key: 'a', target_entity_ref: 'k:ns/n' }]); + await knex('refresh_state_references').insert([ + { source_key: 'a', target_entity_ref: 'k:ns/n' }, + ]); await knex('relations').insert([ { originating_entity_id: 'my-id', diff --git a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts index da1497bb1e..582a109f9e 100644 --- a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts +++ b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts @@ -712,7 +712,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - }); async function page(limit: number, offset?: number): Promise { @@ -781,7 +780,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - }); async function page( @@ -848,7 +846,6 @@ describe.each(databases.eachSupportedId())( const catalog = new DefaultEntitiesCatalog({ database: knex, logger: mockServices.logger.mock(), - }); async function page(order: 'asc' | 'desc'): Promise { From 640f6b37a315506e1bbeb83f0052524129bbd461 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 16:36:28 +0200 Subject: [PATCH 03/17] Fix literal control chars in scriptProtocolPattern regex MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The conflict resolution replaced unicode escapes with literal control character bytes, causing GitHub to render the file as binary. Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../operations/stitcher/performStitching.ts | Bin 9992 -> 10002 bytes 1 file changed, 0 insertions(+), 0 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index c68a277a0fb41083043aa184d20797b732cfb934..e96eab811c2e12656ebed4b771fca483537b4119 100644 GIT binary patch delta 26 dcmeD1o8-5FpPe_R)Bp%{L6o7}W+C=(@&Ie;2hjik delta 16 XcmbQ_*WtH;pPiXOSAMe$`!{(2Dslw3 From ed2a9fe492a2d0939641660bdeb174d524086fa1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 17:19:35 +0200 Subject: [PATCH 04/17] Replace placeholder final_entities insert with upsert MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Now that immediate mode is removed, the two-step pattern of inserting a placeholder row and then updating it is unnecessary. Replace with a single upsert that creates or updates the final_entities row in one operation, removing a round-trip and simplifying the flow. Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../stitcher/performStitching.test.ts | 64 --------------- .../operations/stitcher/performStitching.ts | 79 ++++++------------- 2 files changed, 26 insertions(+), 117 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index 7c1ab0fd01..44c8a74cc4 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -317,70 +317,6 @@ describe.each(databases.eachSupportedId())( databaseId => { const logger = mockServices.logger.mock(); - 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('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('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, - entityRef: 'k:ns/n', - }), - ).resolves.toBe('abandoned'); - - expect(stitchLogger.debug).toHaveBeenCalledWith( - 'Skipping stitching of k:ns/n, conflict', - expect.anything(), - ); - }); - it('stitches when final_entities row already exists', async () => { const knex = await databases.init(databaseId); await applyDatabaseMigrations(knex); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index e96eab811c..17cec3cf84 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -34,10 +34,7 @@ import { 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') @@ -79,34 +76,10 @@ export async function performStitching(options: { return 'abandoned'; } - // Ensure that a final_entities row exists for this entity. - try { - await knex('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) { @@ -236,31 +209,31 @@ export async function performStitching(options: { // to write the search index. const searchEntries = buildEntitySearch(entityId, entity); - let updateQuery = knex('final_entities') - .update({ + // Guard against concurrent stitchers by checking that the stitch_ticket + // in stitch_queue still matches what we were given. + if (stitchTicket) { + const ticketValid = await knex('stitch_queue') + .where('entity_ref', entityRef) + .where('stitch_ticket', stitchTicket) + .first(); + if (!ticketValid) { + logger.debug( + `Entity ${entityRef} is already stitched, skipping write.`, + ); + return 'abandoned'; + } + } + + await knex('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); - - // Guard against concurrent stitchers by checking that the stitch_ticket - // in stitch_queue still matches what we were given. - if (stitchTicket) { - updateQuery = updateQuery.whereExists( - knex('stitch_queue') - .where('stitch_queue.entity_ref', entityRef) - .where('stitch_queue.stitch_ticket', stitchTicket) - .select(knex.raw('1')), - ); - } - - const amountOfRowsChanged = await updateQuery; - - if (amountOfRowsChanged === 0) { - logger.debug(`Entity ${entityRef} is already stitched, skipping write.`); - return 'abandoned'; - } + .onConflict('entity_id') + .merge(['final_entity', 'hash', 'last_updated_at']); await syncSearchRows(knex, entityId, searchEntries); From 4282877ba94518e85dcc6d6af3c8942610222e7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 18:13:16 +0200 Subject: [PATCH 05/17] Restore log-once guard for immediate mode deprecation warning MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- plugins/catalog-backend/src/stitching/types.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/plugins/catalog-backend/src/stitching/types.ts b/plugins/catalog-backend/src/stitching/types.ts index 4ca171de29..4d055513b1 100644 --- a/plugins/catalog-backend/src/stitching/types.ts +++ b/plugins/catalog-backend/src/stitching/types.ts @@ -27,6 +27,8 @@ export type StitchingStrategy = { stitchTimeout: HumanDuration; }; +let immediateDeprecationLogged = false; + export function stitchingStrategyFromConfig( config: Config, options?: { logger?: LoggerService }, @@ -35,7 +37,8 @@ export function stitchingStrategyFromConfig( 'catalog.stitchingStrategy.mode', ); - if (strategyMode === 'immediate') { + if (strategyMode === 'immediate' && !immediateDeprecationLogged) { + immediateDeprecationLogged = true; 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.", ); From bcbc92d2ad71ab8b64a75ab33d98d3a3cb57cf6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 21:45:02 +0200 Subject: [PATCH 06/17] Bump changeset to minor for config schema change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .changeset/remove-immediate-stitching.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/remove-immediate-stitching.md b/.changeset/remove-immediate-stitching.md index df48c8f633..07110b1d99 100644 --- a/.changeset/remove-immediate-stitching.md +++ b/.changeset/remove-immediate-stitching.md @@ -1,5 +1,5 @@ --- -'@backstage/plugin-catalog-backend': patch +'@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. From 7bab21f1ef2652f742f8464bc65f4a846ea794dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 22:05:39 +0200 Subject: [PATCH 07/17] Remove redundant entity existence check in performStitching MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The initial SELECT entity_id FROM refresh_state check is redundant — the processedResult query in the Promise.all already handles the missing-entity case. This saves one database round trip per stitch. Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../src/database/operations/stitcher/performStitching.ts | 9 --------- 1 file changed, 9 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index 17cec3cf84..6929d3ca62 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -67,15 +67,6 @@ export async function performStitching(options: { let removeFromStitchQueueOnCompletion = true; try { - const entityResult = await knex('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'; - } - // 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 From b8731fc82b8757855beac8d97fce4671bfa6646f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 22:38:52 +0200 Subject: [PATCH 08/17] Remove unused DbRefreshStateRow import and logger declaration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../database/operations/stitcher/performStitching.test.ts | 2 -- .../src/database/operations/stitcher/performStitching.ts | 6 +----- 2 files changed, 1 insertion(+), 7 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index 44c8a74cc4..9924f75b9d 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -315,8 +315,6 @@ it.each(databases.eachSupportedId())( describe.each(databases.eachSupportedId())( 'performStitching edge cases, %p', databaseId => { - const logger = mockServices.logger.mock(); - it('stitches when final_entities row already exists', async () => { const knex = await databases.init(databaseId); await applyDatabaseMigrations(knex); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index 6929d3ca62..aa5572f4e6 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -26,11 +26,7 @@ import { SerializedError } from '@backstage/errors'; import { Knex } from 'knex'; import { createHash } from 'node:crypto'; import stableStringify from 'fast-json-stable-stringify'; -import { - DbFinalEntitiesRow, - DbRefreshStateRow, - DbStitchQueueRow, -} from '../../tables'; +import { DbFinalEntitiesRow, DbStitchQueueRow } from '../../tables'; import { buildEntitySearch } from './buildEntitySearch'; import { markDeferredStitchCompleted } from './markDeferredStitchCompleted'; import { syncSearchRows } from './syncSearchRows'; From 5e6af9b7f60a375aa4f3bbce7994737a5b15c045 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 21 May 2026 17:12:25 +0200 Subject: [PATCH 09/17] fix(catalog-backend): fix stitcher error message and warn on unknown strategy mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- plugins/catalog-backend/src/stitching/DefaultStitcher.ts | 2 +- plugins/catalog-backend/src/stitching/types.ts | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/plugins/catalog-backend/src/stitching/DefaultStitcher.ts b/plugins/catalog-backend/src/stitching/DefaultStitcher.ts index 335d82f173..a6f794a0aa 100644 --- a/plugins/catalog-backend/src/stitching/DefaultStitcher.ts +++ b/plugins/catalog-backend/src/stitching/DefaultStitcher.ts @@ -80,7 +80,7 @@ export class DefaultStitcher { async start() { if (this.stopFunc) { - throw new Error('Processing engine is already started'); + throw new Error('Stitcher is already started'); } const { pollingInterval, stitchTimeout } = this.strategy; diff --git a/plugins/catalog-backend/src/stitching/types.ts b/plugins/catalog-backend/src/stitching/types.ts index 4d055513b1..5e11f17fab 100644 --- a/plugins/catalog-backend/src/stitching/types.ts +++ b/plugins/catalog-backend/src/stitching/types.ts @@ -42,6 +42,10 @@ export function stitchingStrategyFromConfig( 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.", ); + } 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.`, + ); } const pollingIntervalKey = 'catalog.stitchingStrategy.pollingInterval'; From e864f77e35dea8bbefb2f23a8912228d35ee7d70 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 21 May 2026 17:18:23 +0200 Subject: [PATCH 10/17] fix(catalog-backend): make stitch ticket guard atomic on PostgreSQL and SQLite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On PostgreSQL and SQLite, the stitch ticket check is now part of the upsert WHERE clause, eliminating the TOCTOU window between the check and the write. MySQL does not support ON CONFLICT ... DO UPDATE ... WHERE, so it retains the separate SELECT with a negligible race window. SQLite misreports affected rows for blocked upserts, so a post-write hash verification is used there instead of the row count. Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../operations/stitcher/performStitching.ts | 44 +++++++++++++++++-- 1 file changed, 40 insertions(+), 4 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index aa5572f4e6..bbfe46a417 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -196,9 +196,15 @@ export async function performStitching(options: { // to write the search index. const searchEntries = buildEntitySearch(entityId, entity); - // Guard against concurrent stitchers by checking that the stitch_ticket - // in stitch_queue still matches what we were given. - if (stitchTicket) { + const isMySQL = String(knex.client.config.client).includes('mysql'); + + // 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 that + // syntax, so it falls back to a separate check with a negligible + // TOCTOU window (self-corrects on the next ~1s stitch cycle). + if (stitchTicket && isMySQL) { const ticketValid = await knex('stitch_queue') .where('entity_ref', entityRef) .where('stitch_ticket', stitchTicket) @@ -211,7 +217,7 @@ export async function performStitching(options: { } } - await knex('final_entities') + let upsert = knex('final_entities') .insert({ entity_id: entityId, entity_ref: entityRef, @@ -222,6 +228,36 @@ export async function performStitching(options: { .onConflict('entity_id') .merge(['final_entity', 'hash', 'last_updated_at']); + if (stitchTicket && !isMySQL) { + upsert = upsert.where( + knex.raw( + 'exists (select 1 from stitch_queue where entity_ref = ? and stitch_ticket = ?)', + [entityRef, stitchTicket], + ), + ); + } + + const rowsAffected = await upsert; + + // PostgreSQL correctly reports 0 rows when the WHERE blocks the + // merge. SQLite always reports 1, so fall back to a hash check. + if (stitchTicket && !isMySQL) { + const isSQLite = String(knex.client.config.client).includes('sqlite'); + const blocked = isSQLite + ? !(await knex('final_entities') + .where('entity_id', entityId) + .where('hash', hash) + .select(knex.raw('1')) + .first()) + : rowsAffected === 0; + if (blocked) { + logger.debug( + `Entity ${entityRef} is already stitched, skipping write.`, + ); + return 'abandoned'; + } + } + await syncSearchRows(knex, entityId, searchEntries); return 'changed'; From 8d38538a8213051b0487e47ab1ec48ca7d92e529 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 21 May 2026 18:02:09 +0200 Subject: [PATCH 11/17] =?UTF-8?q?fix(catalog-backend):=20fix=20tsc=20error?= =?UTF-8?q?=20=E2=80=94=20Knex=20insert=20returns=20number[],=20not=20numb?= =?UTF-8?q?er?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../src/database/operations/stitcher/performStitching.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index bbfe46a417..93ad4364da 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -249,7 +249,7 @@ export async function performStitching(options: { .where('hash', hash) .select(knex.raw('1')) .first()) - : rowsAffected === 0; + : rowsAffected[0] === 0; if (blocked) { logger.debug( `Entity ${entityRef} is already stitched, skipping write.`, From e335055c40abdbcff905ae89c336401a136e868a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 21 May 2026 18:07:12 +0200 Subject: [PATCH 12/17] fix(catalog-backend): use hash check for stitch ticket guard instead of unreliable row counts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit INSERT return values differ across database engines (row IDs, row counts, or empty arrays), making them unsuitable for detecting blocked upserts. Use a post-write hash verification instead, which works reliably on all engines. Adds test coverage for stale ticket rejection across all supported database engines (MySQL 8, PostgreSQL 14/18, SQLite 3). Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../stitcher/performStitching.test.ts | 93 +++++++++++++++++++ .../operations/stitcher/performStitching.ts | 34 +++---- 2 files changed, 110 insertions(+), 17 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index 9924f75b9d..a3324c2881 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -363,5 +363,98 @@ 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('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 knex('stitch_queue') + .where('entity_ref', 'k:ns/n') + .select('stitch_ticket') + .first() + )?.stitch_ticket; + + const result1 = await performStitching({ + knex, + logger: mockServices.logger.mock(), + entityRef: 'k:ns/n', + stitchTicket: validTicket, + }); + expect(result1).toBe('changed'); + + const afterFirst = await knex('final_entities'); + expect(afterFirst).toHaveLength(1); + const firstHash = afterFirst[0].hash; + + // Now change the processed entity and mark for stitching again + await knex('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 knex('stitch_queue') + .where('entity_ref', 'k:ns/n') + .select('stitch_ticket') + .first() + )?.stitch_ticket; + + // 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('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('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 }); + }); }, ); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index 93ad4364da..786929c595 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -196,14 +196,15 @@ export async function performStitching(options: { // to write the search index. const searchEntries = buildEntitySearch(entityId, entity); - const isMySQL = String(knex.client.config.client).includes('mysql'); - // 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 that - // syntax, so it falls back to a separate check with a negligible - // TOCTOU window (self-corrects on the next ~1s stitch cycle). + // 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 (stitchTicket && isMySQL) { const ticketValid = await knex('stitch_queue') .where('entity_ref', entityRef) @@ -237,20 +238,19 @@ export async function performStitching(options: { ); } - const rowsAffected = await upsert; + await upsert; - // PostgreSQL correctly reports 0 rows when the WHERE blocks the - // merge. SQLite always reports 1, so fall back to a hash check. + // 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 (stitchTicket && !isMySQL) { - const isSQLite = String(knex.client.config.client).includes('sqlite'); - const blocked = isSQLite - ? !(await knex('final_entities') - .where('entity_id', entityId) - .where('hash', hash) - .select(knex.raw('1')) - .first()) - : rowsAffected[0] === 0; - if (blocked) { + const written = await knex('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.`, ); From eac181f1b56d4ebb499ab6584d22e534f37526b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 21 May 2026 22:38:04 +0200 Subject: [PATCH 13/17] fix(catalog-backend): fix immediate mode logging on repeated calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A second call with mode='immediate' would incorrectly log "Unknown strategy mode" because the deprecation flag was already set. Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- plugins/catalog-backend/src/stitching/types.ts | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/plugins/catalog-backend/src/stitching/types.ts b/plugins/catalog-backend/src/stitching/types.ts index 5e11f17fab..5bcbf3f68b 100644 --- a/plugins/catalog-backend/src/stitching/types.ts +++ b/plugins/catalog-backend/src/stitching/types.ts @@ -37,11 +37,13 @@ export function stitchingStrategyFromConfig( 'catalog.stitchingStrategy.mode', ); - if (strategyMode === 'immediate' && !immediateDeprecationLogged) { - immediateDeprecationLogged = true; - 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.", - ); + if (strategyMode === 'immediate') { + if (!immediateDeprecationLogged) { + immediateDeprecationLogged = true; + 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.", + ); + } } 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.`, From f1d243293b323602f9452afded26cc60c273c419 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 22 May 2026 11:42:42 +0200 Subject: [PATCH 14/17] test(catalog-backend): await engine start/stop in integration test harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- plugins/catalog-backend/src/tests/integration.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/plugins/catalog-backend/src/tests/integration.test.ts b/plugins/catalog-backend/src/tests/integration.test.ts index d4235c2c61..f98418ecda 100644 --- a/plugins/catalog-backend/src/tests/integration.test.ts +++ b/plugins/catalog-backend/src/tests/integration.test.ts @@ -360,12 +360,12 @@ 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 From 9f85367cdcce9da4e6c1d3187b59447fc93a0ccb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 22 May 2026 14:40:30 +0200 Subject: [PATCH 15/17] fix(catalog-backend): deduplicate entity IDs when marking for stitching after deletion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../operations/provider/deleteWithEagerPruningOfChildren.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts b/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts index 7d2f699ea9..80efeace59 100644 --- a/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts +++ b/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts @@ -240,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', From 5b348f3ac17706b060b8b64817b14a0c0d6c3138 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 22 May 2026 16:34:06 +0200 Subject: [PATCH 16/17] refactor(catalog-backend): make stitchTicket required in performStitching MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Now that immediate mode is removed, the stitch ticket is always provided by the deferred worker. Remove the optional marker and all the conditional guards that checked for its presence. Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../operations/stitcher/performStitching.test.ts | 8 ++++++++ .../operations/stitcher/performStitching.ts | 13 ++++++------- .../src/stitching/DefaultStitcher.ts | 2 +- 3 files changed, 15 insertions(+), 8 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index a3324c2881..eca65c4985 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -349,12 +349,20 @@ describe.each(databases.eachSupportedId())( }, ]); + await markForStitching({ knex, entityRefs: ['k:ns/n'] }); + const stitchLogger = mockServices.logger.mock(); await expect( performStitching({ knex, logger: stitchLogger, entityRef: 'k:ns/n', + stitchTicket: ( + await knex('stitch_queue') + .where('entity_ref', 'k:ns/n') + .select('stitch_ticket') + .first() + )?.stitch_ticket, }), ).resolves.toBe('changed'); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index 786929c595..061a196d8a 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -52,10 +52,9 @@ export async function performStitching(options: { knex: Knex | Knex.Transaction; logger: LoggerService; 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; // 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 @@ -205,7 +204,7 @@ export async function performStitching(options: { // stitch cycle). const isMySQL = String(knex.client.config.client).includes('mysql'); - if (stitchTicket && isMySQL) { + if (isMySQL) { const ticketValid = await knex('stitch_queue') .where('entity_ref', entityRef) .where('stitch_ticket', stitchTicket) @@ -229,7 +228,7 @@ export async function performStitching(options: { .onConflict('entity_id') .merge(['final_entity', 'hash', 'last_updated_at']); - if (stitchTicket && !isMySQL) { + if (!isMySQL) { upsert = upsert.where( knex.raw( 'exists (select 1 from stitch_queue where entity_ref = ? and stitch_ticket = ?)', @@ -244,7 +243,7 @@ export async function performStitching(options: { // 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 (stitchTicket && !isMySQL) { + if (!isMySQL) { const written = await knex('final_entities') .where('entity_id', entityId) .where('hash', hash) @@ -265,7 +264,7 @@ export async function performStitching(options: { removeFromStitchQueueOnCompletion = false; throw error; } finally { - if (removeFromStitchQueueOnCompletion && stitchTicket) { + if (removeFromStitchQueueOnCompletion) { await markDeferredStitchCompleted({ knex: knex, entityRef, diff --git a/plugins/catalog-backend/src/stitching/DefaultStitcher.ts b/plugins/catalog-backend/src/stitching/DefaultStitcher.ts index a6f794a0aa..61d71f1a40 100644 --- a/plugins/catalog-backend/src/stitching/DefaultStitcher.ts +++ b/plugins/catalog-backend/src/stitching/DefaultStitcher.ts @@ -128,7 +128,7 @@ export class DefaultStitcher { async #stitchOne(options: { entityRef: string; - stitchTicket?: string; + stitchTicket: string; stitchRequestedAt?: DateTime; }) { const track = this.tracker.stitchStart({ From a3d55d7a0e15c13ceb80088bfd43008a63530aea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 22 May 2026 16:49:36 +0200 Subject: [PATCH 17/17] cleanup(catalog-backend): fix blank lines and unsafe ticket fetching in tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove extra blank lines left behind when the stitcher field was removed from engine test constructors. Extract getStitchTicket helper to assert the queue entry exists instead of using optional chaining that could silently pass undefined. Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../stitcher/performStitching.test.ts | 56 +++++++------------ .../DefaultCatalogProcessingEngine.test.ts | 9 --- 2 files changed, 20 insertions(+), 45 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index eca65c4985..c6349379e7 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -90,12 +90,7 @@ it.each(databases.eachSupportedId())( knex, logger, entityRef: 'k:ns/n', - stitchTicket: ( - await knex('stitch_queue') - .where('entity_ref', 'k:ns/n') - .select('stitch_ticket') - .first() - )?.stitch_ticket, + stitchTicket: await getStitchTicket(knex, 'k:ns/n'), }); entities = await knex('final_entities'); @@ -185,12 +180,7 @@ it.each(databases.eachSupportedId())( knex, logger, entityRef: 'k:ns/n', - stitchTicket: ( - await knex('stitch_queue') - .where('entity_ref', 'k:ns/n') - .select('stitch_ticket') - .first() - )?.stitch_ticket, + stitchTicket: await getStitchTicket(knex, 'k:ns/n'), }); entities = await knex('final_entities'); @@ -218,12 +208,7 @@ it.each(databases.eachSupportedId())( knex, logger, entityRef: 'k:ns/n', - stitchTicket: ( - await knex('stitch_queue') - .where('entity_ref', 'k:ns/n') - .select('stitch_ticket') - .first() - )?.stitch_ticket, + stitchTicket: await getStitchTicket(knex, 'k:ns/n'), }); entities = await knex('final_entities'); @@ -357,12 +342,7 @@ describe.each(databases.eachSupportedId())( knex, logger: stitchLogger, entityRef: 'k:ns/n', - stitchTicket: ( - await knex('stitch_queue') - .where('entity_ref', 'k:ns/n') - .select('stitch_ticket') - .first() - )?.stitch_ticket, + stitchTicket: await getStitchTicket(knex, 'k:ns/n'), }), ).resolves.toBe('changed'); @@ -395,12 +375,7 @@ describe.each(databases.eachSupportedId())( // First stitch: create the final_entities row with a valid ticket await markForStitching({ knex, entityRefs: ['k:ns/n'] }); - const validTicket = ( - await knex('stitch_queue') - .where('entity_ref', 'k:ns/n') - .select('stitch_ticket') - .first() - )?.stitch_ticket; + const validTicket = await getStitchTicket(knex, 'k:ns/n'); const result1 = await performStitching({ knex, @@ -427,12 +402,7 @@ describe.each(databases.eachSupportedId())( }); await markForStitching({ knex, entityRefs: ['k:ns/n'] }); - const freshTicket = ( - await knex('stitch_queue') - .where('entity_ref', 'k:ns/n') - .select('stitch_ticket') - .first() - )?.stitch_ticket; + const freshTicket = await getStitchTicket(knex, 'k:ns/n'); // Attempt to stitch with a WRONG ticket (simulating a stale worker) const result2 = await performStitching({ @@ -466,3 +436,17 @@ describe.each(databases.eachSupportedId())( }); }, ); + +async function getStitchTicket( + knex: import('knex').Knex, + entityRef: string, +): Promise { + 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; +} diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts index b473891a7f..bd6eb0bd47 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts @@ -72,7 +72,6 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - createHash: () => hash, scheduler: mockServices.scheduler(), events: mockServices.events.mock(), @@ -142,7 +141,6 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - scheduler: mockServices.scheduler(), createHash: () => hash, events: mockServices.events.mock(), @@ -228,7 +226,6 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - scheduler: mockServices.scheduler(), createHash: () => hash, events: mockServices.events.mock(), @@ -308,7 +305,6 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - scheduler: mockServices.scheduler(), createHash: () => hash, events: mockServices.events.mock(), @@ -370,7 +366,6 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, @@ -488,7 +483,6 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, @@ -596,7 +590,6 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, @@ -682,7 +675,6 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, @@ -773,7 +765,6 @@ describe('DefaultCatalogProcessingEngine', () => { processingDatabase: db, knex: {} as any, orchestrator: orchestrator, - scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100,