Graceful shutdown for RabbitMQ consumers (0.18.0)

[ramhr]

Summary

Additive graceful-shutdown API, inert unless called. Public surface is four things: arnavmq.close(), arnavmq.consumer.inFlight(), connection.isClosed, and ConnectionClosedError (exported from the package entry point, so instanceof works at runtime). Version bump to 0.18.0 (minor, purely additive).

Main focus: the consumer drains already in-flight messages before the connection closes

The point of this PR is that a message whose handler has already started running gets to finish and ack before we tear anything down, instead of being killed mid-flight and redelivered.

arnavmq.close() is two steps:

async close() {
  await this.consumer.stop();   // cancel every subscription, then drain
  await this.connection.close(); // flush channels, then close the socket
}

1. Cancel every subscription by consumerTag, so the broker stops delivering new messages. Always by consumerTag, never by closing the shared channel — DEFAULT_CHANNEL is shared by every default-prefetch consumer on the connection, so closing it would take down unrelated consumers.
2. Drain — wait for every in-flight handler to run to completion and ack. A message the broker had already pushed into the client buffer before cancel-ok landed has not started its handler, so it is rejected/requeued immediately rather than being run; it goes back to the queue for another instance, and nothing is dropped.
3. Close the connection — round-trip Channel.Close/Close-Ok on every cached channel, then close the socket.

The connection stays open through cancel/drain on purpose: a draining handler may still publish() downstream, answer an RPC request it had already received, or complete a new RPC round-trip of its own. All of that keeps working, and anything a handler starts is waited on too. connection.close() is the first point at which publishing starts failing with ConnectionClosedError.

There is no drain timeout. A handler that never finishes means close() never resolves — deliberately left to the orchestrator's own kill grace period rather than this library abandoning in-flight work on a clock. That includes a handler awaiting an RPC response nobody is left to send (the peer is shutting down too, or is served by a consumer this process just cancelled): in that case the pod burns its grace period and gets SIGKILLed. That is an accepted, documented cost, chosen over a timeout that would abandon real work early — by the time the grace period runs out, the acks written by handlers that did finish were flushed to the broker long before, so nothing is at risk beyond a slow exit.

Also fixes a real, reproduced ~50% message-redelivery race on shutdown: channel.ack() is fire-and-forget in AMQP 0-9-1, so tearing down the raw connection right after a successful drain could lose the race against the broker's own disconnect-driven auto-requeue — a message that had been handled and acked would still be redelivered. Fixed by the Channel.Close/Close-Ok round-trip in step 3, which forces the broker to have processed the preceding acks. Channel.Close is bounded at 5s per channel so an unresponsive broker cannot hang shutdown at that step.

Pending RPC requests fail fast, from a channel-close listener

An RPC request still waiting for its response when the channel goes away is rejected with ConnectionClosedError, from a single 'close' listener the producer keeps on the channel.

This is not a shutdown step — earlier revisions of this PR had it as a producer.stop() call in close(), and that was wrong. The reply queue is exclusive to the connection, and every request already went out carrying that queue's name as its replyTo, so once the channel is gone no answer can arrive for those requests ever — not even after a reconnect. Making it reactive means the same code covers a deliberate close() and a connection dropped unexpectedly, with no shutdown-only flag and no hole.

Waiting for those responses instead was considered and rejected: by that point every locally-handled message has already finished, so the response resolves a promise inside a caller that is going away regardless — and with a long rpcTimeout (or rpcTimeout: 0, which arms no timer at all) waiting means blocking shutdown for that duration, or forever, on a peer that may never answer. Callers get a deterministic, catchable failure instead of a hang.

Review instructions

Read the final per-file diff, not the commit sequence. The commits are historical: several later ones reverse decisions made in earlier ones (notably the close() step ordering, and producer.stop(), which was added and then removed entirely). Reviewing commit-by-commit means reviewing designs that are no longer in the branch.

Suggested file order:

1. src/modules/consumer.js — the core of the PR. The subscription registry, _onDelivery()'s in-flight accounting, _cancelSubscription(), _drain(), stop().
2. src/modules/connection.js + src/modules/channels.js — the highest-risk part; please review closely. close()/isClosed and the ack-flush-race fix in Channels.closeAll().
3. src/modules/producer.js — the channel-close listener and the RPC waiter registry.
4. src/modules/arnavmq.js — the two-step close(). Small, but the ordering rationale lives in its JSDoc.
5. types/, test/shutdown-spec.js, README.md — declarations, ~1400 lines of new coverage, docs.

