diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index db9f209c22..9a9c09daaf 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -19,6 +19,7 @@ import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; import { Knex } from 'knex'; +import { Logger } from 'winston'; import * as uuid from 'uuid'; import { DatabaseManager } from './DatabaseManager'; import { DefaultProcessingDatabase } from './DefaultProcessingDatabase'; @@ -29,12 +30,15 @@ import { } from './tables'; describe('Default Processing Database', () => { - const logger = getVoidLogger(); + const defaultLogger = getVoidLogger(); const databases = TestDatabases.create({ ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], }); - async function createDatabase(databaseId: TestDatabaseId) { + async function createDatabase( + databaseId: TestDatabaseId, + logger: Logger = defaultLogger, + ) { const knex = await databases.init(databaseId); await DatabaseManager.createDatabase(knex); return { @@ -74,7 +78,11 @@ describe('Default Processing Database', () => { it.each(databases.eachSupportedId())( 'updates refresh state with varying location keys, %p', async databaseId => { - const { db } = await createDatabase(databaseId); + const mockWarn = jest.fn(); + const { db } = await createDatabase(databaseId, ({ + debug: jest.fn(), + warn: mockWarn, + } as unknown) as Logger); await db.transaction(async tx => { const knexTx = tx as Knex.Transaction; @@ -102,12 +110,14 @@ describe('Default Processing Database', () => { expectedLocationKey: 'x', type: 'd', expectedType: 'c', + expectConflict: true, }, { locationKey: undefined, expectedLocationKey: 'x', type: 'e', expectedType: 'c', + expectConflict: true, }, { locationKey: 'x', @@ -117,6 +127,8 @@ describe('Default Processing Database', () => { }, ]; for (const step of steps) { + mockWarn.mockClear(); + await db.addUnprocessedEntities(tx, { sourceKey: 'testing', entities: [ @@ -127,6 +139,16 @@ describe('Default Processing Database', () => { ], }); + if (step.expectConflict) { + // eslint-disable-next-line jest/no-conditional-expect + expect(mockWarn).toHaveBeenCalledWith( + expect.stringMatching(/^Detected conflicting entityRef/), + ); + } else { + // eslint-disable-next-line jest/no-conditional-expect + expect(mockWarn).not.toHaveBeenCalled(); + } + const entities = await knexTx( 'refresh_state', ).select(); @@ -447,7 +469,7 @@ describe('Default Processing Database', () => { }, kind: 'Location', } as Entity, - locationKey: 'file:/tmp/foobar', + locationKey: 'file:///tmp/foobar', }, ], }), @@ -666,7 +688,7 @@ describe('Default Processing Database', () => { }, kind: 'Location', } as Entity, - locationKey: 'file:/tmp/foobar', + locationKey: 'file:///tmp/foobar', }, ], }); @@ -803,6 +825,73 @@ describe('Default Processing Database', () => { }, 60_000, ); + + it.each(databases.eachSupportedId())( + 'should update the location key during full replace, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + await createLocations(knex, ['location:default/removed']); + await insertRefreshStateRow(knex, { + entity_id: uuid.v4(), + entity_ref: 'location:default/replaced', + unprocessed_entity: '{}', + processed_entity: '{}', + errors: '[]', + next_update_at: '2021-04-01 13:37:00', + last_discovery_at: '2021-04-01 13:37:00', + location_key: 'file:///tmp/old', + }); + + await insertRefRow(knex, { + source_key: 'lols', + target_entity_ref: 'location:default/removed', + }); + await insertRefRow(knex, { + source_key: 'lols', + target_entity_ref: 'location:default/replaced', + }); + + await db.transaction(async tx => { + await db.replaceUnprocessedEntities(tx, { + type: 'full', + sourceKey: 'lols', + items: [ + { + entity: { + apiVersion: '1.0.0', + metadata: { + name: 'replaced', + }, + kind: 'Location', + } as Entity, + locationKey: 'file:///tmp/foobar', + }, + ], + }); + }); + + const currentRefreshState = await knex( + 'refresh_state', + ).select(); + expect(currentRefreshState).toEqual([ + expect.objectContaining({ + entity_ref: 'location:default/replaced', + location_key: 'file:///tmp/foobar', + }), + ]); + + const currentRefRowState = await knex( + 'refresh_state_references', + ).select(); + expect(currentRefRowState).toEqual([ + expect.objectContaining({ + source_key: 'lols', + target_entity_ref: 'location:default/replaced', + }), + ]); + }, + 60_000, + ); }); describe('getProcessableEntities', () => { diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index f5da7850ea..67b83bdb34 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -325,6 +325,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { // Keeps track of the entities that we end up inserting to update refresh_state_references afterwards const stateReferences = new Array(); + const conflictingStateReferences = new Array(); // Upsert all of the unprocessed entities into the refresh_state table, by // their entity ref. @@ -352,8 +353,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { if (refreshResult === 0) { // In the event that we can't update an existing refresh state, we first try to insert a new row - const result = await tx('refresh_state') - .insert({ + try { + await tx('refresh_state').insert({ entity_id: uuid(), entity_ref: entityRef, unprocessed_entity: serializedEntity, @@ -361,14 +362,11 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { location_key: locationKey, next_update_at: tx.fn.now(), last_discovery_at: tx.fn.now(), - }) - .onConflict('entity_ref') - .ignore(); - - // If the row can't be inserted, we have a conflict, but it could be either - // because of a conflicting locationKey or a race with another instance, so check for that too - if (result.length === 0) { - // Check for conflicting entity with same entityRef but different locationKey + }); + } catch (error) { + // If the row can't be inserted, we have a conflict, but it could be either + // because of a conflicting locationKey or a race with another instance, so check + // whether the conflicting entity has the same entityRef but a different locationKey const [conflictingEntity] = await tx( 'refresh_state', ) @@ -380,6 +378,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { this.options.logger.warn( `Detected conflicting entityRef ${entityRef} already referenced by ${conflictingEntity.location_key} and now also ${locationKey}`, ); + conflictingStateReferences.push(entityRef); continue; } } @@ -392,7 +391,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { // Replace all references for the originating entity or source and then create new ones if ('sourceKey' in options) { await tx('refresh_state_references') - .where({ source_key: options.sourceKey }) + .whereNotIn('target_entity_ref', conflictingStateReferences) + .andWhere({ source_key: options.sourceKey }) .delete(); await tx.batchInsert( 'refresh_state_references', @@ -404,7 +404,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ); } else { await tx('refresh_state_references') - .where({ source_entity_ref: options.sourceEntityRef }) + .whereNotIn('target_entity_ref', conflictingStateReferences) + .andWhere({ source_entity_ref: options.sourceEntityRef }) .delete(); await tx.batchInsert( 'refresh_state_references',