From a0138341b837ce27798ca3daaae4ed5835086116 Mon Sep 17 00:00:00 2001 From: Copilot <198982749+Copilot@users.noreply.github.com> Date: Fri, 24 Apr 2026 01:17:51 -0400 Subject: [PATCH] RabbitMQ: automatic reconnect with topology re-assert on connection/channel loss (#120) * Initial plan * Implement resilient RabbitMQ connectivity with reconnect and topology re-assert Co-authored-by: owens1127 <98496129+owens1127@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: owens1127 <98496129+owens1127@users.noreply.github.com> --- src/integrations/rabbitmq/connection.test.ts | 110 +++++++++++++++++ src/integrations/rabbitmq/connection.ts | 107 ++++++++++++---- src/integrations/rabbitmq/queue.test.ts | 122 +++++++++++++++++++ src/integrations/rabbitmq/queue.ts | 47 ++++--- 4 files changed, 347 insertions(+), 39 deletions(-) create mode 100644 src/integrations/rabbitmq/connection.test.ts create mode 100644 src/integrations/rabbitmq/queue.test.ts diff --git a/src/integrations/rabbitmq/connection.test.ts b/src/integrations/rabbitmq/connection.test.ts new file mode 100644 index 00000000..301ae79d --- /dev/null +++ b/src/integrations/rabbitmq/connection.test.ts @@ -0,0 +1,110 @@ +import amqplib from "amqplib" +import { afterAll, beforeEach, describe, expect, mock, spyOn, test } from "bun:test" +import { RabbitConnection, jitteredDelay } from "./connection" + +describe("jitteredDelay", () => { + test("returns a value between 50% and 100% of the exponential delay", () => { + for (let attempt = 0; attempt < 5; attempt++) { + const delay = jitteredDelay(attempt) + const exponential = Math.min(1000 * 2 ** attempt, 30_000) + expect(delay).toBeGreaterThanOrEqual(exponential * 0.5) + expect(delay).toBeLessThanOrEqual(exponential) + } + }) + + test("caps at MAX_RETRY_DELAY_MS", () => { + const delay = jitteredDelay(100) + expect(delay).toBeLessThanOrEqual(30_000) + }) +}) + +describe("RabbitConnection", () => { + const mockChannel = { on: mock(() => {}) } + const mockCreateChannel = mock(() => Promise.resolve(mockChannel)) + + function makeMockConn(onHandler?: (event: string, handler: unknown) => void) { + return { + createChannel: mockCreateChannel, + on: mock(onHandler ?? (() => {})), + close: mock(() => {}) + } + } + + const connectSpy = spyOn(amqplib, "connect") + + beforeEach(() => { + connectSpy.mockReset() + mockCreateChannel.mockClear() + mockCreateChannel.mockResolvedValue(mockChannel) + }) + + afterAll(() => { + connectSpy.mockRestore() + }) + + test("createChannel connects and returns a channel", async () => { + connectSpy.mockResolvedValueOnce(makeMockConn() as never) + + const rabbit = new RabbitConnection({ user: "guest", password: "guest", port: 5672 }) + const channel = await rabbit.createChannel() + + expect(connectSpy).toHaveBeenCalledTimes(1) + expect(channel).toBe(mockChannel as never) + }) + + test("concurrent createChannel calls share a single connect attempt", async () => { + connectSpy.mockResolvedValueOnce(makeMockConn() as never) + + const rabbit = new RabbitConnection({ user: "guest", password: "guest", port: 5672 }) + const [ch1, ch2] = await Promise.all([rabbit.createChannel(), rabbit.createChannel()]) + + expect(connectSpy).toHaveBeenCalledTimes(1) + expect(ch1).toBe(mockChannel as never) + expect(ch2).toBe(mockChannel as never) + }) + + test("reconnects automatically after connection close event", async () => { + let closeHandler: (() => void) | undefined + const firstConn = makeMockConn((event, handler) => { + if (event === "close") closeHandler = handler as () => void + }) + const secondConn = makeMockConn() + + connectSpy + .mockResolvedValueOnce(firstConn as never) + .mockResolvedValueOnce(secondConn as never) + + const rabbit = new RabbitConnection({ user: "guest", password: "guest", port: 5672 }) + await rabbit.createChannel() + expect(connectSpy).toHaveBeenCalledTimes(1) + + // Simulate connection drop + closeHandler?.() + + // Wait for async reconnect to initiate + await new Promise(resolve => setTimeout(resolve, 0)) + + // The reconnect promise is now in-flight; await createChannel which waits for it + await rabbit.createChannel() + expect(connectSpy).toHaveBeenCalledTimes(2) + }) + + test("$disconnect sets destroyed flag preventing future reconnects", async () => { + let closeHandler: (() => void) | undefined + const conn = makeMockConn((event, handler) => { + if (event === "close") closeHandler = handler as () => void + }) + connectSpy.mockResolvedValueOnce(conn as never) + + const rabbit = new RabbitConnection({ user: "guest", password: "guest", port: 5672 }) + await rabbit.createChannel() + + rabbit.$disconnect() + closeHandler?.() + + await new Promise(resolve => setTimeout(resolve, 0)) + + // No additional connect call should have been made after destroy + expect(connectSpy).toHaveBeenCalledTimes(1) + }) +}) diff --git a/src/integrations/rabbitmq/connection.ts b/src/integrations/rabbitmq/connection.ts index a2deef35..1b2281b1 100644 --- a/src/integrations/rabbitmq/connection.ts +++ b/src/integrations/rabbitmq/connection.ts @@ -1,42 +1,101 @@ +import { Logger } from "@/lib/utils/logging" import amqp from "amqplib" +const logger = new Logger("RABBITMQ") + +const INITIAL_RETRY_DELAY_MS = parseInt(process.env.RABBITMQ_RETRY_INITIAL_DELAY_MS ?? "1000") +const MAX_RETRY_DELAY_MS = parseInt(process.env.RABBITMQ_RETRY_MAX_DELAY_MS ?? "30000") + +export function jitteredDelay(attempt: number): number { + const exponential = Math.min(INITIAL_RETRY_DELAY_MS * 2 ** attempt, MAX_RETRY_DELAY_MS) + return exponential * (0.5 + 0.5 * Math.random()) +} + export class RabbitConnection { - private user: string - private password: string - private port: number - private isReady = false - private isConnecting = false - private conn: Promise = Promise.resolve(null) - - constructor(args: { user: string; password: string; port: string | number }) { + private readonly user: string + private readonly password: string + private readonly port: number + private readonly heartbeat: number + private conn: amqp.Connection | null = null + private connectPromise: Promise | null = null + private destroyed = false + + constructor(args: { + user: string + password: string + port: string | number + heartbeat?: number + }) { this.user = args.user this.password = args.password this.port = parseInt(args.port.toString()) + this.heartbeat = args.heartbeat ?? parseInt(process.env.RABBITMQ_HEARTBEAT ?? "60") } - private async connect() { - if (!this.isConnecting && !this.isReady) { - this.isConnecting = true - this.conn = amqp - .connect(`amqp://${this.user}:${this.password}@localhost:${this.port}`) - .finally(() => { - this.isConnecting = false - this.isReady = true - }) - await this.conn + private async connectWithRetry(): Promise { + let attempt = 0 + while (!this.destroyed) { + if (attempt > 0) { + const delay = jitteredDelay(attempt - 1) + logger.info("RABBITMQ_RECONNECTING", { attempt, delayMs: Math.round(delay) }) + await new Promise(resolve => setTimeout(resolve, delay)) + } + try { + const conn = await amqp.connect( + `amqp://${this.user}:${this.password}@localhost:${this.port}`, + { heartbeat: this.heartbeat } + ) + this.conn = conn + logger.info("RABBITMQ_CONNECTED", { attempt }) + conn.on("close", () => this.handleConnectionLoss("close")) + conn.on("error", err => + this.handleConnectionLoss( + "error", + err instanceof Error ? err : new Error(String(err)) + ) + ) + return + } catch (err) { + logger.warn( + "RABBITMQ_CONNECT_FAILED", + err instanceof Error ? err : new Error(String(err)), + { attempt } + ) + attempt++ + } } } - async createChannel() { - await this.connect() - const conn = await this.conn + private handleConnectionLoss(event: string, err?: Error): void { + if (this.conn) { + logger.warn("RABBITMQ_CONNECTION_LOST", err ?? null, { event }) + this.conn = null + } + if (!this.destroyed && !this.connectPromise) { + this.connectPromise = this.connectWithRetry().finally(() => { + this.connectPromise = null + }) + } + } + + async createChannel(): Promise { + if (!this.conn) { + if (!this.connectPromise) { + this.connectPromise = this.connectWithRetry().finally(() => { + this.connectPromise = null + }) + } + await this.connectPromise + } + const conn = this.conn if (!conn) { throw new Error("Failed to connect to RabbitMQ") } - return await conn.createChannel() + return conn.createChannel() } - $disconnect() { - this.conn.then(conn => conn?.close()) + $disconnect(): void { + this.destroyed = true + this.conn?.close() } } diff --git a/src/integrations/rabbitmq/queue.test.ts b/src/integrations/rabbitmq/queue.test.ts new file mode 100644 index 00000000..c4ea6988 --- /dev/null +++ b/src/integrations/rabbitmq/queue.test.ts @@ -0,0 +1,122 @@ +import amqplib from "amqplib" +import { afterAll, beforeEach, describe, expect, mock, spyOn, test } from "bun:test" +import { RabbitConnection } from "./connection" +import { RabbitQueue } from "./queue" + +describe("RabbitQueue", () => { + const mockAssertQueue = mock(() => + Promise.resolve({ queue: "test_queue", messageCount: 0, consumerCount: 0 }) + ) + const mockSendToQueue = mock(() => true) + + function makeMockChannel(onHandler?: (event: string, handler: unknown) => void) { + return { + assertQueue: mockAssertQueue, + sendToQueue: mockSendToQueue, + on: mock(onHandler ?? (() => {})) + } + } + + const mockCreateChannel = mock(() => Promise.resolve(makeMockChannel())) + const mockConn = { + createChannel: mockCreateChannel, + on: mock(() => {}), + close: mock(() => {}) + } + + const connectSpy = spyOn(amqplib, "connect") + + beforeEach(() => { + connectSpy.mockReset() + connectSpy.mockResolvedValue(mockConn as never) + mockAssertQueue.mockClear() + mockSendToQueue.mockClear() + mockCreateChannel.mockClear() + mockCreateChannel.mockResolvedValue(makeMockChannel()) + }) + + afterAll(() => { + connectSpy.mockRestore() + }) + + test("asserts queue topology when channel is first created", async () => { + const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 }) + const queue = new RabbitQueue({ queueName: "test_queue", connection }) + + await queue.send(1) + + expect(mockAssertQueue).toHaveBeenCalledTimes(1) + expect(mockAssertQueue).toHaveBeenCalledWith("test_queue", { durable: true }) + }) + + test("sends message after topology assertion", async () => { + const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 }) + const queue = new RabbitQueue({ queueName: "test_queue", connection }) + + await queue.send(42) + + expect(mockSendToQueue).toHaveBeenCalledTimes(1) + expect(mockSendToQueue).toHaveBeenCalledWith( + "test_queue", + Buffer.from("42"), + expect.objectContaining({ contentType: "text/plain" }) + ) + }) + + test("reuses channel for multiple sends without re-asserting", async () => { + const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 }) + const queue = new RabbitQueue({ queueName: "test_queue", connection }) + + await queue.send(1) + await queue.send(2) + await queue.send(3) + + expect(mockAssertQueue).toHaveBeenCalledTimes(1) + expect(mockSendToQueue).toHaveBeenCalledTimes(3) + }) + + test("re-asserts queue topology after channel loss", async () => { + let closeHandler: (() => void) | undefined + mockCreateChannel.mockResolvedValue( + makeMockChannel((event, handler) => { + if (event === "close") closeHandler = handler as () => void + }) + ) + + const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 }) + const queue = new RabbitQueue({ queueName: "test_queue", connection }) + + await queue.send(1) + expect(mockAssertQueue).toHaveBeenCalledTimes(1) + + // Simulate channel close + closeHandler?.() + + await queue.send(2) + expect(mockAssertQueue).toHaveBeenCalledTimes(2) + expect(mockSendToQueue).toHaveBeenCalledTimes(2) + }) + + test("sends JSON objects", async () => { + const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 }) + const queue = new RabbitQueue<{ id: number }>({ queueName: "test_queue", connection }) + + await queue.sendJson({ id: 99 }) + + expect(mockSendToQueue).toHaveBeenCalledWith( + "test_queue", + Buffer.from(JSON.stringify({ id: 99 })), + expect.objectContaining({ contentType: "application/json" }) + ) + }) + + test("concurrent sends share a single channel creation", async () => { + const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 }) + const queue = new RabbitQueue({ queueName: "test_queue", connection }) + + await Promise.all([queue.send(1), queue.send(2), queue.send(3)]) + + expect(mockAssertQueue).toHaveBeenCalledTimes(1) + expect(mockSendToQueue).toHaveBeenCalledTimes(3) + }) +}) diff --git a/src/integrations/rabbitmq/queue.ts b/src/integrations/rabbitmq/queue.ts index 85d5da41..98ba3dc0 100644 --- a/src/integrations/rabbitmq/queue.ts +++ b/src/integrations/rabbitmq/queue.ts @@ -8,9 +8,8 @@ type Primitive = string | number | bigint | boolean export class RabbitQueue> { readonly queueName: string - private isReady = false - private isConnecting = false - private channel: Promise = Promise.resolve(null) + private channel: amqp.Channel | null = null + private channelPromise: Promise | null = null private conn: RabbitConnection constructor(args: { queueName: string; connection: RabbitConnection }) { @@ -18,14 +17,37 @@ export class RabbitQueue> { this.conn = args.connection } - private async connect() { - if (!this.isConnecting && !this.isReady) { - this.isConnecting = true - this.channel = this.conn.createChannel().finally(() => { - this.isReady = true - this.isConnecting = false + private async createChannel(): Promise { + const ch = await this.conn.createChannel() + ch.on("close", () => this.handleChannelLoss("close")) + ch.on("error", err => + this.handleChannelLoss( + "error", + err instanceof Error ? err : new Error(String(err)) + ) + ) + await ch.assertQueue(this.queueName, { durable: true }) + logger.info("RABBITMQ_QUEUE_READY", { queue: this.queueName }) + this.channel = ch + return ch + } + + private handleChannelLoss(event: string, err?: Error): void { + logger.warn("RABBITMQ_CHANNEL_LOST", err ?? null, { queue: this.queueName, event }) + this.channel = null + this.channelPromise = null + } + + private async getChannel(): Promise { + if (this.channel) { + return this.channel + } + if (!this.channelPromise) { + this.channelPromise = this.createChannel().finally(() => { + this.channelPromise = null }) } + return this.channelPromise } private static primitiveToBuffer(value: Primitive): Buffer { @@ -42,12 +64,7 @@ export class RabbitQueue> { private async sendBuffer(buffer: Buffer, contentType: string) { try { - await this.connect() - const channel = await this.channel - if (!channel) { - throw new Error("Failed to create channel") - } - + const channel = await this.getChannel() return channel.sendToQueue(this.queueName, buffer, { contentType })