paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 46 Packages Projects Releases Wiki Activity

catalog: move the stitch queue concerns into a dedicated stitch_queue table

Browse Source 
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Fredrik Adelöw <freben@spotify.com>
This commit is contained in:
Fredrik Adelöw
2026-02-22 16:51:21 +01:00
committed by Fredrik Adelöw
parent 182cd749d4
commit 7416e8b0c0
13 changed files with 507 additions and 237 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/gentle-tables-march.md
+5
View File
@@ -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.
plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js
+150
View File
@@ -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');
};
plugins/catalog-backend/report.sql.md
+13 -3
View File
@@ -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
plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts
+36 -60
View File
@@ -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<DbStitchQueueRow>('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<DbStitchQueueRow>('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<DbStitchQueueRow>('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!));
},
);
});
plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts
+7 -11
View File
@@ -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<DbRefreshStateRow>('refresh_state').select(
let itemsQuery = knex<DbStitchQueueRow>('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<DbRefreshStateRow>('refresh_state')
await knex<DbStitchQueueRow>('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),
}));
}
plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts
+15 -15
View File
@@ -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<DbRefreshStateRow>('refresh_state').insert([
// Insert stitch_queue row
await knex<DbStitchQueueRow>('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<DbRefreshStateRow>('refresh_state').select(
return knex<DbStitchQueueRow>('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([]);
},
);
});
plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts
+8 -11
View File
@@ -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<void> {
const { knex, entityRef, stitchTicket } = option;
await knex<DbRefreshStateRow>('refresh_state')
.update({
next_stitch_at: null,
next_stitch_ticket: null,
})
await knex<DbStitchQueueRow>('stitch_queue')
.where('entity_ref', '=', entityRef)
.andWhere('next_stitch_ticket', '=', stitchTicket);
.andWhere('stitch_ticket', '=', stitchTicket)
.delete();
}
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts
+53 -50
View File
@@ -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<DbStitchQueueRow>('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<DbRefreshStateRow>('refresh_state')
.select('entity_id', 'next_stitch_at', 'next_stitch_ticket')
.orderBy('entity_id', 'asc');
return knex<DbStitchQueueRow>('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<DbRefreshStateRow>('refresh_state')
.select('entity_ref', 'next_stitch_at', 'next_stitch_ticket')
.whereNotNull('next_stitch_at')
const finalState = await knex<DbStitchQueueRow>('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();
});
},
);
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts
+34 -14
View File
@@ -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<DbRefreshStateRow>('refresh_state')
.update({
next_stitch_at: knex.fn.now(),
next_stitch_ticket: ticket,
})
.whereIn('entity_ref', chunk);
if (chunk.length > 0) {
await knex<DbStitchQueueRow>('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<DbRefreshStateRow>('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<DbRefreshStateRow>('refresh_state')
.select('entity_ref')
.whereIn('entity_id', chunk);
if (refreshStateRows.length > 0) {
await knex<DbStitchQueueRow>('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 {
plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts
+17 -28
View File
@@ -100,7 +100,7 @@ describe('deleteOrphanedEntities', () => {
async function refreshState(knex: Knex) {
return await knex<DbRefreshStateRow>('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 },
plugins/catalog-backend/src/database/tables.ts
+40 -41
View File
@@ -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;
plugins/catalog-backend/src/stitching/progressTracker.ts
+4 -4
View File
@@ -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<DbRefreshStateRow>('refresh_state')
.count({ count: '*' })
.whereNotNull('next_stitch_at');
const total = await knex<DbStitchQueueRow>('stitch_queue').count({
count: '*',
});
result.observe(Number(total[0].count));
});
plugins/catalog-backend/src/tests/migrations.test.ts
+125
View File
@@ -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();
},
);
});