diff --git a/.changeset/breezy-schools-compete.md b/.changeset/breezy-schools-compete.md new file mode 100644 index 0000000000..b6e82eff25 --- /dev/null +++ b/.changeset/breezy-schools-compete.md @@ -0,0 +1,7 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Fixed an issue where sometimes entities would have stale relations "stuck" and +not getting removed as expected, after the other end of the relation had stopped +referring to them. diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts index c7cd6b7026..251388f0d2 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts @@ -221,7 +221,7 @@ describe('Default Processing Database', () => { }, ]; - await db.transaction(tx => + let updateResult = await db.transaction(tx => db.updateProcessedEntity(tx, { id, processedEntity, @@ -231,7 +231,7 @@ describe('Default Processing Database', () => { }), ); - const savedRelations = await knex('relations') + let savedRelations = await knex('relations') .where({ originating_entity_id: id }) .select(); expect(savedRelations.length).toBe(1); @@ -241,6 +241,36 @@ describe('Default Processing Database', () => { type: 'memberOf', target_entity_ref: 'component:default/foo', }); + expect(updateResult.previous.relations).toEqual([]); + + updateResult = await db.transaction(tx => + db.updateProcessedEntity(tx, { + id, + processedEntity, + resultHash: '', + relations: relations, + deferredEntities: [], + }), + ); + + savedRelations = await knex('relations') + .where({ originating_entity_id: id }) + .select(); + expect(savedRelations.length).toBe(1); + expect(savedRelations[0]).toEqual({ + originating_entity_id: id, + source_entity_ref: 'component:default/foo', + type: 'memberOf', + target_entity_ref: 'component:default/foo', + }); + expect(updateResult.previous.relations).toEqual([ + { + originating_entity_id: expect.any(String), + source_entity_ref: 'component:default/foo', + type: 'memberOf', + target_entity_ref: 'component:default/foo', + }, + ]); }, 60_000, ); diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index c9fe0da0a3..dd28cf1290 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -68,7 +68,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { async updateProcessedEntity( txOpaque: Transaction, options: UpdateProcessedEntityOptions, - ): Promise { + ): Promise<{ previous: { relations: DbRelationsRow[] } }> { const tx = txOpaque as Knex.Transaction; const { id, @@ -108,9 +108,20 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { }); // Delete old relations - await tx('relations') - .where({ originating_entity_id: id }) - .delete(); + let previousRelationRows: DbRelationsRow[]; + if (tx.client.config.client.includes('sqlite3')) { + previousRelationRows = await tx('relations') + .select('*') + .where({ originating_entity_id: id }); + await tx('relations') + .where({ originating_entity_id: id }) + .delete(); + } else { + previousRelationRows = await tx('relations') + .where({ originating_entity_id: id }) + .delete() + .returning('*'); + } // Batch insert new relations const relationRows: DbRelationsRow[] = relations.map( @@ -126,6 +137,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { this.deduplicateRelations(relationRows), BATCH_SIZE, ); + + return { + previous: { + relations: previousRelationRows, + }, + }; } async updateProcessedEntityErrors( diff --git a/plugins/catalog-backend/src/database/types.ts b/plugins/catalog-backend/src/database/types.ts index 1affb14b45..ce45a88938 100644 --- a/plugins/catalog-backend/src/database/types.ts +++ b/plugins/catalog-backend/src/database/types.ts @@ -19,6 +19,7 @@ import { JsonObject } from '@backstage/types'; import { DateTime } from 'luxon'; import { EntityRelationSpec } from '../api'; import { DeferredEntity } from '../processing/types'; +import { DbRelationsRow } from './tables'; /** * An abstraction for transactions of the underlying database technology. @@ -125,7 +126,7 @@ export interface ProcessingDatabase { updateProcessedEntity( txOpaque: Transaction, options: UpdateProcessedEntityOptions, - ): Promise; + ): Promise<{ previous: { relations: DbRelationsRow[] } }>; /** * Updates the cache associated with an entity. diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts index b93ee758d7..10de36a1be 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts @@ -91,6 +91,10 @@ describe('DefaultCatalogProcessingEngine', () => { }, ], }); + db.listParents.mockResolvedValue({ entityRefs: [] }); + db.updateProcessedEntity.mockResolvedValue({ + previous: { relations: [] }, + }); await engine.start(); await waitForExpect(() => { @@ -153,6 +157,10 @@ describe('DefaultCatalogProcessingEngine', () => { }, ], }); + db.listParents.mockResolvedValue({ entityRefs: [] }); + db.updateProcessedEntity.mockImplementation(async () => ({ + previous: { relations: [] }, + })); await engine.start(); await waitForExpect(() => { @@ -213,6 +221,10 @@ describe('DefaultCatalogProcessingEngine', () => { items: [{ ...refreshState, resultHash: 'NOT RIGHT' }], }) .mockResolvedValue({ items: [] }); + db.updateProcessedEntity.mockImplementation(async () => ({ + previous: { relations: [] }, + })); + await engine.start(); await waitForExpect(() => { @@ -285,6 +297,10 @@ describe('DefaultCatalogProcessingEngine', () => { items: [refreshState], }) .mockResolvedValue({ items: [] }); + db.listParents.mockResolvedValue({ entityRefs: [] }); + db.updateProcessedEntity.mockImplementation(async () => ({ + previous: { relations: [] }, + })); await waitForExpect(() => { expect(db.updateEntityCache).toBeCalledTimes(1); @@ -319,4 +335,115 @@ describe('DefaultCatalogProcessingEngine', () => { await engine.stop(); }); + + it('should stitch both the previous and new sources when relations change', async () => { + const engine = new DefaultCatalogProcessingEngine( + getVoidLogger(), + db, + orchestrator, + stitcher, + () => hash, + 100, + ); + + db.transaction.mockImplementation(cb => cb((() => {}) as any)); + + const entity = { + apiVersion: '1', + kind: 'k', + metadata: { name: 'me', namespace: 'ns' }, + }; + const processableEntity = { + entityRef: 'foo', + id: '1', + unprocessedEntity: entity, + resultHash: '', + state: [] as any, + nextUpdateAt: DateTime.now(), + lastDiscoveryAt: DateTime.now(), + }; + + db.listParents.mockResolvedValue({ entityRefs: [] }); + db.getProcessableEntities + .mockResolvedValueOnce({ + items: [processableEntity], + }) + .mockResolvedValueOnce({ + items: [processableEntity], + }); + db.updateProcessedEntity + .mockImplementationOnce(async () => ({ + previous: { relations: [] }, + })) + .mockImplementationOnce(async () => ({ + previous: { + relations: [ + { + originating_entity_id: '', + type: 't', + source_entity_ref: 'k:ns/other1', + target_entity_ref: 'k:ns/me', + }, + { + originating_entity_id: '', + type: 't', + source_entity_ref: 'k:ns/other2', + target_entity_ref: 'k:ns/me', + }, + ], + }, + })); + + orchestrator.process + .mockResolvedValueOnce({ + ok: true, + completedEntity: entity, + relations: [ + { + type: 't', + source: { kind: 'k', namespace: 'ns', name: 'other1' }, + target: { kind: 'k', namespace: 'ns', name: 'me' }, + }, + { + type: 't', + source: { kind: 'k', namespace: 'ns', name: 'other2' }, + target: { kind: 'k', namespace: 'ns', name: 'me' }, + }, + ], + errors: [], + deferredEntities: [], + state: {}, + }) + .mockResolvedValueOnce({ + ok: true, + completedEntity: entity, + relations: [ + { + type: 't', + source: { kind: 'k', namespace: 'ns', name: 'other2' }, + target: { kind: 'k', namespace: 'ns', name: 'me' }, + }, + { + type: 't', + source: { kind: 'k', namespace: 'ns', name: 'other3' }, + target: { kind: 'k', namespace: 'ns', name: 'me' }, + }, + ], + errors: [], + deferredEntities: [], + state: {}, + }); + + await engine.start(); + await waitForExpect(() => { + expect(stitcher.stitch).toBeCalledTimes(2); + }); + expect([...stitcher.stitch.mock.calls[0][0]]).toEqual( + expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other2']), + ); + expect([...stitcher.stitch.mock.calls[1][0]]).toEqual( + expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other3']), + ); + await engine.stop(); + }); }); diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index fb42af778b..b7ef5c9344 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -169,24 +169,43 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } result.completedEntity.metadata.uid = id; + let oldRelationSources: Set; await this.processingDatabase.transaction(async tx => { - await this.processingDatabase.updateProcessedEntity(tx, { - id, - processedEntity: result.completedEntity, - resultHash, - errors: errorsString, - relations: result.relations, - deferredEntities: result.deferredEntities, - locationKey, - }); + const { previous } = + await this.processingDatabase.updateProcessedEntity(tx, { + id, + processedEntity: result.completedEntity, + resultHash, + errors: errorsString, + relations: result.relations, + deferredEntities: result.deferredEntities, + locationKey, + }); + oldRelationSources = new Set( + previous.relations.map(r => r.source_entity_ref), + ); }); + const newRelationSources = new Set( + result.relations.map(relation => + stringifyEntityRef(relation.source), + ), + ); + const setOfThingsToStitch = new Set([ stringifyEntityRef(result.completedEntity), - ...result.relations.map(relation => - stringifyEntityRef(relation.source), - ), ]); + newRelationSources.forEach(r => { + if (!oldRelationSources.has(r)) { + setOfThingsToStitch.add(r); + } + }); + oldRelationSources!.forEach(r => { + if (!newRelationSources.has(r)) { + setOfThingsToStitch.add(r); + } + }); + await this.stitcher.stitch(setOfThingsToStitch); track.markSuccessfulWithChanges(setOfThingsToStitch.size);