From ed2a9fe492a2d0939641660bdeb174d524086fa1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 17:19:35 +0200 Subject: [PATCH] Replace placeholder final_entities insert with upsert MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Now that immediate mode is removed, the two-step pattern of inserting a placeholder row and then updating it is unnecessary. Replace with a single upsert that creates or updates the final_entities row in one operation, removing a round-trip and simplifying the flow. Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../stitcher/performStitching.test.ts | 64 --------------- .../operations/stitcher/performStitching.ts | 79 ++++++------------- 2 files changed, 26 insertions(+), 117 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index 7c1ab0fd01..44c8a74cc4 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -317,70 +317,6 @@ describe.each(databases.eachSupportedId())( databaseId => { const logger = mockServices.logger.mock(); - it('handles conflicts with past stitches', async () => { - if (databaseId === 'MYSQL_8') { - // MySQL doesn't handle conflicts in the same way as the other two, most - // likely due to the conflict probably being handled with a merged even - // if it's for the entity_ref. This probably means that MySQL will have - // inconsistencies in the final_entities table in some cases, but they - // should heal fairly quickly. - return; - } - const knex = await databases.init(databaseId); - await applyDatabaseMigrations(knex); - - // Drop the foreign key constraint so we can insert a conflicting entity - // where the ID is inconsistent with the entity ref across refresh_state - // and final_entities - await knex.schema.alterTable('final_entities', table => { - table.dropForeign('entity_id'); - }); - - await knex('refresh_state').insert([ - { - entity_id: 'other-id', - entity_ref: 'k:ns/n', - unprocessed_entity: JSON.stringify({}), - processed_entity: JSON.stringify({ - apiVersion: 'a', - kind: 'k', - metadata: { - name: 'n', - namespace: 'ns', - }, - spec: { - k: 'v', - }, - }), - errors: '[]', - next_update_at: knex.fn.now(), - last_discovery_at: knex.fn.now(), - }, - ]); - await knex('final_entities').insert([ - { - entity_id: 'my-id', - entity_ref: 'k:ns/n', - hash: '', - final_entity: JSON.stringify({}), - }, - ]); - - const stitchLogger = mockServices.logger.mock(); - await expect( - performStitching({ - knex, - logger: stitchLogger, - entityRef: 'k:ns/n', - }), - ).resolves.toBe('abandoned'); - - expect(stitchLogger.debug).toHaveBeenCalledWith( - 'Skipping stitching of k:ns/n, conflict', - expect.anything(), - ); - }); - it('stitches when final_entities row already exists', async () => { const knex = await databases.init(databaseId); await applyDatabaseMigrations(knex); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index e96eab811c..17cec3cf84 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -34,10 +34,7 @@ import { import { buildEntitySearch } from './buildEntitySearch'; import { markDeferredStitchCompleted } from './markDeferredStitchCompleted'; import { syncSearchRows } from './syncSearchRows'; -import { - LoggerService, - isDatabaseConflictError, -} from '@backstage/backend-plugin-api'; +import { LoggerService } from '@backstage/backend-plugin-api'; function generateStableHash(entity: Entity) { return createHash('sha1') @@ -79,34 +76,10 @@ export async function performStitching(options: { return 'abandoned'; } - // Ensure that a final_entities row exists for this entity. - try { - await knex('final_entities') - .insert({ - entity_id: entityResult[0].entity_id, - hash: '', - entity_ref: entityRef, - }) - .onConflict('entity_id') - .ignore(); - } catch (error) { - // It's possible to hit a race where a refresh_state table delete + insert - // is done just after we read the entity_id from it. This conflict is safe - // to ignore because the current stitching operation will be triggered by - // the old entry, and the new entry will trigger it's own stitching that - // will update the entity. - if (isDatabaseConflictError(error)) { - logger.debug(`Skipping stitching of ${entityRef}, conflict`, error); - return 'abandoned'; - } - - throw error; - } - - // Selecting from refresh_state and final_entities should yield exactly - // one row (except in abnormal cases where the stitch was invoked for - // something that didn't exist at all, in which case it's zero rows). - // The join with the temporary incoming_references still gives one row. + // Selecting from refresh_state (with an optional left join to + // final_entities for the previous hash) should yield exactly one row, + // except in abnormal cases where the entity was deleted between the + // stitch request and now. const [processedResult, relationsResult] = await Promise.all([ knex .with('incoming_references', function incomingReferences(builder) { @@ -236,31 +209,31 @@ export async function performStitching(options: { // to write the search index. const searchEntries = buildEntitySearch(entityId, entity); - let updateQuery = knex('final_entities') - .update({ + // Guard against concurrent stitchers by checking that the stitch_ticket + // in stitch_queue still matches what we were given. + if (stitchTicket) { + const ticketValid = await knex('stitch_queue') + .where('entity_ref', entityRef) + .where('stitch_ticket', stitchTicket) + .first(); + if (!ticketValid) { + logger.debug( + `Entity ${entityRef} is already stitched, skipping write.`, + ); + return 'abandoned'; + } + } + + await knex('final_entities') + .insert({ + entity_id: entityId, + entity_ref: entityRef, final_entity: JSON.stringify(entity), hash, last_updated_at: knex.fn.now(), }) - .where('entity_id', entityId); - - // Guard against concurrent stitchers by checking that the stitch_ticket - // in stitch_queue still matches what we were given. - if (stitchTicket) { - updateQuery = updateQuery.whereExists( - knex('stitch_queue') - .where('stitch_queue.entity_ref', entityRef) - .where('stitch_queue.stitch_ticket', stitchTicket) - .select(knex.raw('1')), - ); - } - - const amountOfRowsChanged = await updateQuery; - - if (amountOfRowsChanged === 0) { - logger.debug(`Entity ${entityRef} is already stitched, skipping write.`); - return 'abandoned'; - } + .onConflict('entity_id') + .merge(['final_entity', 'hash', 'last_updated_at']); await syncSearchRows(knex, entityId, searchEntries);