From bf59528fb902bc66ad00fb314b6cf54cc2e449c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 5 Mar 2026 23:03:38 +0100 Subject: [PATCH] catalog: remove stitch_ticket from final_entities, use two-ticket model in stitch_queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the single stitch_ticket column in final_entities with a two-ticket model (latest_ticket + active_ticket) in stitch_queue for optimistic concurrency control. When picking up stitch work, active_ticket is set to latest_ticket. Before writing results, the stitcher checks that latest_ticket hasn't changed. This fully decouples stitching concurrency from final_entities. Co-Authored-By: Claude Opus 4.6 Signed-off-by: Fredrik Adelöw --- .changeset/gentle-tables-march.md | 2 +- .../20260215000000_move_stitch_queue.js | 42 +++++++-- plugins/catalog-backend/report.sql.md | 6 +- .../getDeferredStitchableEntities.test.ts | 6 +- .../stitcher/getDeferredStitchableEntities.ts | 7 +- .../markDeferredStitchCompleted.test.ts | 6 +- .../stitcher/markDeferredStitchCompleted.ts | 2 +- .../stitcher/markForStitching.test.ts | 36 ++++---- .../operations/stitcher/markForStitching.ts | 8 +- .../stitcher/performStitching.test.ts | 92 ++++++++++--------- .../operations/stitcher/performStitching.ts | 29 ++++-- .../util/deleteOrphanedEntities.test.ts | 1 - .../catalog-backend/src/database/tables.ts | 18 ++-- .../providers/DefaultLocationStore.test.ts | 2 +- .../GenericScmEventRefreshProvider.test.ts | 2 +- .../service/DefaultEntitiesCatalog.test.ts | 2 - .../request/applyEntityFilterToQuery.test.ts | 1 - .../applyPredicateEntityFilterToQuery.test.ts | 1 - .../src/tests/migrations.test.ts | 33 ++++++- 19 files changed, 181 insertions(+), 115 deletions(-) diff --git a/.changeset/gentle-tables-march.md b/.changeset/gentle-tables-march.md index 0aa13cfb4e..91f85c7117 100644 --- a/.changeset/gentle-tables-march.md +++ b/.changeset/gentle-tables-march.md @@ -2,4 +2,4 @@ '@backstage/plugin-catalog-backend': patch --- -Moved stitch queue columns (`next_stitch_at`, `next_stitch_ticket`) from `refresh_state` into a dedicated `stitch_queue` table with `entity_ref` as the primary key. When a stitch completes successfully, the corresponding row is deleted from the queue. The migration handles existing data and is fully reversible. +Moved stitch queue concerns out of `refresh_state` and `final_entities` into a dedicated `stitch_queue` table with `entity_ref` as the primary key. The stitch queue uses a two-ticket model (`latest_ticket` and `active_ticket`) for optimistic concurrency control. When a stitch completes successfully and the ticket hasn't changed, the corresponding row is deleted from the queue. The migration handles existing data and is fully reversible. diff --git a/plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js b/plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js index da998b6031..92becfe257 100644 --- a/plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js +++ b/plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js @@ -34,9 +34,13 @@ exports.up = async function up(knex) { .notNullable() .comment('The entity ref that needs stitching'); table - .string('stitch_ticket') + .string('latest_ticket') .notNullable() - .comment('Random value distinguishing stitch requests'); + .comment('Changes with every new stitch request'); + table + .string('active_ticket') + .nullable() + .comment('Set when a stitcher picks up this item for processing'); table .dateTime('next_stitch_at') .notNullable() @@ -59,7 +63,7 @@ exports.up = async function up(knex) { await knex('stitch_queue').insert( batch.map(row => ({ entity_ref: row.entity_ref, - stitch_ticket: row.next_stitch_ticket || '', + latest_ticket: row.next_stitch_ticket || '', next_stitch_at: row.next_stitch_at, })), ); @@ -77,6 +81,15 @@ exports.up = async function up(knex) { table.dropColumn('next_stitch_ticket'); }); } + + // Step 4: Remove stitch_ticket from final_entities (now managed via stitch_queue) + if (isSQLite) { + await knex.raw('ALTER TABLE final_entities DROP COLUMN stitch_ticket'); + } else { + await knex.schema.alterTable('final_entities', table => { + table.dropColumn('stitch_ticket'); + }); + } }; /** @@ -85,7 +98,16 @@ exports.up = async function up(knex) { exports.down = async function down(knex) { const isSQLite = knex.client.config.client.includes('sqlite'); - // Step 1: Add back the columns to refresh_state + // Step 1: Add back stitch_ticket to final_entities + await knex.schema.alterTable('final_entities', table => { + table + .text('stitch_ticket') + .notNullable() + .defaultTo('') + .comment('Optimistic concurrency ticket for stitching'); + }); + + // Step 2: Add back the columns to refresh_state await knex.schema.alterTable('refresh_state', table => { table .dateTime('next_stitch_at') @@ -97,7 +119,7 @@ exports.down = async function down(knex) { .comment('Random value distinguishing stitch requests'); }); - // Step 1b: Create partial index separately + // Step 2b: Create partial index separately if (isSQLite) { await knex.raw(` CREATE INDEX refresh_state_next_stitch_at_idx @@ -112,13 +134,13 @@ exports.down = async function down(knex) { }); } - // Step 2: Migrate stitch requests back from stitch_queue to refresh_state + // Step 3: Migrate stitch requests back from stitch_queue to refresh_state if (isSQLite) { const pendingStitches = await knex .select({ entityRef: 'stitch_queue.entity_ref', nextStitchAt: 'stitch_queue.next_stitch_at', - stitchTicket: 'stitch_queue.stitch_ticket', + latestTicket: 'stitch_queue.latest_ticket', }) .from('stitch_queue'); @@ -126,7 +148,7 @@ exports.down = async function down(knex) { await knex('refresh_state') .update({ next_stitch_at: row.nextStitchAt, - next_stitch_ticket: row.stitchTicket, + next_stitch_ticket: row.latestTicket, }) .where('entity_ref', row.entityRef); } @@ -138,13 +160,13 @@ exports.down = async function down(knex) { .from('stitch_queue') .whereRaw('stitch_queue.entity_ref = refresh_state.entity_ref'), next_stitch_ticket: knex - .select('stitch_queue.stitch_ticket') + .select('stitch_queue.latest_ticket') .from('stitch_queue') .whereRaw('stitch_queue.entity_ref = refresh_state.entity_ref'), }) .whereIn('entity_ref', knex.select('entity_ref').from('stitch_queue')); } - // Step 3: Drop the stitch_queue table + // Step 4: Drop the stitch_queue table await knex.schema.dropTable('stitch_queue'); }; diff --git a/plugins/catalog-backend/report.sql.md b/plugins/catalog-backend/report.sql.md index b06ef0a79b..bd63a6ee0c 100644 --- a/plugins/catalog-backend/report.sql.md +++ b/plugins/catalog-backend/report.sql.md @@ -3,7 +3,7 @@ > Do not edit this file. It is a report generated by `yarn build:api-reports` > [!WARNING] -> Failed to migrate down from '20241003170511_alter_target_in_locations.js' +> Failed to migrate down from '20260215000000_move_stitch_queue.js' ## Sequences @@ -19,7 +19,6 @@ | `final_entity` | `text` | true | - | - | | `hash` | `character varying` | false | 255 | - | | `last_updated_at` | `timestamp with time zone` | true | - | - | -| `stitch_ticket` | `text` | false | - | - | ### Indices @@ -137,9 +136,10 @@ | Column | Type | Nullable | Max Length | Default | | ---------------- | -------------------------- | -------- | ---------- | ------- | +| `active_ticket` | `character varying` | true | 255 | - | | `entity_ref` | `character varying` | false | 255 | - | +| `latest_ticket` | `character varying` | false | 255 | - | | `next_stitch_at` | `timestamp with time zone` | false | - | - | -| `stitch_ticket` | `character varying` | false | 255 | - | ### Indices diff --git a/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts index 1f6137f91e..ac3cfc0cfb 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts @@ -35,17 +35,17 @@ describe('getDeferredStitchableEntities', () => { await knex('stitch_queue').insert([ { entity_ref: 'k:ns/future_stitch_time', - stitch_ticket: 't1', + latest_ticket: 't1', next_stitch_at: '2037-01-01T00:00:00.000', }, { entity_ref: 'k:ns/past_stitch_time', - stitch_ticket: 't3', + latest_ticket: 't3', next_stitch_at: '1971-01-01T00:00:00.000', }, { entity_ref: 'k:ns/past_stitch_time_again', - stitch_ticket: 't4', + latest_ticket: 't4', next_stitch_at: '1972-01-01T00:00:00.000', }, ]); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts index 2d12ee4f6a..4adce0fc0d 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts @@ -31,7 +31,7 @@ import { DbStitchQueueRow } from '../../tables'; * * This assumes that the stitching strategy is set to deferred. * - * They are expected to already have the stitch_ticket set (by + * They are expected to already have the latest_ticket set (by * markForStitching) so that their tickets can be returned with each item. * * All returned items have their next_stitch_at updated to be moved forward by @@ -55,7 +55,7 @@ export async function getDeferredStitchableEntities(options: { let itemsQuery = knex('stitch_queue').select( 'entity_ref', 'next_stitch_at', - 'stitch_ticket', + 'latest_ticket', ); // This avoids duplication of work because of race conditions and is @@ -81,11 +81,12 @@ export async function getDeferredStitchableEntities(options: { ) .update({ next_stitch_at: nowPlus(knex, stitchTimeout), + active_ticket: knex.ref('latest_ticket'), }); return items.map(i => ({ entityRef: i.entity_ref, - stitchTicket: i.stitch_ticket, + stitchTicket: i.latest_ticket, stitchRequestedAt: timestampToDateTime(i.next_stitch_at), })); } diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts index dbcbba13df..325740394a 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts @@ -34,7 +34,7 @@ describe('markDeferredStitchCompleted', () => { await knex('stitch_queue').insert([ { entity_ref: 'k:ns/n', - stitch_ticket: 'the-ticket', + latest_ticket: 'the-ticket', next_stitch_at: '1971-01-01T00:00:00.000', }, ]); @@ -43,7 +43,7 @@ describe('markDeferredStitchCompleted', () => { return knex('stitch_queue').select( 'entity_ref', 'next_stitch_at', - 'stitch_ticket', + 'latest_ticket', ); } @@ -57,7 +57,7 @@ describe('markDeferredStitchCompleted', () => { { entity_ref: 'k:ns/n', next_stitch_at: expect.anything(), - stitch_ticket: 'the-ticket', + latest_ticket: 'the-ticket', }, ]); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts index bb3e2be6f9..2d80a62496 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts @@ -38,6 +38,6 @@ export async function markDeferredStitchCompleted(option: { await knex('stitch_queue') .where('entity_ref', '=', entityRef) - .andWhere('stitch_ticket', '=', stitchTicket) + .andWhere('latest_ticket', '=', stitchTicket) .delete(); } diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts index 2b43f3f281..88a0ab027a 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts @@ -76,14 +76,14 @@ describe('markForStitching', () => { await knex('stitch_queue').insert([ { entity_ref: 'k:ns/four', - stitch_ticket: 'old', + latest_ticket: 'old', next_stitch_at: '1971-01-01T00:00:00.000', }, ]); async function result() { return knex('stitch_queue') - .select('entity_ref', 'next_stitch_at', 'stitch_ticket') + .select('entity_ref', 'next_stitch_at', 'latest_ticket') .orderBy('entity_ref', 'asc'); } @@ -93,7 +93,7 @@ describe('markForStitching', () => { { entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - stitch_ticket: 'old', + latest_ticket: 'old', }, ]); @@ -111,7 +111,7 @@ describe('markForStitching', () => { { entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - stitch_ticket: 'old', + latest_ticket: 'old', }, ]); @@ -129,12 +129,12 @@ describe('markForStitching', () => { { entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - stitch_ticket: 'old', + latest_ticket: 'old', }, { entity_ref: 'k:ns/one', next_stitch_at: expect.anything(), - stitch_ticket: expect.anything(), + latest_ticket: expect.anything(), }, ]); @@ -152,17 +152,17 @@ describe('markForStitching', () => { { entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - stitch_ticket: 'old', + latest_ticket: 'old', }, { entity_ref: 'k:ns/one', next_stitch_at: expect.anything(), - stitch_ticket: expect.anything(), + latest_ticket: expect.anything(), }, { entity_ref: 'k:ns/two', next_stitch_at: expect.anything(), - stitch_ticket: expect.anything(), + latest_ticket: expect.anything(), }, ]); @@ -180,29 +180,29 @@ describe('markForStitching', () => { { entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - stitch_ticket: expect.anything(), + latest_ticket: expect.anything(), }, { entity_ref: 'k:ns/one', next_stitch_at: expect.anything(), - stitch_ticket: expect.anything(), + latest_ticket: expect.anything(), }, { entity_ref: 'k:ns/three', next_stitch_at: expect.anything(), - stitch_ticket: expect.anything(), + latest_ticket: expect.anything(), }, { entity_ref: 'k:ns/two', next_stitch_at: expect.anything(), - stitch_ticket: expect.anything(), + latest_ticket: expect.anything(), }, ]); // Entity 4's ticket should have been updated (was 'old', now something else) const final = await result(); const entity4Final = final.find(r => r.entity_ref === 'k:ns/four'); - expect(entity4Final?.stitch_ticket).not.toEqual('old'); + expect(entity4Final?.latest_ticket).not.toEqual('old'); }, ); @@ -260,28 +260,24 @@ describe('markForStitching', () => { final_entity: '{}', entity_ref: 'k:ns/one', hash: 'old', - stitch_ticket: 'old', }, { entity_id: '2', final_entity: '{}', entity_ref: 'k:ns/two', hash: 'old', - stitch_ticket: 'old', }, { entity_id: '3', final_entity: '{}', entity_ref: 'k:ns/three', hash: 'old', - stitch_ticket: 'old', }, { entity_id: '4', final_entity: '{}', entity_ref: 'k:ns/four', hash: 'old', - stitch_ticket: 'old', }, ]); @@ -563,13 +559,13 @@ describe('markForStitching', () => { // Verify final state - all entities should have been marked for stitching const finalState = await knex('stitch_queue') - .select('entity_ref', 'next_stitch_at', 'stitch_ticket') + .select('entity_ref', 'next_stitch_at', 'latest_ticket') .orderBy('entity_ref'); expect(finalState.length).toBeGreaterThan(0); finalState.forEach(row => { expect(row.next_stitch_at).not.toBeNull(); - expect(row.stitch_ticket).not.toBeNull(); + expect(row.latest_ticket).not.toBeNull(); }); }, ); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts index 9e22a65e7b..7c67ada1a2 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts @@ -92,12 +92,12 @@ export async function markForStitching(options: { .insert( chunk.map(ref => ({ entity_ref: ref, - stitch_ticket: ticket, + latest_ticket: ticket, next_stitch_at: knex.fn.now(), })), ) .onConflict('entity_ref') - .merge(['next_stitch_at', 'stitch_ticket']); + .merge(['next_stitch_at', 'latest_ticket']); } }, knex); } @@ -114,12 +114,12 @@ export async function markForStitching(options: { .insert( refreshStateRows.map(row => ({ entity_ref: row.entity_ref, - stitch_ticket: ticket, + latest_ticket: ticket, next_stitch_at: knex.fn.now(), })), ) .onConflict('entity_ref') - .merge(['next_stitch_at', 'stitch_ticket']); + .merge(['next_stitch_at', 'latest_ticket']); } }, knex); } diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index 985146229a..15fc8b71d2 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -24,6 +24,7 @@ import { DbRelationsRow, DbSearchRow, } from '../../tables'; +import { markForStitching } from './markForStitching'; import { performStitching } from './performStitching'; jest.setTimeout(60_000); @@ -82,15 +83,29 @@ describe('performStitching', () => { }, ]); + const deferredStrategy = { + mode: 'deferred' as const, + pollingInterval: { seconds: 1 }, + stitchTimeout: { seconds: 1 }, + }; + + await markForStitching({ + knex, + strategy: deferredStrategy, + entityRefs: ['k:ns/n'], + }); + await performStitching({ knex, logger, - strategy: { - mode: 'deferred', - pollingInterval: { seconds: 1 }, - stitchTimeout: { seconds: 1 }, - }, + strategy: deferredStrategy, entityRef: 'k:ns/n', + stitchTicket: ( + await knex('stitch_queue') + .where('entity_ref', 'k:ns/n') + .select('latest_ticket') + .first() + )?.latest_ticket, }); entities = await knex('final_entities'); @@ -171,15 +186,23 @@ describe('performStitching', () => { ); // Re-stitch without any changes + await markForStitching({ + knex, + strategy: deferredStrategy, + entityRefs: ['k:ns/n'], + }); + await performStitching({ knex, logger, - strategy: { - mode: 'deferred', - pollingInterval: { seconds: 1 }, - stitchTimeout: { seconds: 1 }, - }, + strategy: deferredStrategy, entityRef: 'k:ns/n', + stitchTicket: ( + await knex('stitch_queue') + .where('entity_ref', 'k:ns/n') + .select('latest_ticket') + .first() + )?.latest_ticket, }); entities = await knex('final_entities'); @@ -198,15 +221,23 @@ describe('performStitching', () => { }, ]); + await markForStitching({ + knex, + strategy: deferredStrategy, + entityRefs: ['k:ns/n'], + }); + await performStitching({ knex, logger, - strategy: { - mode: 'deferred', - pollingInterval: { seconds: 1 }, - stitchTimeout: { seconds: 1 }, - }, + strategy: deferredStrategy, entityRef: 'k:ns/n', + stitchTicket: ( + await knex('stitch_queue') + .where('entity_ref', 'k:ns/n') + .select('latest_ticket') + .first() + )?.latest_ticket, }); entities = await knex('final_entities'); @@ -342,7 +373,6 @@ describe('performStitching', () => { entity_id: 'my-id', entity_ref: 'k:ns/n', hash: '', - stitch_ticket: 'old-ticket', final_entity: JSON.stringify({}), }, ]); @@ -365,7 +395,7 @@ describe('performStitching', () => { ); it.each(databases.eachSupportedId())( - 'replaces existing stitch ticket %p', + 'stitches when final_entities row already exists %p', async databaseId => { const knex = await databases.init(databaseId); await applyDatabaseMigrations(knex); @@ -396,23 +426,10 @@ describe('performStitching', () => { entity_id: 'my-id', entity_ref: 'k:ns/n', hash: '', - stitch_ticket: 'old-ticket', final_entity: JSON.stringify({}), }, ]); - await expect( - knex('final_entities').select([ - 'entity_id', - 'stitch_ticket', - ]), - ).resolves.toEqual([ - { - entity_id: 'my-id', - stitch_ticket: expect.stringContaining('old-ticket'), - }, - ]); - const stitchLogger = mockServices.logger.mock(); await expect( performStitching({ @@ -423,17 +440,10 @@ describe('performStitching', () => { }), ).resolves.toBe('changed'); - await expect( - knex('final_entities').select([ - 'entity_id', - 'stitch_ticket', - ]), - ).resolves.toEqual([ - { - entity_id: 'my-id', - stitch_ticket: expect.not.stringContaining('old-ticket'), - }, - ]); + const entities = await knex('final_entities'); + expect(entities.length).toBe(1); + expect(entities[0].hash).not.toBe(''); + expect(entities[0].final_entity).toBeDefined(); }, ); }); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index 32c0d4ccb3..8d96acc029 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -23,12 +23,12 @@ import { import { AlphaEntity, EntityStatusItem } from '@backstage/catalog-model/alpha'; import { SerializedError } from '@backstage/errors'; import { Knex } from 'knex'; -import { v4 as uuid } from 'uuid'; import { StitchingStrategy } from '../../../stitching/types'; import { DbFinalEntitiesRow, DbRefreshStateRow, DbSearchRow, + DbStitchQueueRow, } from '../../tables'; import { buildEntitySearch } from './buildEntitySearch'; import { markDeferredStitchCompleted } from './markDeferredStitchCompleted'; @@ -56,7 +56,7 @@ export async function performStitching(options: { stitchTicket?: string; }): Promise<'changed' | 'unchanged' | 'abandoned'> { const { knex, logger, entityRef } = options; - const stitchTicket = options.stitchTicket ?? uuid(); + const stitchTicket = options.stitchTicket; // In deferred mode, the entity is removed from the stitch queue on ANY // completion, except when an exception is thrown. In the latter case, the @@ -73,17 +73,16 @@ export async function performStitching(options: { return 'abandoned'; } - // Insert stitching ticket that will be compared before inserting the final entity. + // 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, - stitch_ticket: stitchTicket, }) .onConflict('entity_id') - .merge(['stitch_ticket']); + .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 @@ -231,14 +230,26 @@ export async function performStitching(options: { // to write the search index. const searchEntries = buildEntitySearch(entityId, entity); - const amountOfRowsChanged = await knex('final_entities') + let updateQuery = knex('final_entities') .update({ final_entity: JSON.stringify(entity), hash, last_updated_at: knex.fn.now(), }) - .where('entity_id', entityId) - .where('stitch_ticket', stitchTicket); + .where('entity_id', entityId); + + // In deferred mode, guard against concurrent stitchers by checking that + // the latest_ticket in stitch_queue still matches what we were given. + if (options.strategy.mode === 'deferred' && stitchTicket) { + updateQuery = updateQuery.whereExists( + knex('stitch_queue') + .where('stitch_queue.entity_ref', entityRef) + .where('stitch_queue.latest_ticket', stitchTicket) + .select(knex.raw('1')), + ); + } + + const amountOfRowsChanged = await updateQuery; if (amountOfRowsChanged === 0) { logger.debug(`Entity ${entityRef} is already stitched, skipping write.`); @@ -255,7 +266,7 @@ export async function performStitching(options: { removeFromStitchQueueOnCompletion = false; throw error; } finally { - if (removeFromStitchQueueOnCompletion) { + if (removeFromStitchQueueOnCompletion && stitchTicket) { await markDeferredStitchCompleted({ knex: knex, entityRef, diff --git a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts index bcf2cc4674..48e90d1431 100644 --- a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts +++ b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts @@ -70,7 +70,6 @@ describe('deleteOrphanedEntities', () => { entity_id: `id-${ref}`, hash: 'original', entity_ref: ref, - stitch_ticket: '', }); } } diff --git a/plugins/catalog-backend/src/database/tables.ts b/plugins/catalog-backend/src/database/tables.ts index cb4b6deff4..bab0f7b20d 100644 --- a/plugins/catalog-backend/src/database/tables.ts +++ b/plugins/catalog-backend/src/database/tables.ts @@ -135,7 +135,6 @@ export type DbRelationsRow = { export type DbFinalEntitiesRow = { entity_id: string; hash: string; - stitch_ticket: string; final_entity?: string; last_updated_at?: string | Date; entity_ref: string; @@ -155,11 +154,18 @@ export type DbStitchQueueRow = { */ entity_ref: string; /** - * A random value distinguishing stitch requests. Used for optimistic - * concurrency: when a stitch completes, the row is only deleted if the - * ticket still matches the one that was read at the start. + * A random value that changes with every new stitch request. Used for + * optimistic concurrency: when a stitch completes, the row is only deleted + * if this ticket still matches the active_ticket (meaning no new request + * came in while stitching was in progress). */ - stitch_ticket: string; + latest_ticket: string; + /** + * Set when a stitcher picks up this item for processing. The stitcher + * compares this against latest_ticket before writing results; if they + * differ, a new stitch request arrived and the current work is abandoned. + */ + active_ticket?: string | null; /** * The point in time when this entity should next be stitched. * @@ -175,7 +181,7 @@ export type DbStitchQueueRow = { * future. * * Only when a stitch run is completed successfully, AND it's found that the - * stitch ticket has not changed since the start (which means that no new + * latest ticket has not changed since the start (which means that no new * request has been made behind our backs), does the row get deleted. */ next_stitch_at: string | Date; diff --git a/plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts b/plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts index e1bd8ec386..39f65a2b9e 100644 --- a/plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts +++ b/plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts @@ -225,7 +225,7 @@ describe('DefaultLocationStore', () => { final_entity: '{}', hash: 'hash', last_updated_at: new Date(), - stitch_ticket: '', + entity_ref: 'k:ns/n', }); diff --git a/plugins/catalog-backend/src/providers/GenericScmEventRefreshProvider.test.ts b/plugins/catalog-backend/src/providers/GenericScmEventRefreshProvider.test.ts index e6c1c701d3..0b4d2ad1cd 100644 --- a/plugins/catalog-backend/src/providers/GenericScmEventRefreshProvider.test.ts +++ b/plugins/catalog-backend/src/providers/GenericScmEventRefreshProvider.test.ts @@ -86,7 +86,7 @@ describe('GenericScmEventRefreshProvider', () => { entity_id: id, entity_ref: `k:ns/${id}`, hash: 'h', - stitch_ticket: '', + final_entity: '{}', }); } diff --git a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts index f9e41b679c..3a184676df 100644 --- a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts +++ b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts @@ -80,7 +80,6 @@ describe('DefaultEntitiesCatalog', () => { entity_ref: entityRef, final_entity: entityJson, hash: 'h', - stitch_ticket: '', }); for (const parent of parents) { @@ -118,7 +117,6 @@ describe('DefaultEntitiesCatalog', () => { entity_ref: entityRef, final_entity: entityJson, hash: 'h', - stitch_ticket: '', }); for (const row of buildEntitySearch(id, entity)) { diff --git a/plugins/catalog-backend/src/service/request/applyEntityFilterToQuery.test.ts b/plugins/catalog-backend/src/service/request/applyEntityFilterToQuery.test.ts index ea64f5b400..9b540c95b7 100644 --- a/plugins/catalog-backend/src/service/request/applyEntityFilterToQuery.test.ts +++ b/plugins/catalog-backend/src/service/request/applyEntityFilterToQuery.test.ts @@ -98,7 +98,6 @@ describe.each(databases.eachSupportedId())( entity_ref: entityRef, final_entity: entityJson, hash: 'h', - stitch_ticket: '', }); const search = await buildEntitySearch(id, entity); diff --git a/plugins/catalog-backend/src/service/request/applyPredicateEntityFilterToQuery.test.ts b/plugins/catalog-backend/src/service/request/applyPredicateEntityFilterToQuery.test.ts index cea9d961f7..6755ed5cc2 100644 --- a/plugins/catalog-backend/src/service/request/applyPredicateEntityFilterToQuery.test.ts +++ b/plugins/catalog-backend/src/service/request/applyPredicateEntityFilterToQuery.test.ts @@ -105,7 +105,6 @@ describe.each(databases.eachSupportedId())( entity_ref: entityRef, final_entity: entityJson, hash: 'h', - stitch_ticket: '', }); const search = await buildEntitySearch(id, entity); diff --git a/plugins/catalog-backend/src/tests/migrations.test.ts b/plugins/catalog-backend/src/tests/migrations.test.ts index 3cb3cc8fc3..6335321512 100644 --- a/plugins/catalog-backend/src/tests/migrations.test.ts +++ b/plugins/catalog-backend/src/tests/migrations.test.ts @@ -74,7 +74,6 @@ describe('migrations', () => { .insert({ entity_id: 'i1', hash: 'h', - stitch_ticket: '', final_entity: '{}', entity_ref: 'k:ns/n1', }) @@ -967,6 +966,22 @@ describe('migrations', () => { }, ]); + // Insert final_entities rows (with stitch_ticket column that will be dropped) + await knex('final_entities').insert([ + { + entity_id: 'id1', + entity_ref: 'component:default/with-stitch', + hash: 'h1', + stitch_ticket: 'old-ticket-1', + }, + { + entity_id: 'id2', + entity_ref: 'component:default/no-stitch', + hash: 'h2', + stitch_ticket: 'old-ticket-2', + }, + ]); + // Verify initial state - stitch_queue table should NOT exist yet const preTableExists = await knex.schema.hasTable('stitch_queue'); expect(preTableExists).toBe(false); @@ -983,21 +998,25 @@ describe('migrations', () => { expect(refreshStateColumnInfo.next_stitch_at).toBeUndefined(); expect(refreshStateColumnInfo.next_stitch_ticket).toBeUndefined(); + // Verify stitch_ticket column was removed from final_entities + const finalEntitiesColumnInfo = await knex('final_entities').columnInfo(); + expect(finalEntitiesColumnInfo.stitch_ticket).toBeUndefined(); + // Verify data was migrated correctly to stitch_queue const stitchQueueAfterUp = await knex('stitch_queue') .orderBy('entity_ref') - .select('entity_ref', 'stitch_ticket', 'next_stitch_at'); + .select('entity_ref', 'latest_ticket', 'next_stitch_at'); // Only entities with pending stitches should be in stitch_queue (id1 and id3) expect(stitchQueueAfterUp).toEqual([ { entity_ref: 'component:default/orphan-stitch', - stitch_ticket: 'ticket-3', + latest_ticket: 'ticket-3', next_stitch_at: expect.anything(), }, { entity_ref: 'component:default/with-stitch', - stitch_ticket: 'ticket-1', + latest_ticket: 'ticket-1', next_stitch_at: expect.anything(), }, ]); @@ -1009,6 +1028,12 @@ describe('migrations', () => { const revertedTableExists = await knex.schema.hasTable('stitch_queue'); expect(revertedTableExists).toBe(false); + // Verify stitch_ticket column was restored to final_entities + const revertedFinalEntitiesColumnInfo = await knex( + 'final_entities', + ).columnInfo(); + expect(revertedFinalEntitiesColumnInfo.stitch_ticket).not.toBeUndefined(); + // Verify next_stitch_at and next_stitch_ticket columns were restored to refresh_state const revertedRefreshColumnInfo = await knex( 'refresh_state',