Decide completeness by par2 set for releases without a counter
59% of settled releases had complete = NULL: without an [i/N] counter the settler left it unknown, and the "phase 2 reads par2 sets for those" step was never built. A release is now complete when every file its par2 sets describe is there by name or by its first-16-KiB hash, or, for an upload renamed after its par2 was made, when it has as many non-par2 files and bytes. Decided when a par2 index is probed, after a par2 merge, and before the pruner drops a set's file list; the batch store, settler and merger keep a decided value instead of resetting it to NULL. releases:par2-complete decides the releases probed before this. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
f75097d7ac
commit
ebe9f8bb36
10 files changed
+187
-12
No files matched your search
+8
-4
@@ -189,10 +189,14 @@ dropping garbage; 28 TB of compressed overview traffic. Throughput goal:
|
||||
number; merged when the neighbouring range is done, then written as a
|
||||
chunk. Never staged per segment.
|
||||
|
||||
8. **Settling**. A release is settled when the ranges covering its posting
|
||||
span are done and no chunk arrived for a day: completeness is computed
|
||||
then (expected vs found files and segments). Phase 2 picks its work from
|
||||
settled releases.
|
||||
8. **Settling**. A release is settled when nothing was added to it for 6
|
||||
hours (30 minutes once its last post is that old: history), and its
|
||||
completeness decided: complete when every file its [i/N] counter
|
||||
announced was found. Without a counter, by its par2 sets once their index
|
||||
is probed (`par2_completeness`): every file a set describes is there by
|
||||
name or by its first 16 KiB, or, renamed, as many files and bytes; decided
|
||||
again after a par2 merge and before the pruner drops a set's file list.
|
||||
Segments aren't compared. Phase 2 picks its work from settled releases.
|
||||
|
||||
9. **NZB and API**. NZB built from a release's chunks; a minimal Newznab
|
||||
search (names) and get. Web UI later.
|
||||
|
||||
@@ -237,8 +237,9 @@ class BatchStore {
|
||||
last_article = greatest(r.last_article, a.last_article),
|
||||
readable = r.readable OR a.readable,
|
||||
has_par2 = r.has_par2 OR a.has_par2,
|
||||
-- As it stands (the counter's files all found); settling makes it final.
|
||||
complete = CASE WHEN coalesce(r.file_total, a.file_total) IS NULL THEN NULL ELSE r.files + a.files >= coalesce(r.file_total, a.file_total) END,
|
||||
-- As it stands (the counter's files all found); settling makes it final. Without a counter, what its par2
|
||||
-- set decided stays (app/services/par2_completeness.ts).
|
||||
complete = CASE WHEN coalesce(r.file_total, a.file_total) IS NULL THEN r.complete ELSE r.files + a.files >= coalesce(r.file_total, a.file_total) END,
|
||||
last_seen_at = now()
|
||||
FROM a WHERE r.id = a.id`,
|
||||
[JSON.stringify(updates)]
|
||||
|
||||
@@ -21,8 +21,8 @@ class NzbService {
|
||||
return rows.sort((a, b) => rank(a.source) - rank(b.source)).map((r) => ({ password: r.password, source: r.source }))
|
||||
}
|
||||
|
||||
async files(releaseId: number): Promise<ChunkFile[]> {
|
||||
const chunks = await db.from('release_chunks').where('release_id', releaseId).orderBy('id').select('payload')
|
||||
async files(releaseId: number, client: any = db): Promise<ChunkFile[]> {
|
||||
const chunks = await client.from('release_chunks').where('release_id', releaseId).orderBy('id').select('payload')
|
||||
const files = new Map<string, ChunkFile>()
|
||||
for (const chunk of chunks) {
|
||||
for (const file of decodeChunk(chunk.payload)) files.set(`${file.name}\n${file.segments[0]?.[1] ?? ''}`, file)
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
import NzbService from '#services/nzb_service'
|
||||
|
||||
/** A file a par2 set describes (par2_files). */
|
||||
export interface SetFile {
|
||||
name: string
|
||||
length: number
|
||||
hash16k: string
|
||||
}
|
||||
|
||||
const base = (name: string) => name.replace(/^.*[\\/]/, '').toLowerCase()
|
||||
|
||||
/**
|
||||
* Whether a release holds every file its par2 sets describe (REBUILD.md, part 8): for one without an [i/N] counter,
|
||||
* whose completeness the settler leaves unknown. A described file is there when a file of the release has its name,
|
||||
* or a probed file of it its first 16 KiB and size (file_hashes). An upload renamed after its par2 was made (no
|
||||
* described name among its files) is complete when it has as many files besides par2 as the set describes, and as
|
||||
* many bytes (its articles' bytes, a little over the files' with yEnc).
|
||||
*/
|
||||
export function par2Complete(set: SetFile[], files: { name: string; size: number }[], hashes: { hash16k: string; size: number }[]): boolean {
|
||||
const names = new Set(files.map((f) => base(f.name)))
|
||||
const hashed = new Set(hashes.map((h) => `${h.hash16k}:${h.size}`))
|
||||
const byName = set.filter((f) => names.has(base(f.name))).length
|
||||
if (set.every((f) => names.has(base(f.name)) || hashed.has(`${f.hash16k}:${f.length}`))) return true
|
||||
if (byName > 0) return false
|
||||
const data = files.filter((f) => !/\.par2$/i.test(f.name))
|
||||
return data.length >= set.length && data.reduce((n, f) => n + f.size, 0) >= set.reduce((n, f) => n + f.length, 0)
|
||||
}
|
||||
|
||||
/**
|
||||
* Decides `complete` for the settled releases among `releaseIds` that have no counter and a par2 set with its file
|
||||
* list (par2_files; the pruner drops it once the set is done, so it decides before). How many it decided. `client`:
|
||||
* a transaction that moved chunks (a merge) reads them as moved.
|
||||
*/
|
||||
export async function decidePar2Completeness(client: any, releaseIds: (number | string)[]): Promise<number> {
|
||||
if (releaseIds.length === 0) return 0
|
||||
const { rows: releases } = await client.rawQuery(
|
||||
`SELECT id FROM releases WHERE id = ANY(?) AND file_total IS NULL AND settled_at IS NOT NULL`,
|
||||
[releaseIds.map(String)]
|
||||
)
|
||||
let decided = 0
|
||||
for (const { id } of releases) {
|
||||
const { rows: set } = await client.rawQuery(
|
||||
`SELECT DISTINCT f.file_id, f.name, f.length::float8 AS length, f.hash16k FROM par2_sets s JOIN par2_files f ON f.set_id = s.set_id WHERE s.release_id = ?`,
|
||||
[id]
|
||||
)
|
||||
if (set.length === 0) continue
|
||||
const { rows: hashes } = await client.rawQuery(`SELECT hash16k, size::float8 AS size FROM file_hashes WHERE release_id = ?`, [id])
|
||||
const files = (await NzbService.files(Number(id), client)).map((f) => ({ name: f.name, size: f.segments.reduce((n, [, , bytes]) => n + bytes, 0) }))
|
||||
await client.rawQuery(`UPDATE releases SET complete = ? WHERE id = ?`, [par2Complete(set, files, hashes), id])
|
||||
decided++
|
||||
}
|
||||
return decided
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import { looksRandom } from '#lib/release/random'
|
||||
import { SidecarFolder } from '#services/sidecar_folder'
|
||||
import { Par2Pruner } from '#services/par2_pruner'
|
||||
import { pruneSevenZipTails } from '#services/seven_zip_tails'
|
||||
import { decidePar2Completeness } from '#services/par2_completeness'
|
||||
|
||||
/** Releases of a set further apart than this (posted) aren't one upload: a repost is merged separately (decision d). */
|
||||
const MAX_SPREAD_MS = 48 * 3600 * 1000
|
||||
@@ -371,12 +372,13 @@ export class Par2Merger {
|
||||
WHERE t.id = ?`,
|
||||
[name ?? '', others, target.id]
|
||||
)
|
||||
// Completeness as the settler decides it, once settled.
|
||||
// Completeness as the settler decides it, once settled; without a counter, by the par2 sets it now holds.
|
||||
await trx.rawQuery(
|
||||
`UPDATE releases SET complete = CASE WHEN file_total IS NULL THEN NULL ELSE files >= file_total END WHERE id = ? AND settled_at IS NOT NULL`,
|
||||
`UPDATE releases SET complete = CASE WHEN file_total IS NULL THEN complete ELSE files >= file_total END WHERE id = ? AND settled_at IS NOT NULL`,
|
||||
[target.id]
|
||||
)
|
||||
await trx.rawQuery(`DELETE FROM releases WHERE id = ANY(?)`, [others])
|
||||
await decidePar2Completeness(trx, [target.id])
|
||||
return others.length
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import { decidePar2Completeness } from '#services/par2_completeness'
|
||||
|
||||
/** A set this long past its last holder's par2 has had its merges: one posted later is a repost (decision d). */
|
||||
const SETTLED_MS = 2 * 86400_000
|
||||
@@ -86,6 +87,11 @@ export class Par2Pruner {
|
||||
for (const reason of ['complete', 'judged', 'final'] as const) {
|
||||
const ids = done(reason)
|
||||
round[reason] += ids.length
|
||||
// The holders' completeness while the file lists are there (app/services/par2_completeness.ts).
|
||||
if (reason !== 'judged' && ids.length > 0) {
|
||||
const { rows: holders } = await db.rawQuery(`SELECT DISTINCT release_id FROM par2_sets WHERE set_id = ANY(?::char(32)[])`, [ids])
|
||||
await decidePar2Completeness(db, holders.map((h: { release_id: string }) => h.release_id))
|
||||
}
|
||||
round.rows += await this.drop(ids)
|
||||
}
|
||||
// Past the last set: the next batch starts the next pass.
|
||||
|
||||
@@ -8,6 +8,7 @@ import { YencError } from '#lib/yenc/index'
|
||||
import { inspect, innerName, PROBE_VERSION, type Inspection } from '#lib/probe/index'
|
||||
import { NntpShare } from '#services/nntp_shares'
|
||||
import { DiskSpace } from '#services/disk_space'
|
||||
import { decidePar2Completeness } from '#services/par2_completeness'
|
||||
|
||||
interface Candidate {
|
||||
id: number
|
||||
@@ -306,6 +307,8 @@ export class Prober {
|
||||
[jsonb(done.map((o) => ({ release_id: o.candidate.releaseId, candidate_id: o.candidate.id, kind: o.candidate.kind, result: { ...o.inspection, par2: undefined } })))]
|
||||
)
|
||||
await this.storeForMerge(trx, done)
|
||||
// A par2 index read: the release's completeness by its set, when it has no counter.
|
||||
await decidePar2Completeness(trx, [...new Set(done.filter((o) => o.inspection!.par2).map((o) => o.candidate.releaseId))])
|
||||
// Each touched release from all its inspections (earlier rounds may have stored others).
|
||||
const ids = [...new Set(done.map((o) => o.candidate.releaseId))]
|
||||
const { rows } = await trx.rawQuery(
|
||||
|
||||
@@ -15,7 +15,8 @@ const BATCH = 50_000
|
||||
* Settles releases (REBUILD.md, part 8): once nothing has been added for
|
||||
* QUIET_HOURS (QUIET_HISTORY_MINUTES once its last post is that old), its completeness is decided: complete when every file the
|
||||
* [i/N] counter announced was found, unknown without a counter (phase 2
|
||||
* reads par2 sets for those). Phase 2 picks its work from settled releases.
|
||||
* decides those by their par2 sets: app/services/par2_completeness.ts).
|
||||
* Phase 2 picks its work from settled releases.
|
||||
*/
|
||||
class Settler {
|
||||
/** Up to `batches` batches of BATCH (all there are without). */
|
||||
@@ -25,7 +26,7 @@ class Settler {
|
||||
// Their probe candidates become claimable (Prober.claim), the unreadable ones first.
|
||||
const { rows } = await db.rawQuery(
|
||||
`WITH settled AS (
|
||||
UPDATE releases SET settled_at = now(), complete = CASE WHEN file_total IS NULL THEN NULL ELSE files >= file_total END
|
||||
UPDATE releases SET settled_at = now(), complete = CASE WHEN file_total IS NULL THEN complete ELSE files >= file_total END
|
||||
WHERE id IN (SELECT id FROM releases WHERE settled_at IS NULL AND last_seen_at < now() - make_interval(mins => ?)
|
||||
AND (last_seen_at < now() - make_interval(hours => ?) OR coalesce(last_posted_at, posted_at) < now() - make_interval(hours => ?)) LIMIT ?)
|
||||
RETURNING id, readable, files, bytes, poster, sidecar_only),
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import type { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import { decidePar2Completeness } from '#services/par2_completeness'
|
||||
|
||||
/** Releases per batch. */
|
||||
const BATCH = 1000
|
||||
|
||||
/**
|
||||
* Decides completeness by par2 set (app/services/par2_completeness.ts) for the releases that had their par2 read
|
||||
* before it was built (2026-10-23): every holder of a set whose file list is still there, by release id. Once.
|
||||
*/
|
||||
export default class ReleasesPar2Complete extends BaseCommand {
|
||||
static commandName = 'releases:par2-complete'
|
||||
static description = 'Decides completeness by their par2 sets for settled releases without a counter, then exits'
|
||||
static options: CommandOptions = { startApp: true }
|
||||
|
||||
async run() {
|
||||
let after = 0
|
||||
let looked = 0
|
||||
let decided = 0
|
||||
for (;;) {
|
||||
const { rows } = await db.rawQuery(
|
||||
`SELECT DISTINCT s.release_id FROM par2_sets s
|
||||
WHERE s.release_id > ? AND EXISTS (SELECT 1 FROM par2_files f WHERE f.set_id = s.set_id)
|
||||
ORDER BY s.release_id LIMIT ?`,
|
||||
[after, BATCH]
|
||||
)
|
||||
if (rows.length === 0) break
|
||||
const ids = rows.map((r: { release_id: string }) => Number(r.release_id))
|
||||
decided += await decidePar2Completeness(db, ids)
|
||||
looked += ids.length
|
||||
after = ids.at(-1)!
|
||||
if (looked % 50_000 < BATCH) this.logger.info(`${looked} releases with a par2 set looked at, ${decided} decided (up to id ${after}).`)
|
||||
}
|
||||
this.logger.info(`Done: ${looked} releases with a par2 set looked at, ${decided} decided.`)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
import { test } from '@japa/runner'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import testUtils from '@adonisjs/core/services/test_utils'
|
||||
import BatchStore from '#services/batch_store'
|
||||
import { decidePar2Completeness, par2Complete } from '#services/par2_completeness'
|
||||
import { RangeIndexer } from '#lib/edge/index'
|
||||
import { DEFAULT_RULES } from '#lib/rules/index'
|
||||
import type { Overview } from '#lib/nntp/index'
|
||||
|
||||
const GROUP = 'alt.binaries.test'
|
||||
|
||||
/** A file posted without an [i/N] counter: the release gets no file_total. */
|
||||
function file(name: string, parts: number, at: number): Overview[] {
|
||||
return Array.from({ length: parts }, (_, i) => [at + i, {
|
||||
'subject': `"${name}" yEnc (${i + 1}/${parts})`, 'from': 'p <p@x>', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000',
|
||||
'message-id': `<${name}-${i + 1}@test>`, ':bytes': '700000', 'xref': `news.example ${GROUP}:${at + i}`,
|
||||
}] as Overview)
|
||||
}
|
||||
|
||||
async function store(start: number, overviews: Overview[]) {
|
||||
const g = (await db.from('groups').where('name', GROUP).first()) ?? (await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('*'))[0]
|
||||
const [n] = await db.table('nodes').insert({ name: `n${start}` }).returning('id')
|
||||
const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: start + 999, next_id: start, status: 'leased', node_id: n.id }).returning('id')
|
||||
const indexer = new RangeIndexer({ group: GROUP, start, end: start + 999, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } } })
|
||||
indexer.add(overviews, start + 1000)
|
||||
indexer.finish()
|
||||
await BatchStore.store(Number(r.id), Number(n.id), indexer.take())
|
||||
}
|
||||
|
||||
const setFile = (name: string, length: number, hash16k = name.padEnd(32, '0').slice(0, 32)) => ({ name, length, hash16k })
|
||||
|
||||
test.group('par2 completeness', (group) => {
|
||||
group.each.setup(() => testUtils.db().withGlobalTransaction())
|
||||
|
||||
test('every described file by name or by hash; a renamed upload by its files and bytes', ({ assert }) => {
|
||||
const set = [setFile('a.mkv', 1000), setFile('a.nfo', 10)]
|
||||
const files = [{ name: 'a.mkv', size: 1100 }, { name: 'a.nfo', size: 12 }, { name: 'a.par2', size: 50 }]
|
||||
assert.isTrue(par2Complete(set, files, []))
|
||||
assert.isFalse(par2Complete(set, files.slice(0, 1), []))
|
||||
// The nfo renamed, but its probe hashed it.
|
||||
assert.isTrue(par2Complete(set, [files[0], { name: 'x.nfo', size: 12 }], [{ hash16k: set[1].hash16k, size: 10 }]))
|
||||
// Renamed after the par2 was made (random names): two data files and enough bytes; one missing, or too small.
|
||||
assert.isTrue(par2Complete(set, [{ name: 'Qx81.bin', size: 1100 }, { name: 'Zr02.bin', size: 12 }, { name: 'Qx.par2', size: 50 }], []))
|
||||
assert.isFalse(par2Complete(set, [{ name: 'Qx81.bin', size: 1100 }, { name: 'Qx.par2', size: 50 }], []))
|
||||
assert.isFalse(par2Complete(set, [{ name: 'Qx81.bin', size: 500 }, { name: 'Zr02.bin', size: 12 }], []))
|
||||
})
|
||||
|
||||
test('decided once settled with a par2 set, and kept when more files come', async ({ assert }) => {
|
||||
await store(1, [...file('Some.Upload.mkv', 3, 10), ...file('Some.Upload.par2', 1, 20)])
|
||||
const release = await db.from('releases').firstOrFail()
|
||||
assert.isNull(release.file_total)
|
||||
await db.table('par2_files').multiInsert([setFile('Some.Upload.mkv', 2_000_000), setFile('Some.Upload.nfo', 100)].map((f, i) => ({ set_id: 's'.repeat(32), file_id: String(i).repeat(32), ...f })))
|
||||
await db.table('par2_sets').insert({ set_id: 's'.repeat(32), release_id: release.id, files: 2 })
|
||||
// Not settled yet: left alone.
|
||||
assert.equal(await decidePar2Completeness(db, [release.id]), 0)
|
||||
await db.from('releases').where('id', release.id).update({ settled_at: new Date() })
|
||||
assert.equal(await decidePar2Completeness(db, [release.id]), 1)
|
||||
assert.isFalse((await db.from('releases').where('id', release.id).firstOrFail()).complete)
|
||||
// The nfo comes in a later range: complete, and the batch store's own update keeps it.
|
||||
await store(1001, file('Some.Upload.nfo', 1, 1010))
|
||||
await decidePar2Completeness(db, [release.id])
|
||||
await store(2001, file('Some.Upload.sfv', 1, 2010))
|
||||
const after = await db.from('releases').where('id', release.id).firstOrFail()
|
||||
assert.equal(after.files, 4)
|
||||
assert.isTrue(after.complete)
|
||||
})
|
||||
})
|
||||
Reference in new issue
Block a user