From 7416e8b0c04b21d23b540ba3768fc5902a1daa54 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Sun, 22 Feb 2026 16:51:21 +0100 Subject: [PATCH 1/3] catalog: move the stitch queue concerns into a dedicated stitch_queue table MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 Signed-off-by: Fredrik Adelöw --- .changeset/gentle-tables-march.md | 5 + .../20260215000000_move_stitch_queue.js | 150 ++++++++++++++++++ plugins/catalog-backend/report.sql.md | 16 +- .../getDeferredStitchableEntities.test.ts | 96 +++++------ .../stitcher/getDeferredStitchableEntities.ts | 18 +-- .../markDeferredStitchCompleted.test.ts | 30 ++-- .../stitcher/markDeferredStitchCompleted.ts | 19 +-- .../stitcher/markForStitching.test.ts | 103 ++++++------ .../operations/stitcher/markForStitching.ts | 48 ++++-- .../util/deleteOrphanedEntities.test.ts | 45 ++---- .../catalog-backend/src/database/tables.ts | 81 +++++----- .../src/stitching/progressTracker.ts | 8 +- .../src/tests/migrations.test.ts | 125 +++++++++++++++ 13 files changed, 507 insertions(+), 237 deletions(-) create mode 100644 .changeset/gentle-tables-march.md create mode 100644 plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js diff --git a/.changeset/gentle-tables-march.md b/.changeset/gentle-tables-march.md new file mode 100644 index 0000000000..0aa13cfb4e --- /dev/null +++ b/.changeset/gentle-tables-march.md @@ -0,0 +1,5 @@ +--- +'@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. diff --git a/plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js b/plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js new file mode 100644 index 0000000000..da998b6031 --- /dev/null +++ b/plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js @@ -0,0 +1,150 @@ +/* + * Copyright 2026 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 + +/** + * Creates a dedicated stitch_queue table and moves the stitch queue columns + * from refresh_state into it. The stitch queue semantically is a separate + * concern from the refresh state. + * + * @param {import('knex').Knex} knex + */ +exports.up = async function up(knex) { + const isSQLite = knex.client.config.client.includes('sqlite'); + + // Step 1: Create the stitch_queue table + await knex.schema.createTable('stitch_queue', table => { + table + .string('entity_ref', 255) + .primary() + .notNullable() + .comment('The entity ref that needs stitching'); + table + .string('stitch_ticket') + .notNullable() + .comment('Random value distinguishing stitch requests'); + table + .dateTime('next_stitch_at') + .notNullable() + .index('stitch_queue_next_stitch_at_idx') + .comment('Timestamp of when entity should be stitched'); + }); + + // Step 2: Migrate existing stitch requests from refresh_state to stitch_queue + const pendingStitches = await knex + .select({ + entity_ref: 'refresh_state.entity_ref', + next_stitch_at: 'refresh_state.next_stitch_at', + next_stitch_ticket: 'refresh_state.next_stitch_ticket', + }) + .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, + stitch_ticket: row.next_stitch_ticket || '', + next_stitch_at: row.next_stitch_at, + })), + ); + } + + // Step 3: Remove next_stitch_at and next_stitch_ticket columns from refresh_state + if (isSQLite) { + await knex.raw('DROP INDEX IF EXISTS refresh_state_next_stitch_at_idx'); + await knex.raw('ALTER TABLE refresh_state DROP COLUMN next_stitch_at'); + await knex.raw('ALTER TABLE refresh_state DROP COLUMN next_stitch_ticket'); + } else { + await knex.schema.alterTable('refresh_state', table => { + table.dropIndex([], 'refresh_state_next_stitch_at_idx'); + table.dropColumn('next_stitch_at'); + table.dropColumn('next_stitch_ticket'); + }); + } +}; + +/** + * @param {import('knex').Knex} knex + */ +exports.down = async function down(knex) { + const isSQLite = knex.client.config.client.includes('sqlite'); + + // Step 1: Add back the columns to refresh_state + await knex.schema.alterTable('refresh_state', table => { + table + .dateTime('next_stitch_at') + .nullable() + .comment('Timestamp of when entity should be stitched'); + table + .string('next_stitch_ticket') + .nullable() + .comment('Random value distinguishing stitch requests'); + }); + + // Step 1b: Create partial index separately + if (isSQLite) { + await knex.raw(` + CREATE INDEX refresh_state_next_stitch_at_idx + ON refresh_state (next_stitch_at) + WHERE next_stitch_at IS NOT NULL + `); + } else { + await knex.schema.alterTable('refresh_state', table => { + table.index('next_stitch_at', 'refresh_state_next_stitch_at_idx', { + predicate: knex.whereNotNull('next_stitch_at'), + }); + }); + } + + // Step 2: 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', + }) + .from('stitch_queue'); + + for (const row of pendingStitches) { + await knex('refresh_state') + .update({ + next_stitch_at: row.nextStitchAt, + next_stitch_ticket: row.stitchTicket, + }) + .where('entity_ref', row.entityRef); + } + } else { + await knex('refresh_state') + .update({ + next_stitch_at: knex + .select('stitch_queue.next_stitch_at') + .from('stitch_queue') + .whereRaw('stitch_queue.entity_ref = refresh_state.entity_ref'), + next_stitch_ticket: knex + .select('stitch_queue.stitch_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 + await knex.schema.dropTable('stitch_queue'); +}; diff --git a/plugins/catalog-backend/report.sql.md b/plugins/catalog-backend/report.sql.md index 4f76d2a3f5..b06ef0a79b 100644 --- a/plugins/catalog-backend/report.sql.md +++ b/plugins/catalog-backend/report.sql.md @@ -76,8 +76,6 @@ | `errors` | `text` | false | - | - | | `last_discovery_at` | `timestamp with time zone` | false | - | - | | `location_key` | `text` | true | - | - | -| `next_stitch_at` | `timestamp with time zone` | true | - | - | -| `next_stitch_ticket` | `character varying` | true | 255 | - | | `next_update_at` | `timestamp with time zone` | false | - | - | | `processed_entity` | `text` | true | - | - | | `result_hash` | `text` | true | - | - | @@ -87,7 +85,6 @@ ### Indices - `refresh_state_entity_ref_uniq` (`entity_ref`) unique -- `refresh_state_next_stitch_at_idx` (`next_stitch_at`) - `refresh_state_next_update_at_idx` (`next_update_at`) - `refresh_state_pkey` (`entity_id`) unique primary @@ -135,3 +132,16 @@ - `search_entity_id_idx` (`entity_id`) - `search_key_original_value_idx` (`key`, `original_value`) - `search_key_value_idx` (`key`, `value`) + +## Table `stitch_queue` + +| Column | Type | Nullable | Max Length | Default | +| ---------------- | -------------------------- | -------- | ---------- | ------- | +| `entity_ref` | `character varying` | false | 255 | - | +| `next_stitch_at` | `timestamp with time zone` | false | - | - | +| `stitch_ticket` | `character varying` | false | 255 | - | + +### Indices + +- `stitch_queue_next_stitch_at_idx` (`next_stitch_at`) +- `stitch_queue_pkey` (`entity_ref`) unique primary 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 b66dc74ef2..1f6137f91e 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts @@ -16,6 +16,7 @@ import { TestDatabases } from '@backstage/backend-test-utils'; import { applyDatabaseMigrations } from '../../migrations'; +import { DbStitchQueueRow } from '../../tables'; import { getDeferredStitchableEntities } from './getDeferredStitchableEntities'; jest.setTimeout(60_000); @@ -29,56 +30,27 @@ describe('getDeferredStitchableEntities', () => { const knex = await databases.init(databaseId); await applyDatabaseMigrations(knex); - await knex - .insert([ - { - entity_id: '1', - entity_ref: 'k:ns/no_stitch_time', - unprocessed_entity: '{}', - processed_entity: '{}', - errors: '[]', - next_update_at: knex.fn.now(), - last_discovery_at: knex.fn.now(), - next_stitch_at: null, - next_stitch_ticket: null, - }, - { - entity_id: '2', - entity_ref: 'k:ns/future_stitch_time', - unprocessed_entity: '{}', - processed_entity: '{}', - errors: '[]', - next_update_at: knex.fn.now(), - last_discovery_at: knex.fn.now(), - next_stitch_at: '2037-01-01T00:00:00.000', - next_stitch_ticket: 't1', - }, - { - entity_id: '3', - entity_ref: 'k:ns/past_stitch_time', - unprocessed_entity: '{}', - processed_entity: '{}', - errors: '[]', - next_update_at: knex.fn.now(), - last_discovery_at: knex.fn.now(), - next_stitch_at: '1971-01-01T00:00:00.000', - next_stitch_ticket: 't3', - }, - { - entity_id: '4', - entity_ref: 'k:ns/past_stitch_time_again', - unprocessed_entity: '{}', - processed_entity: '{}', - errors: '[]', - next_update_at: knex.fn.now(), - last_discovery_at: knex.fn.now(), - next_stitch_at: '1972-01-01T00:00:00.000', - next_stitch_ticket: 't4', - }, - ]) - .into('refresh_state'); + // Insert stitch_queue rows - no need for refresh_state rows since + // stitch_queue is a standalone table + await knex('stitch_queue').insert([ + { + entity_ref: 'k:ns/future_stitch_time', + stitch_ticket: 't1', + next_stitch_at: '2037-01-01T00:00:00.000', + }, + { + entity_ref: 'k:ns/past_stitch_time', + stitch_ticket: 't3', + next_stitch_at: '1971-01-01T00:00:00.000', + }, + { + entity_ref: 'k:ns/past_stitch_time_again', + stitch_ticket: 't4', + next_stitch_at: '1972-01-01T00:00:00.000', + }, + ]); - const rowsBefore = await knex('refresh_state'); + const rowsBefore = await knex('stitch_queue'); const items = await getDeferredStitchableEntities({ knex, @@ -86,7 +58,7 @@ describe('getDeferredStitchableEntities', () => { stitchTimeout: { seconds: 2 }, }); - const rowsAfter = await knex('refresh_state'); + const rowsAfter = await knex('stitch_queue'); expect(items).toEqual([ { @@ -96,17 +68,21 @@ describe('getDeferredStitchableEntities', () => { }, ]); - const hitRowBefore = rowsBefore.filter(r => r.entity_id === '3')[0] - .next_stitch_at; - const hitRowAfter = rowsAfter.filter(r => r.entity_id === '3')[0] - .next_stitch_at; - const missRowBefore = rowsBefore.filter(r => r.entity_id === '4')[0] - .next_stitch_at; - const missRowAfter = rowsAfter.filter(r => r.entity_id === '4')[0] - .next_stitch_at; + const hitRowBefore = rowsBefore.filter( + r => r.entity_ref === 'k:ns/past_stitch_time', + )[0].next_stitch_at; + const hitRowAfter = rowsAfter.filter( + r => r.entity_ref === 'k:ns/past_stitch_time', + )[0].next_stitch_at; + const missRowBefore = rowsBefore.filter( + r => r.entity_ref === 'k:ns/past_stitch_time_again', + )[0].next_stitch_at; + const missRowAfter = rowsAfter.filter( + r => r.entity_ref === 'k:ns/past_stitch_time_again', + )[0].next_stitch_at; - expect(+new Date(hitRowAfter)).toBeGreaterThan(+new Date(hitRowBefore)); - expect(+new Date(missRowAfter)).toEqual(+new Date(missRowBefore)); + expect(+new Date(hitRowAfter!)).toBeGreaterThan(+new Date(hitRowBefore!)); + expect(+new Date(missRowAfter!)).toEqual(+new Date(missRowBefore!)); }, ); }); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts index 05c3f4d2eb..2d12ee4f6a 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts @@ -18,7 +18,7 @@ import { durationToMilliseconds, HumanDuration } from '@backstage/types'; import { Knex } from 'knex'; import { DateTime } from 'luxon'; import { timestampToDateTime } from '../../conversion'; -import { DbRefreshStateRow } from '../../tables'; +import { DbStitchQueueRow } from '../../tables'; // TODO(freben): There is no retry counter or similar. If items start // perpetually crashing during stitching, they'll just get silently retried over @@ -31,7 +31,7 @@ import { DbRefreshStateRow } from '../../tables'; * * This assumes that the stitching strategy is set to deferred. * - * They are expected to already have the next_stitch_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 @@ -52,10 +52,10 @@ export async function getDeferredStitchableEntities(options: { > { const { knex, batchSize, stitchTimeout } = options; - let itemsQuery = knex('refresh_state').select( + let itemsQuery = knex('stitch_queue').select( 'entity_ref', 'next_stitch_at', - 'next_stitch_ticket', + 'stitch_ticket', ); // This avoids duplication of work because of race conditions and is @@ -66,8 +66,6 @@ export async function getDeferredStitchableEntities(options: { } const items = await itemsQuery - .whereNotNull('next_stitch_at') - .whereNotNull('next_stitch_ticket') .where('next_stitch_at', '<=', knex.fn.now()) .orderBy('next_stitch_at', 'asc') .limit(batchSize); @@ -76,21 +74,19 @@ export async function getDeferredStitchableEntities(options: { return []; } - await knex('refresh_state') + await knex('stitch_queue') .whereIn( 'entity_ref', items.map(i => i.entity_ref), ) - // avoid race condition where someone completes a stitch right between these statements - .whereNotNull('next_stitch_ticket') .update({ next_stitch_at: nowPlus(knex, stitchTimeout), }); return items.map(i => ({ entityRef: i.entity_ref, - stitchTicket: i.next_stitch_ticket!, - stitchRequestedAt: timestampToDateTime(i.next_stitch_at!), + 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 b0957e0ec9..dbcbba13df 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts @@ -17,7 +17,7 @@ import { TestDatabases } from '@backstage/backend-test-utils'; import { applyDatabaseMigrations } from '../../migrations'; import { markDeferredStitchCompleted } from './markDeferredStitchCompleted'; -import { DbRefreshStateRow } from '../../tables'; +import { DbStitchQueueRow } from '../../tables'; jest.setTimeout(60_000); @@ -30,44 +30,44 @@ describe('markDeferredStitchCompleted', () => { const knex = await databases.init(databaseId); await applyDatabaseMigrations(knex); - await knex('refresh_state').insert([ + // Insert stitch_queue row + await knex('stitch_queue').insert([ { - entity_id: '1', entity_ref: 'k:ns/n', - unprocessed_entity: '{}', - processed_entity: '{}', - errors: '[]', - next_update_at: knex.fn.now(), - last_discovery_at: knex.fn.now(), + stitch_ticket: 'the-ticket', next_stitch_at: '1971-01-01T00:00:00.000', - next_stitch_ticket: 'the-ticket', }, ]); async function result() { - return knex('refresh_state').select( + return knex('stitch_queue').select( + 'entity_ref', 'next_stitch_at', - 'next_stitch_ticket', + 'stitch_ticket', ); } + // Wrong ticket should not delete the row await markDeferredStitchCompleted({ knex, entityRef: 'k:ns/n', stitchTicket: 'the-wrong-ticket', }); await expect(result()).resolves.toEqual([ - { next_stitch_at: expect.anything(), next_stitch_ticket: 'the-ticket' }, + { + entity_ref: 'k:ns/n', + next_stitch_at: expect.anything(), + stitch_ticket: 'the-ticket', + }, ]); + // Correct ticket should delete the row await markDeferredStitchCompleted({ knex, entityRef: 'k:ns/n', stitchTicket: 'the-ticket', }); - await expect(result()).resolves.toEqual([ - { next_stitch_at: null, next_stitch_ticket: null }, - ]); + await expect(result()).resolves.toEqual([]); }, ); }); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts index d1c7c3a6b8..bb3e2be6f9 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts @@ -15,7 +15,7 @@ */ import { Knex } from 'knex'; -import { DbRefreshStateRow } from '../../tables'; +import { DbStitchQueueRow } from '../../tables'; /** * Marks a single entity as having been stitched. @@ -24,10 +24,10 @@ import { DbRefreshStateRow } from '../../tables'; * * This assumes that the stitching strategy is set to deferred. * - * The timestamp and ticket are only reset if the ticket hasn't changed. If it - * has, it means that a new stitch request has been made, and the entity should - * be stitched once more some time in the future - or is indeed already being - * stitched concurrently with ourselves. + * The row is only deleted from stitch_queue if the ticket hasn't changed. If + * it has, it means that a new stitch request has been made, and the entity + * should be stitched once more some time in the future - or is indeed already + * being stitched concurrently with ourselves. */ export async function markDeferredStitchCompleted(option: { knex: Knex | Knex.Transaction; @@ -36,11 +36,8 @@ export async function markDeferredStitchCompleted(option: { }): Promise { const { knex, entityRef, stitchTicket } = option; - await knex('refresh_state') - .update({ - next_stitch_at: null, - next_stitch_ticket: null, - }) + await knex('stitch_queue') .where('entity_ref', '=', entityRef) - .andWhere('next_stitch_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 ff450e9ccf..2b43f3f281 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts @@ -17,7 +17,11 @@ import { TestDatabases } from '@backstage/backend-test-utils'; import { applyDatabaseMigrations } from '../../migrations'; import { markForStitching } from './markForStitching'; -import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables'; +import { + DbFinalEntitiesRow, + DbRefreshStateRow, + DbStitchQueueRow, +} from '../../tables'; jest.setTimeout(60_000); @@ -39,8 +43,6 @@ describe('markForStitching', () => { errors: '[]', next_update_at: knex.fn.now(), last_discovery_at: knex.fn.now(), - next_stitch_at: null, - next_stitch_ticket: null, }, { entity_id: '2', @@ -50,8 +52,6 @@ describe('markForStitching', () => { errors: '[]', next_update_at: knex.fn.now(), last_discovery_at: knex.fn.now(), - next_stitch_at: null, - next_stitch_ticket: null, }, { entity_id: '3', @@ -61,8 +61,6 @@ describe('markForStitching', () => { errors: '[]', next_update_at: knex.fn.now(), last_discovery_at: knex.fn.now(), - next_stitch_at: null, - next_stitch_ticket: null, }, { entity_id: '4', @@ -72,19 +70,34 @@ describe('markForStitching', () => { errors: '[]', next_update_at: knex.fn.now(), last_discovery_at: knex.fn.now(), + }, + ]); + // Entity 4 has an existing stitch_queue row with old stitch data + await knex('stitch_queue').insert([ + { + entity_ref: 'k:ns/four', + stitch_ticket: 'old', next_stitch_at: '1971-01-01T00:00:00.000', - next_stitch_ticket: 'old', }, ]); async function result() { - return knex('refresh_state') - .select('entity_id', 'next_stitch_at', 'next_stitch_ticket') - .orderBy('entity_id', 'asc'); + return knex('stitch_queue') + .select('entity_ref', 'next_stitch_at', 'stitch_ticket') + .orderBy('entity_ref', 'asc'); } + // Initially only entity 4 has a stitch_queue row const original = await result(); + expect(original).toEqual([ + { + entity_ref: 'k:ns/four', + next_stitch_at: expect.anything(), + stitch_ticket: 'old', + }, + ]); + // Calling with empty set should not create any new rows await markForStitching({ knex, strategy: { @@ -95,16 +108,14 @@ describe('markForStitching', () => { entityRefs: new Set(), }); await expect(result()).resolves.toEqual([ - { entity_id: '1', next_stitch_at: null, next_stitch_ticket: null }, - { entity_id: '2', next_stitch_at: null, next_stitch_ticket: null }, - { entity_id: '3', next_stitch_at: null, next_stitch_ticket: null }, { - entity_id: '4', + entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - next_stitch_ticket: 'old', + stitch_ticket: 'old', }, ]); + // Mark entity 1 - should create a new stitch_queue row await markForStitching({ knex, strategy: { @@ -116,19 +127,18 @@ describe('markForStitching', () => { }); await expect(result()).resolves.toEqual([ { - entity_id: '1', + entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - next_stitch_ticket: expect.anything(), + stitch_ticket: 'old', }, - { entity_id: '2', next_stitch_at: null, next_stitch_ticket: null }, - { entity_id: '3', next_stitch_at: null, next_stitch_ticket: null }, { - entity_id: '4', + entity_ref: 'k:ns/one', next_stitch_at: expect.anything(), - next_stitch_ticket: 'old', + stitch_ticket: expect.anything(), }, ]); + // Mark entity 2 - should create another new stitch_queue row await markForStitching({ knex, strategy: { @@ -140,23 +150,23 @@ describe('markForStitching', () => { }); await expect(result()).resolves.toEqual([ { - entity_id: '1', + entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - next_stitch_ticket: expect.anything(), + stitch_ticket: 'old', }, { - entity_id: '2', + entity_ref: 'k:ns/one', next_stitch_at: expect.anything(), - next_stitch_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, - { entity_id: '3', next_stitch_at: null, next_stitch_ticket: null }, { - entity_id: '4', + entity_ref: 'k:ns/two', next_stitch_at: expect.anything(), - next_stitch_ticket: 'old', + stitch_ticket: expect.anything(), }, ]); + // Mark entities 3 and 4 by ID - entity 3 creates new row, entity 4 updates existing await markForStitching({ knex, strategy: { @@ -168,35 +178,31 @@ describe('markForStitching', () => { }); await expect(result()).resolves.toEqual([ { - entity_id: '1', + entity_ref: 'k:ns/four', next_stitch_at: expect.anything(), - next_stitch_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, { - entity_id: '2', + entity_ref: 'k:ns/one', next_stitch_at: expect.anything(), - next_stitch_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, { - entity_id: '3', + entity_ref: 'k:ns/three', next_stitch_at: expect.anything(), - next_stitch_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, { - entity_id: '4', + entity_ref: 'k:ns/two', next_stitch_at: expect.anything(), - next_stitch_ticket: expect.anything(), + stitch_ticket: expect.anything(), }, ]); - // It overwrites timers and tickets if they existed before + // Entity 4's ticket should have been updated (was 'old', now something else) const final = await result(); - for (let i = 0; i < final.length; ++i) { - expect(original[i].next_stitch_at).not.toEqual(final[i].next_stitch_at); - expect(original[i].next_stitch_ticket).not.toEqual( - final[i].next_stitch_ticket, - ); - } + const entity4Final = final.find(r => r.entity_ref === 'k:ns/four'); + expect(entity4Final?.stitch_ticket).not.toEqual('old'); }, ); @@ -461,8 +467,6 @@ describe('markForStitching', () => { errors: '[]', next_update_at: knex.fn.now(), last_discovery_at: knex.fn.now(), - next_stitch_at: null, - next_stitch_ticket: null, })), ); @@ -558,15 +562,14 @@ describe('markForStitching', () => { expect(deadlockErrors).toEqual([]); // Verify final state - all entities should have been marked for stitching - const finalState = await knex('refresh_state') - .select('entity_ref', 'next_stitch_at', 'next_stitch_ticket') - .whereNotNull('next_stitch_at') + const finalState = await knex('stitch_queue') + .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.next_stitch_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 e4172e8cdf..9e22a65e7b 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts @@ -18,7 +18,11 @@ import { Knex } from 'knex'; import splitToChunks from 'lodash/chunk'; import { v4 as uuid } from 'uuid'; import { StitchingStrategy } from '../../../stitching/types'; -import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables'; +import { + DbFinalEntitiesRow, + DbRefreshStateRow, + DbStitchQueueRow, +} from '../../tables'; import { retryOnDeadlock } from '../../util'; const UPDATE_CHUNK_SIZE = 100; // Smaller chunks reduce contention @@ -77,30 +81,46 @@ export async function markForStitching(options: { }, knex); } } else if (mode === 'deferred') { - // It's OK that this is shared across refresh state rows; it just needs to + // It's OK that this is shared across stitch_queue rows; it just needs to // be uniquely generated for every new stitch request. const ticket = uuid(); - // Update by primary key in deterministic order to avoid deadlocks for (const chunk of entityRefs) { await retryOnDeadlock(async () => { - await knex('refresh_state') - .update({ - next_stitch_at: knex.fn.now(), - next_stitch_ticket: ticket, - }) - .whereIn('entity_ref', chunk); + if (chunk.length > 0) { + await knex('stitch_queue') + .insert( + chunk.map(ref => ({ + entity_ref: ref, + stitch_ticket: ticket, + next_stitch_at: knex.fn.now(), + })), + ) + .onConflict('entity_ref') + .merge(['next_stitch_at', 'stitch_ticket']); + } }, knex); } for (const chunk of entityIds) { await retryOnDeadlock(async () => { - await knex('refresh_state') - .update({ - next_stitch_at: knex.fn.now(), - next_stitch_ticket: ticket, - }) + // Look up entity_refs from refresh_state by entity_id + const refreshStateRows = await knex('refresh_state') + .select('entity_ref') .whereIn('entity_id', chunk); + + if (refreshStateRows.length > 0) { + await knex('stitch_queue') + .insert( + refreshStateRows.map(row => ({ + entity_ref: row.entity_ref, + stitch_ticket: ticket, + next_stitch_at: knex.fn.now(), + })), + ) + .onConflict('entity_ref') + .merge(['next_stitch_at', 'stitch_ticket']); + } }, knex); } } else { 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 0f4f5f0c3d..bcf2cc4674 100644 --- a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts +++ b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts @@ -100,7 +100,7 @@ describe('deleteOrphanedEntities', () => { async function refreshState(knex: Knex) { return await knex('refresh_state') .orderBy('entity_ref') - .select('entity_ref', 'result_hash', 'next_stitch_at'); + .select('entity_ref', 'result_hash'); } async function finalEntities(knex: Knex) { @@ -110,11 +110,16 @@ describe('deleteOrphanedEntities', () => { 'final_entities.entity_id', 'refresh_state.entity_id', ) + .leftOuterJoin( + 'stitch_queue', + 'stitch_queue.entity_ref', + 'refresh_state.entity_ref', + ) .orderBy('refresh_state.entity_ref') .select({ entity_ref: 'refresh_state.entity_ref', hash: 'final_entities.hash', - next_stitch_at: 'refresh_state.next_stitch_at', + next_stitch_at: 'stitch_queue.next_stitch_at', }); } @@ -178,19 +183,11 @@ describe('deleteOrphanedEntities', () => { await insertRelation(knex, 'E7', 'E6'); await expect(run(knex, { mode: 'immediate' })).resolves.toEqual(5); await expect(refreshState(knex)).resolves.toEqual([ - { entity_ref: 'E1', result_hash: 'original', next_stitch_at: null }, - { - entity_ref: 'E2', - result_hash: 'force-stitching', - next_stitch_at: null, - }, - { - entity_ref: 'E7', - result_hash: 'force-stitching', - next_stitch_at: null, - }, - { entity_ref: 'E8', result_hash: 'original', next_stitch_at: null }, - { entity_ref: 'E9', result_hash: 'original', next_stitch_at: null }, + { entity_ref: 'E1', result_hash: 'original' }, + { entity_ref: 'E2', result_hash: 'force-stitching' }, + { entity_ref: 'E7', result_hash: 'force-stitching' }, + { entity_ref: 'E8', result_hash: 'original' }, + { entity_ref: 'E9', result_hash: 'original' }, ]); await expect(finalEntities(knex)).resolves.toEqual([ { entity_ref: 'E1', hash: 'original', next_stitch_at: null }, @@ -276,19 +273,11 @@ describe('deleteOrphanedEntities', () => { }), ).resolves.toEqual(5); await expect(refreshState(knex)).resolves.toEqual([ - { entity_ref: 'E1', result_hash: 'original', next_stitch_at: null }, - { - entity_ref: 'E2', - result_hash: 'original', - next_stitch_at: expect.anything(), - }, - { - entity_ref: 'E7', - result_hash: 'original', - next_stitch_at: expect.anything(), - }, - { entity_ref: 'E8', result_hash: 'original', next_stitch_at: null }, - { entity_ref: 'E9', result_hash: 'original', next_stitch_at: null }, + { entity_ref: 'E1', result_hash: 'original' }, + { entity_ref: 'E2', result_hash: 'original' }, + { entity_ref: 'E7', result_hash: 'original' }, + { entity_ref: 'E8', result_hash: 'original' }, + { entity_ref: 'E9', result_hash: 'original' }, ]); 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 33e6250413..cb4b6deff4 100644 --- a/plugins/catalog-backend/src/database/tables.ts +++ b/plugins/catalog-backend/src/database/tables.ts @@ -81,47 +81,6 @@ export type DbRefreshStateRow = { * continuously gets moved forward as items are picked up for processing. */ next_update_at: string | Date; - /** - * If a stitch has been requested, this is the point in time that that last - * happened. - * - * @remarks - * - * Each time that a request is made, this timestamp is updated to the current - * time, overwriting the previous value if applicable. - * - * When the stitch loop runs and picks up an entity, this timestamp is not - * immediately reset. It's instead moved forward in time by a certain amount, - * which means that if the stitcher for some reason fails (eg if the process - * crashes or gets shut down), the entity will be picked up again in the - * 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 - * request has been made behind our backs), does the timestamp (and the - * ticket) get reset. - */ - next_stitch_at?: string | Date | null; - /** - * If a stitch has been requested, this is the unique ticket that was chosen - * to mark the last request. - * - * @remarks - * - * Each time that a request is made, a new random ticket is chosen, - * overwriting the previous value if applicable. - * - * When the stitch loop runs and picks up an entity, this column is left - * unchanged. This means that if the stitcher for some reason fails (eg if the - * process crashes or gets shut down), the entity will be picked up again in - * the 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 - * request has been made behind our backs), does the ticket (and the - * timestamp) get reset. - */ - next_stitch_ticket?: string | null; /** * The last time that this entity was emitted by somebody (the entity provider * or a parent entity). @@ -182,6 +141,46 @@ export type DbFinalEntitiesRow = { entity_ref: string; }; +/** + * Represents the stitch_queue table. + * + * @remarks + * + * Each row represents a pending stitch request for an entity. When a stitch + * completes successfully and the ticket hasn't changed, the row is deleted. + */ +export type DbStitchQueueRow = { + /** + * The entity ref that needs stitching (primary key). + */ + 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. + */ + stitch_ticket: string; + /** + * The point in time when this entity should next be stitched. + * + * @remarks + * + * Each time that a request is made, this timestamp is updated to the current + * time, overwriting the previous value if applicable. + * + * When the stitch loop runs and picks up an entity, this timestamp is not + * immediately reset. It's instead moved forward in time by a certain amount, + * which means that if the stitcher for some reason fails (eg if the process + * crashes or gets shut down), the entity will be picked up again in the + * 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 + * request has been made behind our backs), does the row get deleted. + */ + next_stitch_at: string | Date; +}; + export type DbSearchRow = { entity_id: string; key: string; diff --git a/plugins/catalog-backend/src/stitching/progressTracker.ts b/plugins/catalog-backend/src/stitching/progressTracker.ts index f460a17a51..31a405d9cd 100644 --- a/plugins/catalog-backend/src/stitching/progressTracker.ts +++ b/plugins/catalog-backend/src/stitching/progressTracker.ts @@ -17,7 +17,7 @@ import { stringifyError } from '@backstage/errors'; import { Knex } from 'knex'; import { DateTime } from 'luxon'; -import { DbRefreshStateRow } from '../database/tables'; +import { DbStitchQueueRow } from '../database/tables'; import { createCounterMetric } from '../util/metrics'; import { LoggerService } from '@backstage/backend-plugin-api'; import { MetricsService } from '@backstage/backend-plugin-api/alpha'; @@ -54,9 +54,9 @@ export function progressTracker( { description: 'Number of entities currently in the stitching queue' }, ); stitchingQueueCount.addCallback(async result => { - const total = await knex('refresh_state') - .count({ count: '*' }) - .whereNotNull('next_stitch_at'); + const total = await knex('stitch_queue').count({ + count: '*', + }); result.observe(Number(total[0].count)); }); diff --git a/plugins/catalog-backend/src/tests/migrations.test.ts b/plugins/catalog-backend/src/tests/migrations.test.ts index faf7351135..3cb3cc8fc3 100644 --- a/plugins/catalog-backend/src/tests/migrations.test.ts +++ b/plugins/catalog-backend/src/tests/migrations.test.ts @@ -917,4 +917,129 @@ describe('migrations', () => { await knex.destroy(); }, ); + + it.each(databases.eachSupportedId())( + '20260215000000_move_stitch_queue.js, %p', + async databaseId => { + const knex = await databases.init(databaseId); + + // Run migrations up to just before the target migration + await migrateUntilBefore(knex, '20260215000000_move_stitch_queue.js'); + + // Insert rows into refresh_state with stitch queue data + // Before migration, refresh_state has next_stitch_at and next_stitch_ticket columns + const stitchTime1 = new Date('2026-01-15T12:00:00.000Z'); + const stitchTime3 = new Date('2026-01-16T12:00:00.000Z'); + + await knex('refresh_state').insert([ + { + entity_id: 'id1', + entity_ref: 'component:default/with-stitch', + unprocessed_entity: '{}', + processed_entity: '{}', + errors: '[]', + next_update_at: knex.fn.now(), + last_discovery_at: knex.fn.now(), + next_stitch_at: stitchTime1, + next_stitch_ticket: 'ticket-1', + }, + { + entity_id: 'id2', + entity_ref: 'component:default/no-stitch', + unprocessed_entity: '{}', + processed_entity: '{}', + errors: '[]', + next_update_at: knex.fn.now(), + last_discovery_at: knex.fn.now(), + next_stitch_at: null, + next_stitch_ticket: null, + }, + { + entity_id: 'id3', + entity_ref: 'component:default/orphan-stitch', + unprocessed_entity: '{}', + processed_entity: '{}', + errors: '[]', + next_update_at: knex.fn.now(), + last_discovery_at: knex.fn.now(), + next_stitch_at: stitchTime3, + next_stitch_ticket: 'ticket-3', + }, + ]); + + // Verify initial state - stitch_queue table should NOT exist yet + const preTableExists = await knex.schema.hasTable('stitch_queue'); + expect(preTableExists).toBe(false); + + // Run the migration + await migrateUpOnce(knex); + + // Verify stitch_queue table was created + const postTableExists = await knex.schema.hasTable('stitch_queue'); + expect(postTableExists).toBe(true); + + // Verify next_stitch_at and next_stitch_ticket columns were removed from refresh_state + const refreshStateColumnInfo = await knex('refresh_state').columnInfo(); + expect(refreshStateColumnInfo.next_stitch_at).toBeUndefined(); + expect(refreshStateColumnInfo.next_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'); + + // 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', + next_stitch_at: expect.anything(), + }, + { + entity_ref: 'component:default/with-stitch', + stitch_ticket: 'ticket-1', + next_stitch_at: expect.anything(), + }, + ]); + + // Run the down migration + await migrateDownOnce(knex); + + // Verify stitch_queue table was dropped + const revertedTableExists = await knex.schema.hasTable('stitch_queue'); + expect(revertedTableExists).toBe(false); + + // Verify next_stitch_at and next_stitch_ticket columns were restored to refresh_state + const revertedRefreshColumnInfo = await knex( + 'refresh_state', + ).columnInfo(); + expect(revertedRefreshColumnInfo.next_stitch_at).not.toBeUndefined(); + expect(revertedRefreshColumnInfo.next_stitch_ticket).not.toBeUndefined(); + + // Verify data was migrated back to refresh_state + const refreshStateAfterDown = await knex('refresh_state') + .orderBy('entity_id') + .select( + 'entity_id', + 'entity_ref', + 'next_stitch_at', + 'next_stitch_ticket', + ); + + // id1 should have stitch data restored + const id1RefreshRow = refreshStateAfterDown.find( + r => r.entity_id === 'id1', + ); + expect(id1RefreshRow?.next_stitch_at).not.toBeNull(); + expect(id1RefreshRow?.next_stitch_ticket).toBe('ticket-1'); + + // id2 should have no stitch data + const id2RefreshRow = refreshStateAfterDown.find( + r => r.entity_id === 'id2', + ); + expect(id2RefreshRow?.next_stitch_at).toBeNull(); + + await knex.destroy(); + }, + ); }); 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 2/3] 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', From 874456a2493bfcaa264e54517c903ee17df211e1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 6 Mar 2026 11:37:01 +0100 Subject: [PATCH 3/3] catalog: simplify stitch_queue to single ticket, fix down migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove unused active_ticket column and rename latest_ticket back to stitch_ticket. Use knex.batchInsert in the up migration. Fix the down migration to restore stitch_ticket on final_entities without a server-side default, matching the original schema exactly. 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 | 5 +-- .../getDeferredStitchableEntities.test.ts | 6 +-- .../stitcher/getDeferredStitchableEntities.ts | 7 ++-- .../markDeferredStitchCompleted.test.ts | 6 +-- .../stitcher/markDeferredStitchCompleted.ts | 2 +- .../stitcher/markForStitching.test.ts | 32 +++++++------- .../operations/stitcher/markForStitching.ts | 8 ++-- .../stitcher/performStitching.test.ts | 12 +++--- .../operations/stitcher/performStitching.ts | 4 +- .../util/deleteOrphanedEntities.test.ts | 11 +++++ .../catalog-backend/src/database/tables.ts | 14 ++----- .../providers/DefaultLocationStore.test.ts | 1 - .../GenericScmEventRefreshProvider.test.ts | 1 - .../src/tests/migrations.test.ts | 6 +-- 16 files changed, 81 insertions(+), 78 deletions(-) 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(), }, ]);