feat!: upgrade to bunny 0.6 - #55
Open
trearcul wants to merge 2 commits into
Open
Conversation
There was a problem hiding this comment.
Pull request overview
Upgrades the bundle to Bunny 0.6’s Fiber-based API while preserving synchronous bundle operations.
Changes:
- Wraps RabbitMQ I/O in
Connection::run()and updates Bunny APIs. - Reworks consumer lifecycle, failure propagation, and connection cleanup.
- Removes
read_write_timeoutand updates dependencies and documentation.
Reviewed changes
Copilot reviewed 27 out of 27 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
composer.json |
Upgrades Bunny and ReactPHP dependencies. |
composer-dependency-analyser.php |
Adjusts dependency analysis. |
phpunit.xml.dist |
Updates test DSN options. |
src/Configuration/Connection.php |
Removes read/write timeout configuration. |
src/ConsumerRunner.php |
Reworks Fiber-based consumption and stopping. |
src/DependencyInjection/Configuration.php |
Removes the obsolete YAML option. |
src/EventListener/DisconnectConnection.php |
Adds termination-time disconnection. |
src/Exception/CannotCreateChannel.php |
Removes obsolete promise-type errors. |
src/Exception/ConfigurationFailed.php |
Preserves underlying exceptions. |
src/Exception/ConnectionFailed.php |
Adds channel-closure failure reporting. |
src/Exception/OperationFailed.php |
Removes obsolete promise-type errors. |
src/RabbitMQ/BunnyConnection.php |
Implements Bunny 0.6 connection lifecycle. |
src/RabbitMQ/Connection.php |
Adds run() and interface channel types. |
src/RabbitMQ/Operation/AcknowledgeOperation.php |
Uses the new acknowledgment API. |
src/RabbitMQ/Operation/GetOperation.php |
Runs message retrieval inside a Fiber. |
src/RabbitMQ/Operation/PublishOperation.php |
Updates publishing and transactions. |
src/RabbitMQ/Operation/RejectOperation.php |
Uses the new nack API. |
src/Resources/config/services.yaml |
Registers the disconnect subscriber. |
src/SetupAction.php |
Updates topology operations and error handling. |
tests/ConsumerRunnerTest.php |
Tests limits and consumer exceptions. |
tests/EventListener/DisconnectConnectionTest.php |
Tests termination subscriptions. |
tests/RabbitMQ/ConfigurationTest.php |
Removes obsolete timeout assertions. |
tests/RabbitMQ/ConfigurationTest.yaml |
Removes the obsolete DSN parameter. |
tests/RabbitMQ/ThrowingConsumer.php |
Adds a failing consumer fixture. |
docs/Installation.md |
Documents upgrade requirements. |
docs/Producing.md |
Documents producer cleanup behavior. |
docs/Setup.md |
Updates the example DSN. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #55 +/- ##
===========================================
+ Coverage 51.98% 68.87% +16.88%
===========================================
Files 29 29
Lines 654 816 +162
===========================================
+ Hits 340 562 +222
+ Misses 314 254 -60 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 26 out of 26 changed files in this pull request and generated 2 comments.
Suppressed comments (1)
src/ConsumerRunner.php:112
- When
Consumer::consume()throws before acknowledging or rejecting this delivery, the code rejects$stoppedand later only cancels the consumer. AMQPbasic.canceldoes not requeue deliveries that are already unacknowledged, and this channel remains cached/open, so the failed message can stay invisible to every consumer until the whole connection is eventually closed. On the failure path, explicitly settle the current delivery or close/evict the channel (while preserving the original consumer exception) so outstanding deliveries are requeued beforerun()returns.
} catch (Throwable $error) {
// The callback runs in its own Fiber, so throwing here would only surface
// as an unhandled promise rejection. Hand the failure to the awaited
// promise instead to let it propagate out of run().
$fail($error);
trearcul
force-pushed
the
bunny-0.6
branch
2 times, most recently
from
August 24, 2026 15:16
6b01c4c to
4dd55a8
Compare
trearcul
force-pushed
the
bunny-0.6
branch
2 times, most recently
from
August 25, 2026 12:35
763ff9e to
8c36abb
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 33 out of 33 changed files in this pull request and generated 1 comment.
Suppressed comments (1)
src/ConsumerRunner.php:70
- A connection loss during the
consume()handshake still hangs this call. Bunny leaves the protocol wait pending when the socket closes; because the channel listeners and timeout are installed only afterconsume()returns, neither can settle this Fiber if the connection drops beforebasic.consume-ok. Please supervise the subscription handshake itself (for example, register failure handling first and race an asyncconsume()task against close/error/timeout), or ensure the underlying pending wait is rejected.
$consumeOk = $channel->consume(
Bunny 0.6 is a rewrite: it dropped the sync/async duality and runs on a
ReactPHP event loop using PHP Fibers, so every Client/Channel call has to
execute inside a Fiber, and failures are reported by throwing instead of by
returning *OkFrame protocol objects.
All broker I/O is funnelled through the new Connection::run(Closure), which
wraps the operation in React\Async\await(async(...)), and does so even when a
Fiber is already current - running the operation on the calling Fiber instead
leaves PHP unable to switch out of contexts that forbid it, such as a signal
handler. Callers keep a plain synchronous API and must never drive the loop
themselves. The *OkFrame return checks in SetupAction and ConsumerRunner become
try/catch blocks that pass the cause through as $previous, and the instanceof
PromiseInterface guards are gone now that there is a single return type.
ConsumerRunner no longer polls Client::run() in a while loop; it awaits a
Deferred settled by the message-count limit or a Loop timer. Bunny reports
asynchronous failures as 'error'/'close' events, which Evenement silently drops
when nobody listens, so the runner listens on both and wraps Consumer::consume()
in a try/catch - without that, a throwing consumer, a broker-closed channel or a
lost connection left the consume loop blocked forever with nothing but an
unhandled promise rejection on stderr. Both events are needed: a broker that
closes the channel emits 'close' and then 'error' with the reply code and text,
while a connection that goes away emits only 'close'. Since the first rejection
is the one that counts, the 'close' fallback is held back by a tick so that the
error saying why gets there first. Every settle goes through
Loop::futureTick(), because resuming the awaiting Fiber from inside a delivery
callback's own Fiber never lets React's scheduler hand the result back to
run()'s caller.
While connected, Bunny keeps a heartbeat timer on the loop that would keep a
php-fpm worker or console process alive after the work is done. The new
DisconnectConnection subscriber closes the connection on kernel.terminate and
console.terminate, covering producers in both HTTP and CLI contexts.
BunnyConnection also drops a cached channel on its 'close' event, since 0.6
throws ChannelException('Channel is closed') for every later call on a channel
the broker closed and a single 404 publish would otherwise poison the connection
for the rest of the process, and it replaces a client whose connect() failed,
because Bunny leaves such a client reporting itself as connected while refusing
to connect again.
One upstream limitation remains: bunny 0.6.0-alpha.4 cannot recover in-process
from a first connection attempt that failed, even with a brand-new Client. A
connection that was established and then lost reconnects on the next operation.
BREAKING CHANGE:
* Requires bunny/bunny 0.6, currently available only as an alpha pre-release,
so consuming projects have to allow that stability.
* The read_write_timeout option is gone, as Bunny 0.6 has no read/write
timeout. The DSN query parameter is merely ignored, but the YAML key now
fails the container build and has to be removed.
* RabbitMQ\Connection gained run(Closure): mixed - templated over what the
closure returns, so callers keep the type they passed in - and getChannel()
and getTransactionalChannel() now return Bunny\ChannelInterface rather than
Bunny\Channel, so custom implementations need updating.
Signed-off-by: Pavel Vondrak <pavel.vondrak@hotmail.cz>
trearcul
force-pushed
the
bunny-0.6
branch
2 times, most recently
from
August 25, 2026 15:06
55f5ab4 to
dc03f2a
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 33 out of 33 changed files in this pull request and generated 5 comments.
Suppressed comments (1)
src/RabbitMQ/BunnyConnection.php:164
- A configured timeout of
0is documented as disabling only the time bound, but this early return also skips the clienterrorlistener below. If the socket fails while an operation is awaiting a Bunny reply,runWithoutTimeout()has nothing that can resume the Fiber, so the call hangs even though Bunny emitted the failure. Keep the error race active when no timer is configured; only omit the timer itself.
if ($this->operationTimeout <= 0.0) {
return $this->runWithoutTimeout($operation);
}
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 34 out of 34 changed files in this pull request and generated 2 comments.
Suppressed comments (1)
src/ConsumerRunner.php:139
- When
consume()throws before acknowledging this delivery, this path only rejects the runner's stop promise. Cleanup then sendsbasic.cancel, which removes the consumer but does not requeue already-delivered unacknowledged messages, and the cached channel remains open. If the caller catches the propagated exception and continues, the failed message stays invisible indefinitely. Explicitly settle the current delivery or close/invalidate the channel on this failure path before propagating the error, while accounting for consumers that may have acknowledged before throwing.
} catch (Throwable $error) {
// The callback runs in its own Fiber, so throwing here would only
// surface as an unhandled promise rejection. Hand the failure to
// the awaited promise instead to let it propagate out of run().
$fail($error);
trearcul
force-pushed
the
bunny-0.6
branch
2 times, most recently
from
August 26, 2026 08:41
9c58d11 to
39cf308
Compare
Comment on lines
+59
to
+68
| if (! $event instanceof ConsoleTerminateEvent) { | ||
| return; | ||
| } | ||
|
|
||
| // Belt and braces for a command that is on its way out: nothing should be left on the loop | ||
| // after a local teardown, but a stray timer there would hold the process in React's own | ||
| // shutdown autorun with no way to notice. Console only - `Loop::stop()` is process-wide and | ||
| // permanent, and a php-fpm worker serves further requests after `kernel.terminate`: their | ||
| // first await() would resume a scheduler Fiber sitting in a loop that no longer runs. | ||
| Loop::stop(); |
| $this->runningOperations++; | ||
|
|
||
| try { | ||
| $result = await(race([async($operation)(), $failed->promise()])); |
trearcul
force-pushed
the
bunny-0.6
branch
2 times, most recently
from
August 26, 2026 15:42
02f8283 to
74ac56f
Compare
A producer process could stop making progress for as long as it ran. Publishing goes through Connection::run(), which awaits on the event loop and suspends it again the moment the operation settles - so between two publishes nothing drives the loop, no heartbeat frame can be written, and a broker with heartbeat=60 hangs up on a command that spends minutes on work of its own. A firewall silently blackholing an idle socket looks the same from here. The next operation then awaited a reply that could never arrive. Bunny settles every protocol wait from an incoming frame and rejects none of them when the socket dies (Connection::$awaitList), and it reports connection-level failures as an 'error' event that nobody listened to, so the awaited promise simply stayed pending. A sync command was observed stuck in hrtimer_nanosleep for 20 hours. Even a socket the broker had closed cleanly only got as far as a fatal AssertionError from React's scheduler, one operation later. Connection::run() now races the operation against the client's 'error' event and against a timer of operation_timeout seconds, and throws OperationFailed rather than waiting forever. The client is discarded on either, because a timed-out operation is still parked inside the loop and an error means the connection is gone. ConsumerRunner is the one caller that must not be bounded - its loop runs until the consumer's own message or time limit - so it takes the new runWithoutTimeout(). Not everything that comes out of that await is a failure of the connection, though. An exception from the caller's own closure leaves by the same door, and Bunny re-emits every channel error onto the client (Client::channel()), so a client 'error' is no proof either. Discarding on those cost a reconnect and whatever was still unflushed elsewhere on the connection - measured with rabbitmqctl list_connections, a 404 publish took the connection from 2 to 0. So the client is now replaced on a timeout, where the operation is parked inside the loop for good, or when it can no longer be disconnected, which is the state Bunny leaves it in when it tears the client down itself for a connection that is gone. A channel the broker closed is replaced on its own 'close' event, as before. Only the wait for messages is exempt there. Getting to it opens a channel and sends basic.qos and basic.consume, each of which waits for a reply frame and so would hang for good on a socket that dies mid-handshake; the runner's own watch for channel failures is no help however early it is installed, since a Fiber stuck in the handshake never reaches the await() it rejects. That startup therefore runs inside a bounded run() of its own. A prefetch the broker refuses still arrives as ConfigurationFailed: Bunny's await list rejects the pending basic.qos-ok before that frame reaches the channel and the client, so the operation's own catch gets there first, and an integration test pins that order. SetupAction goes the other way and takes a run() per declaration rather than one around the whole topology. operation_timeout is what a single round trip may take, and a topology of hundreds of items would otherwise run out of it while every one of them was answered promptly. It also puts the timeout where it can be reported: run() raises it from outside the operation's Fiber, which stays suspended in the frame it is waiting for, so the per-item catch never saw it and setup() leaked OperationFailed instead of naming the exchange or queue that hung. GetOperation is split the same way, for the first of those reasons: a read asks for as many basic.get round trips as it wants messages, and one budget shared between them ends a read that was answered promptly throughout - 5000 gets, each about a tenth of a millisecond, ran out of a 0.5s bound - taking the connection with it. The timer and listener a run() per get adds cost 0.52s against 0.83s for 2000 gets locally, which is noise beside any real network round trip. Since the loop is frozen between operations, a connection idle for longer than the heartbeat it promised cannot have sent one and is assumed closed: run() replaces it up front instead of publishing into a hole. Whether a call is nested - and so already driving the loop, in which case the connection stays - is counted rather than read off Fiber::getCurrent(): that stands for "inside an operation of ours" only until the application brings Fibers of its own, as anything built on React does, and such a caller would have had the check skipped for the rest of the process. The teardown behind all of this closes the client locally (RAW_CONNECTION_INACTIVE) rather than exchanging connection.close with a broker that may be gone - which is also the only path that reaches Bunny's Connection::disconnect(), the one place that cancels the heartbeat timer that would otherwise keep the process alive with no stream left to wake it. It is the fallback on every failure path, and what the console listener uses before stopping the loop; kernel.terminate must not stop it, as a php-fpm worker serves further requests in the same process. That stop is not the belt and braces it looks like. One socket the teardown cannot reach is the one Bunny abandons: Client::connect() neither closes the connection nor rolls the state back when the handshake throws, so the client stays Connecting, refuses to be disconnected, and its socket is left registered on the event loop. A process holding one cannot exit, since React's shutdown blocks in stream_select() with nothing to wake it - a command that could not reach the broker ran until it was killed 25 seconds later, and exits at once with the loop stopped. Closing that socket belongs upstream, in that catch; stopping the loop is what a terminating command can do about it from here, at the price of dropping work a later console.terminate listener put on the loop without awaiting it. A subprocess test holds that line in place, since a process that hangs cannot be told from one that exits from the inside. disconnect() itself asks for the connection.close handshake first, and falls back to that local teardown. Not for the courtesy: a publish() sits in React's write buffer until the loop turns again, and closing the stream locally reaches React's close(), which discards the buffer rather than flushing it like end(). Every producer that published and then let kernel.terminate close the connection therefore lost its message and left "client unexpectedly closed TCP connection" in the broker log - the same for an acknowledge issued as the last thing a consumer does. The handshake awaits a reply, so the loop turns and the buffer goes out ahead of it. Bounded by run() and skipped for a stale connection, so a broker that has gone away still cannot hold a terminating process: with the broker paused, disconnect() returns after exactly operation_timeout and the process exits. A heartbeat of zero is refused outright. Bunny arms the heartbeat timer whatever the interval is and re-arms it with the same value, so 0 - the AMQP way of switching heartbeats off - leaves a timer that is due again the moment it fires: it spins the event loop at a full core for the whole of every operation and floods the broker with heartbeat frames (0.78s of CPU for a one-second await, against 0.00s at 60). The guard sits in Configuration\Connection, which is where the DSN parameter, the container key and a hand-built configuration all meet - including the blind cast that turns any non-numeric value into a zero. A run ends between messages, never in the middle of one. A handler that awaits anything - a batch publish committing, a get, any round trip - suspends its own Fiber and turns the loop, which is exactly where the maxSeconds timer falls due: ending the run there had run() return, the consumer cancelled and the connection closed by kernel.terminate or console.terminate while that handler was still parked. It then resumed into a connection that was gone, so its acknowledge was lost and the message redelivered - and had it thrown, the rejection landed on a promise settled long ago, which react/promise drops, so the command exited successfully having half-processed a message. Measured: the runner returned at 0.34s, the handler acknowledged at 1.05s, the message went back on the queue, and the process then hung because that late acknowledge reconnected and armed a fresh heartbeat timer. The stop is therefore recorded while a message is being handled and the delivery callback settles it on its way out; only one is ever in flight, since Bunny queues deliveries and runs them with a concurrency of 1. A failure is not held back the same way - it comes from a channel that errored or closed, and a handler parked on a dead socket never resumes at all, so waiting for it would block the run for good. A consumer whose consume() blocks for longer than the heartbeat loses its connection, and this cannot fix that: the loop only turns while something awaits it, and a delivery callback is synchronous PHP. The broker hangs up after two missed intervals, and an acknowledge issued after that reports success into a dead socket - Channel::ack() awaits no reply - so the message is redelivered and the handler runs again. Reconnecting for it would be worse, since delivery tags belong to the channel that is gone. What is left is to say so: the fallback the runner reports it through no longer claims the broker closed the channel, and docs/Consuming.md spells out that such a handler needs a heartbeat above twice its worst case, or must stop blocking the loop. A batch reports its failure as OperationFailed however that failure arrives, so handleAll() wraps around run() rather than inside the operation. The failure does not always come back through the operation's Fiber: the await list resumes the commit with the broker's channel.close, so the catch does run, but the best-effort txRollback() then awaits a reply of its own - and that suspension hands the read loop the very frame that caused all this. It reaches the channel, whose 'error' the client re-emits, and run() loses the race while the operation is still parked in the rollback, leaving the OperationFailed after it in an orphaned promise. A batch published to a missing exchange leaked Bunny's ChannelException. Exceptions of the bundle's own are rethrown unwrapped there, so a connection that could not be opened and an operation the timeout ended keep saying which they were. Worth knowing: a single publish() is still not on the wire when handle() returns. The next operation flushes it, and so does the disconnect above, but a message published into a connection that goes away before either - a process that publishes and then spends minutes on work of its own - is lost. handleAll() is transactional and does force the flush. BREAKING CHANGE: * RabbitMQ\Connection gained runWithoutTimeout(Closure): mixed, so custom implementations need updating. Only the consume loop should use it; everything else wants the bounded run(). * Operations that used to wait indefinitely now throw Exception\OperationFailed after operation_timeout seconds (default 30, and configurable per DSN parameter or YAML key). Any value of 0 or below restores the old behaviour. Callers that report failures in their own terms keep doing so with it as the cause: Exception\ConnectionFailed for connecting, Exception\ConfigurationFailed for topology setup. * heartbeat has to be positive now. Zero, which in AMQP switches heartbeats off, throws Exception\ConfigurationFailed - Bunny 0.6 cannot turn them off, and spins the event loop instead. Configure a long interval for as few heartbeats as possible.
| } | ||
|
|
||
| return $channel; | ||
| $this->lastOperationAt = microtime(true); |
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.
Bunny 0.6 is a rewrite: it dropped the sync/async duality and runs on a
ReactPHP event loop using PHP Fibers, so every Client/Channel call has to
execute inside a Fiber, and failures are reported by throwing instead of by
returning *OkFrame protocol objects.
All broker I/O is funnelled through the new Connection::run(Closure), which
wraps the operation in React\Async\await(async(...)), and does so even when a
Fiber is already current - running the operation on the calling Fiber instead
leaves PHP unable to switch out of contexts that forbid it, such as a signal
handler. Callers keep a plain synchronous API and must never drive the loop
themselves. The *OkFrame return checks in SetupAction and ConsumerRunner become
try/catch blocks that pass the cause through as $previous, and the instanceof
PromiseInterface guards are gone now that there is a single return type.
ConsumerRunner no longer polls Client::run() in a while loop; it awaits a
Deferred settled by the message-count limit or a Loop timer. Bunny reports
asynchronous failures as 'error'/'close' events, which Evenement silently drops
when nobody listens, so the runner listens on both and wraps Consumer::consume()
in a try/catch - without that, a throwing consumer, a broker-closed channel or a
lost connection left the consume loop blocked forever with nothing but an
unhandled promise rejection on stderr. Both events are needed: a broker that
closes the channel emits 'close' and then 'error' with the reply code and text,
while a connection that goes away emits only 'close'. Since the first rejection
is the one that counts, the 'close' fallback is held back by a tick so that the
error saying why gets there first. Every settle goes through
Loop::futureTick(), because resuming the awaiting Fiber from inside a delivery
callback's own Fiber never lets React's scheduler hand the result back to
run()'s caller.
While connected, Bunny keeps a heartbeat timer on the loop that would keep a
php-fpm worker or console process alive after the work is done. The new
DisconnectConnection subscriber closes the connection on kernel.terminate and
console.terminate, covering producers in both HTTP and CLI contexts.
BunnyConnection also drops a cached channel on its 'close' event, since 0.6
throws ChannelException('Channel is closed') for every later call on a channel
the broker closed and a single 404 publish would otherwise poison the connection
for the rest of the process, and it replaces a client whose connect() failed,
because Bunny leaves such a client reporting itself as connected while refusing
to connect again.
One upstream limitation remains: bunny 0.6.0-alpha.4 cannot recover in-process
from a first connection attempt that failed, even with a brand-new Client. A
connection that was established and then lost reconnects on the next operation.
Breaking changes
bunny/bunny0.6, currently available only as an alpha pre-release, so consumingprojects have to allow that stability.
read_write_timeoutoption is gone, as Bunny 0.6 has no read/write timeout. The DSN queryparameter is merely ignored, but the YAML key now fails the container build and has to be removed.
RabbitMQ\Connectiongained two required methods, so a custom implementation does not load untilit has both:
run(Closure): mixed- templated over what the closure returns, so callers keep thetype they passed in - and
runWithoutTimeout(Closure): mixed, which only the consume loop shoulduse; everything else wants the bounded
run().RabbitMQ\Connection::getChannel()andgetTransactionalChannel()now returnBunny\ChannelInterfacerather thanBunny\Channel.Exception\OperationFailedafteroperation_timeoutseconds (default 30, and configurable per DSN parameter or YAML key). Set itto
0for the old behaviour. Callers that report failures in their own terms keep doing so withit as the cause:
Exception\ConnectionFailedfor connecting,Exception\ConfigurationFailedfortopology setup.
heartbeathas to be positive now. Zero, which in AMQP switches heartbeats off, throwsException\ConfigurationFailed- Bunny 0.6 cannot turn them off, and spins the event loopinstead. Configure a long interval for as few heartbeats as possible.