From e22dff2b471ef41ce16dda4e42b77bad7cb9fd9d Mon Sep 17 00:00:00 2001 From: "nekomimi.pet" Date: Wed, 7 Oct 2026 21:52:04 -0400 Subject: [PATCH] fix(main-app): ping idle redis connections so haproxy stops cutting them every 180 s main-app's shared Bun RedisClient, firehose's cache-invalidation publisher and hosting's revalidate queue client sit idle behind haproxy redis_write (timeout client/server 180s), so each was cut and reconnected every 180 s. Add startRedisKeepalive to @wispplace/constants (unref'd 60 s PING, skipped while not ready or a ping is outstanding, stopped on close) and use it for the three clients. --- .../lib/cache-invalidation.keepalive.test.ts | 71 ++++++++++++++ .../src/lib/cache-invalidation.ts | 12 +++ .../lib/revalidate-queue.keepalive.test.ts | 68 ++++++++++++++ .../src/lib/revalidate-queue.ts | 10 ++ apps/main-app/src/lib/redis.test.ts | 79 ++++++++++++++++ apps/main-app/src/lib/redis.ts | 29 ++++-- packages/@wispplace/constants/src/index.ts | 1 + .../constants/src/redis-keepalive.test.ts | 92 +++++++++++++++++++ .../constants/src/redis-keepalive.ts | 55 +++++++++++ 9 files changed, 411 insertions(+), 6 deletions(-) create mode 100644 apps/firehose-service/src/lib/cache-invalidation.keepalive.test.ts create mode 100644 apps/hosting-service/src/lib/revalidate-queue.keepalive.test.ts create mode 100644 apps/main-app/src/lib/redis.test.ts create mode 100644 packages/@wispplace/constants/src/redis-keepalive.test.ts create mode 100644 packages/@wispplace/constants/src/redis-keepalive.ts diff --git a/apps/firehose-service/src/lib/cache-invalidation.keepalive.test.ts b/apps/firehose-service/src/lib/cache-invalidation.keepalive.test.ts new file mode 100644 index 0000000..53df6c5 --- /dev/null +++ b/apps/firehose-service/src/lib/cache-invalidation.keepalive.test.ts @@ -0,0 +1,71 @@ +import { afterEach, beforeEach, expect, jest, mock, test } from 'bun:test' +import { config } from '../config' + +class FakeIORedis { + static instances: FakeIORedis[] = [] + status = 'ready' + pings = 0 + quitCalled = false + + constructor() { + FakeIORedis.instances.push(this) + } + + on(): this { + return this + } + + async ping(): Promise { + this.pings++ + return 'PONG' + } + + async eval(): Promise { + return [1, '1-0'] + } + + async quit(): Promise { + this.quitCalled = true + this.status = 'end' + return 'OK' + } +} + +mock.module('ioredis', () => ({ default: FakeIORedis })) + +const { closeCacheInvalidationPublisher, enqueueSiteRevalidation } = await import('./cache-invalidation') +const originalRedisUrl = config.redisUrl + +beforeEach(() => { + jest.useFakeTimers() + config.redisUrl = 'redis://redis.test:6379' + FakeIORedis.instances = [] +}) + +afterEach(async () => { + await closeCacheInvalidationPublisher() + config.redisUrl = originalRedisUrl + jest.useRealTimers() +}) + +test('pings the idle publisher connection inside the 180 s proxy idle timeout', async () => { + expect(await enqueueSiteRevalidation('did:plc:test', 'site', 'storage-miss:index.html')).toBe('enqueued') + const [publisher] = FakeIORedis.instances + + jest.advanceTimersByTime(180_000) + expect(publisher?.pings).toBe(3) + + publisher!.status = 'reconnecting' + jest.advanceTimersByTime(60_000) + expect(publisher?.pings).toBe(3) +}) + +test('stops pinging once the publisher is closed', async () => { + await enqueueSiteRevalidation('did:plc:test', 'site', 'storage-miss:index.html') + const [publisher] = FakeIORedis.instances + await closeCacheInvalidationPublisher() + + jest.advanceTimersByTime(180_000) + expect(publisher?.quitCalled).toBe(true) + expect(publisher?.pings).toBe(0) +}) diff --git a/apps/firehose-service/src/lib/cache-invalidation.ts b/apps/firehose-service/src/lib/cache-invalidation.ts index 6f28ba6..06d0b58 100644 --- a/apps/firehose-service/src/lib/cache-invalidation.ts +++ b/apps/firehose-service/src/lib/cache-invalidation.ts @@ -11,6 +11,7 @@ import { publishCacheInvalidationEvent, revalidationQuarantineKey, revalidationSiteVersionKey, + startRedisKeepalive, } from '@wispplace/constants' import { createLogger } from '@wispplace/observability' import Redis from 'ioredis' @@ -20,6 +21,7 @@ import { isSiteDeleteTombstoneReason } from './revalidate-queue' const logger = createLogger('firehose-service') let publisher: Redis | null = null +let stopPublisherKeepalive: (() => void) | null = null let publisherReadyTimeoutMs = 5_000 let loggedMissingRedis = false @@ -83,6 +85,14 @@ function getPublisher(): Redis | null { publisher.on('ready', () => { logger.info('[CacheInvalidation] Redis publisher connected') }) + + // Publishes are rare, so without a ping haproxy cuts the idle connection + // every 180 s and the next publish has to wait for the reconnect. + const created = publisher + stopPublisherKeepalive = startRedisKeepalive(() => (created.status === 'ready' ? created.ping() : undefined), { + onError: (err) => + logger.warn('[CacheInvalidation] Redis publisher keepalive failed', { errorKind: revalidationErrorKind(err) }), + }) } return publisher @@ -288,6 +298,8 @@ export async function closeCacheInvalidationPublisher(): Promise { if (publisher) { const toClose = publisher publisher = null + stopPublisherKeepalive?.() + stopPublisherKeepalive = null await toClose.quit() } } diff --git a/apps/hosting-service/src/lib/revalidate-queue.keepalive.test.ts b/apps/hosting-service/src/lib/revalidate-queue.keepalive.test.ts new file mode 100644 index 0000000..8479a55 --- /dev/null +++ b/apps/hosting-service/src/lib/revalidate-queue.keepalive.test.ts @@ -0,0 +1,68 @@ +import { afterEach, beforeEach, expect, jest, mock, test } from 'bun:test' + +class FakeIORedis { + static instances: FakeIORedis[] = [] + status = 'ready' + pings = 0 + quitCalled = false + + constructor() { + FakeIORedis.instances.push(this) + } + + on(): this { + return this + } + + async ping(): Promise { + this.pings++ + return 'PONG' + } + + async eval(): Promise { + return [1, '1-0'] + } + + async quit(): Promise { + this.quitCalled = true + this.status = 'end' + return 'OK' + } +} + +mock.module('ioredis', () => ({ default: FakeIORedis })) + +process.env.REDIS_URL = 'redis://redis.test:6379' +const { closeRevalidateQueue, enqueueRevalidate } = await import('./revalidate-queue') + +beforeEach(() => { + jest.useFakeTimers() + FakeIORedis.instances = [] +}) + +afterEach(async () => { + await closeRevalidateQueue() + jest.useRealTimers() +}) + +test('pings the idle revalidate queue connection inside the 180 s proxy idle timeout', async () => { + await enqueueRevalidate('did:plc:test', 'site', 'storage-miss:index.html') + const [client] = FakeIORedis.instances + + jest.advanceTimersByTime(180_000) + expect(client?.pings).toBe(3) + + client!.status = 'reconnecting' + jest.advanceTimersByTime(60_000) + expect(client?.pings).toBe(3) +}) + +test('stops pinging once the queue is closed', async () => { + await enqueueRevalidate('did:plc:test', 'site', 'storage-miss:index.html') + const [client] = FakeIORedis.instances + await closeRevalidateQueue() + + jest.advanceTimersByTime(180_000) + expect(client?.quitCalled).toBe(true) + expect(client?.pings).toBe(0) +}) diff --git a/apps/hosting-service/src/lib/revalidate-queue.ts b/apps/hosting-service/src/lib/revalidate-queue.ts index e99e5cb..9e65c0b 100644 --- a/apps/hosting-service/src/lib/revalidate-queue.ts +++ b/apps/hosting-service/src/lib/revalidate-queue.ts @@ -5,6 +5,7 @@ import { type RevalidateQueueClient, type RevalidateReasonCategory, revalidateReasonCategory, + startRedisKeepalive, } from '@wispplace/constants' import Redis from 'ioredis' import { recordRevalidateResult } from './revalidate-metrics' @@ -30,6 +31,7 @@ const storageMissDedupeTtlSeconds = parseBoundedPositiveInt( ) let client: Redis | null = null +let stopKeepalive: (() => void) | null = null let loggedMissingRedis = false function parseBoundedPositiveInt(value: string | undefined, fallback: number, maximum: number): number { @@ -73,6 +75,12 @@ function getRedisClient(): Redis | null { client.on('ready', () => { console.log(`[Revalidate] Redis connected, stream: ${streamName}`) }) + + // Enqueues are rare, so without a ping haproxy cuts the idle connection every 180 s. + const created = client + stopKeepalive = startRedisKeepalive(() => (created.status === 'ready' ? created.ping() : undefined), { + onError: (err) => console.warn(`[Revalidate] Redis keepalive failed (${redisErrorKind(err)})`), + }) } return client @@ -128,6 +136,8 @@ export async function closeRevalidateQueue(): Promise { if (client) { const toClose = client client = null + stopKeepalive?.() + stopKeepalive = null await toClose.quit() } } diff --git a/apps/main-app/src/lib/redis.test.ts b/apps/main-app/src/lib/redis.test.ts new file mode 100644 index 0000000..d1043dc --- /dev/null +++ b/apps/main-app/src/lib/redis.test.ts @@ -0,0 +1,79 @@ +import { afterEach, beforeEach, describe, expect, jest, test } from 'bun:test' +import type { RedisClient } from 'bun' +import { closeRedisClient, getConnectedRedisClient, getRedisClient, setRedisClientFactoryForTests } from './redis' + +class FakeRedisClient { + connected = false + commands: string[] = [] + closed = false + onconnect: (() => void) | null = null + onclose: ((error: Error) => void) | null = null + + async connect(): Promise { + this.connected = true + } + + send(command: string): Promise { + this.commands.push(command) + return Promise.resolve('PONG') + } + + close(): void { + this.closed = true + this.connected = false + } +} + +describe('main-app Redis keepalive', () => { + const originalRedisUrl = Bun.env.REDIS_URL + let created: FakeRedisClient[] + + beforeEach(() => { + jest.useFakeTimers() + Bun.env.REDIS_URL = 'redis://redis.test:6379' + created = [] + setRedisClientFactoryForTests(() => { + const fake = new FakeRedisClient() + created.push(fake) + return fake as unknown as RedisClient + }) + }) + + afterEach(() => { + closeRedisClient() + setRedisClientFactoryForTests() + jest.useRealTimers() + if (originalRedisUrl === undefined) delete Bun.env.REDIS_URL + else Bun.env.REDIS_URL = originalRedisUrl + }) + + test('pings an idle connection well inside the 180 s proxy idle timeout', async () => { + await getConnectedRedisClient() + const [client] = created + + jest.advanceTimersByTime(59_000) + expect(client?.commands).toEqual([]) + + jest.advanceTimersByTime(1_000) + expect(client?.commands).toEqual(['PING']) + + jest.advanceTimersByTime(120_000) + expect(client?.commands).toEqual(['PING', 'PING', 'PING']) + }) + + test('skips pings while the client is reconnecting', () => { + getRedisClient() + jest.advanceTimersByTime(180_000) + expect(created[0]?.commands).toEqual([]) + }) + + test('stops pinging once the client is closed', async () => { + await getConnectedRedisClient() + const [client] = created + closeRedisClient() + + jest.advanceTimersByTime(180_000) + expect(client?.closed).toBe(true) + expect(client?.commands).toEqual([]) + }) +}) diff --git a/apps/main-app/src/lib/redis.ts b/apps/main-app/src/lib/redis.ts index fa86991..1fd4567 100644 --- a/apps/main-app/src/lib/redis.ts +++ b/apps/main-app/src/lib/redis.ts @@ -1,3 +1,4 @@ +import { startRedisKeepalive } from '@wispplace/constants' import { createLogger } from '@wispplace/observability' import { RedisClient } from 'bun' @@ -5,6 +6,17 @@ const logger = createLogger('main-app:redis') let client: RedisClient | null = null let connectionPromise: Promise | null = null +let stopKeepalive: (() => void) | null = null + +type RedisClientFactory = (url: string) => RedisClient +const defaultRedisClientFactory: RedisClientFactory = (url) => new RedisClient(url) +let redisClientFactory: RedisClientFactory = defaultRedisClientFactory + +/** Test seam for exercising the shared client without a live Redis server. */ +export function setRedisClientFactoryForTests(factory?: RedisClientFactory): void { + if (client) throw new Error('Cannot replace an active Redis client') + redisClientFactory = factory ?? defaultRedisClientFactory +} /** Returns the shared Redis client, creating it lazily. Returns null if REDIS_URL is not set. */ export function getRedisClient(): RedisClient | null { @@ -13,13 +25,20 @@ export function getRedisClient(): RedisClient | null { if (!client) { logger.info('[Redis] Connecting') - const created = new RedisClient(redisUrl) + const created = redisClientFactory(redisUrl) created.onconnect = () => logger.info('[Redis] Connected') created.onclose = (error) => { if (client === created) connectionPromise = null if (error) logger.error('[Redis] Disconnected with error', error) } client = created + // Writes are rare, so without a ping haproxy cuts the idle connection every 180 s. + stopKeepalive = startRedisKeepalive(() => (created.connected ? created.send('PING', []) : undefined), { + onError: (error) => + logger.warn('[Redis] Keepalive ping failed', { + errorName: error instanceof Error ? error.name : 'UnknownError', + }), + }) } return client @@ -34,11 +53,7 @@ export async function getConnectedRedisClient(): Promise { .connect() .then(() => target) .catch((error) => { - if (client === target) { - target.close() - client = null - connectionPromise = null - } + if (client === target) closeRedisClient() throw error }) return await connectionPromise @@ -48,5 +63,7 @@ export function closeRedisClient(): void { const target = client client = null connectionPromise = null + stopKeepalive?.() + stopKeepalive = null target?.close() } diff --git a/packages/@wispplace/constants/src/index.ts b/packages/@wispplace/constants/src/index.ts index 0830d89..faf0dd3 100644 --- a/packages/@wispplace/constants/src/index.ts +++ b/packages/@wispplace/constants/src/index.ts @@ -93,4 +93,5 @@ export const DEFAULT_IGNORE_PATTERNS: string[] = [ export * from './cache-events' // AT Protocol OAuth permission sets for the place.wisp.* namespace export * from './oauth-scopes' +export * from './redis-keepalive' export * from './revalidate-events' diff --git a/packages/@wispplace/constants/src/redis-keepalive.test.ts b/packages/@wispplace/constants/src/redis-keepalive.test.ts new file mode 100644 index 0000000..f0a31e7 --- /dev/null +++ b/packages/@wispplace/constants/src/redis-keepalive.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, test } from 'bun:test' +import { REDIS_KEEPALIVE_INTERVAL_MS, type RedisKeepaliveScheduler, startRedisKeepalive } from './redis-keepalive' + +function manualScheduler() { + const state = { tick: () => {}, intervalMs: 0, cancelled: false } + const schedule: RedisKeepaliveScheduler = (callback, intervalMs) => { + state.tick = callback + state.intervalMs = intervalMs + return () => { + state.cancelled = true + } + } + return { state, schedule } +} + +const settle = () => new Promise((resolve) => setImmediate(resolve)) + +describe('startRedisKeepalive', () => { + test('pings at an interval well under the 180 s haproxy idle timeout', () => { + const { state, schedule } = manualScheduler() + let pings = 0 + startRedisKeepalive( + () => { + pings++ + return Promise.resolve('PONG') + }, + { schedule }, + ) + + expect(state.intervalMs).toBe(REDIS_KEEPALIVE_INTERVAL_MS) + expect(REDIS_KEEPALIVE_INTERVAL_MS * 2).toBeLessThan(180_000) + state.tick() + expect(pings).toBe(1) + }) + + test('does not stack pings behind one that is still outstanding', async () => { + const { state, schedule } = manualScheduler() + let pings = 0 + let resolvePing = () => {} + startRedisKeepalive( + () => { + pings++ + return new Promise((resolve) => { + resolvePing = resolve + }) + }, + { schedule }, + ) + + state.tick() + state.tick() + expect(pings).toBe(1) + resolvePing() + await settle() + state.tick() + expect(pings).toBe(2) + }) + + test('skips a tick when ping returns undefined', () => { + const { state, schedule } = manualScheduler() + const errors: unknown[] = [] + startRedisKeepalive(() => undefined, { schedule, onError: (error) => errors.push(error) }) + state.tick() + expect(errors).toEqual([]) + }) + + test('reports rejected and thrown pings without an unhandled rejection', async () => { + const { state, schedule } = manualScheduler() + const errors: unknown[] = [] + let calls = 0 + startRedisKeepalive( + () => { + calls++ + if (calls === 1) return Promise.reject(new Error('Connection closed')) + throw new Error('Stream not writeable') + }, + { schedule, onError: (error) => errors.push(error) }, + ) + + state.tick() + await settle() + state.tick() + expect(errors.map((error) => (error as Error).message)).toEqual(['Connection closed', 'Stream not writeable']) + }) + + test('returns the scheduler cancel function', () => { + const { state, schedule } = manualScheduler() + const stop = startRedisKeepalive(() => undefined, { schedule }) + stop() + expect(state.cancelled).toBe(true) + }) +}) diff --git a/packages/@wispplace/constants/src/redis-keepalive.ts b/packages/@wispplace/constants/src/redis-keepalive.ts new file mode 100644 index 0000000..32428ec --- /dev/null +++ b/packages/@wispplace/constants/src/redis-keepalive.ts @@ -0,0 +1,55 @@ +/** + * The Redis path through haproxy closes connections idle for 180 s + * (`timeout client` / `timeout server`). TCP keepalive does not reset those + * application timeouts, so a client that can sit idle must send a command. + */ +export const REDIS_KEEPALIVE_INTERVAL_MS = 60_000 + +/** Schedules a repeating callback and returns its cancel function. */ +export type RedisKeepaliveScheduler = (callback: () => void, intervalMs: number) => () => void + +export interface RedisKeepaliveOptions { + intervalMs?: number + schedule?: RedisKeepaliveScheduler + /** Called with a failed ping; the client's own reconnect handles recovery. */ + onError?: (error: unknown) => void +} + +// The keepalive must never hold the event loop open during shutdown. +const scheduleUnrefInterval: RedisKeepaliveScheduler = (callback, intervalMs) => { + const timer = setInterval(callback, intervalMs) + timer.unref() + return () => clearInterval(timer) +} + +/** + * Send `ping` every interval so a proxy idle timeout never cuts the connection. + * `ping` returns undefined to skip a tick (e.g. while the client reconnects); + * a tick is also skipped while the previous ping is still outstanding. + * Do not use on a subscriber-mode connection unless its client accepts PING there. + * + * @returns stop function; idempotent + */ +export function startRedisKeepalive( + ping: () => PromiseLike | undefined, + { intervalMs = REDIS_KEEPALIVE_INTERVAL_MS, schedule = scheduleUnrefInterval, onError }: RedisKeepaliveOptions = {}, +): () => void { + let inFlight = false + return schedule(() => { + if (inFlight) return + let pending: PromiseLike | undefined + try { + pending = ping() + } catch (error) { + onError?.(error) + return + } + if (!pending) return + inFlight = true + Promise.resolve(pending) + .catch((error: unknown) => onError?.(error)) + .finally(() => { + inFlight = false + }) + }, intervalMs) +} -- 2.54.0