diff --git a/.changeset/gentle-tables-march.md b/.changeset/gentle-tables-march.md new file mode 100644 index 0000000000..5f74be9a24 --- /dev/null +++ b/.changeset/gentle-tables-march.md @@ -0,0 +1,5 @@ +--- +'@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_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 new file mode 100644 index 0000000000..7dd11db78b --- /dev/null +++ b/plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js @@ -0,0 +1,174 @@ +/* + * 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('Changes with every new stitch request'); + 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'); + + 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) { + 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'); + }); + } + + // 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'); + }); + } +}; + +/** + * @param {import('knex').Knex} knex + */ +exports.down = async function down(knex) { + const isSQLite = knex.client.config.client.includes('sqlite'); + + // Step 1: Add back stitch_ticket to final_entities + await knex.schema.alterTable('final_entities', table => { + table + .text('stitch_ticket') + .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 => { + 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 2b: 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 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', + latestTicket: '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.latestTicket, + }) + .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 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 4f76d2a3f5..dd5892b7fb 100644 --- a/plugins/catalog-backend/report.sql.md +++ b/plugins/catalog-backend/report.sql.md @@ -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 @@ -76,8 +75,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 +84,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 +131,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..f7fd9a0695 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'); }, ); @@ -254,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', }, ]); @@ -461,8 +463,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 +558,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/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index 985146229a..41931a8b7e 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('stitch_ticket') + .first() + )?.stitch_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('stitch_ticket') + .first() + )?.stitch_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('stitch_ticket') + .first() + )?.stitch_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..95fe880eda 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 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.stitch_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 0f4f5f0c3d..c61c425a30 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: '', }); } } @@ -100,7 +99,13 @@ 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 stitchQueue(knex: Knex) { + return await knex('stitch_queue') + .orderBy('entity_ref') + .select('entity_ref'); } async function finalEntities(knex: Knex) { @@ -110,11 +115,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,20 +188,13 @@ 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(stitchQueue(knex)).resolves.toEqual([]); await expect(finalEntities(knex)).resolves.toEqual([ { entity_ref: 'E1', hash: 'original', next_stitch_at: null }, { @@ -276,19 +279,15 @@ 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(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 33e6250413..43068c8df8 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). @@ -176,12 +135,52 @@ 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; }; +/** + * 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 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 (meaning no new request came in while + * stitching was in progress). + */ + 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/providers/DefaultLocationStore.test.ts b/plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts index d1cd6e7c5e..4f32c58816 100644 --- a/plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts +++ b/plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts @@ -271,7 +271,6 @@ 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..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', - 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/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..4a469ab07e 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', }) @@ -917,4 +916,155 @@ 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', + }, + ]); + + // 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); + + // 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 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'); + + // 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 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', + ).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(); + }, + ); });