Queue Concurrency Not Working

[devhammed]

Hello there,

Thanks again for this package but I configured it like this:

import env from '#start/env'
import app from '@adonisjs/core/services/app'
import { exponentialBackoff } from '@boringnode/queue'
import { redis } from '@boringnode/queue/drivers/redis_adapter'
import { sync } from '@boringnode/queue/drivers/sync_adapter'
import { type QueueManagerConfig } from '@boringnode/queue/types'

const queueConfig: QueueManagerConfig = {
  default: env.get('QUEUE_CONNECTION'),

  adapters: {
    sync: sync(),
    redis: redis({
      host: env.get('REDIS_HOST'),
      port: env.get('REDIS_PORT'),
      password: env.get('REDIS_PASSWORD', ''),
      db: env.get('REDIS_DB', 0),
      keyPrefix: env.get('REDIS_KEY_PREFIX', 'flowshub:'),
      retryStrategy: (times) => (times > 10 ? null : times * 50),
    }),
  },

  retry: {
    maxRetries: 3,
    backoff: exponentialBackoff({
      baseDelay: '1s',
      maxDelay: '1m',
      multiplier: 2,
      jitter: true,
    }),
  },

  worker: {
    timeout: env.get('QUEUE_TIMEOUT', '24h'),
    concurrency: env.get('QUEUE_CONCURRENCY', 20),
    idleDelay: env.get('QUEUE_IDLE_DELAY', '2s'),
  },

  defaultJobOptions: {
    removeOnComplete: false,
    removeOnFail: false,
  },

  locations: ['./app/jobs/**/*.{js,ts}'],

  jobFactory: async (JobClass) => {
    return app.container.make(JobClass)
  },
}

export default queueConfig

But jobs are still waiting for the current one to complete before starting, even though I have a concurrency limit of 20:

[RomainLanz]

Hey @devhammed! 👋🏻

Can you share your job's code?

[devhammed]

@RomainLanz I can't really do that, but even with a minimal example that just logs Hello World after some delays, they don't run at the same time.

[RomainLanz]

I cannot reproduce your issue locally.

Are you sure you weren't using the sync adapter at that time?

[devhammed]

@RomainLanz I always use Redis. I cannot send the actual job because it has some database operations which cannot be easily reproduced but this example should give you an idea:

Dispatching sample:

import HelloWorldJob from '#jobs/hello_world_job'
import type { HttpContext } from '@adonisjs/core/http'

export default class HomeController {
  async index({ inertia }: HttpContext) {
    await HelloWorldJob.dispatch({ index: 0 })

    await new Promise((resolve) => setTimeout(resolve, 1000))

    await HelloWorldJob.dispatch({ index: 1 })

    await new Promise((resolve) => setTimeout(resolve, 1000))

    await HelloWorldJob.dispatch({ index: 2 })

    await new Promise((resolve) => setTimeout(resolve, 1000))

    await HelloWorldJob.dispatch({ index: 3 })

    return inertia.render('marketing/home')
  }
}

The job:

import logger from '@adonisjs/core/services/logger'
import { Job } from '@boringnode/queue'
import { DateTime } from 'luxon'

export type HelloWorldPayload = {
  index: number
}

export default class HelloWorldJob extends Job<HelloWorldPayload> {
  async execute() {
    console.log(`HelloWorld Processing ${this.payload.index} @ ${DateTime.now()}`)

    await new Promise((resolve) => setTimeout(resolve, 10_000))

    console.log(`HelloWorld Processed ${this.payload.index} @ ${DateTime.now()}`)
  }

  async failed(error: Error) {
    logger.error(error, 'HelloWorld failed')
  }
}

The output:

You can see that even with a concurrency of 5, index 1 had to wait for index 0 to finish executing before processing, and then somehow 2 and 3 were able to perform at the same time.

[thetutlage]

Did you start the worker after pushing all the jobs to the queue? Or the worker was always running as the jobs were getting enqueued?

[RomainLanz]

Can you try 0.3.4?

https://github.com/boringnode/queue/releases/tag/v0.3.4

[devhammed]

Did you start the worker after pushing all the jobs to the queue? Or the worker was always running as the jobs were getting enqueued?

The worker is always running, I wrote a command for it:

import { BaseCommand, flags } from '@adonisjs/core/ace'
import string from '@adonisjs/core/helpers/string'
import type { CommandOptions } from '@adonisjs/core/types/ace'
import { Worker } from '@boringnode/queue'
import type { QueueManagerConfig } from '@boringnode/queue/types'
import { setTimeout } from 'node:timers/promises'

