diff --git a/app/services/leases.ts b/app/services/leases.ts index 27e9407..1a53222 100644 --- a/app/services/leases.ts +++ b/app/services/leases.ts @@ -67,7 +67,14 @@ class Leases { const range = await this.leased(rangeId, nodeId) const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find((s) => s.from === from) if (span) await Spool.removeChunk(rangeId, span) - await db.from('ranges').where('id', rangeId).update({ status: 'planned', node_id: null }) + const plain: number[] = (range.plain_chunks ?? []).map(Number) + if (plain.includes(from)) { + // Fetched plain (deflated here) and still not inflating: not the server's compression; set aside. + await db.from('ranges').where('id', rangeId).update({ status: 'failed', node_id: null, error: `chunk ${from} doesn't inflate, also fetched plain` }) + return + } + // The server's compressed reply is cut short (the same bytes every time): fetched plain next time. + await db.rawQuery(`UPDATE ranges SET status = 'planned', node_id = NULL, plain_chunks = array_append(plain_chunks, ?::bigint) WHERE id = ?`, [from, rangeId]) } /** A stopping worker gives its range back; the next one resumes it. */ diff --git a/app/services/nntp_pool.ts b/app/services/nntp_pool.ts index 8aef5a4..622076b 100644 --- a/app/services/nntp_pool.ts +++ b/app/services/nntp_pool.ts @@ -75,7 +75,41 @@ class NntpPool { } /** Closes every connection, also those in use: their commands fail. */ + private plain: Promise | null = null + private plainQueue: Promise = Promise.resolve() + + /** + * One connection without XFEATURE COMPRESS GZIP, opened when first needed, + * one command at a time: for chunks whose compressed reply the server + * cuts short (Leases.corrupt). Its replies are deflated here. + */ + async usePlain(work: (client: XoverClient) => Promise): Promise { + const run = this.plainQueue.then(async () => { + this.plain ??= XoverClient.connect({ + host: env.get('NNTP_HOST')!, + port: env.get('NNTP_PORT', 563), + user: env.get('NNTP_USER'), + password: env.get('NNTP_PASSWORD'), + secure: env.get('NNTP_SECURE', true), + compress: false, + timeoutMs: coordinatorConfig.nntpTimeoutMs, + }) + const client = await this.plain + try { + return await work(client) + } catch (error) { + this.plain = null + client.close() + throw error + } + }) + this.plainQueue = run.catch(() => {}) + return run + } + async close() { + void this.plain?.then((c) => c.close()).catch(() => {}) + this.plain = null for (const client of [...this.idle.splice(0), ...this.busy]) client.close() this.busy.clear() this.open = 0 diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index b7948c7..e828078 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -7,6 +7,8 @@ import Spool, { type ChunkSpan } from '#services/spool' interface Target { id: number group: string + /** Chunk starts to fetch without compression (ranges.plain_chunks). */ + plain: Set /** Chunks not spooled yet, in order. */ missing: ChunkSpan[] /** All of the range's chunks, from its start. */ @@ -76,10 +78,11 @@ export class Prefetcher { const { target, span } = next try { const fetchedAt = Date.now() - const data = await NntpPool.use(async (client) => { + const fetch = async (client: Parameters[0]>[0]) => { await client.group(target.group) return client.xoverRaw(span.from, span.to) - }) + } + const data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await NntpPool.use(fetch) this.stats.chunks++ this.stats.articles += span.to - span.from + 1 this.stats.fetchMs += Date.now() - fetchedAt @@ -134,7 +137,7 @@ export class Prefetcher { this.spoolBytes = await Spool.size() } const mine = (q: any) => (this.shard.count > 1 ? q.whereRaw('r.id % ? = ?', [this.shard.count, this.shard.index]) : q) - const rows = (q: any) => mine(q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id')) + const rows = (q: any) => mine(q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id', 'r.plain_chunks')) // Leased ranges, checked against the spool's files whatever spooled_at says: a worker waits for them. const leased = await rows(db.query()).where('r.status', 'leased').orderBy('r.leased_at') // Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease. @@ -154,7 +157,7 @@ export class Prefetcher { for (const span of Spool.chunksOf({ start, end }, Number(row.next_id))) { if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) missing.push(span) } - if (missing.length > 0) targets.push({ id: Number(row.id), group: row.group, missing, total }) + if (missing.length > 0) targets.push({ id: Number(row.id), group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total }) } return targets } diff --git a/commands/edge_bench.ts b/commands/edge_bench.ts index 43489f8..cdb07e9 100644 --- a/commands/edge_bench.ts +++ b/commands/edge_bench.ts @@ -78,7 +78,17 @@ export default class EdgeBench extends BaseCommand { for (let from = this.start; from <= end; from += this.chunk, index++) { const to = Math.min(from + this.chunk - 1, end) const started = Date.now() - const raw = await client.xoverRaw(from, to) + let raw = await client.xoverRaw(from, to) + // Eweka cuts some ranges' compressed reply short (the same bytes every time): fetched plain then. + try { + unzipSync(raw) + } catch { + const plain = await XoverClient.connect({ host: env.get('NNTP_HOST')!, port: env.get('NNTP_PORT')!, user: env.get('NNTP_USER'), password: env.get('NNTP_PASSWORD'), secure: env.get('NNTP_SECURE', true), compress: false, timeoutMs: 300_000 }) + await plain.group(this.group) + raw = await plain.xoverRaw(from, to) + plain.close() + this.logger.warning(`chunk ${index}: the compressed reply didn't inflate; fetched plain`) + } await writeFile(join(this.dir, `${String(index).padStart(4, '0')}-${from}-${to}.gz`), raw) this.logger.info(`chunk ${index}: ${from}–${to}, ${(raw.length / 1e6).toFixed(1)} MB in ${((Date.now() - started) / 1000).toFixed(1)} s`) } diff --git a/database/migrations/1770000001500_ranges_plain_chunks.ts b/database/migrations/1770000001500_ranges_plain_chunks.ts new file mode 100644 index 0000000..2a859be --- /dev/null +++ b/database/migrations/1770000001500_ranges_plain_chunks.ts @@ -0,0 +1,18 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Chunks of a range to fetch without compression: Eweka sends some ranges' + * compressed XOVER truncated, the same bytes every time, while the plain + * reply is whole (LOG.md 2026-10-09 (11)). A worker reports such a chunk; + * its start goes here and the prefetcher fetches it plain. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(`ALTER TABLE ranges ADD COLUMN IF NOT EXISTS plain_chunks bigint[] NOT NULL DEFAULT '{}'`) + this.schema.raw(`ALTER TABLE ranges ADD COLUMN IF NOT EXISTS error text`) + } + + async down() { + this.schema.raw(`ALTER TABLE ranges DROP COLUMN IF EXISTS plain_chunks, DROP COLUMN IF EXISTS error`) + } +} diff --git a/tests/functional/node_api.spec.ts b/tests/functional/node_api.spec.ts index b78f25e..eafa7b7 100644 --- a/tests/functional/node_api.spec.ts +++ b/tests/functional/node_api.spec.ts @@ -98,10 +98,19 @@ test.group('Node API', (group) => { await Promise.all(Spool.chunksOf({ start: 1, end: 300 }).map((span) => Spool.has(rangeId, span))), [true, false, true] ) - const range = await db.from('ranges').where('id', rangeId).first() + 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') }) }) diff --git a/tools/false_end.ts b/tools/false_end.ts new file mode 100644 index 0000000..7553fe4 --- /dev/null +++ b/tools/false_end.ts @@ -0,0 +1,33 @@ +/** + * Whether a compressed XOVER reply that doesn't inflate is the server's: the + * range fetched raw twice (compared with a saved copy), and once uncompressed. + * + * node --env-file=.env --import=@poppinss/ts-exec tools/false_end.ts [saved.gz] + */ +import { readFileSync } from 'node:fs' +import { unzipSync } from 'node:zlib' +import { XoverClient } from '#lib/nntp/index' + +const [group, from, to, saved] = process.argv.slice(2) +const env = process.env +const connect = (compress: boolean) => + XoverClient.connect({ host: env.NNTP_HOST!, port: Number(env.NNTP_PORT ?? 563), user: env.NNTP_USER, password: env.NNTP_PASSWORD, secure: (env.NNTP_SECURE ?? 'true') !== 'false', compress, timeoutMs: 120_000 }) +const gz = await connect(true) +await gz.group(group) +for (let i = 0; i < 2; i++) { + const raw = await gz.xoverRaw(Number(from), Number(to)) + let inflates = 'inflates' + try { + unzipSync(raw) + } catch (e: any) { + inflates = `doesn't inflate (${e.message})` + } + const same = saved ? (Buffer.compare(raw, readFileSync(saved)) === 0 ? ', same bytes as saved' : ', differs from saved') : '' + console.log(`compressed #${i + 1}: ${raw.length} bytes, ${inflates}${same}`) +} +gz.close() +const plain = await connect(false) +await plain.group(group) +const lines = await plain.xover(Number(from), Number(to)) +console.log(`plain: ${lines.length} overview lines`) +plain.close()