What most needs a human eye:

• The absence of a drain timeout. This is the one deliberate way close() can never resolve. The reasoning is above; disagreement with it is the most valuable review finding here.
• In-flight accounting in _onDelivery(). A message counted as in-flight but never decremented hangs close() forever, since there is no timeout. Every path out of the handler — success, throw, reject/requeue, parse failure — must decrement exactly once.
• The buffered-message rejection path. Confirm a message that arrived before cancel-ok really is requeued, not dropped or double-acked.
• The channel flush in connection.close(). Whether Channel.Close/Close-Ok per cached channel is genuinely sufficient to order the preceding acks ahead of the socket teardown, and whether a channel already closed by the broker is handled.
• No-regression on the reconnect path. _isLive() and the isClosed checks now gate resubscribe behavior. test/disconnect-spec.js is the gate and stays green, but the interaction between shutdown state and reconnect logic is worth a read.

What you can safely skip: the API is additive and inert unless close() is called, so existing-behavior regressions are concentrated in consumer.js's consume path and the reconnect logic, not in the new methods themselves.

Verifying locally: npm test needs a real broker — test/docker.js brings one up. Individual specs: npx mocha test/shutdown-spec.js --exit.

Also in this branch

• eslint:recommended is now actually enabled. The flat config passed js.configs.recommended to FlatCompat's recommendedConfig option, which only tells the compat layer how to resolve "eslint:recommended" when some other config extends it — it never enables the rules. The config is now spread into the array directly. This is not cosmetic: it catches the no-unsafe-finally bug this branch shipped in withTimeout and fixed a few commits later (return clearTimeout(timeoutId) inside a finally, which swallowed the raced promise's result), which had been invisible to CI. Existing warnings were left as warnings so CI stays green.
• Consumer cleanup. _consumeQueue() was one 108-line method wrapped around a nested async closure; the delivery path is now four named methods. A redundant _shuttingDown flag was folded into the already-present _stopPromise, and a dead try/catch around Channels.closeAll() (which cannot reject) was removed from Connection._close().

Process note

Built by AI agents against a written design plan, every task run against a real dockerized RabbitMQ broker. History was reorganized into one commit per logical concern.

The follow-up commits came out of a point-by-point human-led audit of an AI-generated PR review, which flagged four real bugs (of which three sub-claims and two proposed fixes were wrong and rejected): a close() step ordering that could deadlock, an RPC waiter registered before its publish could fail, the eslint gap above, and the producer.stop() design that has now been removed. The current design — including "no drain timeout, let the orchestrator kill us" — is a deliberate human decision, taken over the AI reviewer's recommendation.

Test plan

• npm test — 105 passing, 0 failing; test/disconnect-spec.js (the resubscribe/reconnect regression gate) green throughout
• npm run lint (eslint + prettier + tsc over types/) clean — exit 0
• Human review of the ack-flush-race fix and of the no-drain-timeout decision

[roei-bringg]

will it affect the ack? we want an already running message to be acked once finished

[ramhr]

The consumer:

1. stops receiving new messages
2. rejects any prefetched messages not yet started processing
3. finishes processing and ack-ing in progress messages

The producer doesn't need to ack the messages, except when waiting for RPC responses. On that case, id don't block here, and you could get RPC responses that will never be read. This is not too bad as the message was already processed, so I keep it like this here.

Only after both consumer and producer are done, the channels are closed, and finally the connection is closed.

[qodo-code-review]

PR Summary by Qodo

Add graceful RabbitMQ shutdown and acknowledgement flushing

✨ Enhancement 🐞 Bug fix 🧪 Tests 📝 Documentation 🕐 40+ Minutes

AI Description

• Adds idempotent graceful shutdown across consumers, RPC waiters, channels, and connections.
• Flushes channel acknowledgements before socket teardown to prevent shutdown redeliveries.
• Documents and comprehensively tests cancellation, draining, closure, and TypeScript APIs.

Diagram

