diff --git a/app/services/prober.ts b/app/services/prober.ts index e2efaa4..3ec7ea3 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -30,6 +30,15 @@ const TIERS: { kind: string; unreadable: boolean }[] = [ { kind: 'archive', unreadable: false }, ] const FLUSH_MS = 2000 + +/** + * JSON for a jsonb parameter, without NUL characters: jsonb can't hold \u0000 (a name read from an article had one), + * and the whole flush failed with it (LOG.md 2026-10-18 (78)). + */ +function jsonb(value: unknown): string { + return JSON.stringify(value, (_key, v) => (typeof v === 'string' ? v.replaceAll('\0', '') : v)) +} + /** Slices a claim takes from random points of the queue. */ const SLICES = 20 const REPORT_MS = 60_000 @@ -262,16 +271,18 @@ export class Prober { await db.transaction(async (trx) => { await trx.rawQuery( `UPDATE probe_candidates pc SET status = v.status FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, status text) WHERE pc.id = v.id`, - [JSON.stringify(outcomes.map((o) => ({ id: o.candidate.id, status: o.status })))] + [jsonb(outcomes.map((o) => ({ id: o.candidate.id, status: o.status })))] ) // Final answers read from the news server, kept by message-id for the next catalog (a reset). - const fresh = outcomes.filter((o) => !o.cached && o.status !== 'waiting') + // One per message-id: the same article is a candidate of more than one release, and ON CONFLICT can't take a + // row twice in one statement. + const fresh = [...new Map(outcomes.filter((o) => !o.cached && o.status !== 'waiting').map((o) => [o.candidate.messageId, o])).values()] if (fresh.length > 0) { await trx.rawQuery( `INSERT INTO probe_cache (message_id, status, result, version) SELECT v.message_id, v.status, v.result, ? FROM jsonb_to_recordset(?::jsonb) AS v(message_id text, status text, result jsonb) ON CONFLICT (message_id) DO UPDATE SET status = excluded.status, result = excluded.result, version = excluded.version, created_at = now()`, - [PROBE_VERSION, JSON.stringify(fresh.map((o) => ({ message_id: o.candidate.messageId, status: o.status, result: o.inspection ?? null })))] + [PROBE_VERSION, jsonb(fresh.map((o) => ({ message_id: o.candidate.messageId, status: o.status, result: o.inspection ?? null })))] ) } const done = outcomes.filter((o) => o.inspection) @@ -281,7 +292,7 @@ export class Prober { SELECT v.release_id, v.candidate_id, v.kind, v.result FROM jsonb_to_recordset(?::jsonb) AS v(release_id bigint, candidate_id bigint, kind text, result jsonb) ON CONFLICT (release_id, candidate_id) DO UPDATE SET result = excluded.result, created_at = now()`, // A par2 set's file list goes to par2_files, not into every inspection. - [JSON.stringify(done.map((o) => ({ release_id: o.candidate.releaseId, candidate_id: o.candidate.id, kind: o.candidate.kind, result: { ...o.inspection, par2: undefined } })))] + [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) // Each touched release from all its inspections (earlier rounds may have stored others). @@ -305,7 +316,7 @@ export class Prober { await trx.rawQuery( `UPDATE releases r SET passworded = v.passworded, inner_name = v.inner_name, inspected_at = now() FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, passworded boolean, inner_name text) WHERE r.id = v.id`, - [JSON.stringify(updates)] + [jsonb(updates)] ) }) } @@ -318,7 +329,7 @@ export class Prober { `INSERT INTO file_hashes (candidate_id, release_id, hash16k, size) SELECT v.candidate_id, v.release_id, v.hash16k, v.size FROM jsonb_to_recordset(?::jsonb) AS v(candidate_id bigint, release_id bigint, hash16k text, size bigint) ON CONFLICT (candidate_id) DO NOTHING`, - [JSON.stringify(hashed.map((o) => ({ candidate_id: o.candidate.id, release_id: o.candidate.releaseId, hash16k: o.inspection!.hash16k, size: o.inspection!.size })))] + [jsonb(hashed.map((o) => ({ candidate_id: o.candidate.id, release_id: o.candidate.releaseId, hash16k: o.inspection!.hash16k, size: o.inspection!.size })))] ) } for (const o of done) { @@ -328,7 +339,7 @@ export class Prober { `INSERT INTO par2_files (set_id, file_id, hash16k, length, name) SELECT ?, v.file_id, v.hash16k, v.length, v.name FROM jsonb_to_recordset(?::jsonb) AS v(file_id text, hash16k text, length bigint, name text) ON CONFLICT DO NOTHING`, - [set.setId, JSON.stringify(set.files.map((f) => ({ file_id: f.fileId, hash16k: f.hash16k, length: f.length, name: f.name })))] + [set.setId, jsonb(set.files.map((f) => ({ file_id: f.fileId, hash16k: f.hash16k, length: f.length, name: f.name })))] ) await trx.rawQuery(`INSERT INTO par2_sets (set_id, release_id, files) VALUES (?, ?, ?) ON CONFLICT DO NOTHING`, [set.setId, o.candidate.releaseId, set.files.length]) } diff --git a/tests/functional/prober.spec.ts b/tests/functional/prober.spec.ts index 70b27b7..3291cbe 100644 --- a/tests/functional/prober.spec.ts +++ b/tests/functional/prober.spec.ts @@ -23,4 +23,23 @@ test.group('Prober', (group) => { const status = Object.fromEntries((await db.from('probe_candidates').select('kind', 'status')).map((c) => [c.kind, c.status])) assert.deepEqual(status, { 'lone': 'taken', 'par2': 'taken', 'archive': 'taken', '7z-tail': 'waiting' }) }) + + test('a flush stores a name with a NUL in it, and one article that two releases have as a candidate', async ({ assert }) => { + const release = async (key: string) => + Number((await db.table('releases').insert({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'subject', groups: '[]', files: 1 }).returning('id'))[0].id) + const a = await release('fa') + const b = await release('fb') + const candidate = async (releaseId: number) => + Number((await db.table('probe_candidates').insert({ release_id: releaseId, kind: 'archive', message_id: '', file: 'x.rar', status: 'taken' }).returning('id'))[0].id) + const inspection = { yencName: 'x.rar', size: 1000, type: null, names: ['Movie\u0000.mkv'], entries: 1, passworded: false } + const prober = new Prober() as any + for (const releaseId of [a, b]) { + prober.outcomes.push({ candidate: { id: await candidate(releaseId), releaseId, kind: 'archive', messageId: '' }, status: 'done', inspection }) + } + await prober.flush() + assert.deepEqual((await db.from('probe_candidates').select('status')).map((c) => c.status), ['done', 'done']) + const cached = await db.from('probe_cache').where('message_id', '').firstOrFail() + assert.deepEqual(cached.result.names, ['Movie.mkv']) + assert.equal((await db.from('release_inspections').count('* as n'))[0].n, '2') + }) })