Daan MeijerandClaude Opus 5.5
693e6c3244
- seed:plan plans every part of a group's history no range covers (it started below the lowest known range and left a.b.ath's 13 most recent months out); seed_ranges.priority orders leases (seed:plan --priority). - binaries.part_count, kept by the parts insert in the same statement: claimCompleted compares it instead of counting parts every batch. - file_keys: 64-bit hashes of all file keys; FileStore.insert asks it before probing every files partition. - worker:file stores 100 completed binaries per job with set-based queries. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
417 lines
17 KiB
TypeScript
417 lines
17 KiB
TypeScript
import db from '@adonisjs/lucid/services/db'
|
|
import logger from '@adonisjs/core/services/logger'
|
|
import type { QueryClientContract } from '@adonisjs/lucid/types/database'
|
|
import type { Knex } from 'knex'
|
|
import QueueService from '#services/QueueService'
|
|
import NameService from '#services/NameService'
|
|
import type { IngestSegment } from '#lib/ingest/segment'
|
|
|
|
// Header parsing is pure and lives in lib, so seeding nodes need no database (CONCEPT.md D37).
|
|
export { segmentFromHeader, segmentFromYenc, type IngestSegment, type OverviewHeader } from '#lib/ingest/segment'
|
|
|
|
export interface CompletedBinary {
|
|
id: number
|
|
fileKey: string
|
|
}
|
|
|
|
const BINARY_CHUNK = 1000
|
|
/** last_seen_at is only rewritten when it is older than this (it drives GC at 48 h). */
|
|
const TOUCH_MS = 10 * 60 * 1000
|
|
|
|
type Merge = 'least' | 'coalesce' | 'touch'
|
|
|
|
/**
|
|
* Inserts rows keyed by `key` and returns their ids. Existing rows are not
|
|
* locked unless something in them changes (`least`: an earlier value,
|
|
* `coalesce`: a value where there was none, `touch`: a timestamp more than
|
|
* TOUCH_MS newer). Parallel fetch jobs see mostly the same rows, and
|
|
* `INSERT … ON CONFLICT DO UPDATE` made them wait on each other for every one.
|
|
*/
|
|
async function upsertKeyed(
|
|
knex: Knex,
|
|
table: 'binaries' | 'releases',
|
|
key: string,
|
|
rows: Record<string, unknown>[],
|
|
merge: Record<string, [Merge, string]>
|
|
): Promise<Map<string, number>> {
|
|
const ids = new Map<string, number>()
|
|
if (rows.length === 0) return ids
|
|
const inserted: Record<string, any>[] = await knex(table).insert(rows).onConflict(key).ignore().returning(['id', key])
|
|
for (const row of inserted) ids.set(row[key], Number(row.id))
|
|
|
|
const incoming = new Map(rows.filter((r) => !ids.has(r[key] as string)).map((r) => [r[key] as string, r]))
|
|
if (incoming.size === 0) return ids
|
|
const columns = Object.keys(merge)
|
|
const existing: Record<string, any>[] = await knex(table).whereIn(key, [...incoming.keys()]).select('id', key, ...columns)
|
|
|
|
const changes: Record<string, unknown>[] = []
|
|
for (const row of existing) {
|
|
ids.set(row[key], Number(row.id))
|
|
const next = incoming.get(row[key])!
|
|
const changed = columns.some((column) => {
|
|
const [kind] = merge[column]
|
|
const current = row[column]
|
|
const value = next[column]
|
|
if (value === null || value === undefined) return false
|
|
if (kind === 'coalesce') return current === null
|
|
if (kind === 'least') return current === null || value < current
|
|
return current === null || +(value as Date) - +current > TOUCH_MS
|
|
})
|
|
if (changed) changes.push({ id: Number(row.id), ...Object.fromEntries(columns.map((c) => [c, next[c] ?? null])) })
|
|
}
|
|
if (changes.length === 0) return ids
|
|
|
|
const set = columns
|
|
.map((c) => {
|
|
const [kind] = merge[c]
|
|
const fn = kind === 'least' ? 'least' : kind === 'touch' ? 'greatest' : 'coalesce'
|
|
return `${c} = ${fn}(t.${c}, v.${c})`
|
|
})
|
|
.join(', ')
|
|
const types = columns.map((c) => `${c} ${merge[c][1]}`).join(', ')
|
|
changes.sort((a, b) => (a.id as number) - (b.id as number))
|
|
for (let attempt = 1; ; attempt++) {
|
|
try {
|
|
await knex.raw(
|
|
`UPDATE ${table} t SET ${set}
|
|
FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, ${types})
|
|
WHERE t.id = v.id`,
|
|
[JSON.stringify(changes)]
|
|
)
|
|
return ids
|
|
} catch (error: any) {
|
|
// Rare now, but two jobs can still update overlapping rows in a different order.
|
|
if (error.code !== '40P01' || attempt >= 3) throw error
|
|
}
|
|
}
|
|
}
|
|
/** Completed binaries per worker:file job. */
|
|
const FILE_BATCH = 100
|
|
|
|
/** Size-only matching below this is too ambiguous (nfo, sfv, small volumes). */
|
|
export const PAR2_SIZE_ONLY_MIN = 1024 * 1024
|
|
/** A par2 set and its files are posted together; don't match across days. */
|
|
export const PAR2_MATCH_WINDOW_SECONDS = 12 * 60 * 60
|
|
const PART_CHUNK = 2000
|
|
|
|
function chunks<T>(items: T[], size: number): T[][] {
|
|
const result: T[][] = []
|
|
for (let i = 0; i < items.length; i += size) result.push(items.slice(i, i + size))
|
|
return result
|
|
}
|
|
|
|
function validDate(date?: Date) {
|
|
return date && !Number.isNaN(date.getTime()) ? date : null
|
|
}
|
|
|
|
/**
|
|
* Stores segments in `binaries`/`parts` and queues every binary that became
|
|
* complete exactly once. Idempotent: ingesting the same segments again is a
|
|
* no-op. See CONCEPT.md (D12).
|
|
*/
|
|
class IngestService {
|
|
async ingest(segments: IngestSegment[]): Promise<CompletedBinary[]> {
|
|
if (segments.length === 0) return []
|
|
|
|
// Short statements, no enclosing transaction: every step is idempotent, and
|
|
// row locks held for a whole batch made parallel fetch jobs wait on each other.
|
|
const binaryIds = await (async () => {
|
|
const knex = db.connection().getWriteClient()
|
|
const now = new Date()
|
|
const releaseIds = await this.upsertReleases(knex, segments, now)
|
|
|
|
// One row per file_key (ON CONFLICT cannot touch a row twice per statement),
|
|
// sorted so concurrent workers lock rows in the same order and cannot deadlock.
|
|
const binaries = new Map<string, Record<string, unknown>>()
|
|
for (const segment of segments) {
|
|
const postedAt = validDate(segment.postedAt)
|
|
const existing = binaries.get(segment.fileKey)
|
|
if (existing) {
|
|
if (postedAt && (!existing.posted_at || postedAt < (existing.posted_at as Date))) {
|
|
existing.posted_at = postedAt
|
|
}
|
|
continue
|
|
}
|
|
binaries.set(segment.fileKey, {
|
|
file_key: segment.fileKey,
|
|
poster: segment.poster,
|
|
subject: segment.subject,
|
|
name: segment.name,
|
|
total_parts: segment.totalParts,
|
|
groups: JSON.stringify(segment.groups),
|
|
posted_at: postedAt,
|
|
first_seen_at: now,
|
|
last_seen_at: now,
|
|
release_id: segment.release ? releaseIds.get(segment.release.key) : null,
|
|
size: segment.size ?? null,
|
|
md5_16k: segment.md5_16k ?? null,
|
|
})
|
|
}
|
|
const binaryRows = [...binaries.values()].sort((a, b) =>
|
|
(a.file_key as string).localeCompare(b.file_key as string)
|
|
)
|
|
|
|
const idsByKey = new Map<string, number>()
|
|
for (const chunk of chunks(binaryRows, BINARY_CHUNK)) {
|
|
const ids = await upsertKeyed(knex, 'binaries', 'file_key', chunk, {
|
|
last_seen_at: ['touch', 'timestamptz'],
|
|
posted_at: ['least', 'timestamptz'],
|
|
release_id: ['coalesce', 'bigint'],
|
|
md5_16k: ['coalesce', 'varchar'],
|
|
})
|
|
for (const [key, id] of ids) idsByKey.set(key, id)
|
|
}
|
|
|
|
// A binary that is a file already has its segment list in `files`; its
|
|
// header parts were dropped then. Crossposts in other groups, reposts
|
|
// and backfill ranges keep bringing the same articles, and inserting
|
|
// them again grew `parts` to 56M rows nobody reads (LOG.md 2026-10-04).
|
|
const done = new Set<number>()
|
|
const ids = [...idsByKey.values()]
|
|
for (const chunk of chunks(ids, BINARY_CHUNK)) {
|
|
const { rows } = await knex.raw('SELECT id FROM binaries WHERE id = ANY(?) AND file_id IS NOT NULL', [chunk])
|
|
for (const row of rows) done.add(Number(row.id))
|
|
}
|
|
|
|
const partRows = segments
|
|
.map((segment) => ({
|
|
binary_id: idsByKey.get(segment.fileKey)!,
|
|
number: segment.number,
|
|
message_id: segment.messageId,
|
|
bytes: segment.bytes,
|
|
article_number: segment.articleNumber?.toString() ?? null,
|
|
pcrc32: segment.pcrc32 ?? null,
|
|
part_size: segment.partSize ?? null,
|
|
file_crc32: segment.fileCrc32 ?? null,
|
|
md5_16k: segment.md5_16k ?? null,
|
|
}))
|
|
const weakRows = partRows.filter((row, i) => segments[i].weakIdentity && !(done.has(row.binary_id) && row.pcrc32 === null))
|
|
// Body-derived parts (with a per-part CRC) are kept, as worker:file does.
|
|
const newParts = partRows.filter((row) => !(done.has(row.binary_id) && row.pcrc32 === null))
|
|
|
|
newParts.sort((a, b) => a.binary_id - b.binary_id || a.number - b.number)
|
|
// Parts and the binaries' stored-part counts in one statement, so the count can't miss a
|
|
// part (LOG.md 2026-10-06 (7)); claimCompleted compares it instead of counting parts. NULL
|
|
// (binaries from before part_count) is counted once: the statement's own rows aren't in
|
|
// its snapshot, so they're added on top. Concurrent batches can lock binaries in a
|
|
// different order: a deadlock rolls the whole statement back, and it is retried.
|
|
for (const chunk of chunks(newParts, PART_CHUNK)) {
|
|
const insert = knex('parts').insert(chunk).onConflict(['binary_id', 'number']).ignore().returning('binary_id').toSQL()
|
|
for (let attempt = 1; ; attempt++) {
|
|
try {
|
|
await knex.raw(
|
|
`WITH added AS (${insert.sql}),
|
|
counted AS (SELECT binary_id, count(*)::int AS n FROM added GROUP BY binary_id)
|
|
UPDATE binaries b SET part_count = CASE
|
|
WHEN b.part_count IS NULL THEN (SELECT count(*) FROM parts p WHERE p.binary_id = b.id) + c.n
|
|
ELSE b.part_count + c.n END
|
|
FROM counted c WHERE b.id = c.binary_id`,
|
|
insert.bindings as any[]
|
|
)
|
|
break
|
|
} catch (error: any) {
|
|
if (error.code !== '40P01' || attempt >= 5) throw error
|
|
}
|
|
}
|
|
}
|
|
|
|
if (weakRows.length > 0) await this.flagConflicts(db.connection(), weakRows)
|
|
|
|
return [...idsByKey.values()]
|
|
})()
|
|
|
|
// par2 naming needs a size, which only body-derived segments have; for
|
|
// header batches these queries were most of Postgres' work.
|
|
const sized = [...new Set(segments.filter((s) => s.size != null).map((s) => s.fileKey))]
|
|
if (sized.length > 0) {
|
|
const sizedIds = (await db.from('binaries').whereIn('file_key', sized).select('id')).map((r) => Number(r.id))
|
|
await this.matchPar2Releases(sizedIds)
|
|
await NameService.resolve(sizedIds)
|
|
}
|
|
|
|
// After commit, so the last writer of a binary always sees all of its parts.
|
|
const completed = await this.claimCompleted(binaryIds)
|
|
await this.enqueue(completed)
|
|
return completed
|
|
}
|
|
|
|
/**
|
|
* Creates or touches the releases referenced by `segments` and returns
|
|
* their ids by release key. Also used by `release:rebuild`.
|
|
*/
|
|
async upsertReleases(
|
|
knex: Knex,
|
|
segments: Pick<IngestSegment, 'release' | 'poster' | 'groups' | 'postedAt' | 'weakIdentity'>[],
|
|
now = new Date()
|
|
): Promise<Map<string, number>> {
|
|
const releases = new Map<string, Record<string, unknown>>()
|
|
for (const segment of segments) {
|
|
const release = segment.release
|
|
if (!release) continue
|
|
const postedAt = validDate(segment.postedAt)
|
|
const existing = releases.get(release.key)
|
|
if (existing) {
|
|
if (postedAt && (!existing.posted_at || postedAt < (existing.posted_at as Date))) {
|
|
existing.posted_at = postedAt
|
|
}
|
|
// Not every file carries the [i/M] counter (e.g. a par2 index): take it from any that does.
|
|
if (existing.file_total === null && release.fileTotal) existing.file_total = release.fileTotal
|
|
continue
|
|
}
|
|
releases.set(release.key, {
|
|
release_key: release.key,
|
|
name: release.name,
|
|
// Obfuscating posters randomise the poster, so it says nothing about the release.
|
|
poster: segment.weakIdentity ? null : segment.poster,
|
|
file_total: release.fileTotal ?? null,
|
|
source: release.source,
|
|
groups: JSON.stringify(segment.groups),
|
|
posted_at: postedAt,
|
|
first_seen_at: now,
|
|
last_seen_at: now,
|
|
})
|
|
}
|
|
|
|
const ids = new Map<string, number>()
|
|
const rows = [...releases.values()].sort((a, b) =>
|
|
(a.release_key as string).localeCompare(b.release_key as string)
|
|
)
|
|
for (const chunk of chunks(rows, BINARY_CHUNK)) {
|
|
const result = await upsertKeyed(knex, 'releases', 'release_key', chunk, {
|
|
last_seen_at: ['touch', 'timestamptz'],
|
|
posted_at: ['least', 'timestamptz'],
|
|
file_total: ['coalesce', 'integer'],
|
|
source: ['coalesce', 'varchar'],
|
|
})
|
|
for (const [key, id] of result) ids.set(key, id)
|
|
}
|
|
return ids
|
|
}
|
|
|
|
/**
|
|
* Assigns body-derived binaries without a release and without a
|
|
* fingerprint to a known par2 set by size: only when that size is listed
|
|
* once in the set, the file is large, and exactly one set nearby lists it.
|
|
* Binaries with a fingerprint are matched exactly by NameService instead.
|
|
*/
|
|
async matchPar2Releases(binaryIds: number[]) {
|
|
if (binaryIds.length === 0) return
|
|
// Listed-once first (index-only on release_files (release_id, size): most candidate
|
|
// sizes are a set's equal-sized volumes), the release's date only for what's left;
|
|
// OFFSET 0 keeps that order. Was: every size match's date and count, 18 s per batch.
|
|
await db.rawQuery(
|
|
`UPDATE binaries b SET release_id = m.release_id
|
|
FROM (
|
|
SELECT c.id, min(c.release_id) AS release_id
|
|
FROM (
|
|
SELECT b2.id, b2.posted_at, rf.release_id
|
|
FROM binaries b2
|
|
JOIN release_files rf ON rf.size = b2.size
|
|
WHERE b2.id = ANY(?) AND b2.release_id IS NULL AND b2.md5_16k IS NULL AND b2.size >= ?
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM release_files x
|
|
WHERE x.release_id = rf.release_id AND x.size = rf.size AND x.par2_file_id <> rf.par2_file_id)
|
|
OFFSET 0
|
|
) c
|
|
JOIN releases r ON r.id = c.release_id
|
|
WHERE abs(extract(epoch FROM (r.posted_at - c.posted_at))) < ?
|
|
GROUP BY c.id
|
|
HAVING count(DISTINCT c.release_id) = 1
|
|
) m
|
|
WHERE b.id = m.id`,
|
|
[binaryIds, PAR2_SIZE_ONLY_MIN, PAR2_MATCH_WINDOW_SECONDS]
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Flags binaries where a stored part has a different message-id and a
|
|
* different part CRC than the incoming one: two files were merged under
|
|
* one identity. Identical reposts (same CRC) are ignored.
|
|
*/
|
|
private async flagConflicts(
|
|
trx: QueryClientContract,
|
|
rows: {
|
|
binary_id: number
|
|
number: number
|
|
message_id: string
|
|
bytes: number
|
|
article_number: string | null
|
|
pcrc32: string | null
|
|
part_size: number | null
|
|
file_crc32: string | null
|
|
md5_16k: string | null
|
|
}[]
|
|
) {
|
|
const bindings = [
|
|
rows.map((r) => r.binary_id),
|
|
rows.map((r) => r.number),
|
|
rows.map((r) => r.message_id),
|
|
rows.map((r) => r.bytes),
|
|
rows.map((r) => r.article_number),
|
|
rows.map((r) => r.pcrc32),
|
|
rows.map((r) => r.part_size),
|
|
rows.map((r) => r.file_crc32),
|
|
rows.map((r) => r.md5_16k),
|
|
]
|
|
// Keep the extra articles: SplitService separates the volumes from them.
|
|
const conflicts = await trx.rawQuery(
|
|
`INSERT INTO part_conflicts (binary_id, number, message_id, bytes, article_number, pcrc32, part_size, file_crc32, md5_16k)
|
|
SELECT v.binary_id, v.number, v.message_id, v.bytes, v.article_number, v.pcrc32, v.part_size, v.file_crc32, v.md5_16k
|
|
FROM unnest(?::bigint[], ?::int[], ?::text[], ?::int[], ?::bigint[], ?::text[], ?::int[], ?::text[], ?::text[])
|
|
AS v(binary_id, number, message_id, bytes, article_number, pcrc32, part_size, file_crc32, md5_16k)
|
|
JOIN parts p ON p.binary_id = v.binary_id AND p.number = v.number
|
|
WHERE p.message_id <> v.message_id
|
|
-- Same part CRC: the same data posted again (a repost), not another file.
|
|
AND (p.pcrc32 IS NULL OR v.pcrc32 IS NULL OR p.pcrc32 <> v.pcrc32)
|
|
ON CONFLICT DO NOTHING
|
|
RETURNING binary_id`,
|
|
bindings as any
|
|
)
|
|
const binaryIds: number[] = [
|
|
...new Set<number>(conflicts.rows.map((r: { binary_id: string }) => Number(r.binary_id))),
|
|
]
|
|
if (binaryIds.length === 0) return
|
|
const result = await trx.rawQuery(
|
|
`UPDATE binaries SET conflicted_at = now()
|
|
WHERE conflicted_at IS NULL AND id = ANY(?) RETURNING name`,
|
|
[binaryIds]
|
|
)
|
|
for (const row of result.rows) {
|
|
logger.warn(`Binary "${row.name}" has conflicting parts; it will be split into volumes.`)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Marks complete binaries among `binaryIds` as claimed. The single UPDATE
|
|
* locks each row, so concurrent callers never claim the same binary twice.
|
|
*/
|
|
async claimCompleted(binaryIds: number[]): Promise<CompletedBinary[]> {
|
|
if (binaryIds.length === 0) return []
|
|
const result = await db.rawQuery(
|
|
`UPDATE binaries b SET completed_at = now()
|
|
WHERE b.id = ANY(?) AND b.completed_at IS NULL AND b.conflicted_at IS NULL
|
|
AND coalesce(b.part_count, (SELECT count(*) FROM parts p WHERE p.binary_id = b.id)) >= b.total_parts
|
|
RETURNING b.id, b.file_key`,
|
|
[binaryIds]
|
|
)
|
|
return result.rows.map((row: { id: string; file_key: string }) => ({
|
|
id: Number(row.id),
|
|
fileKey: row.file_key,
|
|
}))
|
|
}
|
|
|
|
async enqueue(completed: CompletedBinary[]) {
|
|
if (completed.length === 0) return
|
|
// A job per FILE_BATCH binaries: worker:file stores them with a few set-based queries (LOG.md 2026-10-06 (7)).
|
|
// jobId from the first file_key: BullMQ ignores an add() for an id that already exists.
|
|
const jobs = []
|
|
for (let i = 0; i < completed.length; i += FILE_BATCH) {
|
|
const batch = completed.slice(i, i + FILE_BATCH)
|
|
jobs.push({ name: 'process-files', data: { binaryIds: batch.map((b) => b.id) }, opts: { jobId: `${batch[0].fileKey}-${batch.length}` } })
|
|
}
|
|
await QueueService.fileQueue.addBulk(jobs)
|
|
}
|
|
}
|
|
|
|
export default new IngestService()
|