Skip to content

Commit 85e7b57

Browse files
Copilotowens1127
Copilot
and
owens1127
authored
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>
1 parent 01ca6be commit 85e7b57

4 files changed

Lines changed: 347 additions & 39 deletions

File tree

‎src/integrations/rabbitmq/connection.test.ts‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
import amqplib from "amqplib"
2+
import { afterAll, beforeEach, describe, expect, mock, spyOn, test } from "bun:test"
3+
import { RabbitConnection, jitteredDelay } from "./connection"
4+
5+
describe("jitteredDelay", () => {
6+
test("returns a value between 50% and 100% of the exponential delay", () => {
7+
for (let attempt = 0; attempt < 5; attempt++) {
8+
const delay = jitteredDelay(attempt)
9+
const exponential = Math.min(1000 * 2 ** attempt, 30_000)
10+
expect(delay).toBeGreaterThanOrEqual(exponential * 0.5)
11+
expect(delay).toBeLessThanOrEqual(exponential)
12+
}
13+
})
14+
15+
test("caps at MAX_RETRY_DELAY_MS", () => {
16+
const delay = jitteredDelay(100)
17+
expect(delay).toBeLessThanOrEqual(30_000)
18+
})
19+
})
20+
21+
describe("RabbitConnection", () => {
22+
const mockChannel = { on: mock(() => {}) }
23+
const mockCreateChannel = mock(() => Promise.resolve(mockChannel))
24+
25+
function makeMockConn(onHandler?: (event: string, handler: unknown) => void) {
26+
return {
27+
createChannel: mockCreateChannel,
28+
on: mock(onHandler ?? (() => {})),
29+
close: mock(() => {})
30+
}
31+
}
32+
33+
const connectSpy = spyOn(amqplib, "connect")
34+
35+
beforeEach(() => {
36+
connectSpy.mockReset()
37+
mockCreateChannel.mockClear()
38+
mockCreateChannel.mockResolvedValue(mockChannel)
39+
})
40+
41+
afterAll(() => {
42+
connectSpy.mockRestore()
43+
})
44+
45+
test("createChannel connects and returns a channel", async () => {
46+
connectSpy.mockResolvedValueOnce(makeMockConn() as never)
47+
48+
const rabbit = new RabbitConnection({ user: "guest", password: "guest", port: 5672 })
49+
const channel = await rabbit.createChannel()
50+
51+
expect(connectSpy).toHaveBeenCalledTimes(1)
52+
expect(channel).toBe(mockChannel as never)
53+
})
54+
55+
test("concurrent createChannel calls share a single connect attempt", async () => {
56+
connectSpy.mockResolvedValueOnce(makeMockConn() as never)
57+
58+
const rabbit = new RabbitConnection({ user: "guest", password: "guest", port: 5672 })
59+
const [ch1, ch2] = await Promise.all([rabbit.createChannel(), rabbit.createChannel()])
60+
61+
expect(connectSpy).toHaveBeenCalledTimes(1)
62+
expect(ch1).toBe(mockChannel as never)
63+
expect(ch2).toBe(mockChannel as never)
64+
})
65+
66+
test("reconnects automatically after connection close event", async () => {
67+
let closeHandler: (() => void) | undefined
68+
const firstConn = makeMockConn((event, handler) => {
69+
if (event === "close") closeHandler = handler as () => void
70+
})
71+
const secondConn = makeMockConn()
72+
73+
connectSpy
74+
.mockResolvedValueOnce(firstConn as never)
75+
.mockResolvedValueOnce(secondConn as never)
76+
77+
const rabbit = new RabbitConnection({ user: "guest", password: "guest", port: 5672 })
78+
await rabbit.createChannel()
79+
expect(connectSpy).toHaveBeenCalledTimes(1)
80+
81+
// Simulate connection drop
82+
closeHandler?.()
83+
84+
// Wait for async reconnect to initiate
85+
await new Promise(resolve => setTimeout(resolve, 0))
86+
87+
// The reconnect promise is now in-flight; await createChannel which waits for it
88+
await rabbit.createChannel()
89+
expect(connectSpy).toHaveBeenCalledTimes(2)
90+
})
91+
92+
test("$disconnect sets destroyed flag preventing future reconnects", async () => {
93+
let closeHandler: (() => void) | undefined
94+
const conn = makeMockConn((event, handler) => {
95+
if (event === "close") closeHandler = handler as () => void
96+
})
97+
connectSpy.mockResolvedValueOnce(conn as never)
98+
99+
const rabbit = new RabbitConnection({ user: "guest", password: "guest", port: 5672 })
100+
await rabbit.createChannel()
101+
102+
rabbit.$disconnect()
103+
closeHandler?.()
104+
105+
await new Promise(resolve => setTimeout(resolve, 0))
106+
107+
// No additional connect call should have been made after destroy
108+
expect(connectSpy).toHaveBeenCalledTimes(1)
109+
})
110+
})

