diff --git a/REBUILD.md b/REBUILD.md index 02a1216..8785541 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -247,7 +247,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/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 520f03d..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[] = [] @@ -244,6 +252,21 @@ 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') @@ -389,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()