Keep binaries until inspected; seeding writes binaries only for inspection
Par2 naming needs the fingerprint inspection stores on a binary, and the GC deleted seeded binaries (6 h unfingerprinted, 48 h stale) long before the inspection queue reached them. binaries.awaiting_inspection protects them for up to 7 days; the GC deletes use partial indexes so they don't read past waiting rows. Seeding no longer creates binaries for files that aren't inspected (~85% of a.b.ath), which only churned the table. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
cc861ce8ae
commit
b2cc75fdd0
11 files changed
+149
-35
No files matched your search
@@ -99,6 +99,7 @@ class BinaryFileService {
|
||||
const id = insertedByKey.get(b.file_key) ?? byKey.get(b.file_key)
|
||||
if (id) fileIdOf.set(Number(b.id), id)
|
||||
}
|
||||
const fresher = new Map(fresh.map((b) => [b.file_key, b]))
|
||||
result.saved = inserted.length
|
||||
result.existing = fileIdOf.size - inserted.length
|
||||
|
||||
@@ -110,9 +111,12 @@ class BinaryFileService {
|
||||
`SELECT id FROM binaries WHERE id IN (SELECT unnest(?::bigint[])) ORDER BY file_key FOR UPDATE`,
|
||||
[linked.map((b) => b.id)]
|
||||
)
|
||||
// Files queued for inspection below keep their binary until it has its fingerprint (par2 naming, D16).
|
||||
const queued = new Set(inserted.filter((row) => row.file_key && ContentService.isCandidate(String(fresher.get(row.file_key)?.name ?? ''))).map((row) => row.id))
|
||||
await trx.rawQuery(
|
||||
`UPDATE binaries b SET file_id = v.file_id FROM unnest(?::bigint[], ?::bigint[]) AS v(id, file_id) WHERE b.id = v.id`,
|
||||
[linked.map((b) => b.id), linked.map((b) => fileIdOf.get(Number(b.id))!)]
|
||||
`UPDATE binaries b SET file_id = v.file_id, awaiting_inspection = CASE WHEN v.queued THEN true ELSE b.awaiting_inspection END
|
||||
FROM unnest(?::bigint[], ?::bigint[], ?::boolean[]) AS v(id, file_id, queued) WHERE b.id = v.id`,
|
||||
[linked.map((b) => b.id), linked.map((b) => fileIdOf.get(Number(b.id))!), linked.map((b) => queued.has(fileIdOf.get(Number(b.id))!))]
|
||||
)
|
||||
})
|
||||
// Header-derived parts are in the file's segment list now; body-derived ones (with a CRC) stay for split checks (D20).
|
||||
@@ -120,7 +124,6 @@ class BinaryFileService {
|
||||
}
|
||||
|
||||
// As for one file before: an SFV posted earlier may list them (D34), then inspection.
|
||||
const fresher = new Map(fresh.map((b) => [b.file_key, b]))
|
||||
const added = inserted.flatMap((row) => {
|
||||
const b = row.file_key ? fresher.get(row.file_key) : undefined
|
||||
return b
|
||||
|
||||
@@ -165,6 +165,8 @@ class ContentService {
|
||||
.update({
|
||||
size: db.raw('coalesce(size, ?)', [probe.size]),
|
||||
md5_16k: db.raw('coalesce(md5_16k, ?)', [probe.md5_16k]),
|
||||
// Inspected: the GC may delete it again by its usual rules (LOG.md 2026-10-06 (11)).
|
||||
awaiting_inspection: null,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
+27
-16
@@ -62,27 +62,38 @@ class SeedService {
|
||||
for (const row of await db.from('files').whereIn('file_key', rows.map((r) => r.file_key)).select('id', 'file_key')) {
|
||||
fileIds.set(row.file_key, Number(row.id))
|
||||
}
|
||||
// The completed binary worker:file would have left behind; an existing one just learns its file.
|
||||
const binaries = files.map((file, i) => ({
|
||||
file_key: file.fileKey,
|
||||
poster: file.poster,
|
||||
subject: file.subject,
|
||||
name: file.name,
|
||||
total_parts: file.totalParts,
|
||||
groups: rows[i].groups,
|
||||
posted_at: postedOf(file) ?? null,
|
||||
first_seen_at: now,
|
||||
last_seen_at: now,
|
||||
completed_at: now,
|
||||
release_id: rows[i].release_id,
|
||||
file_id: fileIds.get(file.fileKey) ?? null,
|
||||
}))
|
||||
// A completed binary, as worker:file would have left behind, only for files queued for content
|
||||
// inspection: par2 naming (D16) needs the fingerprint inspection stores on it. For the others
|
||||
// (~85% in a.b.ath) it was only GC churn: 1.6M rows an hour inserted and deleted (LOG.md 2026-10-06 (11)).
|
||||
const insertedKeys = new Set(inserted.map((row) => row.file_key))
|
||||
const binaries = files.flatMap((file, i) => {
|
||||
if (!ContentService.isCandidate(file.name)) return []
|
||||
return [{
|
||||
file_key: file.fileKey,
|
||||
poster: file.poster,
|
||||
subject: file.subject,
|
||||
name: file.name,
|
||||
total_parts: file.totalParts,
|
||||
groups: rows[i].groups,
|
||||
posted_at: postedOf(file) ?? null,
|
||||
first_seen_at: now,
|
||||
last_seen_at: now,
|
||||
completed_at: now,
|
||||
release_id: rows[i].release_id,
|
||||
file_id: fileIds.get(file.fileKey) ?? null,
|
||||
// New files are queued for inspection below; an existing file was queued when it was added.
|
||||
awaiting_inspection: insertedKeys.has(file.fileKey) ? true : null,
|
||||
}]
|
||||
})
|
||||
binaries.sort((a, b) => a.file_key.localeCompare(b.file_key))
|
||||
for (let i = 0; i < binaries.length; i += insertLimits.rows) {
|
||||
await knex('binaries')
|
||||
.insert(binaries.slice(i, i + insertLimits.rows))
|
||||
.onConflict('file_key')
|
||||
.merge({ file_id: knex.raw('coalesce(binaries.file_id, excluded.file_id)') as any })
|
||||
.merge({
|
||||
file_id: knex.raw('coalesce(binaries.file_id, excluded.file_id)') as any,
|
||||
awaiting_inspection: knex.raw('coalesce(binaries.awaiting_inspection, excluded.awaiting_inspection)') as any,
|
||||
})
|
||||
}
|
||||
|
||||
// As worker:file does for a new file: an SFV posted earlier may list it (D34), then inspection.
|
||||
|
||||
+20
-11
@@ -178,19 +178,27 @@ export default class GcScheduler extends BaseCommand {
|
||||
* already live on in `files`; conflicted ones will never become a file.
|
||||
* Completed ones still waiting for worker:file are kept.
|
||||
*/
|
||||
/**
|
||||
* Binaries without new segments for staleAfterSeconds: incomplete ones are given up on,
|
||||
* finished ones have their file. A finished one whose file waits for content inspection
|
||||
* stays until it's inspected (its fingerprint names it from par2 sets, D16), at most
|
||||
* awaitInspectionSeconds (LOG.md 2026-10-06 (11)). Three deletes, one per partial index,
|
||||
* so none reads past millions of waiting rows.
|
||||
*/
|
||||
private async deleteStaleBinaries(): Promise<number> {
|
||||
const query = db
|
||||
.from('binaries')
|
||||
.where('last_seen_at', '<', new Date(Date.now() - gcConfig.binaries.staleAfterSeconds * 1000))
|
||||
.where((q) =>
|
||||
q.whereNull('completed_at').orWhereNotNull('file_id').orWhereNotNull('conflicted_at')
|
||||
)
|
||||
|
||||
if (this.dryRun) {
|
||||
const [{ count }] = await query.count('* as count')
|
||||
return Number(count)
|
||||
const stale = new Date(Date.now() - gcConfig.binaries.staleAfterSeconds * 1000)
|
||||
const waited = new Date(Date.now() - gcConfig.binaries.awaitInspectionSeconds * 1000)
|
||||
const deletes = [
|
||||
db.from('binaries').where('last_seen_at', '<', stale).whereNotNull('file_id').whereRaw('awaiting_inspection IS NOT TRUE'),
|
||||
db.from('binaries').where('last_seen_at', '<', stale).where((q) => q.whereNull('completed_at').orWhereNotNull('conflicted_at')),
|
||||
db.from('binaries').where('last_seen_at', '<', waited).whereRaw('awaiting_inspection IS TRUE'),
|
||||
]
|
||||
let total = 0
|
||||
for (const query of deletes) {
|
||||
if (this.dryRun) total += Number((await query.count('* as count'))[0].count)
|
||||
else total += Number(await query.delete())
|
||||
}
|
||||
return Number(await query.delete())
|
||||
return total
|
||||
}
|
||||
|
||||
private async deleteOldDeferredFiles(): Promise<number> {
|
||||
@@ -283,6 +291,7 @@ export default class GcScheduler extends BaseCommand {
|
||||
SELECT id FROM binaries
|
||||
WHERE last_seen_at < ? AND file_id IS NOT NULL AND completed_at IS NOT NULL
|
||||
AND size IS NULL AND md5_16k IS NULL AND conflicted_at IS NULL
|
||||
AND awaiting_inspection IS NOT TRUE
|
||||
ORDER BY last_seen_at LIMIT ?)`,
|
||||
[cutoff, gcConfig.deleteBatch]
|
||||
)
|
||||
|
||||
@@ -46,6 +46,11 @@ export default {
|
||||
* later, so they go sooner than staleAfterSeconds.
|
||||
*/
|
||||
unfingerprintedAfterSeconds: env.get('GC_UNFINGERPRINTED_SECONDS', 6 * 60 * 60),
|
||||
/**
|
||||
* A finished binary whose file waits for content inspection is kept this long at
|
||||
* most (it needs the fingerprint inspection stores for par2 naming, D16).
|
||||
*/
|
||||
awaitInspectionSeconds: env.get('GC_AWAIT_INSPECTION_SECONDS', 7 * 24 * 60 * 60),
|
||||
},
|
||||
|
||||
/**
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
import { BaseSchema } from '@adonisjs/lucid/schema'
|
||||
|
||||
/**
|
||||
* A completed binary whose file is queued for content inspection keeps its
|
||||
* row until inspection has stored the file's fingerprint on it (LOG.md
|
||||
* 2026-10-06 (11)): par2 naming (D16) matches par2 listings against binary
|
||||
* fingerprints and attaches sets to binaries. With the inspection queue
|
||||
* days behind, the GC deleted them first (after 6 hours unfingerprinted,
|
||||
* 48 hours stale). TRUE while waiting; cleared by inspection.
|
||||
*/
|
||||
export default class extends BaseSchema {
|
||||
async up() {
|
||||
await this.db.rawQuery(`SET LOCAL lock_timeout = '2min'`)
|
||||
await this.db.rawQuery(`ALTER TABLE binaries ADD COLUMN IF NOT EXISTS awaiting_inspection boolean`)
|
||||
}
|
||||
|
||||
async down() {
|
||||
await this.db.rawQuery(`ALTER TABLE binaries DROP COLUMN IF EXISTS awaiting_inspection`)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
import { BaseSchema } from '@adonisjs/lucid/schema'
|
||||
|
||||
/**
|
||||
* Indexes for the GC's deletes of finished binaries, without the ones
|
||||
* waiting for inspection (LOG.md 2026-10-06 (11)): those can be millions
|
||||
* when inspection is behind, and the deletes must not read past them every
|
||||
* sweep. Built CONCURRENTLY: binaries takes writes all the time.
|
||||
*/
|
||||
export default class extends BaseSchema {
|
||||
static disableTransactions = true
|
||||
|
||||
async up() {
|
||||
await this.db.rawQuery(
|
||||
`CREATE INDEX CONCURRENTLY IF NOT EXISTS binaries_done_last_seen_idx ON binaries (last_seen_at)
|
||||
WHERE file_id IS NOT NULL AND awaiting_inspection IS NOT TRUE`
|
||||
)
|
||||
await this.db.rawQuery(
|
||||
`CREATE INDEX CONCURRENTLY IF NOT EXISTS binaries_awaiting_inspection_idx ON binaries (last_seen_at)
|
||||
WHERE awaiting_inspection IS TRUE`
|
||||
)
|
||||
await this.db.rawQuery(
|
||||
`CREATE INDEX CONCURRENTLY IF NOT EXISTS binaries_open_last_seen_idx ON binaries (last_seen_at)
|
||||
WHERE completed_at IS NULL OR conflicted_at IS NOT NULL`
|
||||
)
|
||||
}
|
||||
|
||||
async down() {
|
||||
await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS binaries_open_last_seen_idx`)
|
||||
await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS binaries_awaiting_inspection_idx`)
|
||||
await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS binaries_done_last_seen_idx`)
|
||||
}
|
||||
}
|
||||
+3
-1
@@ -33,8 +33,10 @@ export class AuthAccessTokenSchema extends BaseModel {
|
||||
}
|
||||
|
||||
export class BinarySchema extends BaseModel {
|
||||
static $columns = ['completedAt', 'conflictedAt', 'fileId', 'fileKey', 'firstSeenAt', 'groups', 'id', 'lastSeenAt', 'md516K', 'name', 'parentBinaryId', 'partCount', 'postedAt', 'poster', 'realName', 'releaseId', 'size', 'splitAt', 'subject', 'totalParts'] as const
|
||||
static $columns = ['awaitingInspection', 'completedAt', 'conflictedAt', 'fileId', 'fileKey', 'firstSeenAt', 'groups', 'id', 'lastSeenAt', 'md516K', 'name', 'parentBinaryId', 'partCount', 'postedAt', 'poster', 'realName', 'releaseId', 'size', 'splitAt', 'subject', 'totalParts'] as const
|
||||
$columns = BinarySchema.$columns
|
||||
@column()
|
||||
declare awaitingInspection: boolean | null
|
||||
@column.dateTime()
|
||||
declare completedAt: DateTime | null
|
||||
@column.dateTime()
|
||||
|
||||
@@ -105,6 +105,7 @@ export default await Env.create(new URL('../', import.meta.url), {
|
||||
ARCHIVE_WALK_CONCURRENCY: Env.schema.number.optional(),
|
||||
RELEASE_COMPLETE_INTERVAL_MS: Env.schema.number.optional(),
|
||||
FILE_KEYS_FILL_SECONDS: Env.schema.number.optional(),
|
||||
GC_AWAIT_INSPECTION_SECONDS: Env.schema.number.optional(),
|
||||
ARCHIVE_WALK_MAX_SEGMENTS: Env.schema.number.optional(),
|
||||
DEFERRED_SETTLE_MS: Env.schema.number.optional(),
|
||||
DEFERRED_EXPAND_MAX_PARTS: Env.schema.number.optional(),
|
||||
|
||||
@@ -6,6 +6,7 @@ import BinaryFileService from '#services/BinaryFileService'
|
||||
import QueueService from '#services/QueueService'
|
||||
import FileStore, { segmentsOf } from '#services/FileStore'
|
||||
import { fileKey } from '#lib/subject/index'
|
||||
import ContentService from '#services/ContentService'
|
||||
|
||||
function segment(name: string, number: number, total = 3): IngestSegment {
|
||||
const poster = 'batch-test@example.invalid'
|
||||
@@ -65,4 +66,21 @@ test.group('Binaries to files', (group) => {
|
||||
assert.include(again, { saved: 0, skipped: 2 })
|
||||
assert.deepEqual(again.incomplete, [ids[2]])
|
||||
})
|
||||
|
||||
test('a queued obfuscated file keeps its binary until inspection fingerprints it', async ({ assert }) => {
|
||||
const name = 'Zk3m9QpL2xW8rT5vN1bYq.bin'
|
||||
await IngestService.ingest([1, 2, 3].map((n) => segment(name, n)))
|
||||
const b = await binary(name)
|
||||
const result = await BinaryFileService.store([Number(b.id)])
|
||||
assert.include(result, { saved: 1, inspections: 1 })
|
||||
const waiting = await binary(name)
|
||||
assert.isTrue(waiting.awaiting_inspection)
|
||||
|
||||
// Inspection stores the fingerprint and clears the flag.
|
||||
const task = { kind: 'inspect' as const, fileId: Number(waiting.file_id), name, segments: [], sevenZip: null }
|
||||
await ContentService.apply(task, { size: 300, contentType: null, md5_16k: 'a'.repeat(32), yencName: name })
|
||||
const inspected = await binary(name)
|
||||
assert.isNull(inspected.awaiting_inspection)
|
||||
assert.equal(inspected.md5_16k, 'a'.repeat(32))
|
||||
})
|
||||
})
|
||||
@@ -37,15 +37,26 @@ test.group('SeedService', (group) => {
|
||||
assert.deepEqual(segmentsOf(stored).map((s) => s.number), [1, 2, 3])
|
||||
const release = await db.from('releases').where('id', stored.release_id).first()
|
||||
assert.equal(release.name, segments[0].release!.name) // named as live indexing would
|
||||
const binary = await db.from('binaries').where('file_key', stored.file_key).first()
|
||||
assert.equal(Number(binary.file_id), Number(stored.id))
|
||||
assert.isNotNull(binary.completed_at)
|
||||
// A readable name isn't inspected: no binary (LOG.md 2026-10-06 (11)).
|
||||
assert.isNull(await db.from('binaries').where('file_key', stored.file_key).first())
|
||||
|
||||
// The same files again (a crosspost, a second run): nothing new.
|
||||
const again = await SeedService.store(files)
|
||||
assert.include(again, { files: 0, existing: 2 })
|
||||
})
|
||||
|
||||
test('an obfuscated file keeps a binary until inspection has fingerprinted it', async ({ assert }) => {
|
||||
const aggregator = new FileAggregator<IngestSegment>()
|
||||
const subject = 'Xq8r2KdLm3Zw91PqA7zz [1/1] - "Xq8r2KdLm3Zw91PqA7zz.bin" yEnc (%d/2)'
|
||||
const segments = [1, 2].map((n) => segmentFromHeader(header(subject.replace('%d', String(n)), n), 'alt.binaries.test', n)!)
|
||||
const files = aggregator.add(segments).map(toSeedFile)
|
||||
const result = await SeedService.store(files)
|
||||
assert.include(result, { files: 1, inspections: 1 })
|
||||
const binary = await db.from('binaries').where('file_key', files[0].fileKey).first()
|
||||
assert.isTrue(binary.awaiting_inspection)
|
||||
assert.isNotNull(binary.file_id)
|
||||
})
|
||||
|
||||
test('stores a batch in several INSERTs', async ({ assert }) => {
|
||||
// Postgres takes 65,535 bind parameters per statement: ~5,400 rows of 12 columns.
|
||||
const rows = insertLimits.rows
|
||||
@@ -57,7 +68,7 @@ test.group('SeedService', (group) => {
|
||||
)
|
||||
const files = aggregator.add(segments).map(toSeedFile)
|
||||
assert.include(await SeedService.store(files), { files: 120 })
|
||||
const [{ count }] = await db.from('binaries').whereLike('name', 'dense.%').whereNotNull('file_id').count('* as count')
|
||||
const [{ count }] = await db.from('files').whereLike('filename', 'dense.%').count('* as count')
|
||||
assert.equal(Number(count), 120)
|
||||
} finally {
|
||||
insertLimits.rows = rows
|
||||
|
||||
Reference in new issue
Block a user