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] 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(); - }); - }); + }, + ); });