sequenceDiagram
  actor App
  participant ArnavMQ
  participant Consumer
  participant Producer
  participant Connection
  participant Channels
  participant RabbitMQ
  App->>ArnavMQ: close()
  ArnavMQ->>Consumer: stop()
  Consumer->>RabbitMQ: basic.cancel
  Consumer-->>ArnavMQ: handlers drained
  ArnavMQ->>Producer: stop RPC waiters
  Producer-->>App: reject pending RPCs
  ArnavMQ->>Connection: close()
  Connection->>Channels: closeAll()
  Channels->>RabbitMQ: Channel.Close
  RabbitMQ-->>Channels: Close-Ok
  Connection->>RabbitMQ: close socket
  ArnavMQ-->>App: shutdown complete

Loading

High-Level Assessment

The staged cancel, drain, RPC rejection, channel flush, and socket-close sequence is appropriate. Cancelling by consumer tag preserves shared channels, while Channel.Close/Close-Ok provides the necessary AMQP ordering barrier that raw acknowledgement calls cannot provide. Closing shared channels earlier or imposing a library-level drain timeout would either disrupt unrelated consumers or abandon active work, so the current approach is preferable.

Files changed (14) +1810 / -76

Enhancement (9) +503 / -68

index.jsExport ConnectionClosedError at runtime +6/-0

Export ConnectionClosedError at runtime

• Exposes ConnectionClosedError from the package entry point so callers can use constructor and instanceof checks.

src/index.js

arnavmq.jsOrchestrate top-level graceful shutdown +33/-1

Orchestrate top-level graceful shutdown

• Adds close() to stop consumers, reject producer RPC waiters, and finally close the connection. Exposes consumer shutdown methods and restores the testable ArnavMQ class export.

src/modules/arnavmq.js

connection.jsAdd terminal, idempotent connection closure +91/-0

Add terminal, idempotent connection closure

• Introduces close(), isClosed, and ConnectionClosedError. Connection shutdown waits for cached channels, handles in-progress connection attempts, suppresses teardown failures, and prevents reconnection afterward.

src/modules/connection.js

consumer.jsTrack subscriptions and drain active handlers +242/-56

Track subscriptions and drain active handlers

• Replaces implicit subscription state with explicit records containing consumer tags, channel listeners, cancellation state, and in-flight counts. Adds cancellation, draining, stopping, resubscribe guards, and immediate requeueing of post-cancellation buffered deliveries.

src/modules/consumer.js

utils.jsAdd reusable promise timeout helper +22/-0

Add reusable promise timeout helper

• Adds withTimeout() for bounding channel-open and channel-close waits while reliably clearing its timer.

src/modules/utils.js

index.d.tsExport ConnectionClosedError type +11/-2

Export ConnectionClosedError type

• Re-exports ConnectionClosedError through the package-level TypeScript namespace alongside existing public types.

types/index.d.ts

arnavmq.d.tsDeclare top-level and consumer shutdown APIs +11/-0

Declare top-level and consumer shutdown APIs

• Adds declarations for close(), cancel(), stop(), drain(), and inFlight() on the public ArnavMQ interface.

types/modules/arnavmq.d.ts

connection.d.tsDeclare terminal connection lifecycle APIs +34/-1

Declare terminal connection lifecycle APIs

• Defines ConnectionClosedError and adds close() and isClosed to both the Connection class and singleton connection interface.

types/modules/connection.d.ts

consumer.d.tsDeclare subscription and draining state +53/-8

Declare subscription and draining state

• Adds the Subscription state type and consumer shutdown method declarations. Updates consume and subscribe results to boolean because cancellation can prevent successful registration.

types/modules/consumer.d.ts

Bug fix (2) +101 / -7

channels.jsFlush cached channels before connection teardown +57/-6

Flush cached channels before connection teardown

• Adds bounded, best-effort closure of every cached channel before socket shutdown. Channel.Close/Close-Ok acts as a barrier ensuring prior acknowledgement and rejection frames reached RabbitMQ.

src/modules/channels.js

producer.jsTerminate pending RPC work during shutdown +44/-1

Terminate pending RPC work during shutdown

• Adds idempotent producer stopping that rejects pending RPC waiters and clears their timers. Treats terminal connection closure as non-retryable and prevents RPC queue reinitialization loops.

src/modules/producer.js

Tests (1) +1188 / -0

shutdown-spec.jsCover graceful shutdown and AMQP teardown races +1188/-0

Cover graceful shutdown and AMQP teardown races

