From 4c75f2f0e29262e4cd432dd689ce07c632db6e37 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Wed, 23 Jun 2021 14:16:33 +0200 Subject: [PATCH 01/14] Catalog: Add location key to handle duplicate entities Signed-off-by: Johan Haals --- .../20210302150147_refresh_state.js | 2 + ...210622104022_refresh_state_location_key.js | 35 +++++++++++++++++ .../src/next/ConfigLocationEntityProvider.ts | 5 ++- .../next/DefaultCatalogProcessingEngine.ts | 9 +++-- .../src/next/DefaultLocationStore.ts | 12 +++--- .../database/DefaultProcessingDatabase.ts | 39 ++++++++++++------- .../src/next/database/tables.ts | 1 + .../src/next/database/types.ts | 13 ++++--- .../DefaultCatalogProcessingOrchestrator.ts | 1 - .../processing/ProcessorOutputCollector.ts | 21 +++++++--- .../src/next/processing/types.ts | 7 +++- plugins/catalog-backend/src/next/types.ts | 5 ++- 12 files changed, 110 insertions(+), 40 deletions(-) create mode 100644 plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js diff --git a/plugins/catalog-backend/migrations/20210302150147_refresh_state.js b/plugins/catalog-backend/migrations/20210302150147_refresh_state.js index db66b46305..2752ab0176 100644 --- a/plugins/catalog-backend/migrations/20210302150147_refresh_state.js +++ b/plugins/catalog-backend/migrations/20210302150147_refresh_state.js @@ -65,6 +65,8 @@ exports.up = async function up(knex) { .dateTime('last_discovery_at') // TODO: timezone or change to epoch-millis or similar .notNullable() .comment('The last timestamp of which this entity was discovered'); + // TODO: get migrations to work for this field instead. should be removed before merge. + table.text('location_key').nullable().comment('Location key'); table.unique(['entity_ref'], 'refresh_state_entity_ref_uniq'); table.index('entity_id', 'refresh_state_entity_id_idx'); table.index('entity_ref', 'refresh_state_entity_ref_idx'); diff --git a/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js b/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js new file mode 100644 index 0000000000..b77ff8961c --- /dev/null +++ b/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js @@ -0,0 +1,35 @@ +/* + * Copyright 2020 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// @ts-check + +/** + * @param {import('knex').Knex} knex + */ +exports.up = async function up(knex) { + await knex.schema.alterTable('refresh_state', table => { + table.text('location_key').nullable().comment('').alter(); + }); +}; + +/** + * @param {import('knex').Knex} knex + */ +exports.down = async function down(knex) { + await knex.schema.alterTable('refresh_state', table => { + table.dropColumn('location_key'); + }); +}; diff --git a/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts b/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts index 9218a9c0e1..5e343b0f23 100644 --- a/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts +++ b/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts @@ -16,6 +16,7 @@ import { Config } from '@backstage/config'; import path from 'path'; +import { getEntityLocationRef } from './processing/util'; import { EntityProvider, EntityProviderConnection } from './types'; import { locationSpecToLocationEntity } from './util'; @@ -37,10 +38,12 @@ export class ConfigLocationEntityProvider implements EntityProvider { const entities = locationConfigs.map(location => { const type = location.getString('type'); const target = location.getString('target'); - return locationSpecToLocationEntity({ + const entity = locationSpecToLocationEntity({ type, target: type === 'file' ? path.resolve(target) : target, }); + const emitKey = getEntityLocationRef(entity); + return { entity, emitKey }; }); await this.connection.applyMutation({ diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index a504de0bf3..e3828bdfd8 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -46,7 +46,7 @@ class Connection implements EntityProviderConnection { const db = this.config.processingDatabase; if (mutation.type === 'full') { - this.check(mutation.entities); + this.check(mutation.entities.map(e => e.entity)); await db.transaction(async tx => { await db.replaceUnprocessedEntities(tx, { sourceKey: this.config.id, @@ -57,8 +57,8 @@ class Connection implements EntityProviderConnection { return; } - this.check(mutation.added); - this.check(mutation.removed); + this.check(mutation.added.map(e => e.entity)); + this.check(mutation.removed.map(e => e.entity)); await db.transaction(async tx => { await db.replaceUnprocessedEntities(tx, { sourceKey: this.config.id, @@ -125,7 +125,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { }, processTask: async item => { try { - const { id, state, unprocessedEntity, entityRef } = item; + const { id, state, unprocessedEntity, entityRef, emitKey } = item; const result = await this.orchestrator.process({ entity: unprocessedEntity, state, @@ -171,6 +171,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { errors: errorsString, relations: result.relations, deferredEntities: result.deferredEntities, + emitKey, }); }); diff --git a/plugins/catalog-backend/src/next/DefaultLocationStore.ts b/plugins/catalog-backend/src/next/DefaultLocationStore.ts index 09de77e49f..5c3cefbaf7 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationStore.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationStore.ts @@ -19,6 +19,7 @@ import { ConflictError, NotFoundError } from '@backstage/errors'; import { Knex } from 'knex'; import { v4 as uuid } from 'uuid'; import { DbLocationsRow } from './database/tables'; +import { getEntityLocationRef } from './processing/util'; import { EntityProvider, EntityProviderConnection, @@ -60,10 +61,10 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { return inner; }); - + const entity = locationSpecToLocationEntity(location); await this.connection.applyMutation({ type: 'delta', - added: [locationSpecToLocationEntity(location)], + added: [{ entity, emitKey: getEntityLocationRef(entity) }], removed: [], }); @@ -102,11 +103,11 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { await tx('locations').where({ id }).del(); return location; }); - + const entity = locationSpecToLocationEntity(deleted); await this.connection.applyMutation({ type: 'delta', added: [], - removed: [locationSpecToLocationEntity(deleted)], + removed: [{ entity, emitKey: getEntityLocationRef(entity) }], }); } @@ -124,7 +125,8 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { const locations = await this.locations(); const entities = locations.map(location => { - return locationSpecToLocationEntity(location); + const entity = locationSpecToLocationEntity(location); + return { entity, emitKey: getEntityLocationRef(entity) }; }); await this.connection.applyMutation({ diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 79daee13e6..5bdde1fe78 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -22,6 +22,7 @@ import lodash from 'lodash'; import { v4 as uuid } from 'uuid'; import type { Logger } from 'winston'; import { Transaction } from '../../database'; +import { DeferredEntity } from '../processing/types'; import { DbRefreshStateReferencesRow, DbRefreshStateRow, @@ -63,15 +64,18 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { errors, relations, deferredEntities, + emitKey, } = options; - const refreshResult = await tx('refresh_state') .update({ processed_entity: JSON.stringify(processedEntity), cache: JSON.stringify(state), errors, + emit_key: emitKey, }) - .where('entity_id', id); + .where('entity_id', id) + .andWhere('emit_key', emitKey) + .orWhere('emit_key', ''); if (refreshResult === 0) { throw new NotFoundError(`Processing state not found for ${id}`); } @@ -129,11 +133,11 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { private async createDelta( tx: Knex.Transaction, options: ReplaceUnprocessedEntitiesOptions, - ): Promise<{ toAdd: Entity[]; toRemove: string[] }> { + ): Promise<{ toAdd: DeferredEntity[]; toRemove: string[] }> { if (options.type === 'delta') { return { toAdd: options.added, - toRemove: options.removed.map(e => stringifyEntityRef(e)), + toRemove: options.removed.map(e => stringifyEntityRef(e.entity)), }; } @@ -146,7 +150,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const items = options.items.map(entity => ({ entity, - ref: stringifyEntityRef(entity), + ref: stringifyEntityRef(entity.entity), })); const oldRefsSet = new Set(oldRefs); @@ -281,19 +285,22 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { } if (toAdd.length) { - const state: Knex.DbRecord[] = toAdd.map(entity => ({ - entity_id: uuid(), - entity_ref: stringifyEntityRef(entity), - unprocessed_entity: JSON.stringify(entity), - errors: '', - next_update_at: tx.fn.now(), - last_discovery_at: tx.fn.now(), - })); + const state: Knex.DbRecord[] = toAdd.map( + ({ entity, emitKey }) => ({ + entity_id: uuid(), + entity_ref: stringifyEntityRef(entity), + unprocessed_entity: JSON.stringify(entity), + emit_key: emitKey, + errors: '', + next_update_at: tx.fn.now(), + last_discovery_at: tx.fn.now(), + }), + ); const stateReferences: DbRefreshStateReferencesRow[] = toAdd.map( entity => ({ source_key: options.sourceKey, - target_entity_ref: stringifyEntityRef(entity), + target_entity_ref: stringifyEntityRef(entity.entity), }), ); // TODO(freben): Concurrency? If we did these one by one, a .onConflict().merge would have made sense @@ -313,12 +320,13 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const tx = txOpaque as Knex.Transaction; const stateRows = options.entities.map( - entity => + ({ entity, emitKey }) => ({ entity_id: uuid(), entity_ref: stringifyEntityRef(entity), unprocessed_entity: JSON.stringify(entity), errors: '', + emit_key: emitKey, next_update_at: tx.fn.now(), last_discovery_at: tx.fn.now(), } as Knex.DbRecord), @@ -406,6 +414,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ? JSON.parse(i.cache) : new Map(), errors: i.errors, + emitKey: i.emit_key, } as RefreshStateItem), ), }; diff --git a/plugins/catalog-backend/src/next/database/tables.ts b/plugins/catalog-backend/src/next/database/tables.ts index 93453ba914..a298c37022 100644 --- a/plugins/catalog-backend/src/next/database/tables.ts +++ b/plugins/catalog-backend/src/next/database/tables.ts @@ -29,6 +29,7 @@ export type DbRefreshStateRow = { next_update_at: string; last_discovery_at: string; // remove? errors?: string; + emit_key: string; }; export type DbRefreshStateReferencesRow = { diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index dd342126c4..5e4d0becd3 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -17,10 +17,11 @@ import { Entity, EntityRelationSpec } from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; import { Transaction } from '../../database/types'; +import { DeferredEntity } from '../processing/types'; export type AddUnprocessedEntitiesOptions = { entityRef: string; - entities: Entity[]; + entities: DeferredEntity[]; }; export type AddUnprocessedEntitiesResult = {}; @@ -31,7 +32,8 @@ export type UpdateProcessedEntityOptions = { state?: Map; errors?: string; relations: EntityRelationSpec[]; - deferredEntities: Entity[]; + deferredEntities: DeferredEntity[]; + emitKey: string; }; export type UpdateProcessedEntityErrorsOptions = { @@ -48,6 +50,7 @@ export type RefreshStateItem = { lastDiscoveryAt: string; // remove? state: Map; errors?: string; + emitKey: string; }; export type GetProcessableEntitiesResult = { @@ -57,13 +60,13 @@ export type GetProcessableEntitiesResult = { export type ReplaceUnprocessedEntitiesOptions = | { sourceKey: string; - items: Entity[]; + items: DeferredEntity[]; type: 'full'; } | { sourceKey: string; - added: Entity[]; - removed: Entity[]; + added: DeferredEntity[]; + removed: DeferredEntity[]; type: 'delta'; }; diff --git a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts index 48a40a2f6e..bb23b6afb3 100644 --- a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts @@ -68,7 +68,6 @@ export class DefaultCatalogProcessingOrchestrator async process( request: EntityProcessingRequest, ): Promise { - // TODO: implement dryRun/eager return this.processSingleEntity(request.entity); } diff --git a/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts b/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts index 3f8594bd48..fbb2744668 100644 --- a/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts +++ b/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts @@ -24,7 +24,12 @@ import { import { Logger } from 'winston'; import { CatalogProcessorResult } from '../../ingestion'; import { locationSpecToLocationEntity } from '../util'; -import { getEntityOriginLocationRef, validateEntityEnvelope } from './util'; +import { DeferredEntity } from './types'; +import { + getEntityLocationRef, + getEntityOriginLocationRef, + validateEntityEnvelope, +} from './util'; /** * Helper class for aggregating all of the emitted data from processors. @@ -32,7 +37,7 @@ import { getEntityOriginLocationRef, validateEntityEnvelope } from './util'; export class ProcessorOutputCollector { private readonly errors = new Array(); private readonly relations = new Array(); - private readonly deferredEntities = new Array(); + private readonly deferredEntities = new Array(); private done = false; constructor( @@ -73,6 +78,8 @@ export class ProcessorOutputCollector { return; } + const location = stringifyLocationReference(i.location); + // Note that at this point, we have only validated the envelope part of // the entity data. Annotations are not part of that, so we have to be // defensive. If the annotations were malformed (e.g. were not a valid @@ -81,7 +88,6 @@ export class ProcessorOutputCollector { const annotations = entity.metadata.annotations || {}; if (typeof annotations === 'object' && !Array.isArray(annotations)) { const originLocation = getEntityOriginLocationRef(this.parentEntity); - const location = stringifyLocationReference(i.location); entity = { ...entity, metadata: { @@ -95,11 +101,14 @@ export class ProcessorOutputCollector { }; } - this.deferredEntities.push(entity); + this.deferredEntities.push({ entity, emitKey: location }); } else if (i.type === 'location') { - this.deferredEntities.push( - locationSpecToLocationEntity(i.location, this.parentEntity), + const entity = locationSpecToLocationEntity( + i.location, + this.parentEntity, ); + const emitKey = getEntityLocationRef(entity); + this.deferredEntities.push({ entity, emitKey }); } else if (i.type === 'relation') { this.relations.push(i.relation); } else if (i.type === 'error') { diff --git a/plugins/catalog-backend/src/next/processing/types.ts b/plugins/catalog-backend/src/next/processing/types.ts index e76e789e54..b2a627943b 100644 --- a/plugins/catalog-backend/src/next/processing/types.ts +++ b/plugins/catalog-backend/src/next/processing/types.ts @@ -27,7 +27,7 @@ export type EntityProcessingResult = ok: true; state: Map; completedEntity: Entity; - deferredEntities: Entity[]; + deferredEntities: DeferredEntity[]; relations: EntityRelationSpec[]; errors: Error[]; } @@ -39,3 +39,8 @@ export type EntityProcessingResult = export interface CatalogProcessingOrchestrator { process(request: EntityProcessingRequest): Promise; } + +export type DeferredEntity = { + entity: Entity; + emitKey: string; +}; diff --git a/plugins/catalog-backend/src/next/types.ts b/plugins/catalog-backend/src/next/types.ts index c689f288aa..3b27dd7d10 100644 --- a/plugins/catalog-backend/src/next/types.ts +++ b/plugins/catalog-backend/src/next/types.ts @@ -15,6 +15,7 @@ */ import { Entity, Location, LocationSpec } from '@backstage/catalog-model'; +import { DeferredEntity } from './processing/types'; export interface LocationService { createLocation( @@ -39,8 +40,8 @@ export interface CatalogProcessingEngine { } export type EntityProviderMutation = - | { type: 'full'; entities: Entity[] } - | { type: 'delta'; added: Entity[]; removed: Entity[] }; + | { type: 'full'; entities: DeferredEntity[] } + | { type: 'delta'; added: DeferredEntity[]; removed: DeferredEntity[] }; export interface EntityProviderConnection { applyMutation(mutation: EntityProviderMutation): Promise; From ffaedfb702d2f21168fa0e19013f9c65aadc1528 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 28 Jun 2021 10:37:38 +0200 Subject: [PATCH 02/14] Rename to locationKey, make it optional Signed-off-by: Johan Haals --- .../src/next/DefaultCatalogProcessingEngine.ts | 4 ++-- .../src/next/DefaultLocationService.ts | 12 +++++++++--- .../src/next/DefaultLocationStore.ts | 6 +++--- .../next/database/DefaultProcessingDatabase.ts | 18 +++++++++--------- .../src/next/database/tables.ts | 2 +- .../catalog-backend/src/next/database/types.ts | 4 ++-- .../processing/ProcessorOutputCollector.ts | 6 +++--- .../src/next/processing/types.ts | 2 +- 8 files changed, 30 insertions(+), 24 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index e3828bdfd8..0ae20c94b1 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -125,7 +125,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { }, processTask: async item => { try { - const { id, state, unprocessedEntity, entityRef, emitKey } = item; + const { id, state, unprocessedEntity, entityRef, locationKey } = item; const result = await this.orchestrator.process({ entity: unprocessedEntity, state, @@ -171,7 +171,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { errors: errorsString, relations: result.relations, deferredEntities: result.deferredEntities, - emitKey, + locationKey, }); }); diff --git a/plugins/catalog-backend/src/next/DefaultLocationService.ts b/plugins/catalog-backend/src/next/DefaultLocationService.ts index ceb51a952e..70679a3bd6 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationService.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationService.ts @@ -20,7 +20,10 @@ import { LOCATION_ANNOTATION, ORIGIN_LOCATION_ANNOTATION, } from '@backstage/catalog-model'; -import { CatalogProcessingOrchestrator } from './processing/types'; +import { + CatalogProcessingOrchestrator, + DeferredEntity, +} from './processing/types'; import { LocationService, LocationStore } from './types'; import { locationSpecToMetadataName } from './util'; @@ -73,7 +76,9 @@ export class DefaultLocationService implements LocationService { target: spec.target, }, }; - const unprocessedEntities: Entity[] = [entity]; + const unprocessedEntities: DeferredEntity[] = [ + { entity, locationKey: `${spec.type}:${spec.target}` }, + ]; const entities: Entity[] = []; const state = new Map(); // ignored while (unprocessedEntities.length) { @@ -82,11 +87,12 @@ export class DefaultLocationService implements LocationService { continue; } const processed = await this.orchestrator.process({ - entity: currentEntity, + entity: currentEntity.entity, state, }); if (processed.ok) { + // todo unprocessedEntities.push(...processed.deferredEntities); entities.push(processed.completedEntity); } else { diff --git a/plugins/catalog-backend/src/next/DefaultLocationStore.ts b/plugins/catalog-backend/src/next/DefaultLocationStore.ts index 5c3cefbaf7..c9ffe9a682 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationStore.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationStore.ts @@ -64,7 +64,7 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { const entity = locationSpecToLocationEntity(location); await this.connection.applyMutation({ type: 'delta', - added: [{ entity, emitKey: getEntityLocationRef(entity) }], + added: [{ entity, locationKey: getEntityLocationRef(entity) }], removed: [], }); @@ -107,7 +107,7 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { await this.connection.applyMutation({ type: 'delta', added: [], - removed: [{ entity, emitKey: getEntityLocationRef(entity) }], + removed: [{ entity, locationKey: getEntityLocationRef(entity) }], }); } @@ -126,7 +126,7 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { const entities = locations.map(location => { const entity = locationSpecToLocationEntity(location); - return { entity, emitKey: getEntityLocationRef(entity) }; + return { entity, locationKey: getEntityLocationRef(entity) }; }); await this.connection.applyMutation({ diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 5bdde1fe78..c2bfb4f633 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -64,18 +64,18 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { errors, relations, deferredEntities, - emitKey, + locationKey, } = options; const refreshResult = await tx('refresh_state') .update({ processed_entity: JSON.stringify(processedEntity), cache: JSON.stringify(state), errors, - emit_key: emitKey, + location_key: locationKey, }) .where('entity_id', id) - .andWhere('emit_key', emitKey) - .orWhere('emit_key', ''); + .andWhere('location_key', locationKey) + .orWhere('location_key', ''); if (refreshResult === 0) { throw new NotFoundError(`Processing state not found for ${id}`); } @@ -286,11 +286,11 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { if (toAdd.length) { const state: Knex.DbRecord[] = toAdd.map( - ({ entity, emitKey }) => ({ + ({ entity, locationKey }) => ({ entity_id: uuid(), entity_ref: stringifyEntityRef(entity), unprocessed_entity: JSON.stringify(entity), - emit_key: emitKey, + location_key: locationKey, errors: '', next_update_at: tx.fn.now(), last_discovery_at: tx.fn.now(), @@ -320,13 +320,13 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const tx = txOpaque as Knex.Transaction; const stateRows = options.entities.map( - ({ entity, emitKey }) => + ({ entity, locationKey }) => ({ entity_id: uuid(), entity_ref: stringifyEntityRef(entity), unprocessed_entity: JSON.stringify(entity), errors: '', - emit_key: emitKey, + location_key: locationKey, next_update_at: tx.fn.now(), last_discovery_at: tx.fn.now(), } as Knex.DbRecord), @@ -414,7 +414,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ? JSON.parse(i.cache) : new Map(), errors: i.errors, - emitKey: i.emit_key, + locationKey: i.location_key, } as RefreshStateItem), ), }; diff --git a/plugins/catalog-backend/src/next/database/tables.ts b/plugins/catalog-backend/src/next/database/tables.ts index a298c37022..395694131d 100644 --- a/plugins/catalog-backend/src/next/database/tables.ts +++ b/plugins/catalog-backend/src/next/database/tables.ts @@ -29,7 +29,7 @@ export type DbRefreshStateRow = { next_update_at: string; last_discovery_at: string; // remove? errors?: string; - emit_key: string; + location_key?: string; }; export type DbRefreshStateReferencesRow = { diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index 5e4d0becd3..cd5d2a362d 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -33,7 +33,7 @@ export type UpdateProcessedEntityOptions = { errors?: string; relations: EntityRelationSpec[]; deferredEntities: DeferredEntity[]; - emitKey: string; + locationKey?: string; }; export type UpdateProcessedEntityErrorsOptions = { @@ -50,7 +50,7 @@ export type RefreshStateItem = { lastDiscoveryAt: string; // remove? state: Map; errors?: string; - emitKey: string; + locationKey?: string; }; export type GetProcessableEntitiesResult = { diff --git a/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts b/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts index fbb2744668..507def964a 100644 --- a/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts +++ b/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts @@ -101,14 +101,14 @@ export class ProcessorOutputCollector { }; } - this.deferredEntities.push({ entity, emitKey: location }); + this.deferredEntities.push({ entity, locationKey: location }); } else if (i.type === 'location') { const entity = locationSpecToLocationEntity( i.location, this.parentEntity, ); - const emitKey = getEntityLocationRef(entity); - this.deferredEntities.push({ entity, emitKey }); + const locationKey = getEntityLocationRef(entity); + this.deferredEntities.push({ entity, locationKey }); } else if (i.type === 'relation') { this.relations.push(i.relation); } else if (i.type === 'error') { diff --git a/plugins/catalog-backend/src/next/processing/types.ts b/plugins/catalog-backend/src/next/processing/types.ts index b2a627943b..9e30d07404 100644 --- a/plugins/catalog-backend/src/next/processing/types.ts +++ b/plugins/catalog-backend/src/next/processing/types.ts @@ -42,5 +42,5 @@ export interface CatalogProcessingOrchestrator { export type DeferredEntity = { entity: Entity; - emitKey: string; + locationKey?: string; }; From d0a13d227e9a87677ba312f2114176fabcaab539 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 28 Jun 2021 14:02:57 +0200 Subject: [PATCH 03/14] Catalog: Disallow insertion of conflicting entityRefs Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- .../database/DefaultProcessingDatabase.ts | 93 +++++++++++++------ 1 file changed, 67 insertions(+), 26 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index c2bfb4f633..8d3873a0ad 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -74,8 +74,14 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { location_key: locationKey, }) .where('entity_id', id) - .andWhere('location_key', locationKey) - .orWhere('location_key', ''); + .andWhere(inner => { + if (!locationKey) { + return inner.whereNull('location_key'); + } + return inner + .where('location_key', locationKey) + .orWhereNull('location_key'); + }); if (refreshResult === 0) { throw new NotFoundError(`Processing state not found for ${id}`); } @@ -319,34 +325,69 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ): Promise { const tx = txOpaque as Knex.Transaction; - const stateRows = options.entities.map( - ({ entity, locationKey }) => - ({ - entity_id: uuid(), - entity_ref: stringifyEntityRef(entity), - unprocessed_entity: JSON.stringify(entity), - errors: '', - location_key: locationKey, - next_update_at: tx.fn.now(), - last_discovery_at: tx.fn.now(), - } as Knex.DbRecord), - ); - const stateReferenceRows = stateRows.map( - stateRow => - ({ - source_entity_ref: options.entityRef, - target_entity_ref: stateRow.entity_ref, - } as Knex.DbRecord), - ); + const stateReferenceRows = new Array< + Knex.DbRecord + >(); // Upsert all of the unprocessed entities into the refresh_state table, by // their entity ref. // TODO(freben): Can this be batched somehow? - for (const row of stateRows) { - await tx('refresh_state') - .insert(row) - .onConflict('entity_ref') - .merge(['unprocessed_entity', 'last_discovery_at']); + for (const { entity, locationKey } of options.entities) { + const entityRef = stringifyEntityRef(entity); + const serializedEntity = JSON.stringify(entity); + + const refreshResult = await tx('refresh_state') + .update({ + unprocessed_entity: serializedEntity, + location_key: locationKey, + last_discovery_at: tx.fn.now(), + }) + .where('entity_ref', entityRef) + .andWhere(inner => { + if (!locationKey) { + return inner.whereNull('location_key'); + } + return inner + .where('location_key', locationKey) + .orWhereNull('location_key'); + }); + + if (refreshResult === 0) { + const result = await tx('refresh_state') + .insert({ + entity_id: uuid(), + entity_ref: entityRef, + unprocessed_entity: serializedEntity, + errors: '', + location_key: locationKey, + next_update_at: tx.fn.now(), + last_discovery_at: tx.fn.now(), + }) + .onConflict('entity_ref') + .ignore(); + + if (result.length === 0) { + // Detected conflicting entity with same entityRef but different locationKey + const [conflictingEntity] = await tx( + 'refresh_state', + ) + .where({ entity_ref: entityRef }) + .select(); + + if (conflictingEntity.location_key !== locationKey) { + this.options.logger.warn( + `Detected conflicting entityRef ${entityRef} already referenced by ${conflictingEntity.location_key} and now also ${locationKey}`, + ); + continue; + } + } + } + + // Skipped on locationKey conflict + stateReferenceRows.push({ + source_entity_ref: options.entityRef, + target_entity_ref: entityRef, + }); } // Replace all references for the originating entity before creating new ones From ba3061f53e111648662b901d951866b5677f785d Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 28 Jun 2021 14:04:30 +0200 Subject: [PATCH 04/14] Fix location key migrations Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- .../migrations/20210302150147_refresh_state.js | 2 -- .../20210622104022_refresh_state_location_key.js | 9 ++++++++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/plugins/catalog-backend/migrations/20210302150147_refresh_state.js b/plugins/catalog-backend/migrations/20210302150147_refresh_state.js index 2752ab0176..db66b46305 100644 --- a/plugins/catalog-backend/migrations/20210302150147_refresh_state.js +++ b/plugins/catalog-backend/migrations/20210302150147_refresh_state.js @@ -65,8 +65,6 @@ exports.up = async function up(knex) { .dateTime('last_discovery_at') // TODO: timezone or change to epoch-millis or similar .notNullable() .comment('The last timestamp of which this entity was discovered'); - // TODO: get migrations to work for this field instead. should be removed before merge. - table.text('location_key').nullable().comment('Location key'); table.unique(['entity_ref'], 'refresh_state_entity_ref_uniq'); table.index('entity_id', 'refresh_state_entity_id_idx'); table.index('entity_ref', 'refresh_state_entity_ref_idx'); diff --git a/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js b/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js index b77ff8961c..ca90bf9074 100644 --- a/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js +++ b/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js @@ -21,7 +21,13 @@ */ exports.up = async function up(knex) { await knex.schema.alterTable('refresh_state', table => { - table.text('location_key').nullable().comment('').alter(); + table + .text('location_key') + .nullable() + .comment( + 'An opaque key that uniquely identifies the location of an entity in order to support conflict resolution', + ); + // table.index(['location_key'], 'refresh_state_location_key_idx'); }); }; @@ -31,5 +37,6 @@ exports.up = async function up(knex) { exports.down = async function down(knex) { await knex.schema.alterTable('refresh_state', table => { table.dropColumn('location_key'); + // table.dropIndex([], 'refresh_state_location_key_idx'); }); }; From e4906366a3b257f6ffb686d55f0b4574c3b60765 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 28 Jun 2021 14:05:02 +0200 Subject: [PATCH 05/14] Catalog: Rename emitKey to locationkey Signed-off-by: Johan Haals --- .../catalog-backend/src/next/ConfigLocationEntityProvider.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts b/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts index 5e343b0f23..71ec84dae7 100644 --- a/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts +++ b/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts @@ -42,8 +42,8 @@ export class ConfigLocationEntityProvider implements EntityProvider { type, target: type === 'file' ? path.resolve(target) : target, }); - const emitKey = getEntityLocationRef(entity); - return { entity, emitKey }; + const locationKey = getEntityLocationRef(entity); + return { entity, locationKey }; }); await this.connection.applyMutation({ From 3aa0014263e03c55dbf17f3facac4b046a2d6f35 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Tue, 29 Jun 2021 09:18:20 +0200 Subject: [PATCH 06/14] Consolidate addUnprocessedEntities, compute delta using locationKey Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- .../database/DefaultProcessingDatabase.ts | 122 ++++++++++-------- .../src/next/database/types.ts | 13 +- 2 files changed, 78 insertions(+), 57 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 8d3873a0ad..f5da7850ea 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -89,7 +89,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { // Schedule all deferred entities for future processing. await this.addUnprocessedEntities(tx, { entities: deferredEntities, - entityRef: stringifyEntityRef(processedEntity), + sourceEntityRef: stringifyEntityRef(processedEntity), }); // Update fragments @@ -147,24 +147,43 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { }; } + // Grab all of the existing references from the same source, and their locationKeys as well const oldRefs = await tx( 'refresh_state_references', ) .where({ source_key: options.sourceKey }) - .select('target_entity_ref') - .then(rows => rows.map(r => r.target_entity_ref)); + .leftJoin('refresh_state', { + target_entity_ref: 'entity_ref', + }) + .select(['target_entity_ref', 'location_key']); - const items = options.items.map(entity => ({ - entity, - ref: stringifyEntityRef(entity.entity), + const items = options.items.map(deferred => ({ + deferred, + ref: stringifyEntityRef(deferred.entity), })); - const oldRefsSet = new Set(oldRefs); + const oldRefsSet = new Map( + oldRefs.map(r => [r.target_entity_ref, r.location_key]), + ); const newRefsSet = new Set(items.map(item => item.ref)); - const toAdd = items.filter(item => !oldRefsSet.has(item.ref)); - const toRemove = oldRefs.filter(ref => !newRefsSet.has(ref)); - return { toAdd: toAdd.map(({ entity }) => entity), toRemove }; + const toAdd = new Array(); + const toRemove = oldRefs + .map(row => row.target_entity_ref) + .filter(ref => !newRefsSet.has(ref)); + + for (const item of items) { + if (!oldRefsSet.has(item.ref)) { + // Add any entity that does not exist in the database + toAdd.push(item.deferred); + } else if (oldRefsSet.get(item.ref) !== item.deferred.locationKey) { + // Remove and then re-add any entity that exists, but with a different location key + toRemove.push(item.ref); + toAdd.push(item.deferred); + } + } + + return { toAdd, toRemove }; } async replaceUnprocessedEntities( @@ -291,31 +310,10 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { } if (toAdd.length) { - const state: Knex.DbRecord[] = toAdd.map( - ({ entity, locationKey }) => ({ - entity_id: uuid(), - entity_ref: stringifyEntityRef(entity), - unprocessed_entity: JSON.stringify(entity), - location_key: locationKey, - errors: '', - next_update_at: tx.fn.now(), - last_discovery_at: tx.fn.now(), - }), - ); - - const stateReferences: DbRefreshStateReferencesRow[] = toAdd.map( - entity => ({ - source_key: options.sourceKey, - target_entity_ref: stringifyEntityRef(entity.entity), - }), - ); - // TODO(freben): Concurrency? If we did these one by one, a .onConflict().merge would have made sense - await tx.batchInsert('refresh_state', state, BATCH_SIZE); - await tx.batchInsert( - 'refresh_state_references', - stateReferences, - BATCH_SIZE, - ); + await this.addUnprocessedEntities(tx, { + sourceKey: options.sourceKey, + entities: toAdd, + }); } } @@ -325,17 +323,17 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ): Promise { const tx = txOpaque as Knex.Transaction; - const stateReferenceRows = new Array< - Knex.DbRecord - >(); + // Keeps track of the entities that we end up inserting to update refresh_state_references afterwards + const stateReferences = new Array(); // Upsert all of the unprocessed entities into the refresh_state table, by // their entity ref. - // TODO(freben): Can this be batched somehow? for (const { entity, locationKey } of options.entities) { const entityRef = stringifyEntityRef(entity); const serializedEntity = JSON.stringify(entity); + // We optimistically try to update any existing refresh state first, as this is by far + // the most common case. const refreshResult = await tx('refresh_state') .update({ unprocessed_entity: serializedEntity, @@ -353,6 +351,7 @@ 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({ entity_id: uuid(), @@ -366,14 +365,17 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { .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) { - // Detected conflicting entity with same entityRef but different locationKey + // Check for conflicting entity with same entityRef but different locationKey const [conflictingEntity] = await tx( 'refresh_state', ) .where({ entity_ref: entityRef }) .select(); + // If the location key matches it means we just had a race trigger, which we can safely ignore if (conflictingEntity.location_key !== locationKey) { this.options.logger.warn( `Detected conflicting entityRef ${entityRef} already referenced by ${conflictingEntity.location_key} and now also ${locationKey}`, @@ -384,21 +386,35 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { } // Skipped on locationKey conflict - stateReferenceRows.push({ - source_entity_ref: options.entityRef, - target_entity_ref: entityRef, - }); + stateReferences.push(entityRef); } - // Replace all references for the originating entity before creating new ones - await tx('refresh_state_references') - .where({ source_entity_ref: options.entityRef }) - .delete(); - await tx.batchInsert( - 'refresh_state_references', - stateReferenceRows, - BATCH_SIZE, - ); + // 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 }) + .delete(); + await tx.batchInsert( + 'refresh_state_references', + stateReferences.map(entityRef => ({ + source_key: options.sourceKey, + target_entity_ref: entityRef, + })), + BATCH_SIZE, + ); + } else { + await tx('refresh_state_references') + .where({ source_entity_ref: options.sourceEntityRef }) + .delete(); + await tx.batchInsert( + 'refresh_state_references', + stateReferences.map(entityRef => ({ + source_entity_ref: options.sourceEntityRef, + target_entity_ref: entityRef, + })), + BATCH_SIZE, + ); + } } async getProcessableEntities( diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index cd5d2a362d..132b462751 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -19,10 +19,15 @@ import { JsonObject } from '@backstage/config'; import { Transaction } from '../../database/types'; import { DeferredEntity } from '../processing/types'; -export type AddUnprocessedEntitiesOptions = { - entityRef: string; - entities: DeferredEntity[]; -}; +export type AddUnprocessedEntitiesOptions = + | { + sourceEntityRef: string; + entities: DeferredEntity[]; + } + | { + sourceKey: string; + entities: DeferredEntity[]; + }; export type AddUnprocessedEntitiesResult = {}; From 03f1076fa60023df08950ecc7dba70342062d3cf Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Tue, 29 Jun 2021 11:30:36 +0200 Subject: [PATCH 07/14] Remove addUnprocessedEntities from public interface This is only used internally. Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- plugins/catalog-backend/src/next/database/types.ts | 5 ----- 1 file changed, 5 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index 132b462751..78199900e4 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -78,11 +78,6 @@ export type ReplaceUnprocessedEntitiesOptions = export interface ProcessingDatabase { transaction(fn: (tx: Transaction) => Promise): Promise; - addUnprocessedEntities( - tx: Transaction, - options: AddUnprocessedEntitiesOptions, - ): Promise; - replaceUnprocessedEntities( txOpaque: Transaction, options: ReplaceUnprocessedEntitiesOptions, From 72849d64954c109ca00cd13a0bfdd89832eaf764 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Tue, 29 Jun 2021 11:31:59 +0200 Subject: [PATCH 08/14] Add locationKey tests Signed-off-by: Johan Haals --- .../DefaultProcessingDatabase.test.ts | 215 +++++++++++++++--- 1 file changed, 185 insertions(+), 30 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index a0db3b293f..db9f209c22 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -57,6 +57,105 @@ describe('Default Processing Database', () => { await db('refresh_state').insert(ref); }; + describe('addUprocessedEntities', () => { + function mockEntity(name: string, type: string): Entity { + return { + apiVersion: '1', + kind: 'Component', + metadata: { + name, + }, + spec: { + type, + }, + }; + } + + it.each(databases.eachSupportedId())( + 'updates refresh state with varying location keys, %p', + async databaseId => { + const { db } = await createDatabase(databaseId); + await db.transaction(async tx => { + const knexTx = tx as Knex.Transaction; + + const steps = [ + { + locationKey: undefined, + expectedLocationKey: null, + type: 'a', + expectedType: 'a', + }, + { + locationKey: undefined, + expectedLocationKey: null, + type: 'b', + expectedType: 'b', + }, + { + locationKey: 'x', + expectedLocationKey: 'x', + type: 'c', + expectedType: 'c', + }, + { + locationKey: 'y', + expectedLocationKey: 'x', + type: 'd', + expectedType: 'c', + }, + { + locationKey: undefined, + expectedLocationKey: 'x', + type: 'e', + expectedType: 'c', + }, + { + locationKey: 'x', + expectedLocationKey: 'x', + type: 'f', + expectedType: 'f', + }, + ]; + for (const step of steps) { + await db.addUnprocessedEntities(tx, { + sourceKey: 'testing', + entities: [ + { + entity: mockEntity('1', step.type), + locationKey: step.locationKey, + }, + ], + }); + + const entities = await knexTx( + 'refresh_state', + ).select(); + expect(entities).toEqual([ + expect.objectContaining({ + entity_ref: 'component:default/1', + location_key: step.expectedLocationKey, + }), + ]); + const entity = JSON.parse(entities[0].unprocessed_entity) as Entity; + expect(entity.spec?.type).toEqual(step.expectedType); + + await expect( + knexTx( + 'refresh_state_references', + ).select(), + ).resolves.toEqual([ + expect.objectContaining({ + source_key: 'testing', + target_entity_ref: 'component:default/1', + }), + ]); + } + }); + }, + 60_000, + ); + }); + describe('updateProcessedEntity', () => { let id: string; let processedEntity: Entity; @@ -77,7 +176,7 @@ describe('Default Processing Database', () => { }); it.each(databases.eachSupportedId())( - 'fails when there is no processing state for the entity, %p', + 'fails when an entity is processed with a different locationKey, %p', async databaseId => { const { db } = await createDatabase(databaseId); await db.transaction(async tx => { @@ -95,6 +194,45 @@ describe('Default Processing Database', () => { 60_000, ); + it.each(databases.eachSupportedId())( + 'fails when the locationKey is different, %p', + async databaseId => { + const options = { + id, + processedEntity, + state: new Map(), + relations: [], + deferredEntities: [], + locationKey: 'key', + errors: "['something broke']", + }; + const { knex, db } = await createDatabase(databaseId); + await insertRefreshStateRow(knex, { + entity_id: id, + entity_ref: 'location:default/fakelocation', + unprocessed_entity: '{}', + processed_entity: '{}', + location_key: 'key', + errors: '[]', + next_update_at: '2021-04-01 13:37:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + await db.transaction(tx => db.updateProcessedEntity(tx, options)); + + const entities = await knex( + 'refresh_state', + ).select(); + expect(entities.length).toBe(1); + + await db.transaction(tx => + expect( + db.updateProcessedEntity(tx, { ...options, locationKey: 'fail' }), + ).rejects.toThrow(`Processing state not found for ${id}`), + ); + }, + 60_000, + ); + it.each(databases.eachSupportedId())( 'updates the refresh state entry with the cache, processed entity and errors, %p', async databaseId => { @@ -119,6 +257,7 @@ describe('Default Processing Database', () => { state, relations: [], deferredEntities: [], + locationKey: 'key', errors: "['something broke']", }), ); @@ -132,6 +271,7 @@ describe('Default Processing Database', () => { ); expect(entities[0].cache).toEqual(JSON.stringify(state)); expect(entities[0].errors).toEqual("['something broke']"); + expect(entities[0].location_key).toEqual('key'); }, 60_000, ); @@ -206,11 +346,14 @@ describe('Default Processing Database', () => { const deferredEntities = [ { - apiVersion: '1', - kind: 'Location', - metadata: { - name: 'next', + entity: { + apiVersion: '1', + kind: 'Location', + metadata: { + name: 'next', + }, }, + locationKey: 'mock', }, ]; @@ -227,7 +370,7 @@ describe('Default Processing Database', () => { const refreshStateEntries = await knex( 'refresh_state', ) - .where({ entity_ref: stringifyEntityRef(deferredEntities[0]) }) + .where({ entity_ref: stringifyEntityRef(deferredEntities[0].entity) }) .select(); expect(refreshStateEntries).toHaveLength(1); @@ -297,12 +440,15 @@ describe('Default Processing Database', () => { sourceKey: 'config', items: [ { - apiVersion: '1.0.0', - metadata: { - name: 'new-root', - }, - kind: 'Location', - } as Entity, + entity: { + apiVersion: '1.0.0', + metadata: { + name: 'new-root', + }, + kind: 'Location', + } as Entity, + locationKey: 'file:/tmp/foobar', + }, ], }), ); @@ -411,12 +557,15 @@ describe('Default Processing Database', () => { sourceKey: 'config', items: [ { - apiVersion: '1.0.0', - metadata: { - name: 'new-root', - }, - kind: 'Location', - } as Entity, + entity: { + apiVersion: '1.0.0', + metadata: { + name: 'new-root', + }, + kind: 'Location', + } as Entity, + locationKey: 'file:/tmp/foobar', + }, ], }); }); @@ -510,12 +659,15 @@ describe('Default Processing Database', () => { removed: [], added: [ { - apiVersion: '1.0.0', - metadata: { - name: 'new-root', - }, - kind: 'Location', - } as Entity, + entity: { + apiVersion: '1.0.0', + metadata: { + name: 'new-root', + }, + kind: 'Location', + } as Entity, + locationKey: 'file:/tmp/foobar', + }, ], }); }); @@ -614,12 +766,15 @@ describe('Default Processing Database', () => { added: [], removed: [ { - apiVersion: '1.0.0', - metadata: { - name: 'new-root', - }, - kind: 'Location', - } as Entity, + entity: { + apiVersion: '1.0.0', + metadata: { + name: 'new-root', + }, + kind: 'Location', + } as Entity, + locationKey: 'file:/tmp/foobar', + }, ], }); }); From 057e78ebb1c212c238f71b1e03054bf3b8add795 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 29 Jun 2021 14:07:19 +0200 Subject: [PATCH 09/14] catalog-backend: more tests for DB + fix ref handling during conflicts Co-authored-by: blam Co-authored-by: Johan Haals Signed-off-by: Patrik Oldsberg --- .../DefaultProcessingDatabase.test.ts | 99 ++++++++++++++++++- .../database/DefaultProcessingDatabase.ts | 25 ++--- 2 files changed, 107 insertions(+), 17 deletions(-) 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', From 698e823bbe171ac68b9b27c2d79ea9a03ee49d16 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 29 Jun 2021 16:58:09 +0200 Subject: [PATCH 10/14] catalog-backend: fix unprocessed entity conflict handling when using postgres Co-authored-by: blam Co-authored-by: Johan Haals Signed-off-by: Patrik Oldsberg --- .../database/DefaultProcessingDatabase.ts | 23 ++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 67b83bdb34..d18f194a57 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -354,7 +354,7 @@ 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 try { - await tx('refresh_state').insert({ + let query = tx('refresh_state').insert({ entity_id: uuid(), entity_ref: entityRef, unprocessed_entity: serializedEntity, @@ -363,7 +363,28 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { next_update_at: tx.fn.now(), last_discovery_at: tx.fn.now(), }); + + // TODO(Rugvip): only tested towards Postgres and SQLite + // We have to do this because the only way to detect if there was a conflict with + // SQLite is to catch the error, while Postgres needs to ignore the conflict to not + // break the ongoing transaction. + if (tx.client.config.client !== 'sqlite3') { + query = query.onConflict('entity_ref').ignore(); + } + + const result = await query; + if (result.rowCount === 0) { + throw new ConflictError( + 'Insert failed due to conflicting entity_ref', + ); + } } catch (error) { + if ( + !error.message.contains('UNIQUE constraint failed') && + error.name !== 'ConflictError' + ) { + throw 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 From b74666f46c5db08942f07ce3fb3c6f13999bd703 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 29 Jun 2021 17:46:39 +0200 Subject: [PATCH 11/14] catalog-backend: couple of cleanups and fixes Signed-off-by: Patrik Oldsberg --- .../20210622104022_refresh_state_location_key.js | 2 -- .../src/next/DefaultLocationService.test.ts | 11 +++++++---- .../src/next/DefaultLocationService.ts | 1 - 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js b/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js index ca90bf9074..37ab5e12cb 100644 --- a/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js +++ b/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js @@ -27,7 +27,6 @@ exports.up = async function up(knex) { .comment( 'An opaque key that uniquely identifies the location of an entity in order to support conflict resolution', ); - // table.index(['location_key'], 'refresh_state_location_key_idx'); }); }; @@ -37,6 +36,5 @@ exports.up = async function up(knex) { exports.down = async function down(knex) { await knex.schema.alterTable('refresh_state', table => { table.dropColumn('location_key'); - // table.dropIndex([], 'refresh_state_location_key_idx'); }); }; diff --git a/plugins/catalog-backend/src/next/DefaultLocationService.test.ts b/plugins/catalog-backend/src/next/DefaultLocationService.test.ts index ae2c4e9a20..f965ec3077 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationService.test.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationService.test.ts @@ -48,11 +48,14 @@ describe('DefaultLocationServiceTest', () => { }, deferredEntities: [ { - apiVersion: 'backstage.io/v1alpha1', - kind: 'Component', - metadata: { - name: 'bar', + entity: { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { + name: 'bar', + }, }, + locationKey: 'file:///tmp/mock.yaml', }, ], relations: [], diff --git a/plugins/catalog-backend/src/next/DefaultLocationService.ts b/plugins/catalog-backend/src/next/DefaultLocationService.ts index 70679a3bd6..2820ba14b1 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationService.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationService.ts @@ -92,7 +92,6 @@ export class DefaultLocationService implements LocationService { }); if (processed.ok) { - // todo unprocessedEntities.push(...processed.deferredEntities); entities.push(processed.completedEntity); } else { From 45af985df09a2972449f5014351cb8fb57bc7b9d Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 29 Jun 2021 20:24:58 +0200 Subject: [PATCH 12/14] changesets: add changeset for the location key changes Signed-off-by: Patrik Oldsberg --- .changeset/rude-windows-live.md | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) create mode 100644 .changeset/rude-windows-live.md diff --git a/.changeset/rude-windows-live.md b/.changeset/rude-windows-live.md new file mode 100644 index 0000000000..b8f0e0e158 --- /dev/null +++ b/.changeset/rude-windows-live.md @@ -0,0 +1,17 @@ +--- +'@backstage/plugin-catalog-backend': minor +--- + +Handle entity name conflicts in a deterministic way and avoid crashes due to naming conflicts at startup. + +This is a breaking change for the database and entity provider interfaces of the new catalog. The interfaces with breaking changes are `EntityProvider` and `ProcessingDatabase`, and while it's unlikely that these interfaces have much usage yet, a migration guide is provided below. + +The breaking change to the `EntityProvider` interface lies within the items passed in the `EntityProviderMutation` type. Rather than passing along entities directly, they are now wrapped up in a `DeferredEntity` type, which is a tuple of an `entity` and a `locationKey`. The `entity` houses the entity as it was passed on before, while the `locationKey` is a new concept that is used for conflict resolution within the catalog. + +The `locationKey` is an opaque string that should be unique for each location that an entity could be located at, and undefined if the entity does not have a fixed location. In practice it should be set to the serialized location reference if the entity is stored in Git, for example `https://github.com/backstage/backstage/blob/master/catalog-info.yaml`. A conflict between two entity definitions happen when they have the same entity reference, i.e. kind, namespace, and name. In the event of a conflict the location key will be used according to the following rules to resolve the conflict: + +- If the entity is already present in the database but does not have a location key set, the new entity wins and will override the existing one. +- If the entity is already present in the database the new entity will only win if the location keys of the existing and new entity are the same. +- If the entity is not already present, insert the entity into the database along with the provided location key. + +The breaking change to the `ProcessingDatabase` is similar to the one for the entity provider, as it reflects the switch from `Entity` to `DeferredEntity` in the `ReplaceUnprocessedEntitiesOptions`. In addition, the `addUnprocessedEntities` method has been removed from the `ProcessingDatabase` interface, and the `RefreshStateItem` and `UpdateProcessedEntityOptions` types have received a new optional `locationKey` property. From 27e90f969d990013f0615ac90113aea35b8060d3 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Wed, 30 Jun 2021 10:53:01 +0200 Subject: [PATCH 13/14] catalob-backend: more test fixes Signed-off-by: Patrik Oldsberg --- .../next/ConfigLocationEntityProvider.test.ts | 38 +++++++++++------- .../src/next/DefaultLocationStore.test.ts | 40 +++++++++++-------- .../database/DefaultProcessingDatabase.ts | 2 +- 3 files changed, 48 insertions(+), 32 deletions(-) diff --git a/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.test.ts b/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.test.ts index 37d549c7f2..7592182bad 100644 --- a/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.test.ts +++ b/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.test.ts @@ -41,26 +41,34 @@ describe('ConfigLocationEntityProvider', () => { expect(mockConnection.applyMutation).toHaveBeenCalledWith({ type: 'full', entities: expect.arrayContaining([ - expect.objectContaining({ - spec: { - target: path.join( - resolvePackagePath('@backstage/plugin-catalog-backend'), - './lols.yaml', - ), - type: 'file', - }, - }), + { + entity: expect.objectContaining({ + spec: { + target: path.join( + resolvePackagePath('@backstage/plugin-catalog-backend'), + './lols.yaml', + ), + type: 'file', + }, + }), + locationKey: expect.stringMatching( + /plugins\/catalog-backend\/lols\.yaml$/, + ), + }, ]), }); expect(mockConnection.applyMutation).toHaveBeenCalledWith({ type: 'full', entities: expect.arrayContaining([ - expect.objectContaining({ - spec: { - target: 'https://github.com/backstage/backstage', - type: 'url', - }, - }), + { + entity: expect.objectContaining({ + spec: { + target: 'https://github.com/backstage/backstage', + type: 'url', + }, + }), + locationKey: 'url:https://github.com/backstage/backstage', + }, ]), }); }); diff --git a/plugins/catalog-backend/src/next/DefaultLocationStore.test.ts b/plugins/catalog-backend/src/next/DefaultLocationStore.test.ts index 762a585754..82b08f2e70 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationStore.test.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationStore.test.ts @@ -113,13 +113,17 @@ describe('DefaultLocationStore', () => { type: 'delta', removed: [], added: expect.arrayContaining([ - expect.objectContaining({ - spec: { - target: - 'https://github.com/backstage/demo/blob/master/catalog-info.yml', - type: 'url', - }, - }), + { + entity: expect.objectContaining({ + spec: { + target: + 'https://github.com/backstage/demo/blob/master/catalog-info.yml', + type: 'url', + }, + }), + locationKey: + 'url:https://github.com/backstage/demo/blob/master/catalog-info.yml', + }, ]), }); }, @@ -156,15 +160,19 @@ describe('DefaultLocationStore', () => { expect(connection.applyMutation).toHaveBeenCalledWith({ type: 'delta', added: [], - removed: expect.arrayContaining([ - expect.objectContaining({ - spec: { - target: - 'https://github.com/backstage/demo/blob/master/catalog-info.yml', - type: 'url', - }, - }), - ]), + removed: [ + { + entity: expect.objectContaining({ + spec: { + target: + 'https://github.com/backstage/demo/blob/master/catalog-info.yml', + type: 'url', + }, + }), + locationKey: + 'url:https://github.com/backstage/demo/blob/master/catalog-info.yml', + }, + ], }); }, 60_000, diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index d18f194a57..53025682ac 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -380,7 +380,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { } } catch (error) { if ( - !error.message.contains('UNIQUE constraint failed') && + !error.message.includes('UNIQUE constraint failed') && error.name !== 'ConflictError' ) { throw error; From 3cf22d0a92542dfa81526cdb0482d271201cb359 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Wed, 30 Jun 2021 14:29:31 +0200 Subject: [PATCH 14/14] catalog-backend: review fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Johan Haals Co-authored-by: Fredrik Adelöw Signed-off-by: Patrik Oldsberg --- .../next/database/DefaultProcessingDatabase.test.ts | 8 ++++++-- .../src/next/database/DefaultProcessingDatabase.ts | 13 +++++++++---- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 9a9c09daaf..e449e5580c 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -210,7 +210,9 @@ describe('Default Processing Database', () => { relations: [], deferredEntities: [], }), - ).rejects.toThrow(`Processing state not found for ${id}`); + ).rejects.toThrow( + `Conflicting write of processing result for ${id} with location key 'undefined'`, + ); }); }, 60_000, @@ -249,7 +251,9 @@ describe('Default Processing Database', () => { await db.transaction(tx => expect( db.updateProcessedEntity(tx, { ...options, locationKey: 'fail' }), - ).rejects.toThrow(`Processing state not found for ${id}`), + ).rejects.toThrow( + `Conflicting write of processing result for ${id} with location key 'fail'`, + ), ); }, 60_000, diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 53025682ac..1f1ea4bf84 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -16,7 +16,7 @@ import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; -import { ConflictError, NotFoundError } from '@backstage/errors'; +import { ConflictError } from '@backstage/errors'; import { Knex } from 'knex'; import lodash from 'lodash'; import { v4 as uuid } from 'uuid'; @@ -83,7 +83,9 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { .orWhereNull('location_key'); }); if (refreshResult === 0) { - throw new NotFoundError(`Processing state not found for ${id}`); + throw new ConflictError( + `Conflicting write of processing result for ${id} with location key '${locationKey}'`, + ); } // Schedule all deferred entities for future processing. @@ -372,7 +374,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { query = query.onConflict('entity_ref').ignore(); } - const result = await query; + const result: { /* postgres */ rowCount?: number } = await query; if (result.rowCount === 0) { throw new ConflictError( 'Insert failed due to conflicting entity_ref', @@ -395,7 +397,10 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { .select(); // If the location key matches it means we just had a race trigger, which we can safely ignore - if (conflictingEntity.location_key !== locationKey) { + if ( + !conflictingEntity || + conflictingEntity.location_key !== locationKey + ) { this.options.logger.warn( `Detected conflicting entityRef ${entityRef} already referenced by ${conflictingEntity.location_key} and now also ${locationKey}`, );