usenet-indexer/app/services/prober.ts

import db from '@adonisjs/lucid/services/db'
import logger from '@adonisjs/core/services/logger'
import { setTimeout as sleep } from 'node:timers/promises'
import env from '#start/env'
import coordinatorConfig from '#config/coordinator'
import { MissingArticle, XoverClient } from '#lib/nntp/index'
import { YencError } from '#lib/yenc/index'
import { inspect, innerName, type Inspection } from '#lib/probe/index'

interface Candidate {
  id: number
  releaseId: number
  kind: string
  messageId: string
}

type Outcome = { candidate: Candidate; status: 'done' | 'missing' | 'error' | 'waiting'; inspection?: Inspection }

/** Kinds in the order they're probed: a random name's real name first, then par2 sets, then archives. */
const KINDS = ['lone', 'par2', 'archive']
const FLUSH_MS = 2000
/** Slices a claim takes from random points of the queue. */
const SLICES = 20
const REPORT_MS = 60_000

/**
 * Phase 2's first step (REBUILD.md, part 6): reads the probe queue's
 * articles (settled releases first, by kind) over PROBE_CONNECTIONS
 * connections of its own, without compression, and stores what they show:
 * per release and kind an Inspection, and on the release whether it's
 * passworded and its inner name. One process: fetching (~0.1 s an article)
 * dwarfs parsing (a few ms), so there's no queue between them.
 */
export class Prober {
  private queue: Candidate[] = []
  private outcomes: Outcome[] = []
  private refilling: Promise<void> | null = null
  private stopped = false
  private stats = { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, since: Date.now() }

  async run() {
    // A previous run's claims: one prober at a time.
    const { rowCount } = await db.rawQuery(`UPDATE probe_candidates SET status = 'waiting' WHERE status = 'taken'`)
    if (rowCount) logger.info(`Probing: ${rowCount} candidates from a previous run back in the queue.`)
    const connections = coordinatorConfig.probe.connections
    logger.info(`Probing with ${connections} connections.`)
    const flusher = (async () => {
      while (!this.stopped || this.outcomes.length > 0) {
        await sleep(FLUSH_MS)
        await this.flush().catch((error) => logger.error(`Probing: storing results failed: ${error.message}`))
        this.report()
      }
    })()
    await Promise.all(Array.from({ length: connections }, () => this.lane()))
    // What was claimed and not probed goes back.
    for (const candidate of this.queue.splice(0)) this.outcomes.push({ candidate, status: 'waiting' })
    this.stopped = true
    await flusher
  }

  stop() {
    this.stopped = true
  }

