REBUILD.md: a pending release without a target is exhausted at once

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

REBUILD.md

@@ -932,7 +932,10 @@ f. **Nothing enters the catalog without a usable name** (amended
    11% of the segments and nothing anyone can find; the full history has
    725 M random-named files (LOG.md 2026-10-09). Built 2026-10-26: the
    edge sends an unnamed release with an empty payload and its tool, the
-   batch store writes `pending_releases` (LOG.md (114)); the ladder in the
+   batch store writes `pending_releases` (LOG.md (114)); a finished release
+   the edge found no target for (the edge's `candidates_of`: par2, archive
+   start, 7z tail, else the first random-named file as the lone target;
+   LOG.md (140)) is written exhausted at once; the ladder in the
    prober (`app/services/pending_ladder.ts`, LOG.md (115)): a claim takes
    the oldest waiting releases' next target (par2, archive, lone, 7z-tail
    in that order), at least PENDING_SHARE (25%) of each claim next to the
@@ -950,9 +953,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/batch_store.ts

@@ -9,7 +9,7 @@ import { SIDECAR } from '#services/sidecar_folder'
 import { identityKey, releaseKey } from '#lib/release/keys'
 import { sourceCode } from '#lib/release/source'
 import { uploadTool } from '#lib/rules/index'
-import { byEffort, lockOrder } from '#services/pending_ladder'
+import { PENDING, byEffort, lockOrder } from '#services/pending_ladder'
 import { TEXT_KINDS, TEXT_MAX, textsToStore } from '#lib/release/text_kind'
 
 /**
@@ -213,6 +213,9 @@ class BatchStore {
           file_total: r.fileTotal,
           tool: r.tool ?? (probes[0] ? uploadTool(probes[0].messageId) : 'unknown'),
           candidates: probes,
+          // A finished release without a target is through the ladder before it starts: exhausted, not waiting
+          // (1.57M rows waited forever, never claimed, before the edge gave the hash-named pieces one; LOG.md (140)).
+          status: r.finished !== false && probes.length === 0 ? PENDING.exhausted : PENDING.waiting,
           // The first target's article (by effort): the cheapest probe first within a class (LOG.md (130)).
           target_bytes: byEffort(probes)[0]?.bytes ?? null,
         }
@@ -224,29 +227,32 @@ class BatchStore {
     for (let i = 0; i < rows.length; i += 1000) {
       const { rows: done } = await trx.rawQuery(
         `INSERT INTO pending_releases (release_key, identity_key, group_id, name, poster, posted_at, last_posted_at, first_article, last_article,
-                                       files, segments, bytes, partial, finished, file_total, tool, candidates, klass, target_bytes)
+                                       files, segments, bytes, partial, finished, file_total, tool, candidates, klass, target_bytes, status)
          SELECT decode(v.release_key, 'hex'), decode(v.identity_key, 'hex'), ?, v.name, v.poster, to_timestamp(v.first / 1000.0), to_timestamp(v.last / 1000.0),
                 v.first_article, v.last_article, v.files, v.segments, v.bytes, v.partial, v.finished, v.file_total, v.tool, v.candidates,
-                pending_class(v.name, v.bytes, v.partial), v.target_bytes
+                pending_class(v.name, v.bytes, v.partial), v.target_bytes, v.status
          FROM jsonb_to_recordset(?::jsonb) AS v(release_key text, identity_key text, name text, poster text, first bigint, last bigint, first_article bigint, last_article bigint,
-                                               files int, segments bigint, bytes bigint, partial boolean, finished boolean, file_total int, tool text, candidates jsonb, target_bytes bigint)
+                                               files int, segments bigint, bytes bigint, partial boolean, finished boolean, file_total int, tool text, candidates jsonb, target_bytes bigint, status int)
          ON CONFLICT (release_key) DO UPDATE SET
            files = pending_releases.files + excluded.files, segments = pending_releases.segments + excluded.segments, bytes = pending_releases.bytes + excluded.bytes,
            last_posted_at = greatest(pending_releases.last_posted_at, excluded.last_posted_at), last_article = greatest(pending_releases.last_article, excluded.last_article),
            partial = excluded.partial, finished = excluded.finished, file_total = coalesce(excluded.file_total, pending_releases.file_total),
            candidates = CASE WHEN jsonb_array_length(pending_releases.candidates) = 0 THEN excluded.candidates ELSE pending_releases.candidates END,
            klass = pending_class(pending_releases.name, pending_releases.bytes + excluded.bytes, excluded.partial),
-           target_bytes = coalesce(pending_releases.target_bytes, excluded.target_bytes)
+           target_bytes = coalesce(pending_releases.target_bytes, excluded.target_bytes),
+           status = CASE WHEN pending_releases.status = ? AND excluded.finished
+                              AND jsonb_array_length(pending_releases.candidates) = 0 AND jsonb_array_length(excluded.candidates) = 0
+                         THEN ? ELSE pending_releases.status END
          WHERE NOT pending_releases.finished
-         RETURNING release_key, tool, klass, (xmax = 0) AS inserted`,
-        [groupId, JSON.stringify(rows.slice(i, i + 1000))]
+         RETURNING release_key, tool, klass, (xmax = 0) AS inserted, status`,
+        [groupId, JSON.stringify(rows.slice(i, i + 1000)), PENDING.waiting, PENDING.exhausted]
       )
       written += done.length
       // Each class seen gets its row in the ladder's record, so the claim explores it (pending_classes).
       for (const d of done) {
         const key = `${d.tool}\t${d.klass}`
         const entry = classes.get(key) ?? { tool: d.tool, klass: Number(d.klass), waiting: 0 }
-        if (d.inserted) entry.waiting++
+        if (d.inserted && Number(d.status) === PENDING.waiting) entry.waiting++
         classes.set(key, entry)
       }
     }

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 }>()

edge/src/indexer.rs

@@ -1608,8 +1608,13 @@ fn candidates_of(key: &str, files: &[Vec<Seg>], upload: usize) -> Vec<Candidate>
     if let Some((_, f)) = tail {
         out.push(Candidate { release_key: key.to_string(), upload, kind: "7z-tail", message_id: f.last().unwrap().message_id.clone(), file: f[0].file.name.clone(), bytes: f.last().unwrap().bytes });
     }
-    if out.is_empty() && files.len() == 1 && is_random_file(&files[0][0].file.name) {
-        out.push(c("lone", &files[0][0]));
+    // No set and no archive: the first random-named file's first segment, for its yEnc name or a twin by content
+    // (hash_twins.ts). Any count of files: a hash-named release of several pieces (Bob's, LOG.md (140)) had no
+    // target at all, so the ladder never probed it, never named it and never learned its class.
+    if out.is_empty() {
+        if let Some(f) = files.iter().find(|f| is_random_file(&f[0].file.name)) {
+            out.push(c("lone", &f[0]));
+        }
     }
     out.truncate(4);
     out
@@ -1738,6 +1743,30 @@ mod tests {
         assert_eq!(ix.pending(), 0);
     }
 
+    #[test]
+    fn a_multi_file_random_release_gets_a_lone_target() {
+        let mut priorities = FxHashMap::default();
+        priorities.insert(GROUP.to_string(), 1);
+        let mut ix = RangeIndexer::new(GROUP, 1, 10_000, 1, Rules::defaults(priorities), 5);
+        // Three hash-named pieces of one hash-named upload, five segments each (more than a per-article post): no
+        // archive, no par2, no 7z volume.
+        let mut lines = Vec::new();
+        for (i, name) in ["9e7fc2741cdd450287aff2733ca4c193", "c5ebdd5487f5b132c76e8d5c2572e560", "8114c42c9694380c8b25857e8b3331c3"].iter().enumerate() {
+            for part in 0..5 {
+                lines.push(line(100 + (i * 5 + part) as i64, &format!(r#"b3a75d234aa73891e78a70fc1f5496b1 [{}/3] - "{name}" yEnc ({}/5)"#, i + 1, part + 1), (i * 5 + part) as i64));
+            }
+        }
+        ix.add_text(&lines.join("\n"), 10_001);
+        ix.finish();
+        let batch = ix.take();
+        assert_eq!(batch.releases.len(), 1, "counts {:?}", batch.counts);
+        let r = &batch.releases[0];
+        assert!(!r.readable && r.files == 3 && r.segments == 15, "readable {} files {} segments {}", r.readable, r.files, r.segments);
+        let kinds: Vec<_> = batch.candidates.iter().map(|c| (c.kind, c.file.as_str())).collect();
+        assert_eq!(kinds, vec![("lone", "9e7fc2741cdd450287aff2733ca4c193")]);
+        assert_eq!(batch.candidates[0].message_id, "u100@test");
+    }
+
     #[test]
     fn album_templates_and_names() {
         assert_eq!(album_template("1481439079025_014.jpg.data").as_deref(), Some("#_#.jpg.data"));

tests/functional/batch_store.spec.ts

@@ -538,6 +538,38 @@ test.group('BatchStore', (group) => {
     assert.include(await db.from('pending_releases').firstOrFail(), { files: 2, segments: '5' })
   })
 
+  test('a finished unnamed release without a probe target is exhausted at once, not waiting', async ({
+    assert,
+  }) => {
+    const { rangeId, nodeId } = await rangeIn(GROUP, 1)
+    // Two hash-named pieces: the edge gives the first a lone target; without one the ladder could never claim it.
+    const batch = batchOf(GROUP, 1, [
+      ...upload(
+        10,
+        '9e7fc2741cdd450287aff2733ca4c193',
+        4,
+        'b3a75d234aa73891e78a70fc1f5496b1 [1/2]',
+        '2026-10-05T10:00:00Z'
+      ),
+      ...upload(
+        20,
+        'c5ebdd5487f5b132c76e8d5c2572e560',
+        4,
+        'b3a75d234aa73891e78a70fc1f5496b1 [2/2]',
+        '2026-10-05T10:01:00Z'
+      ),
+    ])
+    assert.lengthOf(batch.releases, 1)
+    assert.isFalse(batch.releases[0].readable)
+    const stored = await BatchStore.store(rangeId, nodeId, { ...batch, candidates: [] })
+    assert.include(stored, { releases: 1, newReleases: 0, candidates: 0, pending: 1 })
+    const pending = await db.from('pending_releases').firstOrFail()
+    assert.include(pending, { files: 2, finished: true, status: 3 })
+    assert.deepEqual(pending.candidates, [])
+    const klass = await db.from('pending_classes').firstOrFail()
+    assert.include(klass, { waiting: 0 })
+  })
+
   test("an album's members stay findable by their own names (release_texts kind member)", async ({
     assert,
   }) => {