‎src/integrations/rabbitmq/connection.ts‎

Lines changed: 83 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,42 +1,101 @@
1+
import { Logger } from "@/lib/utils/logging"
12
import amqp from "amqplib"
23

4+
const logger = new Logger("RABBITMQ")
5+
6+
const INITIAL_RETRY_DELAY_MS = parseInt(process.env.RABBITMQ_RETRY_INITIAL_DELAY_MS ?? "1000")
7+
const MAX_RETRY_DELAY_MS = parseInt(process.env.RABBITMQ_RETRY_MAX_DELAY_MS ?? "30000")
8+
9+
export function jitteredDelay(attempt: number): number {
10+
const exponential = Math.min(INITIAL_RETRY_DELAY_MS * 2 ** attempt, MAX_RETRY_DELAY_MS)
11+
return exponential * (0.5 + 0.5 * Math.random())
12+
}
13+
314
export class RabbitConnection {
4-
private user: string
5-
private password: string
6-
private port: number
7-
private isReady = false
8-
private isConnecting = false
9-
private conn: Promise<amqp.Connection | null> = Promise.resolve(null)
10-
11-
constructor(args: { user: string; password: string; port: string | number }) {
15+
private readonly user: string
16+
private readonly password: string
17+
private readonly port: number
18+
private readonly heartbeat: number
19+
private conn: amqp.Connection | null = null
20+
private connectPromise: Promise<void> | null = null
21+
private destroyed = false
22+
23+
constructor(args: {
24+
user: string
25+
password: string
26+
port: string | number
27+
heartbeat?: number
28+
}) {
1229
this.user = args.user
1330
this.password = args.password
1431
this.port = parseInt(args.port.toString())
32+
this.heartbeat = args.heartbeat ?? parseInt(process.env.RABBITMQ_HEARTBEAT ?? "60")
1533
}
1634

17-
private async connect() {
18-
if (!this.isConnecting && !this.isReady) {
19-
this.isConnecting = true
20-
this.conn = amqp
21-
.connect(`amqp://${this.user}:${this.password}@localhost:${this.port}`)
22-
.finally(() => {
23-
this.isConnecting = false
24-
this.isReady = true
25-
})
26-
await this.conn
35+
private async connectWithRetry(): Promise<void> {
36+
let attempt = 0
37+
while (!this.destroyed) {
38+
if (attempt > 0) {
39+
const delay = jitteredDelay(attempt - 1)
40+
logger.info("RABBITMQ_RECONNECTING", { attempt, delayMs: Math.round(delay) })
41+
await new Promise(resolve => setTimeout(resolve, delay))
42+
}
43+
try {
44+
const conn = await amqp.connect(
45+
`amqp://${this.user}:${this.password}@localhost:${this.port}`,
46+
{ heartbeat: this.heartbeat }
47+
)
48+
this.conn = conn
49+
logger.info("RABBITMQ_CONNECTED", { attempt })
50+
conn.on("close", () => this.handleConnectionLoss("close"))
51+
conn.on("error", err =>
52+
this.handleConnectionLoss(
53+
"error",
54+
err instanceof Error ? err : new Error(String(err))
55+
)
56+
)
57+
return
58+
} catch (err) {
59+
logger.warn(
60+
"RABBITMQ_CONNECT_FAILED",
61+
err instanceof Error ? err : new Error(String(err)),
62+
{ attempt }
63+
)
64+
attempt++
65+
}
2766
}
2867
}
2968

30-
async createChannel() {
31-
await this.connect()
32-
const conn = await this.conn
69+
private handleConnectionLoss(event: string, err?: Error): void {
70+
if (this.conn) {
71+
logger.warn("RABBITMQ_CONNECTION_LOST", err ?? null, { event })
72+
this.conn = null
73+
}
74+
if (!this.destroyed && !this.connectPromise) {
75+
this.connectPromise = this.connectWithRetry().finally(() => {
76+
this.connectPromise = null
77+
})
78+
}
79+
}
80+
81+
async createChannel(): Promise<amqp.Channel> {
82+
if (!this.conn) {
83+
if (!this.connectPromise) {
84+
this.connectPromise = this.connectWithRetry().finally(() => {
85+
this.connectPromise = null
86+
})
87+
}
88+
await this.connectPromise
89+
}
90+
const conn = this.conn
3391
if (!conn) {
3492
throw new Error("Failed to connect to RabbitMQ")
3593
}
36-
return await conn.createChannel()
94+
return conn.createChannel()
3795
}
3896

39-
$disconnect() {
40-
this.conn.then(conn => conn?.close())
97+
$disconnect(): void {
98+
this.destroyed = true
99+
this.conn?.close()
41100
}
42101
}

‎src/integrations/rabbitmq/queue.test.ts‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
import amqplib from "amqplib"
2+
import { afterAll, beforeEach, describe, expect, mock, spyOn, test } from "bun:test"
3+
import { RabbitConnection } from "./connection"
4+
import { RabbitQueue } from "./queue"
5+
6+
describe("RabbitQueue", () => {
7+
const mockAssertQueue = mock(() =>
8+
Promise.resolve({ queue: "test_queue", messageCount: 0, consumerCount: 0 })
9+
)
10+
const mockSendToQueue = mock(() => true)
11+
12+
function makeMockChannel(onHandler?: (event: string, handler: unknown) => void) {
13+
return {
14+
assertQueue: mockAssertQueue,
15+
sendToQueue: mockSendToQueue,
16+
on: mock(onHandler ?? (() => {}))
17+
}
18+
}
19+
20+
const mockCreateChannel = mock(() => Promise.resolve(makeMockChannel()))
21+
const mockConn = {
22+
createChannel: mockCreateChannel,
23+
on: mock(() => {}),
24+
close: mock(() => {})
25+
}
26+
27+
const connectSpy = spyOn(amqplib, "connect")
28+
29+
beforeEach(() => {
30+
connectSpy.mockReset()
31+
connectSpy.mockResolvedValue(mockConn as never)
32+
mockAssertQueue.mockClear()
33+
mockSendToQueue.mockClear()
34+
mockCreateChannel.mockClear()
35+
mockCreateChannel.mockResolvedValue(makeMockChannel())
36+
})
37+
38+
afterAll(() => {
39+
connectSpy.mockRestore()
40+
})
41+
42+
test("asserts queue topology when channel is first created", async () => {
43+
const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 })
44+
const queue = new RabbitQueue<number>({ queueName: "test_queue", connection })
45+
46+
await queue.send(1)
47+
48+
expect(mockAssertQueue).toHaveBeenCalledTimes(1)
49+
expect(mockAssertQueue).toHaveBeenCalledWith("test_queue", { durable: true })
50+
})
51+
52+
test("sends message after topology assertion", async () => {
53+
const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 })
54+
const queue = new RabbitQueue<number>({ queueName: "test_queue", connection })
55+
56+
await queue.send(42)
57+
58+
expect(mockSendToQueue).toHaveBeenCalledTimes(1)
59+
expect(mockSendToQueue).toHaveBeenCalledWith(
60+
"test_queue",
61+
Buffer.from("42"),
62+
expect.objectContaining({ contentType: "text/plain" })
63+
)
64+
})
65+
66+
test("reuses channel for multiple sends without re-asserting", async () => {
67+
const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 })
68+
const queue = new RabbitQueue<number>({ queueName: "test_queue", connection })
69+
70+
await queue.send(1)
71+
await queue.send(2)
72+
await queue.send(3)
73+
74+
expect(mockAssertQueue).toHaveBeenCalledTimes(1)
75+
expect(mockSendToQueue).toHaveBeenCalledTimes(3)
76+
})
77+
78+
test("re-asserts queue topology after channel loss", async () => {
79+
let closeHandler: (() => void) | undefined
80+
mockCreateChannel.mockResolvedValue(
81+
makeMockChannel((event, handler) => {
82+
if (event === "close") closeHandler = handler as () => void
83+
})
84+
)
85+
86+
const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 })
87+
const queue = new RabbitQueue<number>({ queueName: "test_queue", connection })
88+
89+
await queue.send(1)
90+
expect(mockAssertQueue).toHaveBeenCalledTimes(1)
91+
92+
// Simulate channel close
93+
closeHandler?.()
94+
95+
await queue.send(2)
96+
expect(mockAssertQueue).toHaveBeenCalledTimes(2)
97+
expect(mockSendToQueue).toHaveBeenCalledTimes(2)
98+
})
99+
100+
test("sends JSON objects", async () => {
101+
const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 })
102+
const queue = new RabbitQueue<{ id: number }>({ queueName: "test_queue", connection })
103+
104+
await queue.sendJson({ id: 99 })
105+
106+
expect(mockSendToQueue).toHaveBeenCalledWith(
107+
"test_queue",
108+
Buffer.from(JSON.stringify({ id: 99 })),
109+
expect.objectContaining({ contentType: "application/json" })
110+
)
111+
})
112+
113+
test("concurrent sends share a single channel creation", async () => {
114+
const connection = new RabbitConnection({ user: "guest", password: "guest", port: 5672 })
115+
const queue = new RabbitQueue<number>({ queueName: "test_queue", connection })
116+
117+
await Promise.all([queue.send(1), queue.send(2), queue.send(3)])
118+
119+
expect(mockAssertQueue).toHaveBeenCalledTimes(1)
120+
expect(mockSendToQueue).toHaveBeenCalledTimes(3)
121+
})
122+
})

0 commit comments

Comments
 (0)