From 6713e9fe7234359df469e3e32e17fc8837afe27e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= <freben@gmail.com>
Date: Wed, 20 May 2026 00:09:16 +0200
Subject: [PATCH] Apply same transaction fix to getProcessableEntities
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Same FOR UPDATE SKIP LOCKED race: the processing loop's claim query
had the SELECT and UPDATE as separate auto-committed statements.
The existing caller already wraps in a transaction, but this makes
the function self-protecting if called without one.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Fredrik Adelöw <freben@gmail.com>
---
 .../src/database/DefaultProcessingDatabase.ts | 90 +++++++++++--------
 1 file changed, 53 insertions(+), 37 deletions(-)

diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts
index 264c11055d..540e7fea26 100644
--- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts
+++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts
@@ -207,49 +207,65 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     request: { processBatchSize: number },
   ): Promise<GetProcessableEntitiesResult> {
     const knex = maybeTx as Knex.Transaction | Knex;
-
-    let itemsQuery = knex<DbRefreshStateRow>('refresh_state').select([
-      'entity_id',
-      'entity_ref',
-      'unprocessed_entity',
-      'result_hash',
-      'cache',
-      'errors',
-      'location_key',
-      'next_update_at',
-    ]);
-
-    // This avoids duplication of work because of race conditions and is
-    // also fast because locked rows are ignored rather than blocking.
-    // It's only available in MySQL and PostgreSQL
-    if (['mysql', 'mysql2', 'pg'].includes(knex.client.config.client)) {
-      itemsQuery = itemsQuery.forUpdate().skipLocked();
-    }
-
-    const items = await itemsQuery
-      .where('next_update_at', '<=', knex.fn.now())
-      .limit(request.processBatchSize)
-      .orderBy('next_update_at', 'asc');
+    const useLocking = ['mysql', 'mysql2', 'pg'].includes(
+      knex.client.config.client,
+    );
 
     const interval = this.options.refreshInterval();
 
-    const nextUpdateAt = (refreshInterval: number) => {
-      if (knex.client.config.client.includes('sqlite3')) {
-        return knex.raw(`datetime('now', ?)`, [`${refreshInterval} seconds`]);
-      } else if (knex.client.config.client.includes('mysql')) {
-        return knex.raw(`now() + interval ${refreshInterval} second`);
+    const nextUpdateAt = (
+      tx: Knex | Knex.Transaction,
+      refreshInterval: number,
+    ) => {
+      if (tx.client.config.client.includes('sqlite3')) {
+        return tx.raw(`datetime('now', ?)`, [`${refreshInterval} seconds`]);
+      } else if (tx.client.config.client.includes('mysql')) {
+        return tx.raw(`now() + interval ${refreshInterval} second`);
       }
-      return knex.raw(`now() + interval '${refreshInterval} seconds'`);
+      return tx.raw(`now() + interval '${refreshInterval} seconds'`);
     };
 
-    await knex<DbRefreshStateRow>('refresh_state')
-      .whereIn(
-        'entity_ref',
-        items.map(i => i.entity_ref),
-      )
-      .update({
-        next_update_at: nextUpdateAt(interval),
-      });
+    // The SELECT FOR UPDATE SKIP LOCKED + UPDATE must run inside a
+    // single transaction so that the row locks persist until
+    // next_update_at has been bumped.
+    const run = async (tx: Knex | Knex.Transaction) => {
+      const items = await tx<DbRefreshStateRow>('refresh_state')
+        .select([
+          'entity_id',
+          'entity_ref',
+          'unprocessed_entity',
+          'result_hash',
+          'cache',
+          'errors',
+          'location_key',
+          'next_update_at',
+        ])
+        .where('next_update_at', '<=', tx.fn.now())
+        .limit(request.processBatchSize)
+        .orderBy('next_update_at', 'asc')
+        .modify(qb => {
+          if (useLocking) {
+            qb.forUpdate().skipLocked();
+          }
+        });
+
+      if (items.length > 0) {
+        await tx<DbRefreshStateRow>('refresh_state')
+          .whereIn(
+            'entity_ref',
+            items.map(i => i.entity_ref),
+          )
+          .update({
+            next_update_at: nextUpdateAt(tx, interval),
+          });
+      }
+
+      return items;
+    };
+
+    const items = knex.isTransaction
+      ? await run(knex)
+      : await knex.transaction(run);
 
     return {
       items: items.map(