Daan MeijerandClaude Opus 5.5
cfe85abc3e
Eweka sends some ranges' compressed XOVER truncated, the same bytes every time, while the plain reply is whole (a.b.models 3317215-3367214: 1,675,600 bytes that don't inflate, twice; 50,000 lines plain; tools/false_end.ts). Not a false end of reply. The worker's corrupt report removed the chunk and the range was fetched again, with the same bytes: one range was reported 540 times. (Before this morning's change the inflate check waited for more data until the 10-minute timeout, like voyager 1's 'chunk never answered' ranges.) A reported chunk's start goes into ranges.plain_chunks; the prefetcher fetches those over one connection without XFEATURE COMPRESS GZIP (NntpPool.usePlain, opened on demand) and deflates them here. A plain chunk reported again sets the range aside (status failed, error). The capture (edge:bench --capture) fetches such a chunk plain too. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
117 lines
6.4 KiB
TypeScript
117 lines
6.4 KiB
TypeScript
import { test } from '@japa/runner'
|
|
import db from '@adonisjs/lucid/services/db'
|
|
import testUtils from '@adonisjs/core/services/test_utils'
|
|
import { deflateSync } from 'node:zlib'
|
|
import { mkdtemp, rm } from 'node:fs/promises'
|
|
import { tmpdir } from 'node:os'
|
|
import { join } from 'node:path'
|
|
import Node from '#models/node'
|
|
import Spool from '#services/spool'
|
|
import coordinatorConfig from '#config/coordinator'
|
|
import { decodeFrames, fromWire, toWire, RangeIndexer, type Lease } from '#lib/edge/index'
|
|
import { parseOverviews } from '#lib/nntp/index'
|
|
import { unzipSync } from 'node:zlib'
|
|
|
|
const GROUP = 'alt.binaries.test'
|
|
|
|
/** One overview chunk: a two-part release posted at `at`. */
|
|
function chunk(at: number): Buffer {
|
|
const lines = [1, 2].map(
|
|
(n) => `${at + n - 1}\tSome.Show.S01E0${at % 7} [1/1] - "Some.Show.S01E0${at % 7}.1080p.mkv" yEnc (${n}/2)\tp <p@x>\tMon, 05 Oct 2026 10:00:00 +0000\t<a${at}-${n}@test>\t\t700000\t5000\tXref: news.example ${GROUP}:${at + n - 1}`
|
|
)
|
|
// An obfuscated subject: counted under the key '' (which must survive the body parser).
|
|
lines.push(`${at + 2}\tobfuscated\tp <p@x>\tMon, 05 Oct 2026 10:00:00 +0000\t<o${at}@test>\t\t700000\t5000\tXref: news.example ${GROUP}:${at + 2}`)
|
|
return deflateSync(Buffer.from(lines.join('\r\n') + '\r\n'))
|
|
}
|
|
|
|
test.group('Node API', (group) => {
|
|
const config = { ...coordinatorConfig, spool: { ...coordinatorConfig.spool } }
|
|
let dir = ''
|
|
group.each.setup(async () => {
|
|
dir = await mkdtemp(join(tmpdir(), 'spool-'))
|
|
coordinatorConfig.spool.dir = dir
|
|
coordinatorConfig.chunkArticles = 100
|
|
return testUtils.db().withGlobalTransaction()
|
|
})
|
|
group.each.teardown(async () => {
|
|
coordinatorConfig.spool.dir = config.spool.dir
|
|
coordinatorConfig.chunkArticles = config.chunkArticles
|
|
await rm(dir, { recursive: true, force: true })
|
|
})
|
|
|
|
test('lease, stream from the spool, post batches, finish', async ({ client, assert }) => {
|
|
const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
|
|
const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 300, next_id: 1, spooled_chunks: 3 }).returning('id')
|
|
const rangeId = Number(r.id)
|
|
for (const span of Spool.chunksOf({ start: 1, end: 300 })) await Spool.write(rangeId, span, chunk(span.from + 10))
|
|
const node = await Node.create({ name: 'n1' })
|
|
const token = (await Node.accessTokens.create(node)).value!.release()
|
|
|
|
// No token, no lease.
|
|
;(await client.post('/api/v1/node/lease')).assertStatus(401)
|
|
|
|
const lease = await client.post('/api/v1/node/lease').bearerToken(token)
|
|
lease.assertStatus(200)
|
|
const leased = lease.body() as Lease
|
|
assert.deepEqual(leased.range, { id: rangeId, group: GROUP, start: 1, end: 300, next: 1 })
|
|
assert.equal(leased.rules.priorities[GROUP], 1)
|
|
// Nothing else spooled: no second lease.
|
|
;(await client.post('/api/v1/node/lease').bearerToken(token)).assertStatus(204)
|
|
|
|
const stream = await client.get(`/api/v1/node/ranges/${rangeId}/stream`).bearerToken(token)
|
|
stream.assertStatus(200)
|
|
const raw = stream.body() as unknown
|
|
const body: Buffer = Buffer.isBuffer(raw) ? raw : Buffer.from(stream.text(), 'binary')
|
|
const frames = []
|
|
for await (const frame of decodeFrames([body])) frames.push(frame)
|
|
assert.deepEqual(frames.map((f) => [f.from, f.to]), [[1, 100], [101, 200], [201, 300]])
|
|
|
|
const indexer = new RangeIndexer({ group: GROUP, start: 1, end: 300, rules: leased.rules })
|
|
for (const f of frames) indexer.add(parseOverviews(unzipSync(f.data).toString('utf8')), f.to + 1)
|
|
indexer.finish()
|
|
const batch = toWire(indexer.take())
|
|
assert.deepEqual(fromWire(batch).releases.length, 3)
|
|
const posted = await client.post(`/api/v1/node/ranges/${rangeId}/batches`).bearerToken(token).json(batch)
|
|
posted.assertStatus(200)
|
|
assert.include(posted.body(), { releases: 3, newReleases: 3 })
|
|
// The same batch again: refused.
|
|
;(await client.post(`/api/v1/node/ranges/${rangeId}/batches`).bearerToken(token).json(batch)).assertStatus(409)
|
|
|
|
assert.deepEqual((await db.from('rule_counts').where('range_id', rangeId)).map((c) => [c.rule, c.key, Number(c.articles)]), [['obfuscated', '', 3]])
|
|
|
|
;(await client.post(`/api/v1/node/ranges/${rangeId}/finish`).bearerToken(token)).assertStatus(200)
|
|
const range = await db.from('ranges').where('id', rangeId).first()
|
|
assert.equal(range.status, 'done')
|
|
assert.isFalse(await Spool.has(rangeId, { from: 1, to: 100 }))
|
|
})
|
|
test('a chunk that does not inflate: reported, removed from the spool, the range goes back', async ({ client, assert }) => {
|
|
const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
|
|
const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 300, next_id: 1, spooled_chunks: 3 }).returning('id')
|
|
const rangeId = Number(r.id)
|
|
for (const span of Spool.chunksOf({ start: 1, end: 300 })) await Spool.write(rangeId, span, chunk(span.from + 10))
|
|
const node = await Node.create({ name: 'n1' })
|
|
const token = (await Node.accessTokens.create(node)).value!.release()
|
|
;(await client.post('/api/v1/node/lease').bearerToken(token)).assertStatus(200)
|
|
|
|
;(await client.post(`/api/v1/node/ranges/${rangeId}/corrupt`).bearerToken(token).json({ from: 101 })).assertStatus(200)
|
|
assert.deepEqual(
|
|
await Promise.all(Spool.chunksOf({ start: 1, end: 300 }).map((span) => Spool.has(rangeId, span))),
|
|
[true, false, true]
|
|
)
|
|
let range = await db.from('ranges').where('id', rangeId).first()
|
|
assert.equal(range.status, 'planned')
|
|
assert.isNull(range.node_id)
|
|
// Fetched plain next time (the server's compressed reply was cut short).
|
|
assert.deepEqual(range.plain_chunks.map(Number), [101])
|
|
// No longer this node's: a second report is refused.
|
|
;(await client.post(`/api/v1/node/ranges/${rangeId}/corrupt`).bearerToken(token).json({ from: 101 })).assertStatus(409)
|
|
// Leased again and the plain chunk still doesn't inflate: the range is set aside, not leased in a loop.
|
|
await Spool.write(rangeId, { from: 101, to: 200 }, chunk(111))
|
|
;(await client.post('/api/v1/node/lease').bearerToken(token)).assertStatus(200)
|
|
;(await client.post(`/api/v1/node/ranges/${rangeId}/corrupt`).bearerToken(token).json({ from: 101 })).assertStatus(200)
|
|
range = await db.from('ranges').where('id', rangeId).first()
|
|
assert.equal(range.status, 'failed')
|
|
assert.include(range.error, 'also fetched plain')
|
|
})
|
|
})
|