Pending ladder: a hash-named release of several pieces gets a lone target

The edge gave an unnamed release a "lone" probe target only when it had
exactly one file, so a release of several random-named pieces (Bob's,
1.55M rows in five groups) reached the pending store with no target at
all: never claimed, never named by a twin, its class never learned.
candidates_of now takes the first random-named file when there is no
par2, archive or 7z volume. The batch store writes a finished release
without a target as exhausted instead of waiting, and leaves it out of
its class's waiting count. LOG.md (140).

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

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

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,
   }) => {