From 43c17b72a38940721c6847a68896879c533718f5 Mon Sep 17 00:00:00 2001 From: henry <13052648+gitpancake@users.noreply.github.com> Date: Sun, 19 Jul 2026 20:48:07 -0700 Subject: [PATCH] fix(rabbitmq): recover consumer from channel death and broker cancel FlashcastrConsumer only recovered from connection-level `close`. Three paths left it subscribed to nothing while the process stayed healthy: - channel `close` with a live connection only logged, so a broker-side channel error permanently unsubscribed the consumer - amqplib delivers `null` to the consume callback on broker-side cancellation; that was swallowed by the `if (!msg) return` guard - `connect()` had no handshake deadline, so a socket that opened but never finished the AMQP handshake parked the reconnect loop forever with `reconnecting` stuck true Symptom in production: image-engine online for 2 days, TCP to RabbitMQ ESTABLISHED, zero consumers on flash-engine.flash-received, 17,240 messages backed up, and no log output for ~27h. Only a full process restart recovered it. Reconnect now tears down stale handles first, and `closing` suppresses recovery during intentional shutdown so close() doesn't race the reconnect loop. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HioDdCpmbbk6vrmkLt9Psg --- CLAUDE.md | 1 + libs/rabbitmq/src/consumer.ts | 89 ++++++++++++++++++++++++++++++----- 2 files changed, 77 insertions(+), 13 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 4226875..81c3e12 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -64,6 +64,7 @@ All queues have DLQ via `x-dead-letter-exchange: flashcastr.dlx` and `x-max-leng - **No temporal coupling:** Old system waited 3 min between storing and casting. New system is fully event-driven — neynar-engine only receives messages after IPFS CID is populated. - **Idempotency:** DB uses `ON CONFLICT DO UPDATE`, Pinata deduplicates by content hash, neynar-engine checks for existing `cast_hash` before casting. - **Circuit breakers:** image-engine has IPFS circuit breaker (opens after 30 consecutive failures, resets after 5 min). +- **Consumer recovery:** `FlashcastrConsumer` recovers from *channel* death and broker-side consumer cancellation, not just connection loss — a dead channel on a live TCP connection leaves the process "up" with zero consumers and silently backs up the queue (caused a ~27h image-engine outage, 17k backlog). Connect also races an explicit handshake deadline; amqplib's `timeout` only covers TCP. - **Batch processing:** database-engine accumulates messages and batch-inserts (configurable `BATCH_SIZE`, default 50). - **Retry:** neynar-engine runs a periodic retry worker (every 5 min) for failed casts from last 7 days. diff --git a/libs/rabbitmq/src/consumer.ts b/libs/rabbitmq/src/consumer.ts index 000f31c..bf96a3f 100644 --- a/libs/rabbitmq/src/consumer.ts +++ b/libs/rabbitmq/src/consumer.ts @@ -7,9 +7,11 @@ export abstract class FlashcastrConsumer { private channel: Channel | null = null; private reconnecting: boolean = false; private reconnectAttempts: number = 0; + private closing: boolean = false; private static readonly MAX_RECONNECT_DELAY = 30000; private static readonly INITIAL_RECONNECT_DELAY = 1000; private static readonly HEARTBEAT_INTERVAL = 30; + private static readonly CONNECT_TIMEOUT = 20000; protected readonly rabbitUrl: string; protected readonly queue: string; @@ -43,11 +45,58 @@ export abstract class FlashcastrConsumer { ); } + /** + * Drop the current connection/channel without triggering recovery. Listeners + * are detached first so the resulting `close` events don't re-enter reconnect(). + */ + private async teardown(): Promise { + const { connection, channel } = this; + this.connection = null; + this.channel = null; + + if (channel) { + channel.removeAllListeners(); + try { await channel.close(); } catch { /* already gone */ } + } + if (connection) { + connection.removeAllListeners(); + try { await connection.close(); } catch { /* already gone */ } + } + } + + /** + * amqplib's own `timeout` only covers the TCP connect — a broker that accepts + * the socket but never completes the AMQP handshake leaves the promise pending + * forever, which parks the reconnect loop with `reconnecting` stuck true and no + * consumer registered. Race an explicit deadline so a hung handshake is retried. + */ + private async connectWithTimeout(url: string): Promise { + const attempt = connect(url, { timeout: FlashcastrConsumer.CONNECT_TIMEOUT }); + let timer: ReturnType | undefined; + + const deadline = new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error(`AMQP handshake timed out after ${FlashcastrConsumer.CONNECT_TIMEOUT}ms`)), + FlashcastrConsumer.CONNECT_TIMEOUT + ); + }); + + try { + return await Promise.race([attempt, deadline]); + } catch (err) { + // If the handshake completes after we gave up, close it rather than leak the socket. + attempt.then((c) => c.close().catch(() => {})).catch(() => {}); + throw err; + } finally { + clearTimeout(timer); + } + } + private async reconnect(): Promise { - if (this.reconnecting) return; + if (this.reconnecting || this.closing) return; this.reconnecting = true; - while (true) { + while (!this.closing) { this.reconnectAttempts++; const delay = this.getReconnectDelay(); console.log( @@ -56,6 +105,7 @@ export abstract class FlashcastrConsumer { await new Promise((resolve) => setTimeout(resolve, delay)); try { + await this.teardown(); await this.connect(); this.reconnectAttempts = 0; this.reconnecting = false; @@ -69,6 +119,10 @@ export abstract class FlashcastrConsumer { ); } } + + // Only reached when close() flipped `closing` mid-loop — clear the guard so a + // future startConsuming() on this instance isn't permanently locked out. + this.reconnecting = false; } private async connect(): Promise { @@ -77,13 +131,14 @@ export abstract class FlashcastrConsumer { "heartbeat", String(FlashcastrConsumer.HEARTBEAT_INTERVAL) ); - this.connection = await connect(url.toString()); + this.connection = await this.connectWithTimeout(url.toString()); this.connection.on("error", (err) => { console.error(`[${this.serviceName}] Connection error:`, err.message); }); this.connection.on("close", () => { + if (this.closing) return; console.warn(`[${this.serviceName}] Connection closed — initiating reconnect`); this.connection = null; this.channel = null; @@ -103,7 +158,14 @@ export abstract class FlashcastrConsumer { }); this.channel.on("close", () => { - console.warn(`[${this.serviceName}] Channel closed`); + if (this.closing) return; + // A channel can die while the TCP connection stays open (broker-side channel + // error, consumer cancelled, queue deleted). Nothing else fires in that case, + // so without this the process stays "up" with zero consumers registered and + // the queue silently backs up until someone restarts it. + console.warn(`[${this.serviceName}] Channel closed — initiating reconnect`); + this.channel = null; + this.reconnect(); }); const channel = this.channel; @@ -111,7 +173,14 @@ export abstract class FlashcastrConsumer { await channel.consume( this.queue, async (msg) => { - if (!msg) return; + // amqplib delivers null when the broker cancels the consumer (queue deleted, + // node failover). Swallowing it leaves us subscribed to nothing. + if (!msg) { + if (this.closing) return; + console.warn(`[${this.serviceName}] Consumer cancelled by broker — initiating reconnect`); + this.reconnect(); + return; + } try { const content = msg.content.toString(); @@ -145,13 +214,7 @@ export abstract class FlashcastrConsumer { } async close(): Promise { - if (this.channel) { - try { await this.channel.close(); } catch { /* ignore */ } - this.channel = null; - } - if (this.connection) { - try { await this.connection.close(); } catch { /* ignore */ } - this.connection = null; - } + this.closing = true; + await this.teardown(); } }