How to debug / get Unacked messages ?

[geminigeek]

hi,

i have a queue consumer with Prefetch 7 , after running for few hrs, now i have 6 Unacked messages , this is causing the queue to only process 1 job at a time ( the concurrency is now Prefetch - Unacked ) , the unacked messages are stuck at 6 , how to debug this ?
i am running only one consumer for the queue !

if i restart my consumer the unacked count gets cleared, the unacked get back into queue (using x-delivery-limit: 2) and things start to work again normal with concurrency 7 as set.

running 1.0.0-beta.8 with @cloudamqp/amqp-client": "^2.1.0

[baelter]

Hi,

Unacked messages are held by the consumer. So that's where you will have to debug, in your client. Does your consumer used shared resources and/or locks?, do you have a consumer thread work pool?

Can you share code to reproduce the issue?

[geminigeek]

hi,

i am pasting a simple function quite similar to what i am doing ( not actual code) , i am using nodejs and running the consumer with PM2 , pm2 error logs are empty which suggest all errors are caught properly. what i am looking for is a timeout for a consumer or like an abort controller or a function to check time a msg/job is in "Unacked" state

async function consumeFunction(qMsg) {
  try {
    const { id, domain, link, opts } = JSON.parse(qMsg.bodyToString())

    if (!domain) {
      await qMsg.nack()
      return
    }

    const { data, error, errorMessage, statusCode } = await actualJob(
      link,
      opts ?? {},
    )

    if (!error) {
      await processData(data)
      await qMsg.ack() // done processed
      return
    } else {
      // updates for error logs in db
      const updateOptsForError = {
        domain,
        errorMessage: errorMessage ?? "",
        statusCode: statusCode ?? 0,
        link,
        record: id,
      }

      await ErrorLog.create(updateOptsForError)

      if (statusCode && statusCode !== 200) {
        await qMsg.nack(true)
      } else {
        await qMsg.nack()
      }
    }
  } catch (error) {
    console.log("[consume function error] :>> ", error.message)
    // If this fails for a network related reason the msg can stay stuck ?
    await qMsg.nack(true)
  }
}

[geminigeek]

hi,

i am trying to debug this , my new finding is, this happens at resume and start of queue mostly, it seems there must be a way to read queue is now resumed , so consumer can debounce / throttle the first n number of messages to stabilize the actual service the consumers are hitting , in my case my consumer interact over websocket, on start of the consumer i throttle first few messages, but same need to be done on resume , it seems too fast unack and redelivery and again unack for cases like "socket hang up" of websocket might be causing this.