diff --git a/apps/poller/Dockerfile b/apps/poller/Dockerfile index 316e992..1056425 100644 --- a/apps/poller/Dockerfile +++ b/apps/poller/Dockerfile @@ -1,7 +1,32 @@ # syntax=docker/dockerfile:1 # Build from the repo root: # docker build -f apps/poller/Dockerfile . -FROM node:22-alpine + +# RSSHub, taken from the image its own maintainers build and test. +# +# Not `npm install rsshub` into our image, which was the first attempt and is +# the worse one. That package is published only as `1.0.0-master.` — a +# rolling build with no stable version to pin — and pulls 83 direct dependencies +# including `patchright`, a Playwright fork whose install downloads a browser. +# Resolving all of that inside our build makes our deploy depend on their +# dependency tree resolving cleanly on the day we ship, for an app we do not +# develop. Copying their published artifact means our build does no resolution. +FROM diygod/rsshub:latest AS rsshub + +# Debian trixie and Node 24, because that is exactly what RSSHub's own image is +# (`node:24-trixie-slim`) and this container has to run their `node_modules`. +# +# This was `node:22-alpine` and both halves of that were wrong for the job: +# musl cannot load anything in there linked against glibc, and Node 22 cannot +# load a native addon built for Node 24's ABI. Either mismatch produces the same +# unhelpful shape of failure — an image that builds perfectly and a daemon that +# dies on first require — so the base is pinned to theirs rather than to ours. +# +# The cost, stated because it is not free: CI runs Node 22 (matching the web +# image), so the poller now runs a major this repo's CI does not exercise. The +# suite passes on both today, and 22 is the stricter of the two runners, so the +# gap is in the safe direction — but it is a gap. +FROM node:24-trixie-slim ENV PNPM_HOME=/pnpm ENV PATH=$PNPM_HOME:$PATH RUN corepack enable @@ -12,5 +37,18 @@ COPY . . # install is all this needs. RUN pnpm install --frozen-lockfile --prod +# RSSHub's application, at the path apps/poller/src/rsshub.js expects. +# Deliberately outside /repo: it is not part of this workspace, must never enter +# pnpm-lock.yaml, and must never be walked by anything that scans our +# dependencies. +COPY --from=rsshub /app /opt/rsshub + +# The bundled Chromium goes straight back out again — several hundred megabytes +# for routes that need a headless browser to scrape a page. Every route this +# poller asks for is a Twitter one, which is an API call. If a future route does +# need a browser it will fail loudly with a missing-executable error, which is a +# better outcome than carrying the download on every deploy for ever in case. +RUN rm -rf /opt/rsshub/node_modules/.cache/ms-playwright + ENV NODE_ENV=production CMD ["node", "apps/poller/src/index.js"] diff --git a/apps/poller/src/index.js b/apps/poller/src/index.js index 884fbff..da8e576 100644 --- a/apps/poller/src/index.js +++ b/apps/poller/src/index.js @@ -28,6 +28,7 @@ import { deliverAlerts, vapidConfig } from '@rssamplifier/notify'; import { createXRuntime, xEnabled } from '@rssamplifier/social'; import { createRecorder, toEntry, writeFailure } from './log.js'; +import { shouldRunRsshub, startRsshub } from './rsshub.js'; /** * Feed crawler daemon. @@ -603,6 +604,19 @@ try { * that into a five-second answer — the same reason the push half of the alerts * stack prints `push=true` here. */ +/** + * RSSHub, if this process is the one running it. + * + * Started *before* the runtime below, because `createXRuntime` reads + * `RSSHUB_BASE_URL` when it builds the provider and a provider built with no + * base URL declines every request for the life of the process. The daemon does + * not have to be listening yet — the first crawl is a tick away and a refused + * request costs a reschedule, not a source's health — but the address has to be + * known. + */ +const rsshub = shouldRunRsshub(env) ? startRsshub({ env, log }) : null; +if (rsshub) env.RSSHUB_BASE_URL = rsshub.url; + const xRuntime = xEnabled(env) ? await createXRuntime({ env, @@ -618,7 +632,11 @@ const xRuntime = xEnabled(env) log('x-runtime', { enabled: Boolean(xRuntime), providers: xRuntime ? xRuntime.registry.candidates().map((p) => p.name) : [], + // How many X logins are configured. Zero with RSSHub embedded is the state + // worth spotting on a boot line: the daemon starts, answers, and returns + // nothing for every timeline, because it has no session to read X with. sessions: xRuntime ? xRuntime.sessions.size : 0, + rsshub: rsshub ? 'embedded' : (env.RSSHUB_BASE_URL ? 'external' : 'none'), }); /** @@ -1085,6 +1103,12 @@ function shutdown(signal) { // deploy — so this runs several times a day. void writeWorker?.close(); + // The embedded daemon goes down with us. Without this its supervisor would + // see the child die during shutdown and dutifully restart it, and a container + // Railway is trying to stop would keep spawning a new Node process every few + // seconds until it is killed outright. + void rsshub?.stop(); + // Recorded before the buffer is closed, so the live log's last line is the // daemon saying it stopped rather than the log simply going quiet — which is // what a crash looks like. diff --git a/apps/poller/src/rsshub.js b/apps/poller/src/rsshub.js new file mode 100644 index 0000000..6dc289d --- /dev/null +++ b/apps/poller/src/rsshub.js @@ -0,0 +1,248 @@ +import { spawn } from 'node:child_process'; + +/** + * RSSHub, as a daemon inside this container rather than a service beside it. + * + * **Why here and not as its own Railway service.** The only thing in this system + * that collects anything is this process. A separate service would mean a second + * billed container, a private-networking hop, and a surface that has to be kept + * unexposed — all to reach something that has exactly one consumer running in + * the same place. On localhost none of that exists: no domain to forget to + * remove, no IPv6 private-networking trap (Railway's outbound IPv6 is opt-in per + * service and fails as a fast 504, which is a genuinely nasty thing to debug), + * and nothing to bill. + * + * **It is supervised, and its death is never this process's death.** RSSHub is a + * large third-party app talking to a hostile upstream; it will crash. A crash + * must cost one restart and nothing else — not the crawl, not the queue, not the + * 50,000 Reddit feeds that have nothing to do with X. So it is spawned detached + * from the crawler's control flow, restarted with backoff, and if it never comes + * up at all the crawler simply carries on: `fetchXSource` treats an unreachable + * provider as a reschedule rather than as a verdict on the source (see the note + * in packages/social/src/x/fetch.js, and PR #157 for what it cost to get that + * wrong). + * + * **Nothing here is on the request path.** The web service never talks to + * RSSHub — it never collects — so it needs none of this and gets none of it. + */ + +/** Where the daemon listens. Loopback only: nothing outside the container. */ +const HOST = '127.0.0.1'; +const PORT = 1200; + +/** + * Where the image puts RSSHub, and what to run inside it. + * + * `dist/index.mjs` rather than a `bin` — the package publishes none, and their + * own image starts it through an npm script. Reading that script is how this + * path and the header size below were arrived at; guessing either produces a + * daemon that starts and then fails on the first real request. + */ +const DEFAULT_ENTRY = '/opt/rsshub/dist/index.mjs'; + +/** + * RSSHub's own start script raises this, and it is not decoration. + * + * A logged-in X request carries the session cookie in a header, and Node's + * default 16KB header limit is smaller than what X sends back on some + * endpoints. The failure is an opaque parse error partway through a response, + * which reads like a broken upstream rather than a limit of ours. + */ +const NODE_OPTIONS = '--max-http-header-size=32768'; + +/** + * Restart backoff, in milliseconds, by consecutive failure. + * + * Capped and short. RSSHub crashing repeatedly is a configuration problem — a + * missing cookie, a changed upstream — and no amount of patience fixes it; the + * backoff exists so that a crash loop costs a restart a minute rather than a + * core, not because waiting longer is expected to help. + */ +const BACKOFF_MS = [1_000, 5_000, 15_000, 30_000, 60_000]; + +/** + * A run that lasts this long is treated as a success, and resets the backoff. + * + * Without it a process that starts, serves for an hour and then dies is + * indistinguishable from one that has never started, and would inherit the + * backoff of a crash loop it is not in. + */ +const HEALTHY_AFTER_MS = 60_000; + +/** + * Start RSSHub and keep it running. + * + * @param {{ + * env?: Record, + * log?: (event: string, fields?: object) => void, + * spawn?: typeof spawn, + * now?: () => number, + * }} [opts] + * @returns {{ url: string, stop: () => Promise, state: () => object }} + */ +export function startRsshub(opts = {}) { + const env = opts.env ?? process.env; + const log = opts.log ?? (() => {}); + const doSpawn = opts.spawn ?? spawn; + const now = opts.now ?? (() => Date.now()); + + const entry = String(env.RSSHUB_ENTRY ?? DEFAULT_ENTRY); + const url = `http://${HOST}:${PORT}`; + + let child = null; + let failures = 0; + let stopping = false; + let timer = null; + let startedAt = 0; + + const start = () => { + if (stopping) return; + + startedAt = now(); + child = doSpawn(process.execPath, [entry], { + env: { + ...process.env, + NODE_ENV: 'production', + NODE_OPTIONS, + PORT: String(PORT), + // RSSHub binds every interface by default. Inside a worker service with + // no public domain that is already unreachable, but saying loopback + // explicitly means it stays unreachable if this service ever gains one. + LISTEN_INADDR_ANY: '0', + // In-memory cache. Redis is in this project and deliberately not used + // here: it is the crawler's write queue, and handing a third-party app + // the same instance to fill with route caches is a way to lose jobs to + // an eviction policy nobody chose for them. + CACHE_TYPE: 'memory', + // The X login RSSHub collects with. Named differently on each side, so + // the mapping is written down here rather than in an operator's head. + TWITTER_AUTH_TOKEN: authTokens(env), + ...(env.RSSHUB_ACCESS_KEY ? { ACCESS_KEY: String(env.RSSHUB_ACCESS_KEY) } : {}), + }, + // Its own directory, because it resolves routes and configuration + // relative to the application root rather than to the entry file. + cwd: entry.replace(/\/dist\/[^/]+$/, ''), + // Its output is its own; the crawler's log is not the place for a third + // party's request lines. Errors are surfaced through the events below. + stdio: ['ignore', 'ignore', 'pipe'], + }); + + log('rsshub-started', { pid: child.pid, url }); + + // Only the tail of stderr, and only on exit — RSSHub is chatty, and a live + // pipe into the crawler's log buries the crawl. + let stderr = ''; + child.stderr?.on('data', (chunk) => { + stderr = (stderr + String(chunk)).slice(-2000); + }); + + child.on('error', (error) => { + log('rsshub-error', { message: String(error?.message ?? error) }); + }); + + child.on('exit', (code, signal) => { + child = null; + if (stopping) return; + + const lived = now() - startedAt; + if (lived >= HEALTHY_AFTER_MS) failures = 0; + failures += 1; + + const wait = BACKOFF_MS[Math.min(failures - 1, BACKOFF_MS.length - 1)]; + log('rsshub-exited', { + code, + signal, + livedMs: lived, + failures, + restartInMs: wait, + // The last thing it said, truncated. Never its environment: that holds + // the X session cookie. + stderr: stderr.split('\n').filter(Boolean).slice(-3).join(' | ').slice(0, 300) || null, + }); + + timer = setTimeout(start, wait); + timer.unref?.(); + }); + }; + + start(); + + return { + url, + + state: () => ({ running: Boolean(child), pid: child?.pid ?? null, failures, url }), + + async stop() { + stopping = true; + if (timer) clearTimeout(timer); + if (!child) return; + + const dying = child; + dying.kill('SIGTERM'); + + // A grace period, then insist. A supervisor that waits for ever on a + // wedged child turns a deploy into a hung container, and Railway's own + // patience is finite. + await new Promise((resolve) => { + const hard = setTimeout(() => { + dying.kill('SIGKILL'); + resolve(); + }, 5_000); + hard.unref?.(); + dying.on('exit', () => { + clearTimeout(hard); + resolve(); + }); + }); + }, + }; +} + +/** + * Should this process run RSSHub itself? + * + * No when X is off — there is nothing to collect, and a daemon nobody calls is + * memory and a crash loop waiting to be ignored. No when `RSSHUB_BASE_URL` + * already names one, because somebody has pointed this at an instance they run + * and taking that over would be surprising. + * + * @param {Record} [env] + * @returns {boolean} + */ +export function shouldRunRsshub(env = process.env) { + if (String(env.X_ENABLED ?? 'false').toLowerCase() === 'false') return false; + if (String(env.RSSHUB_EMBEDDED ?? '').toLowerCase() === 'false') return false; + return !String(env.RSSHUB_BASE_URL ?? '').trim(); +} + +/** + * The X session cookies, in the spelling RSSHub wants. + * + * Ours is `X_SESSIONS` (structured JSON) because a positional pair of + * comma-separated lists silently mispairs tokens with cookies — see + * packages/social/src/x/sessions.js. RSSHub takes a bare comma-separated list of + * `auth_token` values, so the translation happens here, once. + * + * @param {Record} env + * @returns {string} + */ +export function authTokens(env) { + const raw = String(env.X_SESSIONS ?? '').trim(); + + if (raw) { + try { + const parsed = JSON.parse(raw); + if (Array.isArray(parsed)) { + const tokens = parsed + .map((entry) => String(entry?.authToken ?? entry?.auth_token ?? '').trim()) + .filter(Boolean); + if (tokens.length) return tokens.join(','); + } + } catch { + // Falls through to the flat form rather than throwing: a malformed + // X_SESSIONS must not stop the crawler booting. + } + } + + return String(env.X_AUTH_TOKENS ?? '').trim(); +} diff --git a/apps/poller/test/rsshub.test.js b/apps/poller/test/rsshub.test.js new file mode 100644 index 0000000..fe45e7a --- /dev/null +++ b/apps/poller/test/rsshub.test.js @@ -0,0 +1,162 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import { EventEmitter } from 'node:events'; + +import { startRsshub, shouldRunRsshub, authTokens } from '../src/rsshub.js'; + +/* + * The supervisor, which is the half of this that can be tested without building + * an image. What it must guarantee: RSSHub crashing costs a restart and nothing + * else, a deploy does not leave it respawning into a container that is going + * away, and the X session cookie never reaches a log line. + */ + +/** A child process that does nothing until told to die. */ +function fakeChild(pid = 1234) { + const child = new EventEmitter(); + child.pid = pid; + child.stderr = new EventEmitter(); + child.kill = (signal) => { + child.killed = signal; + queueMicrotask(() => child.emit('exit', null, signal)); + return true; + }; + return child; +} + +/** A spawn that hands back children from a list and records how it was called. */ +function fakeSpawn(children) { + const calls = []; + let n = 0; + const spawn = (cmd, args, opts) => { + calls.push({ cmd, args, opts }); + return children[Math.min(n++, children.length - 1)]; + }; + spawn.calls = calls; + return spawn; +} + +test('the daemon is started with loopback, a fixed port and an in-memory cache', () => { + const spawn = fakeSpawn([fakeChild()]); + const handle = startRsshub({ env: {}, spawn, log: () => {} }); + + assert.equal(handle.url, 'http://127.0.0.1:1200'); + + const { args, opts } = spawn.calls[0]; + + // The three values that were wrong on the first attempt, and that produce a + // daemon which builds and deploys and then does not work: + // + // entry — RSSHub publishes no `bin`; their own image starts `dist/index.mjs` + // through an npm script. `lib/index.js` was a guess and is not there. + // cwd — it resolves routes relative to the application root, not the entry. + // header size — X sends responses whose headers exceed Node's 16KB default, + // and the failure reads like a broken upstream rather than our limit. + assert.equal(args[0], '/opt/rsshub/dist/index.mjs'); + assert.equal(opts.cwd, '/opt/rsshub'); + assert.match(opts.env.NODE_OPTIONS, /--max-http-header-size=32768/); + + assert.equal(opts.env.PORT, '1200'); + assert.equal(opts.env.LISTEN_INADDR_ANY, '0'); + // Redis in this project is the crawler's write queue, not a route cache. + assert.equal(opts.env.CACHE_TYPE, 'memory'); +}); + +test('our X_SESSIONS becomes the spelling RSSHub wants', () => { + const env = { X_SESSIONS: '[{"id":"x-1","authToken":"aaa","ct0":"bbb"},{"id":"x-2","authToken":"ccc","ct0":"ddd"}]' }; + assert.equal(authTokens(env), 'aaa,ccc'); + + // The flat form still works, and a malformed X_SESSIONS falls back to it + // rather than stopping the crawler from booting. + assert.equal(authTokens({ X_AUTH_TOKENS: 'zzz' }), 'zzz'); + assert.equal(authTokens({ X_SESSIONS: '{oops', X_AUTH_TOKENS: 'zzz' }), 'zzz'); + assert.equal(authTokens({}), ''); +}); + +test('a crash is restarted, and the log carries no credential', async () => { + const first = fakeChild(1); + const second = fakeChild(2); + const spawn = fakeSpawn([first, second]); + + const lines = []; + let clock = 0; + + startRsshub({ + env: { X_SESSIONS: '[{"id":"x-1","authToken":"SUPERSECRET","ct0":"CT0SECRET"}]' }, + spawn, + now: () => clock, + log: (event, fields) => lines.push({ event, fields }), + }); + + first.stderr.emit('data', 'Error: something went wrong\n'); + clock += 500; + first.emit('exit', 1, null); + + await new Promise((resolve) => setTimeout(resolve, 5)); + + const exited = lines.find((line) => line.event === 'rsshub-exited'); + assert.ok(exited, 'a crash is reported'); + assert.equal(exited.fields.failures, 1); + assert.ok(exited.fields.restartInMs > 0, 'and a restart is scheduled'); + + // The token is in the child's environment and must be in nothing else. + assert.doesNotMatch(JSON.stringify(lines), /SUPERSECRET|CT0SECRET/); +}); + +test('a long healthy run resets the backoff', async () => { + const children = [fakeChild(1), fakeChild(2)]; + const spawn = fakeSpawn(children); + const lines = []; + let clock = 0; + + startRsshub({ env: {}, spawn, now: () => clock, log: (event, fields) => lines.push({ event, fields }) }); + + // Ran for an hour, then died: that is not a crash loop. + clock += 3_600_000; + children[0].emit('exit', 1, null); + await new Promise((resolve) => setTimeout(resolve, 5)); + + const exited = lines.filter((line) => line.event === 'rsshub-exited'); + assert.equal(exited.at(-1).fields.failures, 1); + assert.equal(exited.at(-1).fields.restartInMs, 1_000, 'back to the shortest backoff'); +}); + +test('stopping does not respawn — a deploy must not fight the supervisor', async () => { + const child = fakeChild(); + const spawn = fakeSpawn([child, fakeChild(2)]); + + const handle = startRsshub({ env: {}, spawn, log: () => {} }); + await handle.stop(); + + assert.equal(child.killed, 'SIGTERM'); + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.equal(spawn.calls.length, 1, 'exactly one spawn, and no restart after stop'); +}); + +test('who runs the daemon, and who does not', () => { + // X off: nothing to collect, so no daemon. + assert.equal(shouldRunRsshub({ X_ENABLED: 'false' }), false); + assert.equal(shouldRunRsshub({}), false); + + // X on and nothing else said: this process runs it. + assert.equal(shouldRunRsshub({ X_ENABLED: 'true' }), true); + + // Somebody has pointed us at an instance they run. Taking that over would be + // surprising, and would put a second RSSHub on the same X sessions. + assert.equal(shouldRunRsshub({ X_ENABLED: 'true', RSSHUB_BASE_URL: 'http://rsshub:1200' }), false); + + // And an explicit opt-out, for a deployment that wants X on with no daemon. + assert.equal(shouldRunRsshub({ X_ENABLED: 'true', RSSHUB_EMBEDDED: 'false' }), false); +}); + +test('the handle reports what it is doing', () => { + const child = fakeChild(4242); + const handle = startRsshub({ env: {}, spawn: fakeSpawn([child]), log: () => {} }); + + assert.deepEqual(handle.state(), { + running: true, + pid: 4242, + failures: 0, + url: 'http://127.0.0.1:1200', + }); +});