Daan MeijerandClaude Fable 5.1
e135076fcd
pending_ladder.ts claims the oldest waiting pending releases' next target (par2, archive, lone, 7z-tail in that order) as at least PENDING_SHARE of each claim; a usable name (the largest named file of a par2 set or a listing, or a lone file's yEnc name, neither packaging nor random) hands the release to the namer, no name the next target or the end of the climb; climbs count on pending_signatures per (tool, group, poster) and (tool, group), and a signature with 100 climbs and 0.5% named or fewer is dropped: its waiting releases are discarded without a probe, the rules' sample excepted. pending_namer.ts reads a named release's articles again from the corpus or by XOVER and writes it to the catalog under the pending key (source weak-stem, the name as inner name) with its chunk and candidates. Migration 6400; config PENDING_*. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
410 lines
20 KiB
TypeScript
410 lines
20 KiB
TypeScript
import db from '@adonisjs/lucid/services/db'
|
|
import { pruneSevenZipTails } from '#services/seven_zip_tails'
|
|
import type { TransactionClientContract } from '@adonisjs/lucid/types/database'
|
|
import type { EdgeBatch, ProbeCandidate, ReleaseChunk } from '#lib/edge/index'
|
|
import ChunkPartitions from '#services/chunk_partitions'
|
|
import CatalogRefs, { groupSetKey, posterText } from '#services/catalog_refs'
|
|
import { decodeChunk } from '#lib/chunks/index'
|
|
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 { TEXT_KINDS, TEXT_MAX, textsToStore } from '#lib/release/text_kind'
|
|
|
|
/**
|
|
* A small new release (at most 3 files, under 1 MB) whose files are all sidecars (.nfo, .sfv, .nzb, …): the sidecar
|
|
* fold's candidates (releases.sidecar_only). Its file list decoded here, small as it is.
|
|
*/
|
|
function sidecarOnly(r: ReleaseChunk): boolean | null {
|
|
if (r.files > 3 || r.bytes >= 1_000_000) return null
|
|
const files = decodeChunk(r.payload)
|
|
return files.length > 0 && files.every((f) => SIDECAR.test(f.name))
|
|
}
|
|
|
|
export class StaleBatch extends Error {}
|
|
|
|
export interface StoredBatch {
|
|
/** Release chunks in the batch. */
|
|
releases: number
|
|
/** Releases the batch started. */
|
|
newReleases: number
|
|
/** Provisional (tail) releases the batch finished. */
|
|
finalized: number
|
|
candidates: number
|
|
/** Unnamed releases written to the pending store instead of the catalog (decision f). */
|
|
pending: number
|
|
}
|
|
|
|
/** A chunk's slot in its batch: the key and the upload the edge numbered it with (ProbeCandidate points at it). */
|
|
const slot = (key: string, upload: number) => `${key}\n${upload}`
|
|
|
|
/**
|
|
* Stores one batch of a leased range (REBUILD.md, "Voyager 3: window indexing", 2 and 4) in one transaction: each
|
|
* release chunk is a whole release the edge's window finished, written once (its poster and group set by id, its
|
|
* keys as bytea, its name into release_texts), with one chunk of its files, its probe candidates, and the batch's
|
|
* rule counts and samples; then the range's resume point moves. A batch that doesn't start at the resume point was
|
|
* sent twice or too late, and is refused (StaleBatch), so the sums stay right.
|
|
*
|
|
* A tail release comes twice: provisionally (finished: false, settled_at null, its candidates waiting) and then
|
|
* finished, under the same release key (the identity, the group and its first article: lib/release/keys.ts), with
|
|
* the files not sent before (a chunk's counts are its own; the finish adds them up); the finish is the one update a
|
|
* release gets from the header pass. A finished release sent again is left as it is.
|
|
*
|
|
* Interim, until blocks are leased (the next step): chunks, counts and samples are keyed by the range's block
|
|
* (ranges.block_id), and a range without one gets a block of its own here.
|
|
*/
|
|
class BatchStore {
|
|
async store(rangeId: number, nodeId: number, batch: EdgeBatch): Promise<StoredBatch> {
|
|
await ChunkPartitions.ensure()
|
|
return db.transaction(async (trx) => {
|
|
// One batch of a range at a time, without holding the range's row for the whole transaction: the
|
|
// prefetcher's and heartbeats' updates of it waited on it (0.5 s on average, NNTP connections idle;
|
|
// LOG.md 2026-10-09 (13)). The resume point is checked again when it moves, at the end.
|
|
await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended('range:' || ?::text, 0))`, [
|
|
rangeId,
|
|
])
|
|
const range = await trx.from('ranges').where('id', rangeId).first()
|
|
if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId)
|
|
throw new StaleBatch(`Range ${rangeId} is not leased to this node`)
|
|
if (Number(range.next_id) !== batch.from)
|
|
throw new StaleBatch(
|
|
`Range ${rangeId} is at ${range.next_id}, the batch starts at ${batch.from}`
|
|
)
|
|
const blockId = await this.blockOf(trx, range)
|
|
const result: StoredBatch = {
|
|
releases: batch.releases.length,
|
|
newReleases: 0,
|
|
finalized: 0,
|
|
candidates: 0,
|
|
pending: 0,
|
|
}
|
|
|
|
// Nothing enters the catalog without a usable name (decision f): an unnamed release (no readable file name)
|
|
// goes to the pending store with its probe targets, the named ones to the catalog.
|
|
const named = batch.releases.filter((r) => r.readable)
|
|
const unnamed = batch.releases.filter((r) => !r.readable)
|
|
const written = await this.writeReleases(
|
|
trx,
|
|
{ groupId: Number(range.group_id), blockId },
|
|
batch.from,
|
|
named,
|
|
batch.candidates
|
|
)
|
|
result.newReleases = written.newReleases
|
|
result.finalized = written.finalized
|
|
result.candidates = written.candidates
|
|
result.pending = await this.writePending(
|
|
trx,
|
|
Number(range.group_id),
|
|
unnamed,
|
|
batch.candidates
|
|
)
|
|
|
|
if (batch.counts.length > 0) {
|
|
await trx.rawQuery(
|
|
`INSERT INTO rule_counts (block_id, rule, key, articles, files)
|
|
SELECT ?, v.rule, v.key, v.articles, v.files FROM jsonb_to_recordset(?::jsonb) AS v(rule text, key text, articles bigint, files bigint)
|
|
ON CONFLICT (block_id, rule, key) DO UPDATE SET articles = rule_counts.articles + excluded.articles, files = rule_counts.files + excluded.files`,
|
|
[blockId, JSON.stringify(batch.counts)]
|
|
)
|
|
}
|
|
|
|
for (const s of batch.samples) {
|
|
await trx.insertQuery().table('rule_samples').insert({
|
|
block_id: blockId,
|
|
rule: s.rule,
|
|
rule_key: s.ruleKey,
|
|
release_key: s.key,
|
|
name: s.name,
|
|
files: s.files,
|
|
segments: s.segments,
|
|
payload: s.payload,
|
|
})
|
|
}
|
|
|
|
const moved = await trx
|
|
.from('ranges')
|
|
.where('id', rangeId)
|
|
.where('status', 'leased')
|
|
.where('node_id', nodeId)
|
|
.where('next_id', batch.from)
|
|
.update({
|
|
next_id: batch.next,
|
|
headers: db.raw('headers + ?', [batch.headers]),
|
|
// The range's posting time span: the batches' trimmed spans, min and max (least and greatest skip nulls).
|
|
posted_low: db.raw('least(posted_low, to_timestamp(?::double precision / 1000))', [
|
|
batch.postedLow ?? null,
|
|
]),
|
|
posted_high: db.raw('greatest(posted_high, to_timestamp(?::double precision / 1000))', [
|
|
batch.postedHigh ?? null,
|
|
]),
|
|
heartbeat_at: new Date(),
|
|
})
|
|
if (Number(moved) !== 1 && !(Array.isArray(moved) && moved.length === 1))
|
|
throw new StaleBatch(`Range ${rangeId} moved on meanwhile`)
|
|
return result
|
|
})
|
|
}
|
|
|
|
/** The range's block; a range without one (no planner yet) is a block of its own, leased as the range is. */
|
|
async blockOf(trx: TransactionClientContract, range: any): Promise<number> {
|
|
if (range.block_id) return Number(range.block_id)
|
|
const [block] = await trx
|
|
.table('blocks')
|
|
.insert({
|
|
group_id: range.group_id,
|
|
start_id: range.start_id,
|
|
end_id: range.end_id,
|
|
next_id: range.next_id,
|
|
status: 'leased',
|
|
priority: range.priority ?? 0,
|
|
node_id: range.node_id,
|
|
leased_at: range.leased_at ?? new Date(),
|
|
heartbeat_at: new Date(),
|
|
})
|
|
.returning('id')
|
|
await trx.from('ranges').where('id', range.id).update({ block_id: block.id })
|
|
return Number(block.id)
|
|
}
|
|
|
|
/**
|
|
* The unnamed releases of a batch into pending_releases (migration 6300, the HDD): keys, counts, the upload
|
|
* tool and the probe targets, no chunk. The pending ladder (REBUILD.md decision f) probes them cheapest first,
|
|
* and a name moves one into the catalog; a provisional tail release sent again adds its counts up.
|
|
*/
|
|
async writePending(
|
|
trx: TransactionClientContract,
|
|
groupId: number,
|
|
releases: ReleaseChunk[],
|
|
candidates: ProbeCandidate[]
|
|
): Promise<number> {
|
|
if (releases.length === 0) return 0
|
|
const targets = new Map<string, { kind: string; messageId: string; file: string }[]>()
|
|
for (const c of candidates) {
|
|
const key = slot(c.releaseKey, c.upload ?? 0)
|
|
const list = targets.get(key) ?? []
|
|
list.push({ kind: c.kind, messageId: c.messageId, file: c.file })
|
|
targets.set(key, list)
|
|
}
|
|
let written = 0
|
|
for (let i = 0; i < releases.length; i += 1000) {
|
|
const rows = releases.slice(i, i + 1000).map((r) => {
|
|
const probes = targets.get(slot(r.key, r.upload)) ?? []
|
|
return {
|
|
release_key: releaseKey(r.key, groupId, r.firstArticle).toString('hex'),
|
|
identity_key: identityKey(r.key).toString('hex'),
|
|
name: r.name.slice(0, TEXT_MAX),
|
|
poster: r.poster || null,
|
|
first: r.postedAt,
|
|
last: r.postedLast ?? r.postedAt,
|
|
first_article: r.firstArticle,
|
|
last_article: r.lastArticle,
|
|
files: r.files,
|
|
segments: r.segments,
|
|
bytes: r.bytes,
|
|
partial: r.partial === true,
|
|
finished: r.finished !== false,
|
|
file_total: r.fileTotal,
|
|
tool: r.tool ?? (probes[0] ? uploadTool(probes[0].messageId) : 'unknown'),
|
|
candidates: probes,
|
|
}
|
|
})
|
|
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)
|
|
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
|
|
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)
|
|
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
|
|
WHERE NOT pending_releases.finished
|
|
RETURNING release_key`,
|
|
[groupId, JSON.stringify(rows)]
|
|
)
|
|
written += done.length
|
|
}
|
|
return written
|
|
}
|
|
|
|
/**
|
|
* The releases of a batch into the catalog, one chunk each, and their probe candidates. Every chunk is one
|
|
* release: inserted, or (a provisional tail release sent again) finished.
|
|
*/
|
|
async writeReleases(
|
|
trx: TransactionClientContract,
|
|
where: { groupId: number; blockId: number },
|
|
batchFrom: number,
|
|
releases: ReleaseChunk[],
|
|
candidates: ProbeCandidate[]
|
|
) {
|
|
const out = { newReleases: 0, finalized: 0, candidates: 0 }
|
|
if (releases.length === 0) return out
|
|
/** Release ids by slot, for the chunks and candidates; a slot not here was a finished release sent again. */
|
|
const ids = new Map<string, number>()
|
|
const finalized = new Set<number>()
|
|
|
|
const posters = await CatalogRefs.posterIds(trx, releases.map((r) => r.poster).filter(Boolean))
|
|
const groupSets = await CatalogRefs.groupSetIds(
|
|
trx,
|
|
releases.map((r) => r.groups)
|
|
)
|
|
const sorted = [...releases].sort((a, b) => a.key.localeCompare(b.key) || a.upload - b.upload)
|
|
for (let i = 0; i < sorted.length; i += 1000) {
|
|
const part = sorted.slice(i, i + 1000)
|
|
const rows = part.map((r, idx) => ({
|
|
idx,
|
|
release_key: releaseKey(r.key, where.groupId, r.firstArticle).toString('hex'),
|
|
identity_key: identityKey(r.key).toString('hex'),
|
|
name: r.name,
|
|
source: sourceCode(r.source),
|
|
poster_id: r.poster ? (posters.get(posterText(r.poster)) ?? null) : null,
|
|
group_set_id: groupSets.get(groupSetKey(r.groups))!,
|
|
first: r.postedAt,
|
|
last: r.postedLast ?? r.postedAt,
|
|
file_total: r.fileTotal,
|
|
files: r.files,
|
|
segments: r.segments,
|
|
bytes: r.bytes,
|
|
first_article: r.firstArticle,
|
|
last_article: r.lastArticle,
|
|
readable: r.readable,
|
|
has_par2: r.hasPar2,
|
|
finished: r.finished !== false,
|
|
edge: r.edge === true,
|
|
partial: r.partial === true,
|
|
sidecar_only: sidecarOnly(r),
|
|
}))
|
|
// Inserted, or a provisional release of the key finished (the chunk's files added up); a finished release
|
|
// sent again (no row returned) is left as it is.
|
|
const { rows: written } = await trx.rawQuery(
|
|
`INSERT INTO releases (release_key, identity_key, group_id, group_set_id, poster_id, name, source, posted_at, last_posted_at, file_total, files, segments, bytes, chunks,
|
|
first_article, last_article, readable, has_par2, complete, edge, sidecar_only, settled_at)
|
|
SELECT decode(v.release_key, 'hex'), decode(v.identity_key, 'hex'), ?, v.group_set_id, v.poster_id, v.name, v.source,
|
|
to_timestamp(v.first / 1000.0), to_timestamp(v.last / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1,
|
|
v.first_article, v.last_article, v.readable, v.has_par2,
|
|
CASE WHEN v.partial THEN false WHEN v.file_total IS NULL THEN NULL ELSE v.files >= v.file_total END,
|
|
v.edge, v.sidecar_only, CASE WHEN v.finished THEN now() END
|
|
FROM jsonb_to_recordset(?::jsonb) AS v(idx int, release_key text, identity_key text, name text, source int, poster_id int, group_set_id int, first bigint, last bigint,
|
|
file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean,
|
|
finished boolean, edge boolean, partial boolean, sidecar_only boolean)
|
|
ON CONFLICT (release_key) DO UPDATE SET
|
|
files = releases.files + excluded.files, segments = releases.segments + excluded.segments, bytes = releases.bytes + excluded.bytes,
|
|
chunks = releases.chunks + CASE WHEN excluded.files > 0 THEN 1 ELSE 0 END,
|
|
posted_at = least(releases.posted_at, excluded.posted_at), last_posted_at = greatest(releases.last_posted_at, excluded.last_posted_at),
|
|
file_total = coalesce(excluded.file_total, releases.file_total),
|
|
first_article = least(releases.first_article, excluded.first_article), last_article = greatest(releases.last_article, excluded.last_article),
|
|
readable = releases.readable OR excluded.readable, has_par2 = releases.has_par2 OR excluded.has_par2,
|
|
complete = CASE WHEN releases.complete IS FALSE AND releases.file_total IS NULL THEN false
|
|
WHEN coalesce(excluded.file_total, releases.file_total) IS NULL THEN releases.complete
|
|
ELSE releases.files + excluded.files >= coalesce(excluded.file_total, releases.file_total) END,
|
|
edge = excluded.edge, sidecar_only = coalesce(releases.sidecar_only, excluded.sidecar_only), settled_at = excluded.settled_at
|
|
WHERE releases.settled_at IS NULL
|
|
RETURNING id, encode(release_key, 'hex') AS release_key, (xmax = 0) AS inserted, settled_at`,
|
|
[where.groupId, JSON.stringify(rows)]
|
|
)
|
|
const byKey = new Map<string, { id: number; inserted: boolean; settled: boolean }>(
|
|
written.map(
|
|
(w: { id: string; release_key: string; inserted: boolean; settled_at: Date | null }) => [
|
|
w.release_key,
|
|
{ id: Number(w.id), inserted: w.inserted, settled: w.settled_at !== null },
|
|
]
|
|
)
|
|
)
|
|
const texts: { release_id: number; kind: number; text: string }[] = []
|
|
for (const row of rows) {
|
|
const w = byKey.get(row.release_key)
|
|
if (!w) continue
|
|
const r = part[row.idx]
|
|
ids.set(slot(r.key, r.upload), w.id)
|
|
if (w.inserted) out.newReleases++
|
|
else if (w.settled) {
|
|
out.finalized++
|
|
finalized.add(w.id)
|
|
}
|
|
texts.push({ release_id: w.id, kind: TEXT_KINDS.name, text: row.name.slice(0, TEXT_MAX) })
|
|
// An album's members stay findable by their own names (the first TEXTS_PER_KIND of them).
|
|
if (r.source === 'album' && r.files > 0) {
|
|
for (const text of textsToStore(decodeChunk(r.payload).map((f) => f.name))) {
|
|
texts.push({ release_id: w.id, kind: TEXT_KINDS.member, text })
|
|
}
|
|
}
|
|
}
|
|
// The name (and an album's members), searchable (release_texts; a text sent again is a no-op).
|
|
if (texts.length > 0) {
|
|
await trx.rawQuery(
|
|
`INSERT INTO release_texts (release_id, kind, text)
|
|
SELECT v.release_id, v.kind, v.text FROM jsonb_to_recordset(?::jsonb) AS v(release_id bigint, kind int, text text) WHERE v.text <> ''
|
|
ON CONFLICT DO NOTHING`,
|
|
[JSON.stringify(texts)]
|
|
)
|
|
}
|
|
}
|
|
|
|
// One write-once chunk per release (a finish adds the files not sent before).
|
|
for (let i = 0; i < sorted.length; i += 500) {
|
|
const rows = sorted
|
|
.slice(i, i + 500)
|
|
.filter((r) => ids.has(slot(r.key, r.upload)) && r.files > 0)
|
|
.map((r) => ({
|
|
release_id: ids.get(slot(r.key, r.upload))!,
|
|
block_id: where.blockId,
|
|
batch_from: batchFrom,
|
|
files: r.files,
|
|
segments: r.segments,
|
|
payload: r.payload,
|
|
}))
|
|
if (rows.length > 0) await trx.insertQuery().table('release_chunks').multiInsert(rows)
|
|
}
|
|
|
|
// A finished release's candidates are claimable at once; a provisional one's wait for its finish.
|
|
const settled = new Set<number>()
|
|
for (const r of sorted) {
|
|
const id = ids.get(slot(r.key, r.upload))
|
|
if (id !== undefined && r.finished !== false) settled.add(id)
|
|
}
|
|
const chunks = new Map(sorted.map((r) => [slot(r.key, r.upload), r]))
|
|
const kept = candidates.filter((c) => ids.has(slot(c.releaseKey, c.upload ?? 0)))
|
|
for (let i = 0; i < kept.length; i += 2000) {
|
|
const rows = kept.slice(i, i + 2000).map((c) => {
|
|
const releaseId = ids.get(slot(c.releaseKey, c.upload ?? 0))!
|
|
const chunk = chunks.get(slot(c.releaseKey, c.upload ?? 0))!
|
|
return {
|
|
release_id: releaseId,
|
|
kind: c.kind,
|
|
message_id: c.messageId,
|
|
file: c.file,
|
|
ready: settled.has(releaseId),
|
|
unreadable: !chunk.readable,
|
|
}
|
|
})
|
|
const added = await trx
|
|
.knexQuery()
|
|
.table('probe_candidates')
|
|
.insert(rows)
|
|
.onConflict(['release_id', 'kind', 'message_id'])
|
|
.ignore()
|
|
.returning('id')
|
|
out.candidates += added.length
|
|
// A later piece of a split 7z: the tails of lower volumes go (app/services/seven_zip_tails.ts).
|
|
const tails = new Set(rows.filter((r) => r.kind === '7z-tail').map((r) => r.release_id))
|
|
if (added.length > 0 && tails.size > 0) await pruneSevenZipTails(trx, [...tails])
|
|
}
|
|
// The finish of a provisional release: its waiting candidates become claimable.
|
|
if (finalized.size > 0) {
|
|
await trx.rawQuery(
|
|
`UPDATE probe_candidates pc SET ready = true, unreadable = NOT r.readable FROM releases r
|
|
WHERE pc.release_id = ANY(?) AND r.id = pc.release_id AND pc.status = 'waiting' AND NOT pc.ready`,
|
|
[[...finalized]]
|
|
)
|
|
}
|
|
|
|
return out
|
|
}
|
|
}
|
|
|
|
export default new BatchStore()
|