Compare commits
5
Commits
eadae82b0c
...
voyager-3
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c6a095e740 | ||
|
|
294f2d73ba | ||
|
|
66980829dc | ||
|
|
f537add573 | ||
|
|
d7171a5ca4 |
No files matched your search
+10
-4
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 }>()
|
||||
|
||||
+31
-2
@@ -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"));
|
||||
|
||||
@@ -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,
|
||||
}) => {
|
||||
|
||||
Reference in new issue
Block a user