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

Merge pull request #33158 from backstage/freben/move-stitch-queue-2

Browse Source 
catalog: move stitch queue into dedicated table
This commit is contained in:
Fredrik Adelöw
2026-03-10 16:21:13 +01:00
committed by GitHub
parent 294f9fe17f 874456a249
commit 2fe57aa333
20 changed files with 640 additions and 301 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 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.
plugins/catalog-backend/migrations/20260215000000_move_stitch_queue.js
+174
View File
@@ -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');
};
plugins/catalog-backend/report.sql.md
+13 -4
View File
@@ -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
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 -54
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');
},
);
@@ -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<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/stitcher/performStitching.test.ts
+51 -41
View File
@@ -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<DbFinalEntitiesRow>('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<DbFinalEntitiesRow>('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<DbFinalEntitiesRow>('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<DbFinalEntitiesRow>('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<DbFinalEntitiesRow>('final_entities').select([
'entity_id',
'stitch_ticket',
]),
).resolves.toEqual([
{
entity_id: 'my-id',
stitch_ticket: expect.not.stringContaining('old-ticket'),
},
]);
const entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
expect(entities[0].hash).not.toBe('');
expect(entities[0].final_entity).toBeDefined();
},
);
});
plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts
+20 -9
View File
@@ -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<DbFinalEntitiesRow>('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<DbFinalEntitiesRow>('final_entities')
let updateQuery = knex<DbFinalEntitiesRow>('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<DbStitchQueueRow>('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,
plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts
+28 -29
View File
@@ -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<DbRefreshStateRow>('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 },
plugins/catalog-backend/src/database/tables.ts
+41 -42
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).
@@ -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;
plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts
-1
View File
@@ -271,7 +271,6 @@ describe('DefaultLocationStore', () => {
final_entity: '{}',
hash: 'hash',
last_updated_at: new Date(),
stitch_ticket: '',
entity_ref: 'k:ns/n',
});
plugins/catalog-backend/src/providers/GenericScmEventRefreshProvider.test.ts
-1
View File
@@ -86,7 +86,6 @@ describe('GenericScmEventRefreshProvider', () => {
entity_id: id,
entity_ref: `k:ns/${id}`,
hash: 'h',
stitch_ticket: '',
final_entity: '{}',
});
}
plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts
-2
View File
@@ -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)) {
plugins/catalog-backend/src/service/request/applyEntityFilterToQuery.test.ts
-1
View File
@@ -98,7 +98,6 @@ describe.each(databases.eachSupportedId())(
entity_ref: entityRef,
final_entity: entityJson,
hash: 'h',
stitch_ticket: '',
});
const search = await buildEntitySearch(id, entity);
plugins/catalog-backend/src/service/request/applyPredicateEntityFilterToQuery.test.ts
-1
View File
@@ -105,7 +105,6 @@ describe.each(databases.eachSupportedId())(
entity_ref: entityRef,
final_entity: entityJson,
hash: 'h',
stitch_ticket: '',
});
const search = await buildEntitySearch(id, entity);
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
+151 -1
View File
@@ -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();
},
);
});