Ensure only canonical are added to RabbitmqBroker.queues #3
Conversation
When a worker is started, it'll eventually start two consumers: one for reading off the queue and another for reading off the delayed queue. When it starts consuming off the delayed queue, it calls
`RabbitmqBroker#consume('queue_name.DQ')`. In tern, 'queue_name.DQ' goes through the declare queue initialization which results in its name being added to the `queues` set. This becomes an issue as later
on in commands such as flush, assumptions are made that only the canonical queue name is present.
|
Hmm. No I don't think this is a good change. It's expected that all queues that might be consumed from are predefined. Are you seeing bugs in dramatiq as a result of this, or are you seeing bugs in our middleware as a result? If bugs in our middleware, the "dramatiq way" seems to be to call |
|
I'm realizing that the previous implementation allows for lazy queue creation 🤔 I'll look at adding some tests to validate that's the case? The problem I'm running into was around the changes in #1. It seems like a lot of the code assumes One option would be to modify the deletion to be more defensive? In theory, it's possible that the queues could be deleted on the RMQ hosts manually or out of band, so we can't assume they're always there. The larger change that I kept avoiding in this PR was refactoring Queue out into a class of its own and adding |
|
Definitely Dramatiq is a bit loosey-goosey with some concepts here, and I see the overlap now in the flush function. To cut the Gordian's knot: What about moving the queue deletion to the test infrastructure, or just using different queue names for testing quorum vs not? |
When a worker is started, it'll eventually start two consumers: one for reading off the queue and another for reading off the delayed queue. When it starts consuming off the delayed queue, it calls
RabbitmqBroker#consume('queue_name.DQ'). In turn,queue_name.DQgoes through the declare queue initialization which results in its name being added to theRabbitmqBroker.queuesset. This becomes an issue as later on in commands such asRabbitmqBroker#flush, where assumptions are made that only the canonical queue name is present.