Twins by content: at most 500 twins read per pair

With the hash typed as the column (f537add) the planner walks every twin of a
pair by index; 55 pairs have over 500 twins, up to 131k, and a flush with such
a pair ran the query for 37 s inside its transaction, the other flushes and
node-api's class counters waiting on it. A LATERAL per pair now reads its
lowest 500 release ids and takes the first named one: 49 ms for 305 pairs with
the five hottest among them. LOG.md (138).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

REBUILD.md

@@ -950,9 +950,12 @@ f. **Nothing enters the catalog without a usable name** (amended
    climb, each by an index (`pending_releases_signature_idx` for a
    poster's rows: tested on the rows, a signature was a scan of the whole
    table; LOG.md (133), (134)). The batch store, the record, the claim
-   and the sweep lock pending rows in one order (`lockOrder`), and a
-   flush that deadlocks anyway keeps its outcomes for the next (LOG.md
-   (134)). The namer
+   and the sweep lock pending rows in one order, by key and the counters
+   after (`lockOrder` for the multi-row statements; the record and the
+   sweep lock their rows with `SELECT … ORDER BY release_key FOR UPDATE`
+   first, since an UPDATE's join or `= ANY(subquery)` locks in plan
+   order), and a flush that deadlocks anyway keeps its outcomes for the
+   next (LOG.md (134), (137)). The namer
    (`app/services/pending_namer.ts`, in the prober's process): a named
    release's article span read again from the corpus (the stored range's
    chunks) or by XOVER on a connection of its own, a chunk of articles

app/services/hash_twins.ts

@@ -5,20 +5,29 @@ import db from '@adonisjs/lucid/services/db'
  * repeated verdict match) in a catalog release that has a name, readable or found inside. The same film posted
  * under scrambles by other tools gets that name without another fetch. Releases with a junk verdict don't count.
  */
+/** Twins read per pair: the hottest pairs have 131k. */
+const TWIN_SCAN = 500
+
 export async function twinNames(
   pairs: { hash16k: string; size: number }[],
   trx: { rawQuery: typeof db.rawQuery } = db
 ): Promise<Map<string, string>> {
   const out = new Map<string, string>()
   if (pairs.length === 0) return out
+  // Per pair the first named twin among its lowest TWIN_SCAN release ids. hash16k is typed as the column
+  // (char(32)): as text the join cast the column and scanned all of file_hashes, 4 s a call; and with the index
+  // the hottest pairs (55 with over 500 twins, up to 131k, all readable) made a call walk every twin, 37 s inside
+  // the flush's transaction (LOG.md (138)). One name is all a pair needs.
   const { rows } = await trx.rawQuery(
-    `SELECT DISTINCT ON (h.hash16k, h.size) h.hash16k, h.size, coalesce(r.inner_name, r.name) AS name
-     FROM jsonb_to_recordset(?::jsonb) AS v(hash16k text, size bigint)
-     JOIN file_hashes h ON h.hash16k = v.hash16k AND h.size = v.size
-     JOIN releases r ON r.id = h.release_id AND (r.readable OR r.inner_name IS NOT NULL)
-     WHERE NOT EXISTS (SELECT 1 FROM release_verdicts x WHERE x.release_id = r.id)
-     ORDER BY h.hash16k, h.size, r.readable DESC, r.id`,
-    [JSON.stringify(pairs)]
+    `SELECT v.hash16k, v.size, t.name
+     FROM jsonb_to_recordset(?::jsonb) AS v(hash16k char(32), size bigint)
+     CROSS JOIN LATERAL (
+       SELECT coalesce(r.inner_name, r.name) AS name
+       FROM (SELECT release_id FROM file_hashes h WHERE h.hash16k = v.hash16k AND h.size = v.size ORDER BY release_id LIMIT ?) h
+       JOIN releases r ON r.id = h.release_id AND (r.readable OR r.inner_name IS NOT NULL)
+       WHERE NOT EXISTS (SELECT 1 FROM release_verdicts x WHERE x.release_id = r.id)
+       ORDER BY r.readable DESC, r.id LIMIT 1) t`,
+    [JSON.stringify(pairs), TWIN_SCAN]
   )
   for (const r of rows) out.set(`${r.hash16k}\t${r.size}`, r.name)
   return out

app/services/pending_ladder.ts

@@ -289,6 +289,12 @@ class PendingLadder {
         final: null,
       }
     })
+    // The rows locked in key order first: the UPDATE's join below takes them in whatever order the plan does.
+    await trx.rawQuery(
+      `SELECT 1 FROM pending_releases WHERE release_key = ANY(SELECT decode(k, 'hex') FROM unnest(?::text[]) AS k)
+       ORDER BY release_key FOR UPDATE`,
+      [rows.map((r) => r.release_key).sort()]
+    )
     await trx.rawQuery(
       `UPDATE pending_releases p SET status = v.status, tries = v.tries, named = coalesce(v.named, p.named),
          inspections = CASE WHEN v.inspection IS NULL THEN p.inspections ELSE p.inspections || v.inspection END,
@@ -470,9 +476,15 @@ class PendingLadder {
       .filter((key: string) => !explored(key, samplePercent))
       .sort()
     if (keys.length === 0) return 0
+    // Locked in key order (the record's lock order, LOG.md (137)): a plain `key = ANY(subquery)` locks in scan
+    // order and deadlocked against a batch upsert; ORDER BY sorts before FOR UPDATE locks.
     const { rows: gone } = await db.rawQuery(
-      `UPDATE pending_releases SET status = ? WHERE status = ? AND release_key = ANY(SELECT decode(k, 'hex') FROM unnest(?::text[]) AS k)
-       RETURNING tool, group_id, klass`,
+      `UPDATE pending_releases p SET status = ?
+       FROM (SELECT release_key FROM pending_releases
+             WHERE status = ? AND release_key = ANY(SELECT decode(k, 'hex') FROM unnest(?::text[]) AS k)
+             ORDER BY release_key FOR UPDATE) k
+       WHERE p.release_key = k.release_key
+       RETURNING p.tool, p.group_id, p.klass`,
       [PENDING.discarded, PENDING.waiting, keys]
     )
     const counts = new Map<string, { tool: string; group_id: number; klass: number; n: number }>()