• Adds extensive unit and real-broker coverage for cancellation races, listener hygiene, draining, idempotency, RPC rejection, terminal connections, channel flush ordering, and redelivery prevention.

test/shutdown-spec.js

Documentation (1) +17 / -0

README.mdDocument graceful-shutdown APIs +17/-0

Document graceful-shutdown APIs

• Adds usage guidance for top-level close and lower-level consumer cancellation, draining, stopping, and in-flight inspection. Clarifies that draining has no timeout and shutdown is idempotent.

README.md

Other (1) +1 / -1

package.jsonRelease graceful shutdown as version 0.18.0 +1/-1

Release graceful shutdown as version 0.18.0

• Bumps the package version from 0.17.2 to 0.18.0 for the additive public API release.

package.json

[qodo-code-review]

Code Review by Qodo

🐞 Bugs (1) 📘 Rule violations (0) 📜 Skill insights (0)

1. Channels escape close barrier ✓ Resolved 🐞 Bug ≡ Correctness

Description

closeAll() snapshots and clears the channel cache before awaiting closure, while an
already-started getChannel() or getDefaultChannel() that passed getConnection() before
close() can resume afterward and insert a new channel into the cleared map. Because that late
channel is absent from the snapshot, _close() can close the raw connection without its pending
publish/ack or other fire-and-forget frames being protected by the Channel.Close-Ok barrier,
reintroducing the shutdown race this PR intends to eliminate.

Code

src/modules/channels.js[R114-118]

