diff --git a/.changeset/gentle-tables-march.md b/.changeset/gentle-tables-march.md index 91f85c7117..5f74be9a24 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 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. +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_ticket` is used 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 92becfe257..7dd11db78b 100644 --- a/plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js +++ b/plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js @@ -34,13 +34,9 @@ exports.up = async function up(knex) { .notNullable() .comment('The entity ref that needs stitching'); table - .string('latest_ticket') + .string('stitch_ticket') .notNullable() .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() @@ -58,16 +54,15 @@ exports.up = async function up(knex) { .from('refresh_state') .whereNotNull('refresh_state.next_stitch_at'); - for (let i = 0; i < pendingStitches.length; i += 1000) { - const batch = pendingStitches.slice(i, i + 1000); - await knex('stitch_queue').insert( - batch.map(row => ({ - entity_ref: row.entity_ref, - latest_ticket: row.next_stitch_ticket || '', - next_stitch_at: row.next_stitch_at, - })), - ); - } + await knex.batchInsert( + 'stitch_queue', + pendingStitches.map(row => ({ + entity_ref: row.entity_ref, + stitch_ticket: row.next_stitch_ticket || '', + next_stitch_at: row.next_stitch_at, + })), + 1000, + ); // Step 3: Remove next_stitch_at and next_stitch_ticket columns from refresh_state if (isSQLite) { @@ -102,10 +97,17 @@ exports.down = async function down(knex) { await knex.schema.alterTable('final_entities', table => { table .text('stitch_ticket') - .notNullable() - .defaultTo('') - .comment('Optimistic concurrency ticket for stitching'); + .nullable() + .comment('Random value representing a unique stitch attempt ticket'); }); + await knex('final_entities').update({ stitch_ticket: '' }); + if (isSQLite) { + // SQLite doesn't support ALTER COLUMN, but the nullable column is fine + } else { + await knex.schema.alterTable('final_entities', table => { + table.text('stitch_ticket').notNullable().alter(); + }); + } // Step 2: Add back the columns to refresh_state await knex.schema.alterTable('refresh_state', table => { @@ -140,7 +142,7 @@ exports.down = async function down(knex) { .select({ entityRef: 'stitch_queue.entity_ref', nextStitchAt: 'stitch_queue.next_stitch_at', - latestTicket: 'stitch_queue.latest_ticket', + latestTicket: 'stitch_queue.stitch_ticket', }) .from('stitch_queue'); @@ -160,7 +162,7 @@ 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.latest_ticket') + .select('stitch_queue.stitch_ticket') .from('stitch_queue') .whereRaw('stitch_queue.entity_ref = refresh_state.entity_ref'), }) diff --git a/plugins/catalog-backend/report.sql.md b/plugins/catalog-backend/report.sql.md index bd63a6ee0c..dd5892b7fb 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 '20260215000000_move_stitch_queue.js' +> Failed to migrate down from '20241003170511_alter_target_in_locations.js' ## Sequences @@ -136,10 +136,9 @@ | 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 ac3cfc0cfb..1f6137f91e 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', - latest_ticket: 't1', + stitch_ticket: 't1', next_stitch_at: '2037-01-01T00:00:00.000', }, { entity_ref: 'k:ns/past_stitch_time', - latest_ticket: 't3', + stitch_ticket: 't3', next_stitch_at: '1971-01-01T00:00:00.000', }, { entity_ref: 'k:ns/past_stitch_time_again', - latest_ticket: 't4', + stitch_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 4adce0fc0d..2d12ee4f6a 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 latest_ticket set (by + * They are expected to already have the stitch_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', - 'latest_ticket', + 'stitch_ticket', ); // This avoids duplication of work because of race conditions and is @@ -81,12 +81,11 @@ 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.latest_ticket, + stitchTicket: i.stitch_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 325740394a..dbcbba13df 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', - latest_ticket: 'the-ticket', + stitch_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', - 'latest_ticket', + 'stitch_ticket', ); } @@ -57,7 +57,7 @@ describe('markDeferredStitchCompleted', () => { { entity_ref: 'k:ns/n', next_stitch_at: expect.anything(), - latest_ticket: 'the-ticket', + stitch_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 2d80a62496..bb3e2be6f9 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('latest_ticket', '=', stitchTicket) + .andWhere('stitch_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 88a0ab027a..f7fd9a0695 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', - latest_ticket: 'old', + stitch_ticket: 'old', next_stitch_at: '1971-01-01T00:00:00.000', }, ]); async function result() { return knex('stitch_queue') - .select('entity_ref', 'next_stitch_at', 'latest_ticket') + .select('entity_ref', 'next_stitch_at', 'stitch_ticket') .orderBy('entity_ref', 'asc'); } @@ -93,7 +93,7 @@ describe('markForStitching', () => { { entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - latest_ticket: 'old', + stitch_ticket: 'old', }, ]); @@ -111,7 +111,7 @@ describe('markForStitching', () => { { entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - latest_ticket: 'old', + stitch_ticket: 'old', }, ]); @@ -129,12 +129,12 @@ describe('markForStitching', () => { { entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - latest_ticket: 'old', + stitch_ticket: 'old', }, { entity_ref: 'k:ns/one', next_stitch_at: expect.anything(), - latest_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, ]); @@ -152,17 +152,17 @@ describe('markForStitching', () => { { entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - latest_ticket: 'old', + stitch_ticket: 'old', }, { entity_ref: 'k:ns/one', next_stitch_at: expect.anything(), - latest_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, { entity_ref: 'k:ns/two', next_stitch_at: expect.anything(), - latest_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, ]); @@ -180,29 +180,29 @@ describe('markForStitching', () => { { entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - latest_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, { entity_ref: 'k:ns/one', next_stitch_at: expect.anything(), - latest_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, { entity_ref: 'k:ns/three', next_stitch_at: expect.anything(), - latest_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, { entity_ref: 'k:ns/two', next_stitch_at: expect.anything(), - latest_ticket: expect.anything(), + stitch_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?.latest_ticket).not.toEqual('old'); + expect(entity4Final?.stitch_ticket).not.toEqual('old'); }, ); @@ -559,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', 'latest_ticket') + .select('entity_ref', 'next_stitch_at', 'stitch_ticket') .orderBy('entity_ref'); expect(finalState.length).toBeGreaterThan(0); finalState.forEach(row => { expect(row.next_stitch_at).not.toBeNull(); - expect(row.latest_ticket).not.toBeNull(); + expect(row.stitch_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 7c67ada1a2..9e22a65e7b 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, - latest_ticket: ticket, + stitch_ticket: ticket, next_stitch_at: knex.fn.now(), })), ) .onConflict('entity_ref') - .merge(['next_stitch_at', 'latest_ticket']); + .merge(['next_stitch_at', 'stitch_ticket']); } }, knex); } @@ -114,12 +114,12 @@ export async function markForStitching(options: { .insert( refreshStateRows.map(row => ({ entity_ref: row.entity_ref, - latest_ticket: ticket, + stitch_ticket: ticket, next_stitch_at: knex.fn.now(), })), ) .onConflict('entity_ref') - .merge(['next_stitch_at', 'latest_ticket']); + .merge(['next_stitch_at', 'stitch_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 15fc8b71d2..41931a8b7e 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -103,9 +103,9 @@ describe('performStitching', () => { stitchTicket: ( await knex('stitch_queue') .where('entity_ref', 'k:ns/n') - .select('latest_ticket') + .select('stitch_ticket') .first() - )?.latest_ticket, + )?.stitch_ticket, }); entities = await knex('final_entities'); @@ -200,9 +200,9 @@ describe('performStitching', () => { stitchTicket: ( await knex('stitch_queue') .where('entity_ref', 'k:ns/n') - .select('latest_ticket') + .select('stitch_ticket') .first() - )?.latest_ticket, + )?.stitch_ticket, }); entities = await knex('final_entities'); @@ -235,9 +235,9 @@ describe('performStitching', () => { stitchTicket: ( await knex('stitch_queue') .where('entity_ref', 'k:ns/n') - .select('latest_ticket') + .select('stitch_ticket') .first() - )?.latest_ticket, + )?.stitch_ticket, }); entities = await knex('final_entities'); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index 8d96acc029..95fe880eda 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -239,12 +239,12 @@ export async function performStitching(options: { .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. + // the stitch_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) + .where('stitch_queue.stitch_ticket', stitchTicket) .select(knex.raw('1')), ); } 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 48e90d1431..c61c425a30 100644 --- a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts +++ b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts @@ -102,6 +102,12 @@ describe('deleteOrphanedEntities', () => { .select('entity_ref', 'result_hash'); } + async function stitchQueue(knex: Knex) { + return await knex('stitch_queue') + .orderBy('entity_ref') + .select('entity_ref'); + } + async function finalEntities(knex: Knex) { return await knex('final_entities') .join( @@ -188,6 +194,7 @@ describe('deleteOrphanedEntities', () => { { entity_ref: 'E8', result_hash: 'original' }, { entity_ref: 'E9', result_hash: 'original' }, ]); + await expect(stitchQueue(knex)).resolves.toEqual([]); await expect(finalEntities(knex)).resolves.toEqual([ { entity_ref: 'E1', hash: 'original', next_stitch_at: null }, { @@ -278,6 +285,10 @@ describe('deleteOrphanedEntities', () => { { entity_ref: 'E8', result_hash: 'original' }, { entity_ref: 'E9', result_hash: 'original' }, ]); + await expect(stitchQueue(knex)).resolves.toEqual([ + { entity_ref: 'E2' }, + { entity_ref: 'E7' }, + ]); await expect(finalEntities(knex)).resolves.toEqual([ { entity_ref: 'E1', hash: 'original', next_stitch_at: null }, { diff --git a/plugins/catalog-backend/src/database/tables.ts b/plugins/catalog-backend/src/database/tables.ts index bab0f7b20d..43068c8df8 100644 --- a/plugins/catalog-backend/src/database/tables.ts +++ b/plugins/catalog-backend/src/database/tables.ts @@ -156,16 +156,10 @@ export type DbStitchQueueRow = { /** * 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). + * if this ticket still matches (meaning no new request came in while + * stitching was in progress). */ - 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; + stitch_ticket: string; /** * The point in time when this entity should next be stitched. * @@ -181,7 +175,7 @@ export type DbStitchQueueRow = { * future. * * Only when a stitch run is completed successfully, AND it's found that the - * latest ticket has not changed since the start (which means that no new + * stitch 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 39f65a2b9e..06cc9269e3 100644 --- a/plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts +++ b/plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts @@ -225,7 +225,6 @@ describe('DefaultLocationStore', () => { final_entity: '{}', hash: 'hash', last_updated_at: new Date(), - 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 0b4d2ad1cd..6e20226677 100644 --- a/plugins/catalog-backend/src/providers/GenericScmEventRefreshProvider.test.ts +++ b/plugins/catalog-backend/src/providers/GenericScmEventRefreshProvider.test.ts @@ -86,7 +86,6 @@ describe('GenericScmEventRefreshProvider', () => { entity_id: id, entity_ref: `k:ns/${id}`, hash: 'h', - final_entity: '{}', }); } diff --git a/plugins/catalog-backend/src/tests/migrations.test.ts b/plugins/catalog-backend/src/tests/migrations.test.ts index 6335321512..4a469ab07e 100644 --- a/plugins/catalog-backend/src/tests/migrations.test.ts +++ b/plugins/catalog-backend/src/tests/migrations.test.ts @@ -1005,18 +1005,18 @@ describe('migrations', () => { // Verify data was migrated correctly to stitch_queue const stitchQueueAfterUp = await knex('stitch_queue') .orderBy('entity_ref') - .select('entity_ref', 'latest_ticket', 'next_stitch_at'); + .select('entity_ref', 'stitch_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', - latest_ticket: 'ticket-3', + stitch_ticket: 'ticket-3', next_stitch_at: expect.anything(), }, { entity_ref: 'component:default/with-stitch', - latest_ticket: 'ticket-1', + stitch_ticket: 'ticket-1', next_stitch_at: expect.anything(), }, ]);