feat(amqp): add confirm option to AMQPChannel - #147
Open
adam-hotait wants to merge 1 commit into
Open
Conversation
There was a problem hiding this comment.
Pull request overview
Adds an opt-in publisher-confirm mode to the AMQP channel abstraction so callers can choose to await broker acknowledgements on publishes (improving backpressure/producer correctness), along with integration tests covering confirm-channel publish behavior.
Changes:
- Add
confirm?: booleantoAMQPChannelOptionsand plumb it into internal channel initialization to create aConfirmChannel. - Update
publishto detect confirm channels at runtime and await broker acks via the callback-based publish API. - Add integration tests and test-layer wiring for confirm-channel publishing and broker error surfacing.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/amqp/test/dependencies.ts | Adds a testConfirmChannel layer configured with confirm: true. |
| packages/amqp/test/AMQPChannel.test.ts | Adds confirm-channel integration tests for publish confirmation and error handling. |
| packages/amqp/src/internal/AMQPChannel.ts | Adds confirm flag to internal service, creates confirm channels when enabled, and awaits broker confirms in publish. |
| packages/amqp/src/AMQPChannel.ts | Exposes confirm?: boolean option and documents behavior. |
| .changeset/confirm-channel.md | Records the new minor feature in the changeset. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+162
to
+165
| sendToQueue: (...params: Parameters<Channel["sendToQueue"]>) => | ||
| internal | ||
| .wrapChannelMethod("sendToQueue", async (channel) => channel.sendToQueue(...params)) | ||
| .pipe(provideInternal), |
There was a problem hiding this comment.
He has a point no? Shouldn't we mirror the publish branching in wrapChannelMethod/sendToQueue?
| * When `true`, the channel is opened in publisher-confirm mode | ||
| * (`connection.createConfirmChannel`). Every `publish` call returns only | ||
| * after the broker has acknowledged the message, providing real | ||
| * backpressure and durability guarantees. Defaults to `false`. |
Comment on lines
+8
to
+9
| giving real backpressure and durability guarantees instead of relying on the | ||
| local socket buffer. Defaults to `false` so existing callers are unaffected. |
When `confirm: true` is passed to `AMQPChannel.layer()`, the underlying amqplib channel is created via `connection.createConfirmChannel` and every `publish` call resolves only after the broker has acknowledged the message (or rejected it via the confirm callback). This gives callers real backpressure and per-message durability guarantees instead of relying on the local socket buffer — which is silently dropped on hard process termination (SIGKILL / OOM / liveness-probe kill) when the event loop is busy enqueueing many publishes. Existing callers are unaffected: `confirm` defaults to `false`, in which case `publish` keeps using the synchronous channel API and returns the amqplib backpressure boolean as before.
adam-hotait
force-pushed
the
feat/confirm-channel
branch
from
May 18, 2026 15:00
21de743 to
1a50780
Compare
charlesarnault
approved these changes
May 19, 2026
Comment on lines
+162
to
+165
| sendToQueue: (...params: Parameters<Channel["sendToQueue"]>) => | ||
| internal | ||
| .wrapChannelMethod("sendToQueue", async (channel) => channel.sendToQueue(...params)) | ||
| .pipe(provideInternal), |
There was a problem hiding this comment.
He has a point no? Shouldn't we mirror the publish branching in wrapChannelMethod/sendToQueue?
gauthierdc
reviewed
May 19, 2026
| HttpTraceContext.toHeaders(span) | ||
| ) | ||
| }), | ||
| try: () => channel.publish(exchange, routingKey, content, finalOptions), |
There was a problem hiding this comment.
It does not fix this case which should theorically also be safe to use with many events
Contributor
Author
There was a problem hiding this comment.
It's because it's opt-in to not make a breaking change
gauthierdc
self-requested a review
May 19, 2026 14:35
gauthierdc
approved these changes
May 19, 2026
4 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
confirm?: booleanflag onAMQPChannelOptions. Whentrue, the underlying amqplib channel is opened viaconnection.createConfirmChanneland everypublishresolves only after the broker has acknowledged (or rejected) the message.ConfirmChannel(viawaitForConfirms) and use the callback form ofchannel.publishin that case; existing non-confirm callers keep using the sync path and the backpressure boolean — defaults are unchanged.publishon a confirm channel only completes once the broker has the message, and that publishing to a non-existent exchange surfaces the broker error.Why
Without publisher confirms,
channel.publishonly enqueues into Node's socket write buffer and returns. If the JS event loop stays busy (e.g. publishing thousands of messages in a tight loop, or under a scheduled backfill), the libuv I/O thread never gets a turn to flush the buffer to the kernel. A SIGKILL (OOM, k8s liveness probe failure, etc.) then drops every queued frame, so the broker — and downstream consumers — never see the messages even though the producer thinks they succeeded.A real production incident on
service-compliancematched this exactly: a recompute job fan-out queued ~14 k publishes per scheduled run, the readiness/liveness probes timed out because the event loop never yielded, kubelet killed the pod with exit 137, and zero events landed in the corresponding RabbitMQ queue between restarts.confirm: truemakes eachpublishawait the broker ack, giving real backpressure and ensuring nothing claims "published" without actually being on the broker.Test plan
pnpm check— typecheck passespnpm build— all packages buildpnpm test --run packages/amqp— all 16 tests pass (including the two new confirm-channel tests)AMQPChannel/AMQPSubscriber/AMQPConnectionsuites are unaffected (the default code path is untouched).