Daan MeijerandClaude Opus 5.5
3b61215f6b
The live tail reads an upload in pieces while it is posted, and each piece queued a 7z-tail candidate for the last volume it had seen so far (release 306905984: seven, for .002-.008). Only the last volume's end holds the 7z listing. When candidates are stored, and when releases are merged, waiting tails below the release's highest volume are deleted. All tails of the highest volume stay: a piece that saw only part of it queued a segment that is not its last. RAR is unaffected: it lists its files in the first volume's header and gets no tail candidate. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
209 lines
13 KiB
TypeScript
209 lines
13 KiB
TypeScript
import { test } from '@japa/runner'
|
|
import db from '@adonisjs/lucid/services/db'
|
|
import testUtils from '@adonisjs/core/services/test_utils'
|
|
import BatchStore, { StaleBatch } from '#services/batch_store'
|
|
import { RangeIndexer } from '#lib/edge/index'
|
|
import { decodeChunk } from '#lib/chunks/index'
|
|
import { DEFAULT_RULES } from '#lib/rules/index'
|
|
import type { Overview } from '#lib/nntp/index'
|
|
|
|
const GROUP = 'alt.binaries.test'
|
|
let id = 0
|
|
|
|
function post(at: number, name: string, parts: number, counter: string, tool: 'ngPost' | 'plain' = 'plain'): Overview[] {
|
|
return Array.from({ length: parts }, (_, i) => {
|
|
id++
|
|
const mid = tool === 'ngPost' ? `${id.toString(16).padStart(32, '0')}@ngPost` : `art${id}@test`
|
|
return [at + i, { 'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`, 'from': 'p <p@x>', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', 'message-id': `<${mid}>`, ':bytes': '700000', 'xref': `news.example ${GROUP}:${at + i}` }] as Overview
|
|
})
|
|
}
|
|
|
|
async function leasedRange() {
|
|
const [group] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
|
|
const [node] = await db.table('nodes').insert({ name: 'test-node' }).returning('id')
|
|
const [range] = await db
|
|
.table('ranges')
|
|
.insert({ group_id: group.id, start_id: 1, end_id: 1_000_000, next_id: 1, status: 'leased', node_id: node.id, rule_version: 1 })
|
|
.returning('id')
|
|
return { rangeId: Number(range.id), nodeId: Number(node.id) }
|
|
}
|
|
|
|
function indexer() {
|
|
return new RangeIndexer({ group: GROUP, start: 1, end: 1_000_000, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } }, edgeArticles: 1000 })
|
|
}
|
|
|
|
/** One file's articles, posted `minutes` apart from `start` (an ISO time), in `group`. */
|
|
function upload(at: number, name: string, parts: number, counter: string, start: string, group = GROUP, minutes = 1): Overview[] {
|
|
return Array.from({ length: parts }, (_, i) => {
|
|
id++
|
|
const date = new Date(Date.parse(start) + i * minutes * 60_000).toUTCString().replace('GMT', '+0000')
|
|
return [at + i, { 'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`, 'from': 'p <p@x>', 'date': date, 'message-id': `<up${id}@test>`, ':bytes': '700000', 'xref': `news.example ${group}:${at + i}` }] as Overview
|
|
})
|
|
}
|
|
|
|
async function rangeIn(groupName: string, start: number, nodeId?: number) {
|
|
const g = (await db.from('groups').where('name', groupName).first()) ?? (await db.table('groups').insert({ name: groupName, priority: groupName === GROUP ? 1 : 2 }).returning('*'))[0]
|
|
const node = nodeId ?? Number((await db.table('nodes').insert({ name: `n${groupName}${start}` }).returning('id'))[0].id)
|
|
const [range] = await db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: start + 99_999, next_id: start, status: 'leased', node_id: node, rule_version: 1 }).returning('id')
|
|
return { rangeId: Number(range.id), nodeId: node }
|
|
}
|
|
|
|
async function storeIn(groupName: string, start: number, overviews: Overview[]) {
|
|
const { rangeId, nodeId } = await rangeIn(groupName, start)
|
|
const ix = new RangeIndexer({ group: groupName, start, end: start + 99_999, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1, 'alt.binaries.other': 2 } }, edgeArticles: 1 })
|
|
ix.add(overviews, start + 100_000)
|
|
ix.finish()
|
|
await BatchStore.store(rangeId, nodeId, ix.take())
|
|
}
|
|
|
|
const releasesOf = () => db.from('releases').orderBy('id').select('release_key', 'identity_key', 'files', 'posted_at', 'last_posted_at')
|
|
|
|
test.group('BatchStore', (group) => {
|
|
group.each.setup(() => testUtils.db().withGlobalTransaction())
|
|
|
|
test('stores releases, their chunks, probe candidates, rule counts and fragments, and moves the resume point', async ({ assert }) => {
|
|
const { rangeId, nodeId } = await leasedRange()
|
|
const ix = indexer()
|
|
ix.add([
|
|
...post(100_000, 'Some.Show.S01E01.1080p.WEB.part1.rar', 2, 'Some.Show.S01E01 [1/2]'),
|
|
...post(100_010, 'Some.Show.S01E01.1080p.WEB.par2', 1, 'Some.Show.S01E01 [2/2]'),
|
|
...post(100_020, 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 3, '[1/1]', 'ngPost'),
|
|
...post(10, 'Edge.Cut.2020.part1.rar', 3, 'Edge.Cut [1/1]').slice(1),
|
|
], 500_000)
|
|
const first = ix.take()
|
|
const stored = await BatchStore.store(rangeId, nodeId, first)
|
|
assert.include(stored, { releases: 1, newReleases: 1, candidates: 2, fragments: 0 })
|
|
|
|
const release = await db.from('releases').first()
|
|
assert.include(release, { files: 2, segments: '3', chunks: 1, readable: true, has_par2: true, file_total: 2 })
|
|
const chunk = await db.from('release_chunks').where('release_id', release.id).first()
|
|
assert.deepEqual(decodeChunk(chunk.payload).map((f) => f.name).sort(), ['Some.Show.S01E01.1080p.WEB.par2', 'Some.Show.S01E01.1080p.WEB.part1.rar'])
|
|
assert.deepEqual((await db.from('probe_candidates').orderBy('kind')).map((c) => c.kind), ['archive', 'par2'])
|
|
assert.deepEqual((await db.from('rule_counts')).map((c) => [c.rule, c.key, Number(c.articles)]), [['garbage', 'ngPost alt.binaries.test', 3]])
|
|
assert.equal(Number((await db.from('ranges').where('id', rangeId).first()).next_id), 500_000)
|
|
|
|
// The same batch again (a retry after a lost reply) is refused, and changes nothing.
|
|
await assert.rejects(() => BatchStore.store(rangeId, nodeId, first), StaleBatch)
|
|
assert.equal((await db.from('releases').first()).files, 2)
|
|
|
|
// The end of the range: the cut-off file near the start edge is kept as a fragment.
|
|
ix.finish()
|
|
const last = ix.take()
|
|
assert.include(await BatchStore.store(rangeId, nodeId, last), { fragments: 1 })
|
|
assert.equal(Number((await db.from('ranges').where('id', rangeId).first()).next_id), 1_000_001)
|
|
})
|
|
|
|
test('a range leased to another node is refused', async ({ assert }) => {
|
|
const { rangeId, nodeId } = await leasedRange()
|
|
const ix = indexer()
|
|
ix.add(post(100_000, 'A.Release.2020.mkv', 1, 'A [1/1]'), 200_000)
|
|
await assert.rejects(() => BatchStore.store(rangeId, nodeId + 1, ix.take()), /not leased to this node/)
|
|
})
|
|
test('a re-upload is a release of its own: in one range, or a later range; a continuation joins', async ({ assert }) => {
|
|
// One range: the same file posted Monday and again Thursday.
|
|
await storeIn(GROUP, 1, [
|
|
...upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/2]', '2026-10-05T10:00:00Z'),
|
|
...upload(20, 'Some.Show.S01E01.1080p.par2', 1, 'Some.Show.S01E01.1080p [2/2]', '2026-10-05T10:05:00Z'),
|
|
...upload(500, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/2]', '2026-10-08T10:00:00Z'),
|
|
])
|
|
let rows = await releasesOf()
|
|
assert.lengthOf(rows, 2)
|
|
assert.equal(rows[1].release_key, `${rows[0].release_key}:2`)
|
|
assert.deepEqual(rows.map((r) => [r.identity_key, r.files]), [[rows[0].release_key, 2], [rows[0].release_key, 1]])
|
|
|
|
// The next range: Thursday's par2 an hour later joins Thursday's release; a Saturday copy is a third.
|
|
await storeIn(GROUP, 100_001, upload(100_010, 'Some.Show.S01E01.1080p.par2', 1, 'Some.Show.S01E01.1080p [2/2]', '2026-10-08T11:00:00Z'))
|
|
await storeIn(GROUP, 200_001, upload(200_010, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/2]', '2026-10-10T10:00:00Z'))
|
|
rows = await releasesOf()
|
|
assert.deepEqual(rows.map((r) => [r.release_key.split(':')[1] ?? '1', r.files]), [['1', 2], ['2', 2], ['3', 1]])
|
|
const chunks = await db.from('release_chunks').count('* as n').first()
|
|
assert.equal(Number(chunks!.n), 4)
|
|
})
|
|
|
|
test('candidates added to a settled release (a later range of its upload) are claimable at once', async ({ assert }) => {
|
|
await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E02.1080p.mkv', 2, 'Some.Show.S01E02.1080p [1/2]', '2026-10-05T10:00:00Z'))
|
|
// Settled before the next range came (the settler marks only what it settles).
|
|
await db.from('releases').update({ settled_at: new Date() })
|
|
await db.from('probe_candidates').update({ ready: true })
|
|
await storeIn(GROUP, 100_001, upload(100_010, 'Some.Show.S01E02.1080p.par2', 1, 'Some.Show.S01E02.1080p [2/2]', '2026-10-05T10:20:00Z'))
|
|
const candidates = await db.from('probe_candidates').orderBy('id').select('kind', 'ready')
|
|
assert.deepEqual(candidates.map((c) => c.kind), ['par2'])
|
|
assert.isTrue(candidates.every((c) => c.ready))
|
|
assert.lengthOf(await releasesOf(), 1)
|
|
})
|
|
|
|
test('a split 7z read in pieces keeps the tail of its highest volume only; RAR gets none', async ({ assert }) => {
|
|
// As the live tail reads it while posted: each piece queues a tail for the last volume it saw so far.
|
|
await storeIn(GROUP, 1, [
|
|
...upload(10, 'TsIgNBewuFbuwWy.7z.001', 2, '[1/4]', '2026-10-23T08:24:00Z'),
|
|
...upload(20, 'TsIgNBewuFbuwWy.7z.002', 2, '[2/4]', '2026-10-23T08:26:00Z'),
|
|
])
|
|
await storeIn(GROUP, 100_001, upload(100_010, 'TsIgNBewuFbuwWy.7z.004', 2, '[4/4]', '2026-10-23T08:30:00Z'))
|
|
// A lower volume posted late: its tail goes at once.
|
|
await storeIn(GROUP, 200_001, upload(200_010, 'TsIgNBewuFbuwWy.7z.003', 2, '[3/4]', '2026-10-23T08:28:00Z'))
|
|
await storeIn(GROUP, 300_001, upload(300_010, 'Some.Show.S01E03.part01.rar', 2, 'Some.Show.S01E03 [1/2]', '2026-10-23T09:00:00Z'))
|
|
await storeIn(GROUP, 400_001, upload(400_010, 'Some.Show.S01E03.part02.rar', 2, 'Some.Show.S01E03 [2/2]', '2026-10-23T09:02:00Z'))
|
|
const candidates = await db.from('probe_candidates').orderBy('file').select('kind', 'file')
|
|
assert.deepEqual(
|
|
candidates.map((c) => [c.kind, c.file]),
|
|
[
|
|
['archive', 'Some.Show.S01E03.part01.rar'],
|
|
['archive', 'TsIgNBewuFbuwWy.7z.001'],
|
|
['7z-tail', 'TsIgNBewuFbuwWy.7z.004'],
|
|
]
|
|
)
|
|
})
|
|
|
|
test('the same upload in another group (multi-posted) is a release of its own', async ({ assert }) => {
|
|
await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:00:00Z'))
|
|
await storeIn('alt.binaries.other', 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:01:00Z', 'alt.binaries.other'))
|
|
const rows = await releasesOf()
|
|
assert.lengthOf(rows, 2)
|
|
assert.equal(rows[0].identity_key, rows[1].identity_key)
|
|
})
|
|
|
|
test('an upload spread over groups (other files in another group) is one release', async ({ assert }) => {
|
|
await storeIn(GROUP, 1, upload(10, 'Some.Album-WEB-2026-FTD.part1.rar', 2, 'Some.Album-WEB-2026-FTD [2/3]', '2026-10-05T10:00:00Z'))
|
|
await storeIn('alt.binaries.other', 1, upload(10, 'Some.Album-WEB-2026-FTD.part2.rar', 2, 'Some.Album-WEB-2026-FTD [3/3]', '2026-10-05T10:00:30Z', 'alt.binaries.other'))
|
|
const rows = await releasesOf()
|
|
assert.lengthOf(rows, 1)
|
|
assert.equal(rows[0].files, 2)
|
|
})
|
|
|
|
test('a release key taken meanwhile by another group\'s writer is skipped for the next number', async ({ assert }) => {
|
|
await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:00:00Z'))
|
|
const [first] = await releasesOf()
|
|
// As if another group's batch numbered the key at the same moment: key:2 exists, not counted for this key.
|
|
const [g] = await db.table('groups').insert({ name: 'alt.binaries.third', priority: 3 }).returning('id')
|
|
await db.table('releases').insert({ release_key: `${first.identity_key}:2`, identity_key: 'someone-else', group_id: g.id, name: 'x', source: 'subject', poster: 'p', groups: JSON.stringify([]), files: 0, segments: 0, bytes: 0, chunks: 0 })
|
|
await storeIn('alt.binaries.other', 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:01:00Z', 'alt.binaries.other'))
|
|
const keys = (await releasesOf()).filter((r) => r.identity_key === first.identity_key).map((r) => r.release_key)
|
|
assert.deepEqual(keys, [first.identity_key, `${first.identity_key}:3`])
|
|
})
|
|
|
|
test('a re-upload is numbered after the highest number, not into the holes merges left', async ({ assert }) => {
|
|
for (let day = 0; day < 4; day++) {
|
|
await storeIn(GROUP, 1 + day * 100_000, upload(10 + day * 100_000, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', `2026-10-0${day * 2 + 1}T10:00:00Z`))
|
|
}
|
|
const [first] = await releasesOf()
|
|
// Merged away: key:2 and key:3 gone, key and key:4 left.
|
|
await db.from('releases').whereIn('release_key', [`${first.identity_key}:2`, `${first.identity_key}:3`]).delete()
|
|
await storeIn(GROUP, 400_001, upload(400_010, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-09T10:00:00Z'))
|
|
const keys = (await releasesOf()).map((r) => r.release_key.split(':')[1] ?? '1')
|
|
assert.deepEqual(keys, ['1', '4', '5'])
|
|
})
|
|
|
|
test('a small release of sidecars only is marked for the sidecar fold', async ({ assert }) => {
|
|
await storeIn(GROUP, 1_000_000, [
|
|
...upload(1_000_000, 'Murdoch Mysteries S8D1.nzb', 1, '[00/80]', '2026-10-05T10:00:00Z'),
|
|
...upload(1_000_010, 'Some.Tool.v1.zip', 1, '[01/01]', '2026-10-05T10:00:00Z'),
|
|
...upload(1_000_020, 'Some.Show.S01E01.part01.rar', 2, '[02/30]', '2026-10-05T10:00:00Z'),
|
|
])
|
|
const flags = new Map((await db.from('releases').select('name', 'sidecar_only')).map((r) => [r.name, r.sidecar_only]))
|
|
assert.isTrue(flags.get('Murdoch Mysteries S8D1'))
|
|
assert.isFalse(flags.get('Some.Tool.v1'))
|
|
// Over 1 MB: not looked at.
|
|
assert.isNull(flags.get('Some.Show.S01E01'))
|
|
})
|
|
})
|