diff --git a/REBUILD.md b/REBUILD.md index ee9b026..694a94e 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -248,7 +248,10 @@ corpus filtered this way no longer serves a sweep of those two rules. Windows by posting time (2026-10-25; voyager 3's test data, LOG.md (104)): `ranges:dates` samples when each active group's articles were posted (one XOVER of 32 articles every 500M articles of the group, 16 … 1000 samples, -the median Date of each slice, kept monotone; `group_dates`) and estimates +the median Date of each slice, kept monotone; `group_dates`; over plain +connections: Eweka cuts a compressed reply at the first ".\r\n" in its +output, which some small replies begin with, `TruncatedReply`, LOG.md +2026-10-25 (105)) and estimates every range's posting date from the curve (`ranges.est_date`; the edges of a window are off by up to one sample interval, three weeks in boneless). `ranges:capture --from 2025-01-01 --to 2026-01-01` marks every active diff --git a/commands/nntp_probe.ts b/commands/nntp_probe.ts index 74c02f4..58bc852 100644 --- a/commands/nntp_probe.ts +++ b/commands/nntp_probe.ts @@ -30,6 +30,14 @@ export default class NntpProbe extends BaseCommand { @flags.boolean({ description: 'Only the span itself and the wider ones, no bisection' }) declare noBisect: boolean + @flags.boolean({ description: 'Without XFEATURE COMPRESS GZIP' }) + declare plain: boolean + + @flags.string({ + description: 'Only spans from the article of these sizes, e.g. "8,16,24,32,48,64"', + }) + declare sizes?: string + private timeoutMs = 20_000 private info = { count: 0, low: 0, high: 0 } @@ -41,10 +49,14 @@ export default class NntpProbe extends BaseCommand { this.info = await client.group(this.group) client.close() this.logger.info( - `${this.group}: ${this.info.count} articles, ${this.info.low}–${this.info.high}.` + `${this.group}: ${this.info.count} articles, ${this.info.low}–${this.info.high}${this.plain ? '; plain' : '; compressed'}.` ) const span = (from: number, to: number) => [Math.max(this.info.low, from), Math.min(this.info.high, to)] as const + if (this.sizes) { + for (const n of this.sizes.split(',').map(Number)) await this.probe(...span(a, a + n - 1)) + return + } this.logger.info('The span and wider ones:') await this.probe(...span(a, a + size - 1)) await this.probe(...span(a, a + size * 32 - 1)) @@ -83,7 +95,7 @@ export default class NntpProbe extends BaseCommand { } catch (error: any) { const timedOut = /timed out/i.test(error.message) this.logger.warning( - ` XOVER ${from}-${to}: ${error.message} after ${Date.now() - startedAt} ms${client ? `, ${client.buffered} bytes buffered` : ''}` + ` XOVER ${from}-${to}: ${error.message} after ${Date.now() - startedAt} ms${client ? `, ${client.buffered} bytes buffered: ${JSON.stringify(client.bufferedHead().toString('latin1'))} (${client.bufferedHead().toString('hex')})` : ''}` ) return timedOut } finally { @@ -98,7 +110,7 @@ export default class NntpProbe extends BaseCommand { user: env.get('NNTP_USER'), password: env.get('NNTP_PASSWORD'), secure: env.get('NNTP_SECURE', true), - compress: true, + compress: !this.plain, timeoutMs: this.timeoutMs, }) } diff --git a/commands/ranges_dates.ts b/commands/ranges_dates.ts index b0017e4..d750142 100644 --- a/commands/ranges_dates.ts +++ b/commands/ranges_dates.ts @@ -1,8 +1,9 @@ import { BaseCommand, flags } from '@adonisjs/core/ace' import type { CommandOptions } from '@adonisjs/core/types/ace' import db from '@adonisjs/lucid/services/db' +import env from '#start/env' import coordinatorConfig from '#config/coordinator' -import NntpPool from '#services/nntp_pool' +import { XoverClient } from '#lib/nntp/index' import { NntpShare } from '#services/nntp_shares' import { estimateDate, sampleGroup, type DateSample } from '#services/group_dates' @@ -39,13 +40,9 @@ export default class RangesDates extends BaseCommand { declare estimateOnly: boolean async run() { - try { - if (!this.estimateOnly) await this.sample() - await this.estimate() - await this.report() - } finally { - await NntpPool.close() - } + if (!this.estimateOnly) await this.sample() + await this.estimate() + await this.report() } private groups(missing = this.missing) { @@ -62,20 +59,26 @@ export default class RangesDates extends BaseCommand { const every = this.every ?? 500_000_000 const connections = this.connections ?? 6 const groups = await this.groups() + // Connections of its own, without compression: Eweka cuts a compressed reply at the first ".\r\n" in its + // output, which 16 of 14k small XOVERs hit (TruncatedReply; LOG.md 2026-10-25 (105)); plain, every one + // answers, and 32 lines are a few kB. Taken from the account as a share (the coordinator and the prober make + // room for them), as many as asked for. + const workers = Math.max(1, Math.min(connections, groups.length)) const share = new NntpShare( 'dates', coordinatorConfig.nntpShares.coordinatorPriority + 1, - 1, - connections - ) - const first = await NntpPool.share(share) - const sharing = setInterval( - () => - NntpPool.share(share).catch((error) => this.logger.warning(`NNTP share: ${error.message}`)), - coordinatorConfig.nntpShares.everyMs + workers, + workers ) + let open = 0 + const sync = () => + share + .sync(workers, open) + .catch((error) => this.logger.warning(`NNTP share: ${error.message}`)) + await sync() + const sharing = setInterval(sync, coordinatorConfig.nntpShares.everyMs) this.logger.info( - `Sampling ${groups.length} groups over ${first.granted} connections, a sample every ${every / 1e6}M articles.` + `Sampling ${groups.length} groups over ${workers} plain connections, a sample every ${every / 1e6}M articles.` ) const startedAt = Date.now() let done = 0 @@ -83,12 +86,32 @@ export default class RangesDates extends BaseCommand { let samples = 0 const queue = [...groups] const worker = async () => { + let client: XoverClient | null = null + const use = async (work: (c: XoverClient) => Promise): Promise => { + if (!client) { + client = await 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, + }) + open++ + } + try { + return await work(client) + } catch (error) { + client.close() + client = null + open-- + throw error + } + } for (let g = queue.shift(); g; g = queue.shift()) { try { - const result = await sampleGroup((work) => NntpPool.use(work), g.name, { - every, - size: 32, - }) + const result = await sampleGroup(use, g.name, { every, size: 32 }) for (const f of result.failed) this.logger.warning(`${g.name} at ${f}`) await db.transaction(async (trx) => { await trx.from('group_dates').where('group_id', g.id).delete() @@ -116,13 +139,16 @@ export default class RangesDates extends BaseCommand { `${done} of ${groups.length} groups: ${samples} samples, ${commands} commands, ${((Date.now() - startedAt) / 1000).toFixed(0)} s.` ) } + if (client) { + ;(client as XoverClient).close() + open-- + } } try { - await Promise.all( - Array.from({ length: Math.max(1, Math.min(connections, groups.length)) }, worker) - ) + await Promise.all(Array.from({ length: workers }, worker)) } finally { clearInterval(sharing) + await share.sync(0, 0).catch(() => {}) } this.logger.info( `Sampled ${groups.length} groups: ${samples} samples, ${commands} commands, ${((Date.now() - startedAt) / 1000).toFixed(0)} s.` diff --git a/lib/nntp/index.ts b/lib/nntp/index.ts index 5f93745..173e379 100644 --- a/lib/nntp/index.ts +++ b/lib/nntp/index.ts @@ -1,4 +1,11 @@ -export { XoverClient, MissingArticle, parseOverviews, type Overview, type XoverClientOptions } from './xover_client.js' +export { + XoverClient, + MissingArticle, + TruncatedReply, + parseOverviews, + type Overview, + type XoverClientOptions, +} from './xover_client.js' export { parseXrefGroups } from './xref.js' export { parseGroupResponse, type GroupInfo } from './group.js' export { hedge, type Request, type Hedged } from './hedge.js' diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index 529ec00..61793ac 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -36,6 +36,14 @@ const DOT_LINE = Buffer.from('.\r\n') */ /** The server doesn't have the article (430): taken down or expired. */ export class MissingArticle extends Error {} +/** + * A compressed reply the server cut short: it ends its compressed output at the first ".\r\n" in it, as if that + * were the terminator, and sends nothing more (Eweka; "78 01 2e 0d 0a" for 16 of 14k small XOVERs, LOG.md + * 2026-10-25 (105)). The same request without compression answers. + */ +export class TruncatedReply extends Error {} +/** How long a compressed reply that doesn't inflate may stay silent before it counts as cut short. */ +const TRUNCATED_MS = 2_000 export class XoverClient { private chunks: Buffer[] = [] @@ -54,6 +62,10 @@ export class XoverClient { get buffered() { return this.length } + /** The first `n` bytes received and not consumed yet (what a reply that stalled left; nntp:probe). */ + bufferedHead(n = 32): Buffer { + return this.buffer().subarray(0, n) + } private lastDataAt = 0 private longestGap = 0 @@ -85,7 +97,11 @@ export class XoverClient { static async connect(options: XoverClientOptions): Promise { const sock = options.secure - ? tls.connect({ host: options.host, port: options.port, servername: options.servername || options.host }) + ? tls.connect({ + host: options.host, + port: options.port, + servername: options.servername || options.host, + }) : net.connect({ host: options.host, port: options.port }) const client = new XoverClient(sock, options.timeoutMs ?? 120_000) // The greeting and login are short: a server holding a connection back (too many open) fails it in 30 s, not timeoutMs. @@ -94,10 +110,14 @@ export class XoverClient { await client.expect(await client.line(), ['200', '201']) if (options.user) { const reply = await client.command(`AUTHINFO USER ${options.user}`) - if (reply.startsWith('381')) await client.expect(await client.command(`AUTHINFO PASS ${options.password ?? ''}`), ['281']) + if (reply.startsWith('381')) + await client.expect(await client.command(`AUTHINFO PASS ${options.password ?? ''}`), [ + '281', + ]) else await client.expect(reply, ['281']) } - if (options.compress) client.compressed = (await client.command('XFEATURE COMPRESS GZIP')).startsWith('290') + if (options.compress) + client.compressed = (await client.command('XFEATURE COMPRESS GZIP')).startsWith('290') } catch (error) { client.close() throw error @@ -127,7 +147,9 @@ export class XoverClient { const reply = await this.command(`LIST ACTIVE ${pattern}`) await this.expect(reply, ['215']) // With XFEATURE COMPRESS GZIP on, Eweka compresses this list too ("215 NewsGroups Follow [COMPRESS=GZIP]"). - const body = reply.includes('COMPRESS=GZIP') ? (await this.compressedRaw(true)).text : await this.plainBody() + const body = reply.includes('COMPRESS=GZIP') + ? (await this.compressedRaw(true)).text + : await this.plainBody() return body .split('\r\n') .filter(Boolean) @@ -154,7 +176,11 @@ export class XoverClient { return deflateAsync(Buffer.from(await this.plainBody()), { level: 1 }) } finally { this.measuring = false - this.lastTiming = { replyMs: repliedAt - sentAt, bodyMs: performance.now() - repliedAt, longestGapMs: this.longestGap } + this.lastTiming = { + replyMs: repliedAt - sentAt, + bodyMs: performance.now() - repliedAt, + longestGapMs: this.longestGap, + } } }) } @@ -180,7 +206,9 @@ export class XoverClient { const reply = await this.command(`XOVER ${from}-${to}`) if (reply.startsWith('420') || reply.startsWith('423')) return [] await this.expect(reply, ['224']) - const data = reply.includes('COMPRESS=GZIP') ? await this.compressedBody() : await this.plainBody() + const data = reply.includes('COMPRESS=GZIP') + ? await this.compressedBody() + : await this.plainBody() return parseOverviews(data) } @@ -199,7 +227,8 @@ export class XoverClient { } private async expect(reply: string, codes: string[]) { - if (!codes.some((code) => reply.startsWith(code))) throw new Error(`NNTP: unexpected reply "${reply}"`) + if (!codes.some((code) => reply.startsWith(code))) + throw new Error(`NNTP: unexpected reply "${reply}"`) } private buffer(): Buffer { @@ -223,10 +252,28 @@ export class XoverClient { if (this.failure && this.length === 0) throw this.failure } + /** As more(), but false when nothing came within `ms`. */ + private async moreWithin(ms: number): Promise { + if (this.failure) throw this.failure + const came = await new Promise((resolve) => { + const timer = setTimeout(() => resolve(false), ms) + this.wake = () => { + clearTimeout(timer) + resolve(true) + } + }) + this.wake = undefined + if (this.failure && this.length === 0) throw this.failure + return came + } + private async line(): Promise { while (true) { const at = this.buffer().indexOf('\r\n') - if (at !== -1) return this.take(at + 2).subarray(0, at).toString('latin1') + if (at !== -1) + return this.take(at + 2) + .subarray(0, at) + .toString('latin1') await this.more() } } @@ -267,7 +314,9 @@ export class XoverClient { */ async bodies(messageIds: string[]): Promise<(Buffer | null)[]> { return this.timed(async () => { - this.sock.write(messageIds.map((m) => `BODY ${m.startsWith('<') ? m : `<${m}>`}\r\n`).join('')) + this.sock.write( + messageIds.map((m) => `BODY ${m.startsWith('<') ? m : `<${m}>`}\r\n`).join('') + ) const out: (Buffer | null)[] = [] for (let i = 0; i < messageIds.length; i++) { const reply = await this.line() @@ -309,7 +358,11 @@ export class XoverClient { return '' } const at = all.indexOf(TERMINATOR, Math.max(0, searched - TERMINATOR.length)) - if (at !== -1) return this.take(at + TERMINATOR.length).subarray(0, at + 2).toString('utf8').replace(/^\.\./gm, '.') + if (at !== -1) + return this.take(at + TERMINATOR.length) + .subarray(0, at + 2) + .toString('utf8') + .replace(/^\.\./gm, '.') searched = all.length await this.more() } @@ -359,7 +412,14 @@ export class XoverClient { this.take(all.length) return { raw, text: inflated.toString('utf8') } } catch { - // ".\r\n" inside the compressed data: wait for more. + // ".\r\n" inside the compressed data: wait for more, but not for the reply timeout: the server may have + // cut its output there (TruncatedReply), and then nothing comes. Raw, the chunk is returned as it is + // and reported by the worker; the text is wanted here. + if (!(await this.moreWithin(TRUNCATED_MS))) { + this.take(all.length) + throw new TruncatedReply(`NNTP: compressed reply cut short after ${all.length} bytes`) + } + continue } } await this.more() diff --git a/tests/unit/nntp/xover_client.spec.ts b/tests/unit/nntp/xover_client.spec.ts index 0ab5b0e..09f00b4 100644 --- a/tests/unit/nntp/xover_client.spec.ts +++ b/tests/unit/nntp/xover_client.spec.ts @@ -1,7 +1,7 @@ import { test } from '@japa/runner' import net from 'node:net' import { deflateSync } from 'node:zlib' -import { MissingArticle, XoverClient, parseOverviews } from '#lib/nntp/index' +import { MissingArticle, TruncatedReply, XoverClient, parseOverviews } from '#lib/nntp/index' import { unzipSync } from 'node:zlib' const LINES = [ @@ -27,10 +27,15 @@ function server(): Promise<{ port: number; close: () => void }> { pending = pending.slice(at + 2) if (line.startsWith('AUTHINFO USER')) sock.write('381 more\r\n') else if (line.startsWith('AUTHINFO PASS')) sock.write('281 ok\r\n') - else if (line === 'XFEATURE COMPRESS GZIP') (gzip = true), sock.write('290 Feature Enabled\r\n') + else if (line === 'XFEATURE COMPRESS GZIP') + ((gzip = true), sock.write('290 Feature Enabled\r\n')) else if (line.startsWith('GROUP')) sock.write('211 2 101 102 a.b.test\r\n') else if (line === 'XOVER 1-1') sock.write('420 No Articles Selected\r\n') - else if (line.startsWith('XOVER')) { + else if (line === 'XOVER 5-5' && gzip) { + // As Eweka cuts a compressed reply whose output has ".\r\n" early: the zlib header, then nothing. + sock.write('224 Overview Information Follows [COMPRESS=GZIP]\r\n') + sock.write(Buffer.from([0x78, 0x01, 0x2e, 0x0d, 0x0a])) + } else if (line.startsWith('XOVER')) { const body = Buffer.from(LINES.join('\r\n') + '\r\n') if (gzip) { sock.write('224 Overview Information Follows [COMPRESS=GZIP]\r\n') @@ -53,19 +58,32 @@ function server(): Promise<{ port: number; close: () => void }> { else if (line.startsWith('BODY')) { sock.write('222 0 body\r\n') // Raw bytes, a dot-stuffed line among them. - send(Buffer.concat([Buffer.from([0x3d, 0x79, 0xff, 0x00, 0x0d, 0x0a]), Buffer.from('..stuffed\r\nlast\r\n.\r\n')])) + send( + Buffer.concat([ + Buffer.from([0x3d, 0x79, 0xff, 0x00, 0x0d, 0x0a]), + Buffer.from('..stuffed\r\nlast\r\n.\r\n'), + ]) + ) } else if (line === 'QUIT') sock.end('205 bye\r\n') } }) }) - srv.listen(0, '127.0.0.1', () => resolve({ port: (srv.address() as net.AddressInfo).port, close: () => srv.close() })) + srv.listen(0, '127.0.0.1', () => + resolve({ port: (srv.address() as net.AddressInfo).port, close: () => srv.close() }) + ) }) } test.group('nntp / XoverClient', () => { test('reads a body by message-id as bytes, raw; 430 is MissingArticle', async ({ assert }) => { const { port, close } = await server() - const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress: false }) + const client = await XoverClient.connect({ + host: '127.0.0.1', + port, + user: 'u', + password: 'p', + compress: false, + }) try { const body = await client.body('one@test') assert.deepEqual([...body.subarray(0, 6)], [0x3d, 0x79, 0xff, 0x00, 0x0d, 0x0a]) @@ -81,7 +99,13 @@ test.group('nntp / XoverClient', () => { test('reads several bodies pipelined, in order, a missing one as null', async ({ assert }) => { const { port, close } = await server() - const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress: false }) + const client = await XoverClient.connect({ + host: '127.0.0.1', + port, + user: 'u', + password: 'p', + compress: false, + }) try { const bodies = await client.bodies(['one@test', '', '']) assert.equal(bodies.length, 3) @@ -105,14 +129,25 @@ test.group('nntp / XoverClient', () => { }) for (const compress of [false, true]) { - test(`fetches overviews ${compress ? 'gzip-compressed' : 'plain'}, and an empty range`, async ({ assert }) => { + test(`fetches overviews ${compress ? 'gzip-compressed' : 'plain'}, and an empty range`, async ({ + assert, + }) => { const { port, close } = await server() - const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress }) + const client = await XoverClient.connect({ + host: '127.0.0.1', + port, + user: 'u', + password: 'p', + compress, + }) try { assert.equal(client.compressed, compress) assert.deepEqual(await client.group('a.b.test'), { count: 2, low: 101, high: 102 }) const overviews = await client.xover(101, 102) - assert.deepEqual(overviews.map(([n]) => n), [101, 102]) + assert.deepEqual( + overviews.map(([n]) => n), + [101, 102] + ) assert.equal(overviews[1][1].subject, 'Show.S01E01 - "a.mkv" yEnc (2/2)') assert.deepEqual(await client.xover(1, 1), []) assert.lengthOf(await client.xover(101, 102), 2) @@ -127,9 +162,43 @@ test.group('nntp / XoverClient', () => { }) } - test('idle time between commands does not count against the reply timeout', async ({ assert }) => { + test('a compressed reply cut short at a ".\\r\\n" in its output fails as TruncatedReply, well before the reply timeout', async ({ + assert, + }) => { const { port, close } = await server() - const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress: true, timeoutMs: 200 }) + const client = await XoverClient.connect({ + host: '127.0.0.1', + port, + user: 'u', + password: 'p', + compress: true, + timeoutMs: 60_000, + }) + try { + await client.group('a.b.test') + const startedAt = Date.now() + await assert.rejects(() => client.xover(5, 5), TruncatedReply) + assert.isBelow(Date.now() - startedAt, 10_000) + // The connection is still usable: the server thinks it answered. + assert.lengthOf(await client.xover(101, 102), 2) + } finally { + client.close() + close() + } + }).timeout(10_000) + + test('idle time between commands does not count against the reply timeout', async ({ + assert, + }) => { + const { port, close } = await server() + const client = await XoverClient.connect({ + host: '127.0.0.1', + port, + user: 'u', + password: 'p', + compress: true, + timeoutMs: 200, + }) try { await client.group('a.b.test') await new Promise((resolve) => setTimeout(resolve, 500)) @@ -141,13 +210,24 @@ test.group('nntp / XoverClient', () => { }) for (const compress of [false, true]) { - test(`hands out overviews as one zlib stream (${compress ? 'as the server sent it' : 'deflated here'})`, async ({ assert }) => { + test(`hands out overviews as one zlib stream (${compress ? 'as the server sent it' : 'deflated here'})`, async ({ + assert, + }) => { const { port, close } = await server() - const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress }) + const client = await XoverClient.connect({ + host: '127.0.0.1', + port, + user: 'u', + password: 'p', + compress, + }) try { await client.group('a.b.test') const raw = await client.xoverRaw(101, 102) - assert.deepEqual(parseOverviews(unzipSync(raw).toString('utf8')).map(([n]) => n), [101, 102]) + assert.deepEqual( + parseOverviews(unzipSync(raw).toString('utf8')).map(([n]) => n), + [101, 102] + ) assert.lengthOf(parseOverviews(unzipSync(await client.xoverRaw(1, 1)).toString('utf8')), 0) } finally { client.close()