diff --git a/.changeset/steady-discord-gateway.md b/.changeset/steady-discord-gateway.md new file mode 100644 index 000000000..1f2385c73 --- /dev/null +++ b/.changeset/steady-discord-gateway.md @@ -0,0 +1,11 @@ +--- +'@roomote/web': patch +--- + +Harden the Discord gateway: quarantined (undeliverable) events now surface in +the Discord settings diagnostics instead of accumulating invisibly, the +durable inbound and dead-letter streams are bounded with capacity pressure +reported before shedding, a single transient Redis blip no longer drops a +healthy Gateway connection or ratchets delivery restarts to the maximum +backoff forever, and a dead gateway supervisor reports to error tracking +instead of only logging. diff --git a/apps/bullmq/src/index.ts b/apps/bullmq/src/index.ts index 5e7e83567..d4349b5f7 100644 --- a/apps/bullmq/src/index.ts +++ b/apps/bullmq/src/index.ts @@ -31,7 +31,7 @@ import { createAdminDashboardMiddleware, resolveAdminDashboardAuth, } from './admin-auth'; -import { initBullMqSentry } from './monitoring/sentry'; +import { captureBullMqMessage, initBullMqSentry } from './monitoring/sentry'; import { getRedis, closeRedis } from './redis'; import { startScheduler } from './scheduler'; import { startSandboxOidcRefreshQueue } from './sandbox-oidc-refresh-queue'; @@ -69,12 +69,25 @@ const redis = getRedis(); initBullMqSentry(); -const discordGatewaySupervisor = startDiscordGatewaySupervisor(redis, { - ...process.env, - ENCRYPTION_KEY: Env.ENCRYPTION_KEY, - R_DISCORD_GATEWAY_SECRET: Env.R_DISCORD_GATEWAY_SECRET, - TRPC_URL: Env.TRPC_URL, -}); +const discordGatewaySupervisor = startDiscordGatewaySupervisor( + redis, + { + ...process.env, + ENCRYPTION_KEY: Env.ENCRYPTION_KEY, + R_DISCORD_GATEWAY_SECRET: Env.R_DISCORD_GATEWAY_SECRET, + TRPC_URL: Env.TRPC_URL, + }, + { + // A dead supervisor silently stops Discord ingestion while this process + // stays green; make sure it pages instead of only logging. + onFatal: (error) => + captureBullMqMessage( + `Discord gateway supervisor stopped unexpectedly: ${error instanceof Error ? error.message : String(error)}`, + undefined, + { component: 'discord-gateway', signal: 'discord-gateway-fatal' }, + ), + }, +); const { schedulerQueue, schedulerWorker, schedulerQueueEvents } = startScheduler(); diff --git a/apps/discord-gateway/src/config.ts b/apps/discord-gateway/src/config.ts index d4573aa3c..f8b0eb321 100644 --- a/apps/discord-gateway/src/config.ts +++ b/apps/discord-gateway/src/config.ts @@ -4,6 +4,8 @@ const DEFAULT_DELIVERY_MAX_ATTEMPTS = 8; const DEFAULT_DELIVERY_MAX_BACKOFF_MS = 30_000; const DEFAULT_LOGIN_RETRY_BASE_MS = 15_000; const DEFAULT_LOGIN_RETRY_MAX_MS = 5 * 60_000; +const DEFAULT_INBOUND_MAX_ENTRIES = 50_000; +const DEFAULT_DEAD_LETTER_MAX_ENTRIES = 1_000; function positiveInteger(value: string | undefined, fallback: number): number { const parsed = Number(value); @@ -33,6 +35,8 @@ export type DiscordGatewayConfig = { deliveryPollMs: number; deliveryMaxAttempts: number; deliveryMaxBackoffMs: number; + inboundMaxEntries: number; + deadLetterMaxEntries: number; loginRetryBaseMs: number; loginRetryMaxMs: number; }; @@ -65,6 +69,14 @@ export function resolveDiscordGatewayConfig( env.DISCORD_GATEWAY_DELIVERY_MAX_BACKOFF_MS, DEFAULT_DELIVERY_MAX_BACKOFF_MS, ), + inboundMaxEntries: positiveInteger( + env.DISCORD_GATEWAY_INBOUND_MAX_ENTRIES, + DEFAULT_INBOUND_MAX_ENTRIES, + ), + deadLetterMaxEntries: positiveInteger( + env.DISCORD_GATEWAY_DEAD_LETTER_MAX_ENTRIES, + DEFAULT_DEAD_LETTER_MAX_ENTRIES, + ), loginRetryBaseMs: positiveInteger( env.DISCORD_GATEWAY_LOGIN_RETRY_BASE_MS, DEFAULT_LOGIN_RETRY_BASE_MS, diff --git a/apps/discord-gateway/src/delivery-loop.test.ts b/apps/discord-gateway/src/delivery-loop.test.ts index 39264c223..9506e37ed 100644 --- a/apps/discord-gateway/src/delivery-loop.test.ts +++ b/apps/discord-gateway/src/delivery-loop.test.ts @@ -245,4 +245,60 @@ describe('Discord delivery worker', () => { }), ); }); + + it('starts the restart backoff over after a long healthy run', async () => { + vi.useFakeTimers(); + try { + const abort = new AbortController(); + const update = vi.fn().mockResolvedValue(undefined); + const runLoop = vi + .fn() + .mockRejectedValueOnce(new Error('blip 1')) + .mockRejectedValueOnce(new Error('blip 2')) + .mockImplementationOnce(async () => { + // A healthy stretch well past restartMaxBackoffMs * 2 before this + // fresh incident. + vi.setSystemTime(Date.now() + 900); + throw new Error('fresh incident'); + }) + .mockImplementationOnce(async () => { + abort.abort(); + }); + + const runPromise = runSupervisedDeliveryLoop({ + queue: {} as DiscordInboundQueue, + forwarder: {} as DiscordApiForwarder, + status: { update } as unknown as GatewayStatusStore, + signal: abort.signal, + pollMs: 100, + restartMaxBackoffMs: 400, + runLoop, + }); + + // First crash sleeps pollMs (100), then the backoff doubles to 200. + await vi.advanceTimersByTimeAsync(0); + expect(runLoop).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(99); + expect(runLoop).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + expect(runLoop).toHaveBeenCalledTimes(2); + + // Second crash in quick succession waits the doubled 200. + await vi.advanceTimersByTimeAsync(199); + expect(runLoop).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(1); + expect(runLoop).toHaveBeenCalledTimes(3); + + // The third run was healthy for 900ms (> 2 * restartMaxBackoffMs), so + // its failure restarts the backoff at pollMs instead of ratcheting on. + await vi.advanceTimersByTimeAsync(99); + expect(runLoop).toHaveBeenCalledTimes(3); + await vi.advanceTimersByTimeAsync(1); + expect(runLoop).toHaveBeenCalledTimes(4); + + await runPromise; + } finally { + vi.useRealTimers(); + } + }); }); diff --git a/apps/discord-gateway/src/delivery-loop.ts b/apps/discord-gateway/src/delivery-loop.ts index 6a77cf1c7..0860f7c07 100644 --- a/apps/discord-gateway/src/delivery-loop.ts +++ b/apps/discord-gateway/src/delivery-loop.ts @@ -155,6 +155,7 @@ export async function runSupervisedDeliveryLoop( let restartDelayMs = params.pollMs; while (!params.signal.aborted) { + const startedAt = Date.now(); try { // This flag represents a running delivery worker, not simply the // presence of an API secret. Any unexpected exit clears it first. @@ -164,6 +165,12 @@ export async function runSupervisedDeliveryLoop( throw new Error('Discord delivery loop exited unexpectedly'); } catch (error) { if (params.signal.aborted) break; + // A long healthy run means this failure is a fresh incident, not part + // of the previous one — start the backoff over instead of ratcheting + // toward the lifetime maximum. + if (Date.now() - startedAt > restartMaxBackoffMs * 2) { + restartDelayMs = params.pollMs; + } await params.status .update({ forwardingReady: false, diff --git a/apps/discord-gateway/src/embedded.test.ts b/apps/discord-gateway/src/embedded.test.ts index 9e47d59f7..74891f788 100644 --- a/apps/discord-gateway/src/embedded.test.ts +++ b/apps/discord-gateway/src/embedded.test.ts @@ -5,6 +5,7 @@ const mocks = vi.hoisted(() => ({ resolveConfig: vi.fn(), run: vi.fn<() => Promise>(), stop: vi.fn<() => Promise>(), + statusUpdate: vi.fn<() => Promise>(), serviceConstructor: vi.fn(), })); @@ -20,6 +21,7 @@ vi.mock('./service', () => ({ run = mocks.run; stop = mocks.stop; + status = { update: mocks.statusUpdate }; }, })); @@ -29,6 +31,7 @@ describe('startDiscordGatewaySupervisor', () => { beforeEach(() => { vi.clearAllMocks(); mocks.resolveConfig.mockReturnValue(mocks.config); + mocks.statusUpdate.mockResolvedValue(undefined); }); it('starts inside the host process and waits for shutdown to finish', async () => { @@ -60,13 +63,25 @@ describe('startDiscordGatewaySupervisor', () => { .mockImplementation(() => undefined); mocks.run.mockRejectedValue(error); mocks.stop.mockResolvedValue(); + const onFatal = vi.fn(); - const supervisor = startDiscordGatewaySupervisor({} as never); + const supervisor = startDiscordGatewaySupervisor({} as never, undefined, { + onFatal, + }); await supervisor.stop(); expect(consoleError).toHaveBeenCalledWith( '[discord-gateway] supervisor stopped unexpectedly: gateway unavailable', ); + // The host is told (so it can page) and the shared status reflects that + // Discord ingestion has stopped, instead of only a console line. + expect(onFatal).toHaveBeenCalledWith(error); + expect(mocks.statusUpdate).toHaveBeenCalledWith( + expect.objectContaining({ + phase: 'error', + lastError: expect.stringContaining('supervisor stopped'), + }), + ); consoleError.mockRestore(); }); }); diff --git a/apps/discord-gateway/src/embedded.ts b/apps/discord-gateway/src/embedded.ts index 5497d0d5f..690e00d35 100644 --- a/apps/discord-gateway/src/embedded.ts +++ b/apps/discord-gateway/src/embedded.ts @@ -15,6 +15,15 @@ export type DiscordGatewaySupervisor = { export function startDiscordGatewaySupervisor( redis: Redis, env: NodeJS.ProcessEnv = process.env, + options: { + /** + * Invoked when the supervisor itself dies. Without it the only signal is + * a console line while the host process stays green and Discord + * ingestion is silently stopped — hosts should report this to their + * error tracker. + */ + onFatal?: (error: unknown) => void; + } = {}, ): DiscordGatewaySupervisor { const service = new DiscordGatewayService( redis, @@ -24,6 +33,15 @@ export function startDiscordGatewaySupervisor( console.error( `[discord-gateway] supervisor stopped unexpectedly: ${error instanceof Error ? error.message : String(error)}`, ); + void service.status + .update({ + phase: 'error', + ready: false, + connected: false, + lastError: `Discord gateway supervisor stopped: ${error instanceof Error ? error.message : String(error)}`, + }) + .catch(() => undefined); + options.onFatal?.(error); }); return { diff --git a/apps/discord-gateway/src/inbound-queue.test.ts b/apps/discord-gateway/src/inbound-queue.test.ts index 7e427f233..7cdf6ae7b 100644 --- a/apps/discord-gateway/src/inbound-queue.test.ts +++ b/apps/discord-gateway/src/inbound-queue.test.ts @@ -13,6 +13,8 @@ function mockRedis(overrides: Record = {}): Redis { eval: vi.fn(), xrange: vi.fn(), hincrby: vi.fn(), + hkeys: vi.fn().mockResolvedValue([]), + hdel: vi.fn(), xlen: vi.fn(), ...overrides, } as unknown as Redis; @@ -39,7 +41,12 @@ describe('DiscordInboundQueue', () => { DISCORD_INBOUND_STREAM_KEY, '86400', JSON.stringify(envelope), + '50000', ); + const enqueueScript = vi.mocked(redis.eval).mock.calls[0]?.[0] as string; + // The durable stream is approximately bounded so a long delivery outage + // cannot grow Redis without limit. + expect(enqueueScript).toContain("'MAXLEN', '~', ARGV[3]"); const script = vi.mocked(redis.eval).mock.calls[0]?.[0] as string; expect(script.indexOf("redis.call('XADD'")).toBeLessThan( script.indexOf("redis.call('SET'"), @@ -143,6 +150,93 @@ describe('DiscordInboundQueue', () => { 'Permanent API rejection', '3', '2026-07-12T13:00:00.000Z', + '1000', ); }); + + it('prunes attempt counters for entries shed by the stream cap', async () => { + const redis = mockRedis({ + // Oldest surviving entry is 200-0: 100-0 and 150-1 were shed by MAXLEN + // without passing through acknowledge/quarantine. + xrange: vi.fn().mockResolvedValue([['200-0', ['envelope', '{}']]]), + hkeys: vi.fn().mockResolvedValue(['100-0', '150-1', '200-0', '250-0']), + hdel: vi.fn().mockResolvedValue(2), + }); + const queue = new DiscordInboundQueue(redis); + + await expect(queue.pruneOrphanedAttempts()).resolves.toBe(2); + expect(redis.hdel).toHaveBeenCalledWith( + DISCORD_INBOUND_ATTEMPTS_KEY, + '100-0', + '150-1', + ); + }); + + it('prunes every attempt counter when the stream is empty', async () => { + const redis = mockRedis({ + xrange: vi.fn().mockResolvedValue([]), + hkeys: vi.fn().mockResolvedValue(['100-0']), + hdel: vi.fn().mockResolvedValue(1), + }); + const queue = new DiscordInboundQueue(redis); + + await expect(queue.pruneOrphanedAttempts()).resolves.toBe(1); + expect(redis.hdel).toHaveBeenCalledWith( + DISCORD_INBOUND_ATTEMPTS_KEY, + '100-0', + ); + }); + + it('never deletes a counter recorded after the snapshot (empty-stream race)', async () => { + // Interleaving under test: HKEYS snapshots an empty hash, an event is + // then enqueued and attempted, and XRANGE still observes the pre-enqueue + // empty stream. The live counter is not in the snapshot, so it survives. + const hkeys = vi.fn().mockResolvedValue([]); + const xrange = vi.fn().mockResolvedValue([]); + const hdel = vi.fn(); + const queue = new DiscordInboundQueue(mockRedis({ hkeys, xrange, hdel })); + + await expect(queue.pruneOrphanedAttempts()).resolves.toBe(0); + expect(hkeys).toHaveBeenCalled(); + // Snapshot-first short-circuits: the stream is never even read, and no + // deletion can touch a counter created after the snapshot. + expect(xrange).not.toHaveBeenCalled(); + expect(hdel).not.toHaveBeenCalled(); + }); + + it('snapshots tracked ids before reading the oldest stream entry', async () => { + const order: string[] = []; + const hkeys = vi.fn(async () => { + order.push('hkeys'); + return ['100-0']; + }); + const xrange = vi.fn(async () => { + order.push('xrange'); + return []; + }); + const hdel = vi.fn().mockResolvedValue(1); + const queue = new DiscordInboundQueue(mockRedis({ hkeys, xrange, hdel })); + + await expect(queue.pruneOrphanedAttempts()).resolves.toBe(1); + expect(order).toEqual(['hkeys', 'xrange']); + }); + + it('applies configured stream bounds and reports dead-letter depth', async () => { + const redis = mockRedis({ + eval: vi.fn().mockResolvedValue('1-0'), + xlen: vi.fn().mockResolvedValueOnce(42).mockResolvedValueOnce(7), + }); + const queue = new DiscordInboundQueue(redis, { + maxEntries: 500, + deadLetterMaxEntries: 50, + }); + + expect(queue.capacity).toBe(500); + await queue.enqueue(envelope); + expect(vi.mocked(redis.eval).mock.calls[0]?.at(-1)).toBe('500'); + + await expect(queue.depth()).resolves.toBe(42); + await expect(queue.deadLetterDepth()).resolves.toBe(7); + expect(redis.xlen).toHaveBeenLastCalledWith(DISCORD_DEAD_LETTER_STREAM_KEY); + }); }); diff --git a/apps/discord-gateway/src/inbound-queue.ts b/apps/discord-gateway/src/inbound-queue.ts index 35711a435..af7713c98 100644 --- a/apps/discord-gateway/src/inbound-queue.ts +++ b/apps/discord-gateway/src/inbound-queue.ts @@ -6,12 +6,14 @@ export const DISCORD_DEAD_LETTER_STREAM_KEY = 'discord:gateway:inbound:dead-letter'; const DEDUPE_PREFIX = 'discord:gateway:dedupe'; const DEDUPE_TTL_SECONDS = 24 * 60 * 60; +const DEFAULT_INBOUND_MAX_ENTRIES = 50_000; +const DEFAULT_DEAD_LETTER_MAX_ENTRIES = 1_000; const ENQUEUE_SCRIPT = ` if redis.call('EXISTS', KEYS[1]) == 1 then return false end -local streamId = redis.call('XADD', KEYS[2], '*', 'envelope', ARGV[2]) +local streamId = redis.call('XADD', KEYS[2], 'MAXLEN', '~', ARGV[3], '*', 'envelope', ARGV[2]) redis.call('SET', KEYS[1], '1', 'EX', ARGV[1]) return streamId `; @@ -29,7 +31,7 @@ if #existing == 0 then return false end local deadLetterId = redis.call( - 'XADD', KEYS[2], '*', + 'XADD', KEYS[2], 'MAXLEN', '~', ARGV[6], '*', 'sourceStreamId', ARGV[1], 'envelope', ARGV[2], 'reason', ARGV[3], @@ -70,8 +72,35 @@ type DiscordInboundQueueEntry = | QueuedDiscordInboundEnvelope | MalformedDiscordInboundEntry; +type DiscordInboundQueueOptions = { + /** + * Approximate cap on the durable inbound stream. During an extended + * delivery outage the oldest undelivered events are shed once the cap is + * exceeded — bounded memory is preferred over an unbounded Redis stream, + * and the status store surfaces the pressure before shedding starts. + */ + maxEntries?: number; + /** Approximate cap on the dead-letter stream. */ + deadLetterMaxEntries?: number; +}; + export class DiscordInboundQueue { - constructor(private readonly redis: Redis) {} + private readonly maxEntries: number; + private readonly deadLetterMaxEntries: number; + + constructor( + private readonly redis: Redis, + options: DiscordInboundQueueOptions = {}, + ) { + this.maxEntries = options.maxEntries ?? DEFAULT_INBOUND_MAX_ENTRIES; + this.deadLetterMaxEntries = + options.deadLetterMaxEntries ?? DEFAULT_DEAD_LETTER_MAX_ENTRIES; + } + + /** Approximate inbound stream cap, for capacity-pressure reporting. */ + get capacity(): number { + return this.maxEntries; + } async enqueue(envelope: DiscordInboundEnvelope): Promise { const dedupeKey = `${DEDUPE_PREFIX}:${envelope.eventType}:${envelope.eventId}`; @@ -82,6 +111,7 @@ export class DiscordInboundQueue { DISCORD_INBOUND_STREAM_KEY, DEDUPE_TTL_SECONDS.toString(), JSON.stringify(envelope), + this.maxEntries.toString(), ); return typeof result === 'string'; @@ -160,6 +190,7 @@ export class DiscordInboundQueue { input.reason.slice(0, 1_000), input.attempts.toString(), (input.quarantinedAt ?? new Date()).toISOString(), + this.deadLetterMaxEntries.toString(), ); return typeof result === 'string'; } @@ -167,4 +198,48 @@ export class DiscordInboundQueue { async depth(): Promise { return this.redis.xlen(DISCORD_INBOUND_STREAM_KEY); } + + async deadLetterDepth(): Promise { + return this.redis.xlen(DISCORD_DEAD_LETTER_STREAM_KEY); + } + + /** + * Deletes attempt counters whose stream entries no longer exist. + * Acknowledge and quarantine clean up their own counters, but entries shed + * by the MAXLEN cap never reach either path, so during a long outage the + * attempts hash would otherwise grow one orphaned field per shed event. + */ + async pruneOrphanedAttempts(): Promise { + // Snapshot the tracked ids BEFORE reading the oldest entry: an event + // enqueued (and attempted) between the two reads is then absent from the + // snapshot, so its live counter can never be mistaken for an orphan when + // the stream looked empty a moment earlier. + const trackedIds = await this.redis.hkeys(DISCORD_INBOUND_ATTEMPTS_KEY); + if (trackedIds.length === 0) { + return 0; + } + const [oldest] = await this.redis.xrange( + DISCORD_INBOUND_STREAM_KEY, + '-', + '+', + 'COUNT', + 1, + ); + const oldestId = oldest?.[0] ?? null; + const orphaned = trackedIds.filter( + (streamId) => + oldestId === null || compareStreamIds(streamId, oldestId) < 0, + ); + if (orphaned.length > 0) { + await this.redis.hdel(DISCORD_INBOUND_ATTEMPTS_KEY, ...orphaned); + } + return orphaned.length; + } +} + +/** Numeric comparison of Redis stream ids ("-"). */ +function compareStreamIds(left: string, right: string): number { + const [leftMs = 0, leftSeq = 0] = left.split('-').map(Number); + const [rightMs = 0, rightSeq = 0] = right.split('-').map(Number); + return leftMs - rightMs || leftSeq - rightSeq; } diff --git a/apps/discord-gateway/src/service.ts b/apps/discord-gateway/src/service.ts index 0fd9bb964..1fa95f7dd 100644 --- a/apps/discord-gateway/src/service.ts +++ b/apps/discord-gateway/src/service.ts @@ -1,4 +1,8 @@ -import { acquireRedisLock, type Redis } from '@roomote/redis'; +import { + acquireRedisLock, + type Redis, + type RedisLockRenewResult, +} from '@roomote/redis'; import { DiscordApiForwarder } from './api-forwarder'; import type { DiscordGatewayConfig } from './config'; @@ -70,7 +74,7 @@ export class DiscordGatewayService { const activeLease = releaseLease; await this.runAsLeader(async () => - activeLease.renew(this.config.leaderLeaseTtlSeconds), + activeLease.renewDetailed(this.config.leaderLeaseTtlSeconds), ); } catch (error) { if (!this.stopped) { @@ -108,8 +112,13 @@ export class DiscordGatewayService { .catch(() => undefined); } - private async runAsLeader(renewLease: () => Promise): Promise { - const queue = new DiscordInboundQueue(this.redis); + private async runAsLeader( + renewLease: () => Promise, + ): Promise { + const queue = new DiscordInboundQueue(this.redis, { + maxEntries: this.config.inboundMaxEntries, + deadLetterMaxEntries: this.config.deadLetterMaxEntries, + }); const session = new DiscordGatewaySession(queue, this.status); this.activeSession = session; @@ -129,31 +138,62 @@ export class DiscordGatewayService { this.config.loginRetryBaseMs, this.config.loginRetryMaxMs, ); - const abortDelivery = () => { delivery.abort.abort(); this.activeDeliveryAbort = delivery.abort; }; + // The lease TTL (30s) leaves room for two renew intervals (10s), so a + // single transient Redis error must not tear down a healthy Gateway + // connection; only a definitive ownership loss or a second consecutive + // failure (past which the lease may genuinely have expired) does. + let renewErrorStreak = 0; + const abandonLeadership = () => { + leaseValid = false; + abortDelivery(); + void session.disconnect(); + }; const leaseTimer = setInterval(() => { void renewLease() - .then((renewed) => { - if (!renewed) { - leaseValid = false; - abortDelivery(); - void session.disconnect(); + .then((result) => { + if (result === 'renewed') { + renewErrorStreak = 0; + return; + } + if (result === 'lost') { + abandonLeadership(); + return; + } + renewErrorStreak += 1; + if (renewErrorStreak >= 2) { + abandonLeadership(); } }) .catch(() => { - leaseValid = false; - abortDelivery(); - void session.disconnect(); + abandonLeadership(); }); }, this.config.leaderLeaseRenewMs); const statusTimer = setInterval(() => { - void queue - .depth() - .then((queueDepth) => this.status.update({ queueDepth })) + void Promise.all([ + queue.depth(), + queue.deadLetterDepth(), + queue.pruneOrphanedAttempts(), + ]) + .then(([queueDepth, deadLetterDepth]) => + this.status.update({ + queueDepth, + deadLetterDepth, + // Surface capacity pressure before the approximate MAXLEN cap + // starts shedding the oldest undelivered events. This is a + // dedicated field with this timer as its only writer, so a + // successful delivery clearing lastError cannot erase it while + // the backlog is still high. + capacityWarning: + queueDepth >= queue.capacity * 0.9 + ? `Inbound event stream is at ${queueDepth}/${queue.capacity} entries; oldest undelivered events will be shed at capacity.` + : undefined, + }), + ) .catch((error) => this.status.update({ lastError: error instanceof Error ? error.message : String(error), diff --git a/apps/discord-gateway/src/status.ts b/apps/discord-gateway/src/status.ts index b52f943b2..b653007d6 100644 --- a/apps/discord-gateway/src/status.ts +++ b/apps/discord-gateway/src/status.ts @@ -24,6 +24,9 @@ type DiscordGatewayStatus = { forwardingReady: boolean; sessionResumed: boolean; queueDepth: number; + deadLetterDepth: number; + /** Set while the inbound stream is close enough to its cap to shed soon. */ + capacityWarning?: string; botUserId?: string; botUsername?: string; lastEventAt?: string; @@ -43,6 +46,7 @@ export class GatewayStatusStore { forwardingReady: false, sessionResumed: false, queueDepth: 0, + deadLetterDepth: 0, updatedAt: new Date().toISOString(), }; diff --git a/apps/web/src/components/settings/DiscordSetupStatus.tsx b/apps/web/src/components/settings/DiscordSetupStatus.tsx index d717f4970..312035a3c 100644 --- a/apps/web/src/components/settings/DiscordSetupStatus.tsx +++ b/apps/web/src/components/settings/DiscordSetupStatus.tsx @@ -83,6 +83,20 @@ export function DiscordSetupStatus({ status }: { status: DiscordCommsStatus }) { : (status.gateway?.lastError ?? 'Discord is still connecting.') } /> + {(status.gateway?.deadLetterDepth ?? 0) > 0 ? ( + + ) : null} + {status.gateway?.capacityWarning ? ( + + ) : null} = { acquired: true; value: T } | { acquired: false }; +/** + * Outcome of a lease renewal that distinguishes definitive ownership loss + * from a transient Redis failure, so holders of long-lived leases can + * tolerate a blip without tearing down still-valid work. + */ +export type RedisLockRenewResult = 'renewed' | 'lost' | 'error'; + export type RedisLockHandle = (() => Promise) & { renew: (ttlSeconds?: number) => Promise; + renewDetailed: (ttlSeconds?: number) => Promise; }; export type ContentionResult = @@ -105,6 +113,21 @@ export async function acquireRedisLock( release.renew = async (ttlSeconds = ttl) => safeRenew(redis, key, ownerId, ttlSeconds); + release.renewDetailed = async (ttlSeconds = ttl) => { + try { + const renewed = await redis.eval( + RENEW_SCRIPT, + 1, + key, + ownerId, + ttlSeconds.toString(), + ); + return renewed === 1 ? 'renewed' : 'lost'; + } catch { + return 'error'; + } + }; + return release; }