+  async closeAll() {
+    const entries = [...this._channels.entries()];
+    this._channels.clear();
+
+    await Promise.all(entries.map(([key, entry]) => closeChannel(key, entry)));

Evidence

getChannel() and getDefaultChannel() have an await boundary after their only closed-state check:
they await getConnection() and then access the mutable channel registry without rechecking
terminal shutdown. Meanwhile, shutdown awaits the connection, snapshots and clears the map, and then
closes the socket; because the map's _get() can insert a new entry after that snapshot, a resumed
channel acquisition can create an entry that closeAll() never closes.

src/modules/connection.js[79-125]
src/modules/connection.js[158-166]
src/modules/channels.js[62-94]
src/modules/channels.js[114-119]
src/modules/producer.js[145-152]
src/modules/connection.js[58-69]
src/modules/connection.js[158-165]
src/modules/channels.js[80-95]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description

Connection shutdown snapshots and clears cached channels, but a channel acquisition already in progress can resume afterward and add a channel that is never gracefully closed or flushed before the raw socket closes.

## Issue Context

`getChannel()` and `getDefaultChannel()` await `getConnection()` before accessing `this._channels`, and `close()` cannot revoke a caller that already passed the initial closed-state check. Closing must atomically prevent new channel creation, including operations that passed `getConnection()` before `close()` set `isClosed`; ensure channel acquisition rechecks terminal shutdown after awaits, or serialize channel creation with the shutdown barrier and close any late-created channel before socket teardown.

## Fix Focus Areas

- src/modules/connection.js[79-125]
- src/modules/connection.js[158-166]
- src/modules/channels.js[80-95]
- src/modules/channels.js[114-118]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Cancellation race leaks listeners 🐞 Bug ☼ Reliability

Description

If cancellation occurs while _initializeChannel() is awaiting getChannel(),
_cancelSubscription() returns before a channel/listener exists; initialization then attaches a new
close listener to the already-cancelled record and no later path removes it. Repeated races on the
shared default channel accumulate inert listeners and can trigger EventEmitter leak warnings.

Code

src/modules/consumer.js[R208-215]

+      const onChannelClose = () => {
+        if (!this._isLive(subscription)) {
+          return;
+        }
+        this._subscribe(subscription);
+      };
+      channel.addListener('close', onChannelClose);
+      subscription.channel = channel;

Evidence

Cancellation marks the record and returns when subscription.channel is still null. After the
pending acquisition resolves, _initializeChannel() attaches the listener without rechecking
liveness, and _subscribe() later returns false without removing it.

src/modules/consumer.js[189-216]
src/modules/consumer.js[401-410]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
A cancellation racing pending channel acquisition can leave an inert close listener attached to the shared channel.

## Issue Context
After `getChannel()` resolves, recheck liveness before attaching the listener, or guarantee cleanup on every early return from `_subscribe()`.

## Fix Focus Areas
- src/modules/consumer.js[199-218]
- src/modules/consumer.js[401-410]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

3. Cancelled subscriptions accumulate ✓ Resolved 🐞 Bug ➹ Performance

Description

Every subscribe() permanently appends a record containing its callback, options, channel, and
consumer tag, while cancel() only marks it and never removes or prunes it. Applications that
dynamically subscribe, cancel, and resubscribe will therefore grow _subscriptions and make
cancelAll()/inFlight() increasingly expensive for the lifetime of the process.

Code

src/modules/consumer.js[R138-148]

+    const subscription = {
+      queue,
+      options,
+      callback,
+      channel: null,
+      consumerTag: null,
+      onChannelClose: null,
+      cancelled: false,
+      inFlightCount: 0, // handlers currently running, not yet acked/rejected.
+    };
+    this._subscriptions.push(subscription);

Evidence

Subscription records are pushed on every registration and all counting/cancellation walks the full
array. Cancellation only flips a flag and cancels the broker tag; there is no removal path, and the
new test explicitly confirms cancelled records remain in the registry.

src/modules/consumer.js[138-148]
src/modules/consumer.js[385-421]
src/modules/consumer.js[463-464]
test/shutdown-spec.js[233-249]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
Cancelled subscription records are retained forever, causing unbounded memory and linear shutdown/counting overhead under repeated subscribe/cancel cycles.

## Issue Context
A cancelled record may need to remain while its handlers are in flight, but it can be removed once cancellation is complete and its count reaches zero.

## Fix Focus Areas
- src/modules/consumer.js[138-148]
- src/modules/consumer.js[385-421]
- src/modules/consumer.js[463-464]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

4. Timeout errors are suppressed ✓ Resolved 🐞 Bug ◔ Observability

Description

withTimeout() returns from its finally block, overriding successful results and rejections from
Promise.race() with undefined, so channel.close() failures and five-second timeouts never
reach closeChannel()'s warning path. Shutdown still waits for the race to settle and closeAll()
can continue, but operators receive no indication that the best-effort ack-flush barrier failed
before the socket closed.

Code

src/modules/utils.js[R38-42]

+  try {
+    return await Promise.race([promise, timeout]);
+  } finally {
+    return clearTimeout(timeoutId);
+  }

Evidence

The helper races the operation against a rejecting timeout, but returning clearTimeout(timeoutId)
from finally determines the async function's result and overrides the awaited race outcome.
Because closeChannel() relies on a rejection from this helper to invoke its warning logger, both a
rejected channel.close() and a timeout are converted into silent successful resolutions.

src/modules/utils.js[32-42]
src/modules/channels.js[18-33]
src/modules/channels.js[114-119]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description

`withTimeout()` returns `clearTimeout(timeoutId)` from `finally`, replacing the fulfilled value or rejection from `Promise.race()`. As a result, channel-close failures and timeout errors silently resolve instead of reaching `closeChannel()`'s warning handler.

## Issue Context

`closeChannel()` deliberately catches and logs failures while allowing `closeAll()` to continue, and shutdown still waits for the race to settle. The helper must preserve the race result or rejection while always clearing its timer; the `finally` block should only clear the timer and must not return a value.

## Fix Focus Areas

- src/modules/utils.js[32-42]
- src/modules/channels.js[18-33]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

View medium (2)

5. Queue filter is ignored ✓ Resolved 🐞 Bug ≡ Correctness

Description

The declared inFlight(queue?) API promises a queue-specific count when a queue is supplied, but
the runtime implementation ignores the argument and always sums every subscription. TypeScript
callers can therefore wait on or report the wrong queue's workload whenever another queue has an
active handler.

Code

src/modules/consumer.js[R463-464]

+  inFlight() {
+    return this._subscriptions.reduce((total, sub) => total + sub.inFlightCount, 0);

Evidence

The public declaration accepts an optional queue and explicitly assigns it queue-filter semantics,
while the runtime method accepts no parameter and unconditionally reduces the entire subscription
registry.

types/modules/consumer.d.ts[90-94]
src/modules/arnavmq.js[73-80]
src/modules/consumer.js[459-464]
src/modules/consumer.js[459-465]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description

The TypeScript API declares that `Consumer.inFlight(queue)` returns the count for only the supplied queue, but the JavaScript implementation accepts no parameter and sums all subscriptions.

## Issue Context

Keep the runtime implementation and public declaration aligned. Either update the implementation to preserve global counting when no queue is supplied and filter subscriptions when one is provided, or remove the unsupported parameter and its queue-specific documentation from the declaration.

## Fix Focus Areas

- src/modules/consumer.js[459-465]
- types/modules/consumer.d.ts[90-94]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

6. Late RPC waiters leak ✓ Resolved 🐞 Bug ☼ Reliability

Description

Producer.stop() scans pending waiters only once and _shuttingDown does not gate RPC setup, so a
checkRpc() already awaiting queue initialization can register a correlation waiter after the scan.
The subsequent publish fails as connection.close() starts, but that failure path never removes the
newly registered waiter or timerless deferred, leaving it and its request data retained permanently.

Code

src/modules/producer.js[R186-190]

+  stop() {
+    if (this._shuttingDown) {
+      return;
+    }
+    this._shuttingDown = true;

Evidence

checkRpc() awaits asynchronous queue initialization before inserting the waiter, while stop()
only iterates entries present at that instant. It explicitly uses _shuttingDown only for
idempotency, and waiter cleanup/timeout setup occurs only after a successful publish, so a
shutdown-induced send failure leaves the late entry behind.

src/modules/producer.js[37-39]
src/modules/producer.js[97-141]
src/modules/producer.js[186-203]
src/modules/producer.js[214-237]
src/modules/arnavmq.js[56-60]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
An RPC operation can insert its waiter after `stop()` has completed its one-time cleanup, then leak that waiter when the shutdown-induced publish failure occurs.

## Issue Context
Gate RPC setup after every relevant await and clean up the correlation entry whenever publishing fails.

## Fix Focus Areas
- src/modules/producer.js[186-203]
- src/modules/producer.js[214-237]
- src/modules/arnavmq.js[56-60]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

Context sources

✅ Compliance rules (platform): 1 rule

Review mode: 🧠 Deep: This is a broad, behaviorally dense shutdown change spanning consumer concurrency, connection/channel teardown, producer RPC rejection, public APIs, and many independent race-sensitive paths where a redundant review could catch subtle defects.

Tip of the day

💡 Did you know, you can commit Qodo's fix in one click with committable suggestions (GitHub & GitLab)

More tips ↗ | Customize Qodo ↗ | Qodo docs ↗

[Copilot]

Pull request overview

Adds graceful RabbitMQ shutdown while preserving in-flight message handling and preventing acknowledged-message redelivery.

Changes:

• Adds consumer cancellation/draining and producer RPC cleanup.
• Adds terminal connection closure with channel flushing.
• Adds public APIs, typings, documentation, and shutdown tests.

Reviewed changes

Copilot reviewed 11 out of 15 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
src/modules/consumer.js	Tracks subscriptions and in-flight handlers.
src/modules/connection.js	Adds terminal connection closure.
src/modules/channels.js	Flushes cached channels before disconnecting.
src/modules/producer.js	Rejects pending RPC waiters.
src/modules/arnavmq.js	Orchestrates graceful shutdown.
src/modules/utils.js	Adds a timeout helper.
src/index.js	Exports ConnectionClosedError.
types/modules/consumer.d.ts	Declares consumer shutdown APIs.
types/modules/connection.d.ts	Declares connection closure APIs.
types/modules/arnavmq.d.ts	Declares top-level shutdown support.
types/index.d.ts	Exports the new error type.
test/shutdown-spec.js	Adds extensive shutdown coverage.
test/docker.js	Extends broker readiness checks.
README.md	Documents graceful shutdown.
package.json	Bumps the version to 0.18.0.

Suppressed comments (1)

src/modules/consumer.js:465

• The new declaration and JSDoc promise queue-scoped counts, but the implementation ignores its argument. With handlers active on queues A and B, inFlight('A') currently returns both counts, which makes the exposed API report incorrect state.

  inFlight() {
    return this._subscriptions.reduce((total, sub) => total + sub.inFlightCount, 0);
  }

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[ramhr]

Add a "reason" field to the ConnectionClosedError so we know if it's closed because we are shutting down or another error.

[yosiat]

The high level logic make sense to me and looks ok.
Having said that, I think the shape of the codebase is overly complex and not because of this change, because of technical debt that is accumulated in this project.

We need to invest time to do internal refactoring here.