changing queueOptions will crash next restart

[christopheblin]

Let's say you have a subscriber like this

@RabbitSubscribe({
  exchange: 'dummy',
  queue: 'my_queue' 
})

You push the app to prod, everything is fine (exchange created if it does not exist, queue created if it does not exist, binding betwen queue and excahnge)

now, you change an option like deadLetterExchange

@RabbitSubscribe({
  exchange: 'dummy',
  queue: 'my_queue',
  queueOptions: {
    deadLetterExchange: 'dead-letter-retry'
  }
})

The app will not start :

Operation failed: QueueDeclare; 406 (PRECONDITION-FAILED) with message "PRECONDITION_FAILED - inequivalent arg 'x-dead-letter-exchange' for queue 'my_queue' in vhost '/': received the value 'dead-letter-retry' of type 'longstr' but current is none"

I know this is the expected behavior from RabbitMq perspective (i.e multiple consumers may be bound to the queue so it is a bit unsafe to delete+recreate the queue automatically during assertQueue)

However, do you think this is the expected behavior with @RabbitSubscribe ?

I mean, if I explicitly specify the queue, it's because I want the same queue for all running replicas ... (besides, options do not really influence the running behavior of the queue from consumer point of view)

Will you accept a PR where we catch the failure of assertQueue to call deleteQueue + assertQueue ?

In any case, what do you suggest as a workaround to detect the situation automatically and to automatically delete the related queues before this lib calls assertQueue ?

[WonderPanda]

These types of situations are tricky for a general purpose library because what might be appropriate for your scenario (in terms of delete + recreate) could be terrible for someone else.

I think that the best option (although it increases API surface) is to have an additional option inside RabbitSubcribe that allows you to specify custom behavior/error handling for when asserting the queue fails.

I'd definitely be open to accepting a PR that gives users more flexibility in this area.

[christopheblin]

WDYT about this :

@RabbitSubscribe({
  exchange: 'dummy',
  queue: 'my_queue',
  queueOptions: {
    deadLetterExchange: 'dead-letter-retry'
  },
  assertQueueErrorHandler: (err, channel, queueName, queueOptions) => {
    if (/*err is precondition failed*/) {
       await channel.deleteQueue(queue)
       const { queue } = await channel.assertQueue(queue, queueOptions);
        return queue
    }
    //the problem is not with a precondition failed, just rethrow
    throw err
  },
})

maybe we can create a forceDeleteAssertQueueErrorHandler

     const {
      exchange,
      routingKey,
      createQueueIfNotExists = true,
      assertQueueErrorHandler = defaultAssertQueueErrorHandler, //will just rethrow the error
    } = subscriptionOptions;

   let actualQueue: string;

    if (createQueueIfNotExists) {
      const queueName = subscriptionOptions.queue || ''
      const queueOptions = subscriptionOptions.queueOptions || undefined
      try {
        const { queue } = await channel.assertQueue(queueName, queueOptions)
        actualQueue = queue;
      } catch (err: unknown) {
         actualQueue = await assertQueueErrorHandler(err, channel, queueName, queueOptions)         
      }
   }

[underfisk]

@christopheblin Can you try the new release version, there was a PR #349 that introduced this feature so let me know if that addresses your concerns

[christopheblin]

@underfisk I am the author of the PR so yes it address my concerns :)

I dont have much time to upgrade the libs in our apps right now, but next app will definitively use the new assertQueueErrorHandler option

I'll close this issue, feel free to reopen for any concerns you may have

[muhaliusman]

Hi @christopheblin, I still get an error even though I've replaced the error handler with forceDeleteAssertQueueErrorHandler. I think the handler is running but the error still occurs.

@Injectable()
export class ReceiptHandler {
  @RabbitSubscribe({
    exchange: 'events',
    routingKey: 'order.received',
    queue: 'sample_queue',
    queueOptions: {
        durable: false,
        deadLetterExchange: 'sample_dlx'
    },
    assertQueueErrorHandler: forceDeleteAssertQueueErrorHandler
  })
  public async pubSubHandler(msg: {}) {
    console.log(`Received message: ${JSON.stringify(msg)}`);
  }
}