diff --git a/.env.dev.example b/.env.dev.example index ec6fcf5..1afb36c 100644 --- a/.env.dev.example +++ b/.env.dev.example @@ -47,6 +47,9 @@ HOSTING_SERVICE_URL=http://127.0.0.1:3001 HOSTING_PORT=3001 BASE_HOST=wisp.localhost +# The local PDS is quiet between deploys; this keeps the firehose connected with ping/pong +# instead of treating silence as a dead relay. Development only, Bun runtime only. +FIREHOSE_SOURCE_MODE=dev-pds FIREHOSE_SERVICE=ws://localhost:3300 FIREHOSE_SERVICE_SECONDARY= HEALTH_PORT=3002 diff --git a/apps/firehose-service/src/config.test.ts b/apps/firehose-service/src/config.test.ts index 6344a87..281ff35 100644 --- a/apps/firehose-service/src/config.test.ts +++ b/apps/firehose-service/src/config.test.ts @@ -120,6 +120,26 @@ describe('resolveConfig', () => { expect(prefixOf('previews')).toBe('previews/') }) + test('treats a quiet source as normal only for an explicit local development PDS', () => { + const devPds = (overrides: Record = {}) => + developmentEnv({ FIREHOSE_SOURCE_MODE: 'dev-pds', FIREHOSE_SERVICE: 'ws://localhost:3300', ...overrides }) + + expect(resolveConfig(developmentEnv()).firehoseSourceMode).toBe('relay') + expect(resolveConfig(devPds()).firehoseSourceMode).toBe('dev-pds') + expect(resolveConfig(developmentEnv({ FIREHOSE_SOURCE_MODE: 'relay' })).firehoseSourceMode).toBe('relay') + + expect(() => resolveConfig(devPds({ FIREHOSE_SOURCE_MODE: 'jetstream' }))).toThrow('Invalid FIREHOSE_SOURCE_MODE') + expect(() => resolveConfig(devPds({ FIREHOSE_SERVICE: 'wss://bsky.network' }))).toThrow( + 'FIREHOSE_SOURCE_MODE dev-pds requires a local PDS', + ) + expect(() => resolveConfig(devPds({ FIREHOSE_SERVICE_SECONDARY: 'wss://relay.example.invalid' }))).toThrow( + 'FIREHOSE_SOURCE_MODE dev-pds cannot use a secondary relay', + ) + expect(() => resolveConfig(productionEnv({ FIREHOSE_SOURCE_MODE: 'dev-pds' }))).toThrow( + 'FIREHOSE_SOURCE_MODE dev-pds requires development or test', + ) + }) + test('rejects relay secrets and insecure production relays without logging raw URLs', () => { for (const firehoseService of [ 'wss://user:secret@example.invalid', diff --git a/apps/firehose-service/src/config.ts b/apps/firehose-service/src/config.ts index 97d01d0..c60927b 100644 --- a/apps/firehose-service/src/config.ts +++ b/apps/firehose-service/src/config.ts @@ -20,6 +20,8 @@ export interface FirehoseConfig { databaseUrl: string // Firehose + /** `dev-pds`: silence is normal (a quiet local PDS), so liveness uses ping/pong, not a message timeout. */ + firehoseSourceMode: 'relay' | 'dev-pds' firehoseService: string firehoseServiceSecondary: string | undefined firehoseMaxConcurrency: number @@ -232,6 +234,7 @@ type ResolvedLeaderConfig = Pick< > type ResolvedFirehoseConfig = Pick< FirehoseConfig, + | 'firehoseSourceMode' | 'firehoseService' | 'firehoseServiceSecondary' | 'firehoseMaxConcurrency' @@ -373,10 +376,30 @@ function resolveLeaderConfig(env: ConfigEnv, production: boolean, redisUrl: stri } } -function resolveFirehoseConfig(env: ConfigEnv, production: boolean): ResolvedFirehoseConfig { +function resolveSourceMode( + env: ConfigEnv, + service: string, + context: EnvironmentContext, +): FirehoseConfig['firehoseSourceMode'] { + const mode = env.FIREHOSE_SOURCE_MODE ?? 'relay' + if (mode !== 'relay' && mode !== 'dev-pds') throw new Error('Invalid FIREHOSE_SOURCE_MODE') + if (mode === 'relay') return mode + if (!context.developmentOrTest) throw new Error('FIREHOSE_SOURCE_MODE dev-pds requires development or test') + const hostname = parseUrl(service, 'FIREHOSE_SERVICE').hostname + if (!(isLocalDevelopmentHost(hostname) || hostname === 'pds' || hostname === 'host.docker.internal')) { + throw new Error('FIREHOSE_SOURCE_MODE dev-pds requires a local PDS') + } + if (env.FIREHOSE_SERVICE_SECONDARY) throw new Error('FIREHOSE_SOURCE_MODE dev-pds cannot use a secondary relay') + return mode +} + +function resolveFirehoseConfig(env: ConfigEnv, context: EnvironmentContext): ResolvedFirehoseConfig { + const { production } = context const secondary = env.FIREHOSE_SERVICE_SECONDARY + const service = validateRelayUrl(env.FIREHOSE_SERVICE ?? DEFAULT_FIREHOSE_SERVICE, 'FIREHOSE_SERVICE', production) return { - firehoseService: validateRelayUrl(env.FIREHOSE_SERVICE ?? DEFAULT_FIREHOSE_SERVICE, 'FIREHOSE_SERVICE', production), + firehoseSourceMode: resolveSourceMode(env, service, context), + firehoseService: service, firehoseServiceSecondary: secondary ? validateRelayUrl(secondary, 'FIREHOSE_SERVICE_SECONDARY', production) : undefined, @@ -436,7 +459,7 @@ export function resolveConfig(env: ConfigEnv = {}, args: readonly string[] = []) const storage = resolveStorageConfig(env, context) const queue = resolveQueueConfig(env) const leader = resolveLeaderConfig(env, context.production, queue.redisUrl) - const firehose = resolveFirehoseConfig(env, context.production) + const firehose = resolveFirehoseConfig(env, context) const mode = resolveModeConfig(env, args, context.production) return { diff --git a/apps/firehose-service/src/lib/firehose.ts b/apps/firehose-service/src/lib/firehose.ts index 4d26aab..c332f21 100644 --- a/apps/firehose-service/src/lib/firehose.ts +++ b/apps/firehose-service/src/lib/firehose.ts @@ -808,6 +808,8 @@ function handleError(error: Error, onTooManyFailures?: () => void): void { function handleStall(onTooManyFailures?: () => void): void { if (!acceptingEvents || durableReplayController.pending || relaySwitchPromise) return + // A local development PDS is quiet between deploys. Its liveness is checked by ping/pong. + if (config.firehoseSourceMode === 'dev-pds') return const silenceMs = Date.now() - lastEventTime if (silenceMs < STALL_THRESHOLD_MS) return @@ -855,6 +857,7 @@ function connect(onTooManyFailures: (() => void) | undefined = activeFailureCall if (isCurrentRelay()) handleError(error, onTooManyFailures) }, getCursor: () => getCurrentSeq(), + maxSilenceMs: config.firehoseSourceMode === 'dev-pds' ? null : undefined, onConnect: () => { if (!acceptingEvents || !isCurrentRelay()) return isConnected = true @@ -874,6 +877,9 @@ function connect(onTooManyFailures: (() => void) | undefined = activeFailureCall return } + if (config.firehoseSourceMode === 'dev-pds') { + throw new Error('FIREHOSE_SOURCE_MODE dev-pds needs the Bun runtime') + } const nodeFirehose = new Firehose({ idResolver, service: activeService, diff --git a/packages/@wispplace/bun-firehose/src/firehose.ts b/packages/@wispplace/bun-firehose/src/firehose.ts index a42ff60..9ef6f23 100644 --- a/packages/@wispplace/bun-firehose/src/firehose.ts +++ b/packages/@wispplace/bun-firehose/src/firehose.ts @@ -123,8 +123,8 @@ export interface BunFirehoseOptions { filterCollections?: string[] unauthenticatedCommits?: boolean getCursor?: () => number | undefined | Promise - /** Force reconnect if no messages received within this many ms (default: 15000) */ - maxSilenceMs?: number + /** Force reconnect if no messages received within this many ms (default: 15000); null checks liveness with ping/pong */ + maxSilenceMs?: number | null } export class BunFirehose { diff --git a/packages/@wispplace/bun-firehose/src/subscription.test.ts b/packages/@wispplace/bun-firehose/src/subscription.test.ts new file mode 100644 index 0000000..8d62680 --- /dev/null +++ b/packages/@wispplace/bun-firehose/src/subscription.test.ts @@ -0,0 +1,108 @@ +import { afterEach, describe, expect, test } from 'bun:test' +import { createHash } from 'node:crypto' +import { createServer, type Server } from 'node:net' +import { BunSubscription } from './subscription' + +const WS_GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11' + +interface Harness { + connects: number + disconnects: number + stop: () => void +} + +const cleanups: Array<() => void> = [] +afterEach(() => { + for (const cleanup of cleanups.splice(0)) cleanup() +}) + +/** A server that accepts connections and never sends a frame, answering pings like any websocket server. */ +function idleServer(): { url: string } { + const server = Bun.serve({ + port: 0, + fetch: (req, srv) => (srv.upgrade(req) ? undefined : new Response('no')), + websocket: { message() {} }, + }) + cleanups.push(() => server.stop(true)) + return { url: `ws://127.0.0.1:${server.port}` } +} + +/** Completes the websocket handshake and then ignores everything, so pings go unanswered. */ +function deafServer(): Promise<{ url: string }> { + return new Promise((resolve) => { + const server: Server = createServer((socket) => { + socket.once('data', (chunk) => { + const key = /Sec-WebSocket-Key: (.+)\r\n/i.exec(chunk.toString())?.[1]?.trim() ?? '' + const accept = createHash('sha1') + .update(key + WS_GUID) + .digest('base64') + socket.write( + `HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`, + ) + socket.on('data', () => {}) + }) + socket.on('error', () => {}) + }) + server.listen(0, '127.0.0.1', () => { + const { port } = server.address() as { port: number } + cleanups.push(() => server.close()) + resolve({ url: `ws://127.0.0.1:${port}` }) + }) + }) +} + +function run( + url: string, + extra: { maxSilenceMs?: number | null; heartbeatIntervalMs?: number; heartbeatTimeoutMs?: number }, +): Harness { + const abort = new AbortController() + const harness: Harness = { connects: 0, disconnects: 0, stop: () => abort.abort() } + const subscription = new BunSubscription({ + service: url, + method: 'test.subscribe', + signal: abort.signal, + validate: () => undefined, + maxReconnectSeconds: 1, + onConnect: () => harness.connects++, + onDisconnect: () => harness.disconnects++, + ...extra, + }) + void (async () => { + for await (const _ of subscription) { + } + })() + cleanups.push(harness.stop) + return harness +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) + +describe('BunSubscription silence handling', () => { + test('reconnects a silent source by default', async () => { + const { url } = idleServer() + const harness = run(url, { maxSilenceMs: 100 }) + + await sleep(500) + + expect(harness.disconnects).toBeGreaterThanOrEqual(1) + }) + + test('keeps an idle but answering source connected when silence is not a failure', async () => { + const { url } = idleServer() + const harness = run(url, { maxSilenceMs: null, heartbeatIntervalMs: 50, heartbeatTimeoutMs: 100 }) + + await sleep(800) + + expect(harness.connects).toBe(1) + expect(harness.disconnects).toBe(0) + }) + + test('reconnects a source that stops answering pings', async () => { + const { url } = await deafServer() + const harness = run(url, { maxSilenceMs: null, heartbeatIntervalMs: 50, heartbeatTimeoutMs: 100 }) + + await sleep(600) + + expect(harness.disconnects).toBeGreaterThanOrEqual(1) + }) +}) diff --git a/packages/@wispplace/bun-firehose/src/subscription.ts b/packages/@wispplace/bun-firehose/src/subscription.ts index f1e11b1..a63b14f 100644 --- a/packages/@wispplace/bun-firehose/src/subscription.ts +++ b/packages/@wispplace/bun-firehose/src/subscription.ts @@ -46,10 +46,20 @@ export interface BunSubscriptionOptions { onConnect?: () => void onDisconnect?: () => void maxReconnectSeconds?: number - /** Force reconnect if no messages received within this many ms (default: 15000 = 15s) */ - maxSilenceMs?: number + /** + * Force reconnect if no messages received within this many ms (default: 15000 = 15s). `null` + * says silence is normal for this source (a quiet local PDS): liveness is then checked with + * websocket ping/pong instead, and pongs never count as messages. + */ + maxSilenceMs?: number | null + /** Ping interval when `maxSilenceMs` is null (default: 5000) */ + heartbeatIntervalMs?: number + /** How long a ping may go unanswered before the connection is dropped (default: 5000) */ + heartbeatTimeoutMs?: number } +type PingWebSocket = WebSocket & { ping(data: string): unknown } + export class BunSubscription { private ws: WebSocket | null = null private reconnectAttempts = 0 @@ -78,7 +88,7 @@ export class BunSubscription { } async *[Symbol.asyncIterator](): AsyncGenerator { - const maxSilenceMs = this.opts.maxSilenceMs ?? 15_000 + const maxSilenceMs = this.opts.maxSilenceMs === undefined ? 15_000 : this.opts.maxSilenceMs while (!this.aborted) { let silenceTimer: ReturnType | null = null @@ -93,16 +103,33 @@ export class BunSubscription { let wsOpen = false let wsClosed = false + const dropSilentConnection = (reason: string) => { + if (!wsClosed && !this.aborted) { + console.warn(`[BunSubscription] ${reason}, forcing reconnect`) + wsClosed = true + this.ws?.close() + resolveMessage?.() + } + } + const resetSilenceTimer = () => { + if (maxSilenceMs === null) return + if (silenceTimer) clearTimeout(silenceTimer) + silenceTimer = setTimeout(() => dropSilentConnection(`No messages for ${maxSilenceMs / 1000}s`), maxSilenceMs) + } + + let pingNonce: string | undefined + const scheduleHeartbeat = () => { if (silenceTimer) clearTimeout(silenceTimer) silenceTimer = setTimeout(() => { - if (!wsClosed && !this.aborted) { - console.warn(`[BunSubscription] No messages for ${maxSilenceMs / 1000}s, forcing reconnect`) - wsClosed = true - this.ws?.close() - resolveMessage?.() + pingNonce = crypto.randomUUID() + silenceTimer = setTimeout(() => dropSilentConnection('No pong'), this.opts.heartbeatTimeoutMs ?? 5_000) + try { + ;(this.ws as PingWebSocket).ping(pingNonce) + } catch { + dropSilentConnection('Ping failed') } - }, maxSilenceMs) + }, this.opts.heartbeatIntervalMs ?? 5_000) } this.ws = new WebSocket(url) @@ -112,7 +139,14 @@ export class BunSubscription { wsOpen = true this.reconnectAttempts = 0 this.opts.onConnect?.() - resetSilenceTimer() + if (maxSilenceMs === null) scheduleHeartbeat() + else resetSilenceTimer() + }) + + this.ws.addEventListener('pong', (event) => { + if (pingNonce === undefined || decodePong((event as MessageEvent).data) !== pingNonce) return + pingNonce = undefined + scheduleHeartbeat() }) this.ws.addEventListener('message', (event) => { @@ -220,6 +254,14 @@ export class BunSubscription { } } +function decodePong(data: unknown): string | undefined { + if (typeof data === 'string') return data + if (data instanceof ArrayBuffer) return new TextDecoder().decode(data) + if (ArrayBuffer.isView(data)) + return new TextDecoder().decode(new Uint8Array(data.buffer, data.byteOffset, data.byteLength)) + return undefined +} + function encodeQueryParams(obj: Record): string { const params = new URLSearchParams() for (const [key, value] of Object.entries(obj)) {