From 2f9bddb93ccdcaddc8a31e410577d7c8d97a7c9e Mon Sep 17 00:00:00 2001 From: "eric.wien" Date: Thu, 11 Jun 2026 19:38:13 +0200 Subject: [PATCH] harden relay: cursor durability, fail-fast Redis, graceful shutdown Apply fixes from extensive code review: - exit on 'exhausted' so a supervisor restarts instead of a zombie process - advance cursor only after a successful Redis publish (no silent skips) - disable ioredis offline queue to avoid unbounded buffering during outages - persist cursor to Redis and resume from it on restart - handle SIGINT/SIGTERM with clean shutdown and final cursor write - add unhandledRejection/uncaughtException handlers - extract relay logic to src/relay.ts and add node:test coverage - enable strict TypeScript, align Node floor (v20+) with @types/node - document delivery semantics (at-least-once, Pub/Sub caveats) in README Co-Authored-By: Claude Fable 5 --- README.md | 16 +++++- package.json | 8 ++- pnpm-lock.yaml | 24 ++++---- src/index.ts | 143 ++++++++++++++++++++++++++++++++++------------ src/relay.test.ts | 77 +++++++++++++++++++++++++ src/relay.ts | 46 +++++++++++++++ tsconfig.json | 5 +- 7 files changed, 263 insertions(+), 56 deletions(-) create mode 100644 src/relay.test.ts create mode 100644 src/relay.ts diff --git a/README.md b/README.md index d319aa8..10fdb28 100644 --- a/README.md +++ b/README.md @@ -7,10 +7,17 @@ Connects to the [Bluesky Jetstream](https://docs.bsky.app/docs/advanced-guides/f 1. Opens a WebSocket connection to one of the Jetstream endpoints (US-East / US-West, with automatic failover across four servers). 2. Filters for `app.bsky.feed.post` events and validates them with Zod via `@vinerima/jetstream-types`. 3. Publishes each validated post-create event as JSON to a configurable Redis Pub/Sub channel. +4. Tracks the `time_us` cursor of the last successfully published event, persists it to Redis, and resumes from it on reconnects and restarts. + +## Delivery semantics + +- **At-least-once across reconnects.** The relay resumes from the cursor of the last event that reached Redis, so events around a disconnect or restart are replayed and consumers may see duplicates. Deduplicate by `time_us` + `did` (or `commit.rev`) if duplicates matter to you. +- **Pub/Sub is fire-and-forget.** Redis Pub/Sub does not buffer for disconnected subscribers — consumers only receive events while subscribed. If consumers must not miss events while offline, put a Redis Streams consumer behind the channel (or adapt the relay to `XADD`). +- **Fail-fast on outages.** While Redis is unreachable, publishes fail immediately (nothing buffers in process memory) and the cursor stops advancing, so the missed window is replayed once Redis recovers. If every Jetstream endpoint stays unreachable through all reconnect cycles, the process exits with code 1 — run it under a supervisor (systemd, Docker `--restart`, pm2) so it gets restarted. ## Prerequisites -- Node.js (v18+) +- Node.js (v20+) - A running Redis instance ## Configuration @@ -19,6 +26,7 @@ Connects to the [Bluesky Jetstream](https://docs.bsky.app/docs/advanced-guides/f |---|---|---| | `REDIS_URL` | `redis://localhost:6379` | Redis connection string | | `REDIS_CHANNEL` | `jetstream:posts` | Pub/Sub channel name | +| `REDIS_CURSOR_KEY` | `jetstream:cursor` | Redis key the cursor is persisted to (every 5s and on shutdown) | ## Running the relay @@ -69,6 +77,12 @@ You can verify events are flowing by subscribing to the channel in a separate te redis-cli SUBSCRIBE jetstream:posts ``` +## Tests + +```bash +pnpm test +``` + ## Dependencies - [`@vinerima/wah`](https://www.npmjs.com/package/@vinerima/wah) — WebSocket client with reconnection, failover, and schema-based message handling diff --git a/package.json b/package.json index 433f8c5..70d54f4 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,11 @@ "main": "dist/index.js", "scripts": { "start": "ts-node src/index.ts", - "build": "tsc" + "build": "tsc", + "test": "node -r ts-node/register --test src/relay.test.ts" + }, + "engines": { + "node": ">=20" }, "author": "vinerima ", "license": "MIT", @@ -17,7 +21,7 @@ "zod": "^4.3.6" }, "devDependencies": { - "@types/node": "^25.5.0", + "@types/node": "^20.19.0", "ts-node": "^10.9.2", "typescript": "^5.9.3" } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 33edc23..40d5b42 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -25,11 +25,11 @@ importers: version: 4.3.6 devDependencies: '@types/node': - specifier: ^25.5.0 - version: 25.5.0 + specifier: ^20.19.0 + version: 20.19.42 ts-node: specifier: ^10.9.2 - version: 10.9.2(@types/node@25.5.0)(typescript@5.9.3) + version: 10.9.2(@types/node@20.19.42)(typescript@5.9.3) typescript: specifier: ^5.9.3 version: 5.9.3 @@ -65,8 +65,8 @@ packages: '@tsconfig/node16@1.0.4': resolution: {integrity: sha512-vxhUy4J8lyeyinH7Azl1pdd43GJhZH/tP2weN8TntQblOY+A0XbT8DJk1/oCPuOOyg/Ja757rG0CgHcWC8OfMA==} - '@types/node@25.5.0': - resolution: {integrity: sha512-jp2P3tQMSxWugkCUKLRPVUpGaL5MVFwF8RDuSRztfwgN1wmqJeMSbKlnEtQqU8UrhTmzEmZdu2I6v2dpp7XIxw==} + '@types/node@20.19.42': + resolution: {integrity: sha512-5L7SUaFC1RyDraj2yRhyBzHTobyXHmohD100CChNtyPyleoq37Mqab5Gn8XEKI04dfN/oqPdpHk38MgcQWHbZg==} '@vinerima/jetstream-types@2.0.0': resolution: {integrity: sha512-OIDqzWIEaTUtlTy4laTG8MNv+AZ+/j05KKdXnxdoZPTqYBz/IyByL80a2a+DMZ3/PNXF36uWgWu+IHb3QvPUDw==} @@ -164,8 +164,8 @@ packages: engines: {node: '>=14.17'} hasBin: true - undici-types@7.18.2: - resolution: {integrity: sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w==} + undici-types@6.21.0: + resolution: {integrity: sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==} v8-compile-cache-lib@3.0.1: resolution: {integrity: sha512-wa7YjyUGfNZngI/vtK0UHAN+lgDCxBPCylVXGp0zu59Fz5aiGtNXaq3DhIov063MorB+VfufLh3JlF2KdTK3xg==} @@ -214,9 +214,9 @@ snapshots: '@tsconfig/node16@1.0.4': {} - '@types/node@25.5.0': + '@types/node@20.19.42': dependencies: - undici-types: 7.18.2 + undici-types: 6.21.0 '@vinerima/jetstream-types@2.0.0(zod@4.3.6)': dependencies: @@ -278,14 +278,14 @@ snapshots: standard-as-callback@2.1.0: {} - ts-node@10.9.2(@types/node@25.5.0)(typescript@5.9.3): + ts-node@10.9.2(@types/node@20.19.42)(typescript@5.9.3): dependencies: '@cspotcode/source-map-support': 0.8.1 '@tsconfig/node10': 1.0.12 '@tsconfig/node12': 1.0.11 '@tsconfig/node14': 1.0.3 '@tsconfig/node16': 1.0.4 - '@types/node': 25.5.0 + '@types/node': 20.19.42 acorn: 8.16.0 acorn-walk: 8.3.5 arg: 4.1.3 @@ -298,7 +298,7 @@ snapshots: typescript@5.9.3: {} - undici-types@7.18.2: {} + undici-types@6.21.0: {} v8-compile-cache-lib@3.0.1: {} diff --git a/src/index.ts b/src/index.ts index 11a8bdb..e0f6d2a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,9 +1,12 @@ import { WebSocketClient } from "@vinerima/wah"; import Redis from "ioredis"; import { jetstreamPostCreateSchema } from "@vinerima/jetstream-types"; +import { createRelay } from "./relay"; const REDIS_URL = process.env.REDIS_URL ?? "redis://localhost:6379"; const CHANNEL = process.env.REDIS_CHANNEL ?? "jetstream:posts"; +const CURSOR_KEY = process.env.REDIS_CURSOR_KEY ?? "jetstream:cursor"; +const CURSOR_PERSIST_INTERVAL_MS = 5000; function log(level: string, message: string, context?: unknown): void { const ts = new Date().toISOString(); @@ -11,7 +14,23 @@ function log(level: string, message: string, context?: unknown): void { fn(`[${ts}] [${level.toUpperCase()}] ${message}`, context ?? ""); } -const redis = new Redis(REDIS_URL); +process.on("unhandledRejection", (reason) => { + log("error", "Unhandled promise rejection", reason); + process.exit(1); +}); + +process.on("uncaughtException", (err) => { + log("error", "Uncaught exception", err); + process.exit(1); +}); + +const redis = new Redis(REDIS_URL, { + // Fail publishes immediately while Redis is down instead of buffering the + // firehose in an unbounded in-memory queue. The cursor stops advancing on + // failed publishes, so the missed window is replayed after recovery. + enableOfflineQueue: false, + lazyConnect: true, +}); redis.on("error", (err) => { log("error", "Redis connection error", err); @@ -21,49 +40,97 @@ redis.on("connect", () => { log("info", `Connected to Redis at ${REDIS_URL}`); }); -const client = new WebSocketClient({ - service: [ - "wss://jetstream1.us-east.bsky.network/subscribe", - "wss://jetstream2.us-east.bsky.network/subscribe", - "wss://jetstream1.us-west.bsky.network/subscribe", - "wss://jetstream2.us-west.bsky.network/subscribe", - ], - queryParams: { - wantedCollections: "app.bsky.feed.post", - }, - reconnect: { - initialDelay: 5000, - maxDelay: 30000, - backoffFactor: 1.5, - maxAttempts: 3, - maxServiceCycles: 2, - }, - logger: { enabled: true }, - pingInterval: 10000, -}); - -let lastTimeUs = 0; +async function main(): Promise { + await redis.connect(); -client.handle(jetstreamPostCreateSchema, async ({ data }) => { + let storedCursor = 0; try { - lastTimeUs = data.time_us; - await redis.publish(CHANNEL, JSON.stringify(data)); + storedCursor = Number(await redis.get(CURSOR_KEY)) || 0; } catch (err) { - log("error", "Failed to publish to Redis", err); + log("error", "Failed to read stored cursor, starting from live", err); } -}); -client.on("open", () => log("info", `Jetstream connected, publishing to channel "${CHANNEL}"`)); -client.on("error", (err) => log("error", "Jetstream error", err)); -client.on("reconnecting", (info) => log("info", "Jetstream reconnecting", info)); + const client = new WebSocketClient({ + service: [ + "wss://jetstream1.us-east.bsky.network/subscribe", + "wss://jetstream2.us-east.bsky.network/subscribe", + "wss://jetstream1.us-west.bsky.network/subscribe", + "wss://jetstream2.us-west.bsky.network/subscribe", + ], + queryParams: { + wantedCollections: "app.bsky.feed.post", + ...(storedCursor > 0 ? { cursor: storedCursor } : {}), + }, + reconnect: { + initialDelay: 5000, + maxDelay: 30000, + backoffFactor: 1.5, + maxAttempts: 3, + maxServiceCycles: 2, + }, + logger: { enabled: true }, + pingInterval: 10000, + }); + + const relay = createRelay({ + channel: CHANNEL, + publish: (channel, message) => redis.publish(channel, message), + updateParams: (params, options) => client.updateParams(params, options), + log, + }); + + client.handle(jetstreamPostCreateSchema, ({ data }) => relay.onPost(data)); -client.on("close", () => { - if (lastTimeUs > 0) { - client.updateParams({ cursor: lastTimeUs }, { immediate: false }); - log("info", `Cursor set to ${lastTimeUs} for reconnection`); + client.on("open", () => log("info", `Jetstream connected, publishing to channel "${CHANNEL}"`)); + client.on("error", (err) => log("error", "Jetstream error", err)); + client.on("reconnecting", (info) => log("info", "Jetstream reconnecting", info)); + client.on("close", () => relay.onClose()); + + async function persistCursor(): Promise { + if (relay.lastTimeUs > 0) { + await redis.set(CURSOR_KEY, String(relay.lastTimeUs)); + } + } + + const persistTimer = setInterval(() => { + persistCursor().catch((err) => log("error", "Failed to persist cursor", err)); + }, CURSOR_PERSIST_INTERVAL_MS); + + async function shutdown(reason: string, code: number): Promise { + log("info", `Shutting down (${reason})`); + clearInterval(persistTimer); + client.close(); + try { + await persistCursor(); + } catch (err) { + log("error", "Failed to persist cursor during shutdown", err); + } + try { + await redis.quit(); + } catch { + // connection may already be gone + } + process.exit(code); } -}); -client.connect(); + process.on("SIGINT", () => void shutdown("SIGINT", 0)); + process.on("SIGTERM", () => void shutdown("SIGTERM", 0)); -log("info", "jetstream-relay started"); + client.on("exhausted", (info) => { + log("error", "All Jetstream reconnect attempts exhausted, exiting so a supervisor can restart", info); + void shutdown("exhausted", 1); + }); + + client.connect(); + log( + "info", + storedCursor > 0 + ? `jetstream-relay started, resuming from cursor ${storedCursor}` + : "jetstream-relay started", + ); +} + +main().catch((err) => { + log("error", "Fatal error during startup", err); + process.exit(1); +}); diff --git a/src/relay.test.ts b/src/relay.test.ts new file mode 100644 index 0000000..fcf1271 --- /dev/null +++ b/src/relay.test.ts @@ -0,0 +1,77 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { createRelay, type RelayDeps } from "./relay"; + +function deps(overrides: Partial = {}): RelayDeps { + return { + channel: "ch", + publish: async () => 1, + updateParams: () => {}, + log: () => {}, + ...overrides, + }; +} + +test("successful publish advances the cursor", async () => { + const published: Array<[string, string]> = []; + const relay = createRelay( + deps({ + publish: async (channel, message) => { + published.push([channel, message]); + return 1; + }, + }), + ); + + await relay.onPost({ time_us: 42 }); + + assert.equal(relay.lastTimeUs, 42); + assert.deepEqual(published, [["ch", JSON.stringify({ time_us: 42 })]]); +}); + +test("failed publish does not advance the cursor", async () => { + const errors: string[] = []; + const relay = createRelay( + deps({ + publish: async () => { + throw new Error("redis down"); + }, + log: (level, message) => { + if (level === "error") errors.push(message); + }, + }), + ); + + await relay.onPost({ time_us: 42 }); + + assert.equal(relay.lastTimeUs, 0); + assert.equal(errors.length, 1); +}); + +test("an older event does not move the cursor backwards", async () => { + const relay = createRelay(deps()); + + await relay.onPost({ time_us: 42 }); + await relay.onPost({ time_us: 41 }); + + assert.equal(relay.lastTimeUs, 42); +}); + +test("close updates the client cursor once events have been published", async () => { + const calls: Array<[Record, { immediate: boolean }]> = []; + const relay = createRelay( + deps({ + updateParams: (params, options) => { + calls.push([params, options]); + }, + }), + ); + + relay.onClose(); + assert.equal(calls.length, 0, "no cursor yet — existing params must be kept"); + + await relay.onPost({ time_us: 42 }); + relay.onClose(); + + assert.deepEqual(calls, [[{ cursor: 42 }, { immediate: false }]]); +}); diff --git a/src/relay.ts b/src/relay.ts new file mode 100644 index 0000000..45c1714 --- /dev/null +++ b/src/relay.ts @@ -0,0 +1,46 @@ +export interface RelayDeps { + channel: string; + publish: (channel: string, message: string) => Promise; + updateParams: ( + params: Record, + options: { immediate: boolean }, + ) => void; + log: (level: string, message: string, context?: unknown) => void; +} + +export interface Relay { + /** time_us of the newest event successfully published to Redis. */ + readonly lastTimeUs: number; + onPost(data: { time_us: number }): Promise; + onClose(): void; +} + +export function createRelay(deps: RelayDeps): Relay { + let lastTimeUs = 0; + + return { + get lastTimeUs() { + return lastTimeUs; + }, + + async onPost(data) { + try { + await deps.publish(deps.channel, JSON.stringify(data)); + // Advance only after the publish succeeded, so a reconnect replays + // anything that never reached Redis. + if (data.time_us > lastTimeUs) { + lastTimeUs = data.time_us; + } + } catch (err) { + deps.log("error", "Failed to publish to Redis", err); + } + }, + + onClose() { + if (lastTimeUs > 0) { + deps.updateParams({ cursor: lastTimeUs }, { immediate: false }); + deps.log("info", `Cursor set to ${lastTimeUs} for reconnection`); + } + }, + }; +} diff --git a/tsconfig.json b/tsconfig.json index e4d4895..66a1f44 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -7,11 +7,10 @@ "target": "ES6", "esModuleInterop": true, "moduleResolution": "node", - "alwaysStrict": true, + "strict": true, "allowUnreachableCode": false, - "strictNullChecks": true, "skipLibCheck": true }, "include": ["./src/**/*.ts"], - "exclude": ["node_modules"] + "exclude": ["node_modules", "src/**/*.test.ts"] } -- 2.51.2