diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index cc5b2bc..7df1826 100644 --- a/app/services/batch_store.ts +++ b/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) } } diff --git a/edge/src/indexer.rs b/edge/src/indexer.rs index 2d16a28..acd98ea 100644 --- a/edge/src/indexer.rs +++ b/edge/src/indexer.rs @@ -1608,8 +1608,13 @@ fn candidates_of(key: &str, files: &[Vec], upload: usize) -> Vec 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")); diff --git a/tests/functional/batch_store.spec.ts b/tests/functional/batch_store.spec.ts index 7091994..efa986f 100644 --- a/tests/functional/batch_store.spec.ts +++ b/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, }) => {