RabbitMQ Coordinator Dynamic Queue Binding#36
Merged
Conversation
First pass at shared connections and using channels to avoid unnecessary connections and re-use existing connections. Should be more efficient and kinder to the RabbitMQ infra. But this doesn't quite work yet. Or, rather, it does... except channels don't properly/correctly close... yet.
We weren't properly cleaning up and closing channels on the muxed connections, which had some pretty negative impact. This makes sure they get cleaned up properly, and just generally makes the RabbitMQ disconnect more robust.
So when the RabbitMQ connection drops and reconnects... all channels that had ever existed on the connection were getting re-connected when the server reconnected, even if they were closed forever ago. This stops that. ... this also has the side-effect of not reconnecting channels that SHOULD have been reconnected, and doesn't reconnect properly until/unless the worker/client reconnects, triggering a new channel to be created. This is next to be fixed here...
We can avoid creating and listening on queues if we just bind to the exchanges and manage the bindings. This seems to actually be a LOT more resilliant and functional than trying to leverage channels.
Also removes the config to enable/disable muxing (as the need for that option doesn't really apply anymore)
This doesn't really do anything, we should be able to create the proxy client and it'll work when the main connection boots back up (and create the subscription then), AND this was really only happening when the dev server reloaded and hit raceconditons that don't normally happen, which results in the old (durable) queues getting created. So not only was it not necessary (the proxy client version handles a late connection), but it was actively annoying (re-creating durable queues).
... and the actual config option we added previously.
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
1 participant
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.
Description
This PR implements connection reuse/multiplexing for RabbitMQ-backed HTTP Polling/Streaming on the Coordinator, fixing the issue where every transient HTTP poll request established a completely new TCP/TLS connection to the RabbitMQ broker.
Instead of standard socket pooling, this PR resolves the connection churn by introducing Dynamic Queue Binding on the Coordinator's single, persistent robust server connection:
RabbitMQServerTransportdeclares exactly one persistent coordinator-wide queue on startup (dffmpeg.coordinator.{uuid}).bind) its single persistent queue to the workers/jobs exchanges using the client's routing key (e.g.worker.{worker_id}). On HTTP disconnect, the coordinator instantly unbinds (unbind) the routing key to prevent blackholing of messages.RabbitMQProxyClientTransport): Refactored the multiplexed client transport into a 100% in-memory, lightweight transport local to the coordinator. It has zero TCP connection handles, zero AMQP channels, and mitigates handle leaks.aio_pika'sRobustQueueautomatically restores all active client bindings natively!_connection_taskautomatically restores and binds all registered clients.Fixes #35 (issue)
Type of change
Breaking Changes
None. This is completely backward-compatible. A migration note has been added to
docs/transports.mdadvising administrators to safely delete legacy durable worker-specific queues from the RabbitMQ broker once.Checklist:
Code Quality
Testing
Architecture & Philosophy