About 5% of seeded headers are fragments. They went through staging, where they waited 30 minutes for parts that were never coming (the neighbouring range is worked on by another node, often hours later) and expired: the staging table grew to 22 GB and the NVMe part of the database reached its 60 GB limit, which stopped leasing and the backfill. Files cut by a range edge can still be completed by the neighbouring range, so fragments within 50k articles of an edge go straight into binaries and parts; the rest, files with articles missing for good, are dropped. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
138 lines
7.8 KiB
TypeScript
138 lines
7.8 KiB
TypeScript
import { test } from '@japa/runner'
|
|
import { gzipSync } from 'node:zlib'
|
|
import db from '@adonisjs/lucid/services/db'
|
|
import testUtils from '@adonisjs/core/services/test_utils'
|
|
import SeedOrchestrator from '#services/SeedOrchestrator'
|
|
import QueueService from '#services/QueueService'
|
|
import indexerConfig from '#config/indexer'
|
|
import SeedFetchPool from '#services/SeedFetchPool'
|
|
import { deflateSync } from 'node:zlib'
|
|
import { segmentFromHeader } from '#lib/ingest/segment'
|
|
import { FileAggregator } from '#lib/aggregate/index'
|
|
import { toFragment, toSeedFile, type SeedBatch } from '#lib/seed/wire'
|
|
import type { IngestSegment } from '#lib/ingest/segment'
|
|
|
|
function header(subject: string, n: number) {
|
|
return { 'subject': subject, 'from': 'api seeder <api@example.invalid>', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', 'message-id': `<api-${n}-${subject.length}@test>`, ':bytes': '500000' }
|
|
}
|
|
|
|
function batch(next: number): SeedBatch {
|
|
const segments = [1, 2].map((n) => segmentFromHeader(header(`Api.Test.2026 - "Api.Test.2026.mkv" yEnc (${n}/2)`, n), 'alt.binaries.test', n)!)
|
|
const lone = segmentFromHeader(header('Api.Test.2026 - "Api.Test.2026.nfo" yEnc (1/3)', 9), 'alt.binaries.test', 9)!
|
|
const aggregator = new FileAggregator<IngestSegment>()
|
|
return { next, headers: 3, obfuscated: 0, deferred: 0, files: aggregator.add(segments).map(toSeedFile), fragments: [toFragment(lone)] }
|
|
}
|
|
|
|
test.group('Seed API', (group) => {
|
|
group.each.setup(() => testUtils.db().withGlobalTransaction())
|
|
group.teardown(() => QueueService.closeAll())
|
|
|
|
test('a node leases a range, posts a gzip batch, and finishes it', async ({ client, assert }) => {
|
|
const token = await SeedOrchestrator.issueToken('test-node')
|
|
await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 })
|
|
const auth = { authorization: `Bearer ${token}` }
|
|
|
|
;(await client.post('/api/v1/seed/lease')).assertStatus(401)
|
|
const leased = await client.post('/api/v1/seed/lease').headers(auth)
|
|
leased.assertStatus(200)
|
|
const range = leased.body().range
|
|
assert.include(range, { group: 'alt.binaries.test', start: 1, end: 100 })
|
|
|
|
// As seed:node sends it: a gzip-compressed JSON body.
|
|
const posted = await fetch(`http://${process.env.HOST}:${process.env.PORT}/api/v1/seed/ranges/${range.id}/batches`, {
|
|
method: 'POST',
|
|
headers: { ...auth, 'content-type': 'application/json', 'content-encoding': 'gzip' },
|
|
body: gzipSync(JSON.stringify(batch(51))),
|
|
})
|
|
assert.equal(posted.status, 200)
|
|
assert.include(((await posted.json()) as any).stored, { files: 1 })
|
|
const file = await db.from('files').where('filename', 'Api.Test.2026.mkv').first()
|
|
assert.equal(file.parts, 2)
|
|
// The lone .nfo segment sits near the range's edge: kept as a binary (not staged) for the neighbouring range.
|
|
assert.isOk(await db.from('binaries').where('name', 'Api.Test.2026.nfo').first())
|
|
|
|
const row = await db.from('seed_ranges').where('id', range.id).first()
|
|
assert.equal(Number(row.next_id), 51)
|
|
assert.equal(Number(row.headers), 3)
|
|
|
|
;(await client.post(`/api/v1/seed/ranges/${range.id}/done`).headers(auth)).assertStatus(200)
|
|
assert.equal((await db.from('seed_ranges').where('id', range.id).first()).status, 'done')
|
|
;(await client.post('/api/v1/seed/lease').headers(auth)).assertStatus(204)
|
|
})
|
|
|
|
test('a quiet lease passes to another node, and the first one gets 409', async ({ client, assert }) => {
|
|
const first = await SeedOrchestrator.issueToken('node-a')
|
|
const second = await SeedOrchestrator.issueToken('node-b')
|
|
await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 })
|
|
const range = (await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${first}` })).body().range
|
|
await db.from('seed_ranges').where('id', range.id).update({ next_id: 40, heartbeat_at: new Date(Date.now() - 60 * 60 * 1000) })
|
|
|
|
const taken = (await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${second}` })).body().range
|
|
assert.equal(taken.id, range.id)
|
|
assert.equal(taken.start, 40)
|
|
const late = await client.post(`/api/v1/seed/ranges/${range.id}/batches`).headers({ authorization: `Bearer ${first}` }).json(batch(60))
|
|
late.assertStatus(409)
|
|
|
|
;(await client.post(`/api/v1/seed/ranges/${range.id}/release`).headers({ authorization: `Bearer ${second}` }).json({ error: 'test' })).assertStatus(200)
|
|
assert.equal((await db.from('seed_ranges').where('id', range.id).first()).status, 'queued')
|
|
})
|
|
|
|
test('no leases while the database is over the disk limit', async ({ client }) => {
|
|
const token = await SeedOrchestrator.issueToken('full-disk')
|
|
await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 })
|
|
const limit = indexerConfig.backfill.maxDefaultGb
|
|
indexerConfig.backfill.maxDefaultGb = 0
|
|
try {
|
|
;(await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${token}` })).assertStatus(204)
|
|
} finally {
|
|
indexerConfig.backfill.maxDefaultGb = limit
|
|
}
|
|
;(await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${token}` })).assertStatus(200)
|
|
})
|
|
|
|
test('a range given back before it started does not use up an attempt', async ({ client, assert }) => {
|
|
const token = await SeedOrchestrator.issueToken('no-news-server')
|
|
await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 })
|
|
const auth = { authorization: `Bearer ${token}` }
|
|
const range = (await client.post('/api/v1/seed/lease').headers(auth)).body().range
|
|
;(await client.post(`/api/v1/seed/ranges/${range.id}/release`).headers(auth).json({ error: 'NNTP login: 502', uncounted: true })).assertStatus(200)
|
|
const row = await db.from('seed_ranges').where('id', range.id).first()
|
|
assert.equal(row.status, 'queued')
|
|
assert.equal(row.attempts, 0)
|
|
})
|
|
|
|
test('relay: a leased range gets its overview chunks through the API', async ({ client, assert }) => {
|
|
const token = await SeedOrchestrator.issueToken('relay-node')
|
|
await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 })
|
|
const auth = { authorization: `Bearer ${token}` }
|
|
const range = (await client.post('/api/v1/seed/lease').headers(auth)).body().range
|
|
const fetch = SeedFetchPool.fetch
|
|
const connections = indexerConfig.seed.nntpConnections
|
|
const asked: unknown[] = []
|
|
SeedFetchPool.fetch = async (group, from, to) => {
|
|
asked.push([group, from, to])
|
|
return deflateSync(Buffer.from('5\tsubject\r\n'))
|
|
}
|
|
try {
|
|
indexerConfig.seed.nntpConnections = 0
|
|
;(await client.get(`/api/v1/seed/ranges/${range.id}/chunk`).qs({ from: 1, to: 50 }).headers(auth)).assertStatus(503)
|
|
indexerConfig.seed.nntpConnections = 1
|
|
const chunk = await client.get(`/api/v1/seed/ranges/${range.id}/chunk`).qs({ from: 50, to: 500 }).headers(auth)
|
|
chunk.assertStatus(200)
|
|
// Clamped to the range.
|
|
assert.deepEqual(asked, [['alt.binaries.test', 50, 100]])
|
|
;(await client.get(`/api/v1/seed/ranges/${range.id}/chunk`).qs({ from: 200, to: 300 }).headers(auth)).assertStatus(400)
|
|
} finally {
|
|
SeedFetchPool.fetch = fetch
|
|
indexerConfig.seed.nntpConnections = connections
|
|
}
|
|
})
|
|
|
|
test('a range reported to its end but never finished is marked done, not leased', async ({ client, assert }) => {
|
|
const token = await SeedOrchestrator.issueToken('late-node')
|
|
const [range] = await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 101 }).returning('id')
|
|
;(await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${token}` })).assertStatus(204)
|
|
assert.equal((await db.from('seed_ranges').where('id', range.id).first()).status, 'done')
|
|
})
|
|
})
|