export default class QueueWork extends BaseCommand {
  static commandName = 'queue:work'

  static description = 'Start processing jobs from the specified queues.'

  static options: CommandOptions = {
    startApp: true,
    staysAlive: true,
  }

  @flags.array({
    description: "Queue names to process (default: ['default']).",
    required: false,
    default: ['default'],
  })
  declare queue: string[]

  @flags.number({
    description: 'Maximum number of jobs to process concurrently.',
    required: false,
  })
  declare concurrency?: number

  @flags.string({
    flagName: 'idle-delay',
    description: 'Delay between queue polls when idle (no jobs available) e.g. 30m, 1h, etc.',
    required: false,
  })
  declare idleDelay?: string

  @flags.string({
    description:
      'Maximum duration a job can run before being timed out. Can be overridden per job via JobOptions.timeout. e.g. 30m, 1h, etc.',
    required: false,
  })
  declare timeout?: string

  @flags.string({
    flagName: 'stalled-threshold',
    description:
      'Duration after which an active job is considered stalled. A stalled job is one that was acquired but the worker stopped responding (e.g., due to a crash).',
    required: false,
  })
  declare stalledThreshold?: string

  @flags.string({
    flagName: 'stalled-interval',
    description: 'How often to check for stalled jobs.',
    required: false,
  })
  declare stalledInterval?: string

  @flags.number({
    flagName: 'max-stalled-count',
    description:
      'Maximum number of times a job can be recovered from stalled state before being marked as failed permanently.',
    required: false,
  })
  declare maxStalledCount?: number

  #worker?: Worker

  prepare() {
    const terminate = async () => {
      await this.terminate()
    }

    const cleanup = async () => {
      this.logger.info('Queue worker shutting down...')

      await this.#worker?.stop()
    }

    this.app.listen('SIGINT', terminate)

    this.app.listen('SIGTERM', terminate)

    this.app.terminating(cleanup)
  }

  async run() {
    const queueConfig = this.app.config.get<QueueManagerConfig>('queue')

    this.#worker = new Worker({
      ...(queueConfig ?? {}),
      worker: {
        ...(queueConfig.worker ?? {}),
        concurrency: this.concurrency ?? queueConfig.worker?.concurrency,
        idleDelay: this.idleDelay ?? queueConfig.worker?.idleDelay,
        timeout: this.timeout ?? queueConfig.worker?.timeout,
        stalledThreshold: this.stalledThreshold ?? queueConfig.worker?.stalledThreshold,
        stalledInterval: this.stalledInterval ?? queueConfig.worker?.stalledInterval,
        maxStalledCount: this.maxStalledCount ?? queueConfig.worker?.maxStalledCount,
      },
    })

    this.logger.info('Queue worker started.')

    let cycle = await this.#getNextCycle()

    while (cycle) {
      switch (cycle.type) {
        case 'started':
          this.logger.info(`Started job ${cycle.job.id}`)
          break
        case 'completed':
          this.logger.success(`Completed job ${cycle.job.id}`)
          break
        case 'idle':
        case 'error':
          if (cycle.type === 'error') {
            this.logger.error(`Error processing job: ${cycle.error}`)
          }

          await setTimeout(string.milliseconds.parse(cycle.suggestedDelay))

          break
      }

      cycle = await this.#getNextCycle()
    }
  }

  async #getNextCycle() {
    try {
      return await this.#worker?.processCycle(this.queue)
    } catch {
      return null
    }
  }
}

[RomainLanz]

I am not sure what you are trying to do by processing the cycle by hand, that may explain some issues.

Have you tried with the latest version? (0.3.4)

[devhammed]

I am not sure what you are trying to do by processing the cycle by hand, that may explain some issues.

Have you tried with the latest version? (0.3.4)

The queue worker is not logging as expected, even when I pass a logger, and I want full transparency on what is going on which is why I went with the processCycle approach.

And yeah, I just tested now with 0.3.4, and it is working!

Thank you for the fix!

[devhammed]

Though you might want to check

queue/src/worker.ts

Line 297 in ef81a3a

yield { type: 'completed', queue: result.completed.queue, job: result.completed.job }

as even though the job completed without any error, there is a false alarm here because maybe you accessed the queue on the wrong object?