  /** One connection's loop: an article at a time, reconnecting after a failure. */
  private async lane() {
    let client: XoverClient | null = null
    while (!this.stopped) {
      const candidate = await this.next()
      if (!candidate) {
        await sleep(30_000)
        continue
      }
      try {
        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,
        })
        const body = await client.body(candidate.messageId)
        this.stats.articles++
        this.stats.bytes += body.length
        try {
          this.outcomes.push({ candidate, status: 'done', inspection: inspect(body) })
        } catch (error: any) {
          // Not yEnc, a broken one, or data a parser chokes on: a final answer, not a connection problem.
          if (!(error instanceof YencError)) logger.warn(`Probing ${candidate.messageId}: ${error.message}`)
          this.outcomes.push({ candidate, status: 'error' })
        }
      } catch (error: any) {
        if (error instanceof MissingArticle) {
          // Taken down or expired: final (Eweka takes ~450 ms per missing article).
          this.stats.articles++
          this.outcomes.push({ candidate, status: 'missing' })
          continue
        }
        logger.warn(`Probing ${candidate.messageId}: ${error.message}`)
        this.outcomes.push({ candidate, status: 'waiting' })
        client?.close()
        client = null
        await sleep(5000)
      }
    }
    client?.close()
  }

  private async next(): Promise<Candidate | undefined> {
    if (this.queue.length < coordinatorConfig.probe.batch / 2) {
      this.refilling ??= this.claim().finally(() => (this.refilling = null))
      if (this.queue.length === 0) await this.refilling
    }
    return this.queue.shift()
  }

  /**
   * Waiting candidates of settled releases, by kind, claimed so another round
   * doesn't take them: SLICES slices from random points of the queue. In id
   * order it went group by group, and some groups' articles take 0.6–3 s
   * where most take 20–60 ms (a.b.zogg 2016: 7 articles/s on 6 connections).
   */
  private async claim() {
    for (const kind of KINDS) {
      const want = coordinatorConfig.probe.batch - this.queue.length
      if (want <= 0) return
      // The starting points chosen here: in the query Postgres drew random() once for all slices.
      const [{ lo, hi }] = (await db.rawQuery(`SELECT min(id) AS lo, max(id) AS hi FROM probe_candidates`)).rows
      const starts = Array.from({ length: SLICES }, () => Number(lo) + Math.floor(Math.random() * (Number(hi) - Number(lo) + 1)))
      const { rows } = await db.rawQuery(
        `UPDATE probe_candidates SET status = 'taken' WHERE id IN (
           SELECT pc.id FROM unnest(?::bigint[]) AS o(start)
           CROSS JOIN LATERAL (
             SELECT pc.id FROM probe_candidates pc JOIN releases r ON r.id = pc.release_id
             WHERE pc.status = 'waiting' AND pc.kind = ? AND r.settled_at IS NOT NULL AND pc.id >= o.start
             ORDER BY pc.id LIMIT ? FOR UPDATE OF pc SKIP LOCKED) pc)
         RETURNING id, release_id, kind, message_id`,
        [starts, kind, Math.ceil(want / SLICES)]
      )
      for (const r of rows) this.queue.push({ id: Number(r.id), releaseId: Number(r.release_id), kind: r.kind, messageId: r.message_id })
    }
  }

  /** The outcomes so far: candidates' status, inspections, and the releases' passworded and inner name. */
  private async flush() {
    const outcomes = this.outcomes.splice(0)
    if (outcomes.length === 0) return
    for (const o of outcomes) if (o.status !== 'waiting') this.stats[o.status]++
    await db.transaction(async (trx) => {
      await trx.rawQuery(
        `UPDATE probe_candidates pc SET status = v.status FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, status text) WHERE pc.id = v.id`,
        [JSON.stringify(outcomes.map((o) => ({ id: o.candidate.id, status: o.status })))]
      )
      const done = outcomes.filter((o) => o.inspection)
      if (done.length === 0) return
      await trx.rawQuery(
        `INSERT INTO release_inspections (release_id, candidate_id, kind, result)
         SELECT v.release_id, v.candidate_id, v.kind, v.result FROM jsonb_to_recordset(?::jsonb) AS v(release_id bigint, candidate_id bigint, kind text, result jsonb)
         ON CONFLICT (release_id, candidate_id) DO UPDATE SET result = excluded.result, created_at = now()`,
        // A par2 set's file list goes to par2_files, not into every inspection.
        [JSON.stringify(done.map((o) => ({ release_id: o.candidate.releaseId, candidate_id: o.candidate.id, kind: o.candidate.kind, result: { ...o.inspection, par2: undefined } })))]
      )
      await this.storeForMerge(trx, done)
      // Each touched release from all its inspections (earlier rounds may have stored others).
      const ids = [...new Set(done.map((o) => o.candidate.releaseId))]
      const { rows } = await trx.rawQuery(
        `SELECT i.release_id, i.kind, i.result, r.name FROM release_inspections i JOIN releases r ON r.id = i.release_id WHERE i.release_id = ANY(?)`,
        [ids]
      )
      const byRelease = new Map<number, { name: string; all: { kind: string; result: Inspection }[] }>()
      for (const r of rows) {
        const id = Number(r.release_id)
        const entry = byRelease.get(id) ?? { name: r.name as string, all: [] as { kind: string; result: Inspection }[] }
        entry.all.push({ kind: r.kind, result: r.result })
        byRelease.set(id, entry)
      }
      const updates = [...byRelease].map(([id, { name, all }]) => ({
        id,
        passworded: all.some((i) => i.result.passworded === true),
        inner_name: innerName(all, name),
      }))
      await trx.rawQuery(
        `UPDATE releases r SET passworded = v.passworded, inner_name = v.inner_name, inspected_at = now()
         FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, passworded boolean, inner_name text) WHERE r.id = v.id`,
        [JSON.stringify(updates)]
      )
    })
  }

  /** What the par2 merge matches (app/services/par2_merger.ts): files' first-16-KiB hashes, par2 sets' file lists. */
  private async storeForMerge(trx: any, done: Outcome[]) {
    const hashed = done.filter((o) => o.inspection!.hash16k)
    if (hashed.length > 0) {
      await trx.rawQuery(
        `INSERT INTO file_hashes (candidate_id, release_id, hash16k, size)
         SELECT v.candidate_id, v.release_id, v.hash16k, v.size FROM jsonb_to_recordset(?::jsonb) AS v(candidate_id bigint, release_id bigint, hash16k text, size bigint)
         ON CONFLICT (candidate_id) DO NOTHING`,
        [JSON.stringify(hashed.map((o) => ({ candidate_id: o.candidate.id, release_id: o.candidate.releaseId, hash16k: o.inspection!.hash16k, size: o.inspection!.size })))]
      )
    }
    for (const o of done) {
      const set = o.inspection!.par2
      if (!set) continue
      await trx.rawQuery(
        `INSERT INTO par2_files (set_id, file_id, hash16k, length, name)
         SELECT ?, v.file_id, v.hash16k, v.length, v.name FROM jsonb_to_recordset(?::jsonb) AS v(file_id text, hash16k text, length bigint, name text)
         ON CONFLICT DO NOTHING`,
        [set.setId, JSON.stringify(set.files.map((f) => ({ file_id: f.fileId, hash16k: f.hash16k, length: f.length, name: f.name })))]
      )
      await trx.rawQuery(`INSERT INTO par2_sets (set_id, release_id, files) VALUES (?, ?, ?) ON CONFLICT DO NOTHING`, [set.setId, o.candidate.releaseId, set.files.length])
    }
  }

  private report() {
    const seconds = (Date.now() - this.stats.since) / 1000
    if (seconds < REPORT_MS / 1000) return
    const { articles, bytes, done, missing, error } = this.stats
    logger.info(
      `Probed ${articles} articles in ${seconds.toFixed(0)} s: ${(articles / seconds).toFixed(1)}/s, ${((bytes * 8) / seconds / 1e6).toFixed(0)} Mbit/s; ` +
        `${done} inspected, ${missing} missing, ${error} not yEnc.`
    )
    this.stats = { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, since: Date.now() }
  }
}