fix(catalog-backend): use hash check for stitch ticket guard instead of unreliable row counts

INSERT return values differ across database engines (row IDs, row
counts, or empty arrays), making them unsuitable for detecting blocked
upserts. Use a post-write hash verification instead, which works
reliably on all engines.

Adds test coverage for stale ticket rejection across all supported
database engines (MySQL 8, PostgreSQL 14/18, SQLite 3).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Fredrik Adelöw <freben@gmail.com>

plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts

@@ -363,5 +363,98 @@ describe.each(databases.eachSupportedId())(
       expect(entities[0].hash).not.toBe('');
       expect(entities[0].final_entity).toBeDefined();
     });
+
+    it('rejects a stale stitch ticket without overwriting', async () => {
+      const knex = await databases.init(databaseId);
+      await applyDatabaseMigrations(knex);
+
+      await knex<DbRefreshStateRow>('refresh_state').insert([
+        {
+          entity_id: 'my-id',
+          entity_ref: 'k:ns/n',
+          unprocessed_entity: JSON.stringify({}),
+          processed_entity: JSON.stringify({
+            apiVersion: 'a',
+            kind: 'k',
+            metadata: { name: 'n', namespace: 'ns' },
+            spec: { original: true },
+          }),
+          errors: '[]',
+          next_update_at: knex.fn.now(),
+          last_discovery_at: knex.fn.now(),
+        },
+      ]);
+
+      // First stitch: create the final_entities row with a valid ticket
+      await markForStitching({ knex, entityRefs: ['k:ns/n'] });
+      const validTicket = (
+        await knex('stitch_queue')
+          .where('entity_ref', 'k:ns/n')
+          .select('stitch_ticket')
+          .first()
+      )?.stitch_ticket;
+
+      const result1 = await performStitching({
+        knex,
+        logger: mockServices.logger.mock(),
+        entityRef: 'k:ns/n',
+        stitchTicket: validTicket,
+      });
+      expect(result1).toBe('changed');
+
+      const afterFirst = await knex<DbFinalEntitiesRow>('final_entities');
+      expect(afterFirst).toHaveLength(1);
+      const firstHash = afterFirst[0].hash;
+
+      // Now change the processed entity and mark for stitching again
+      await knex<DbRefreshStateRow>('refresh_state')
+        .where('entity_id', 'my-id')
+        .update({
+          processed_entity: JSON.stringify({
+            apiVersion: 'a',
+            kind: 'k',
+            metadata: { name: 'n', namespace: 'ns' },
+            spec: { original: false, stale: true },
+          }),
+        });
+
+      await markForStitching({ knex, entityRefs: ['k:ns/n'] });
+      const freshTicket = (
+        await knex('stitch_queue')
+          .where('entity_ref', 'k:ns/n')
+          .select('stitch_ticket')
+          .first()
+      )?.stitch_ticket;
+
+      // Attempt to stitch with a WRONG ticket (simulating a stale worker)
+      const result2 = await performStitching({
+        knex,
+        logger: mockServices.logger.mock(),
+        entityRef: 'k:ns/n',
+        stitchTicket: 'stale-ticket-that-does-not-match',
+      });
+      expect(result2).toBe('abandoned');
+
+      // The final_entities row should still have the FIRST hash,
+      // not be overwritten by the stale worker
+      const afterStale = await knex<DbFinalEntitiesRow>('final_entities');
+      expect(afterStale).toHaveLength(1);
+      expect(afterStale[0].hash).toBe(firstHash);
+
+      // Now stitch with the correct fresh ticket — should succeed
+      const result3 = await performStitching({
+        knex,
+        logger: mockServices.logger.mock(),
+        entityRef: 'k:ns/n',
+        stitchTicket: freshTicket,
+      });
+      expect(result3).toBe('changed');
+
+      const afterFresh = await knex<DbFinalEntitiesRow>('final_entities');
+      expect(afterFresh).toHaveLength(1);
+      expect(afterFresh[0].hash).not.toBe(firstHash);
+      const freshEntity = JSON.parse(afterFresh[0].final_entity!);
+      expect(freshEntity.spec).toEqual({ original: false, stale: true });
+    });
   },
 );

plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts

@@ -196,14 +196,15 @@ export async function performStitching(options: {
     // to write the search index.
     const searchEntries = buildEntitySearch(entityId, entity);
 
-    const isMySQL = String(knex.client.config.client).includes('mysql');
-
     // Guard against concurrent stitchers: if our stitch_ticket no longer
     // matches stitch_queue, another worker has newer data and we should
     // not overwrite it. On PostgreSQL and SQLite, this is done atomically
-    // via a WHERE on the upsert merge path. MySQL does not support that
-    // syntax, so it falls back to a separate check with a negligible
-    // TOCTOU window (self-corrects on the next ~1s stitch cycle).
+    // via a WHERE on the upsert merge path. MySQL does not support
+    // ON CONFLICT ... DO UPDATE ... WHERE, so it uses a separate check
+    // with a negligible TOCTOU window (self-corrects on the next ~1s
+    // stitch cycle).
+    const isMySQL = String(knex.client.config.client).includes('mysql');
+
     if (stitchTicket && isMySQL) {
       const ticketValid = await knex<DbStitchQueueRow>('stitch_queue')
         .where('entity_ref', entityRef)
@@ -237,20 +238,19 @@ export async function performStitching(options: {
       );
     }
 
-    const rowsAffected = await upsert;
+    await upsert;
 
-    // PostgreSQL correctly reports 0 rows when the WHERE blocks the
-    // merge. SQLite always reports 1, so fall back to a hash check.
+    // Verify the write took effect. INSERT return values vary across
+    // database engines (row IDs vs row counts vs empty arrays), so we
+    // check the hash directly — we already know hash !== previousHash
+    // from the check above, so a mismatch means the write was blocked.
     if (stitchTicket && !isMySQL) {
-      const isSQLite = String(knex.client.config.client).includes('sqlite');
-      const blocked = isSQLite
-        ? !(await knex<DbFinalEntitiesRow>('final_entities')
-            .where('entity_id', entityId)
-            .where('hash', hash)
-            .select(knex.raw('1'))
-            .first())
-        : rowsAffected[0] === 0;
-      if (blocked) {
+      const written = await knex<DbFinalEntitiesRow>('final_entities')
+        .where('entity_id', entityId)
+        .where('hash', hash)
+        .select(knex.raw('1'))
+        .first();
+      if (!written) {
         logger.debug(
           `Entity ${entityRef} is already stitched, skipping write.`,
         );