groupConcurrency is exceeded by up to the number of concurrent fetchers #976

Summary

With groupConcurrency: 1 and localConcurrency: 4 on a single node, up to four jobs of the same group run at once. Every worker that fetches at the same moment can claim another job of the group, and those jobs stay active until their handlers return.

The docs warn about races:

there may be brief moments where the limit is slightly exceeded during race conditions when multiple workers fetch jobs simultaneously

The overrun below is neither brief nor slight. It grows with the number of workers that fetch together and lasts as long as the handlers run. The same page recommends groupConcurrency "when you need strict global limits across a distributed deployment". Today it does not give that guarantee.

Version: 12.38.0. The same code is on master (c8ce91d).

Reproduction

Two tests for test/concurrencyGroupTest.ts, written like the existing ones. With the Postgres from docker-compose.yaml, both fail on master, and the other ten tests in the file pass:

docker compose up -d
npx vitest run test/concurrencyGroupTest.ts

FAIL  test/concurrencyGroupTest.ts > groupConcurrency > should limit active jobs per group when several workers fetch at once
AssertionError: expected 2 to be less than or equal to 1
FAIL  test/concurrencyGroupTest.ts > groupConcurrency > should not claim another job of a group while a claim is uncommitted
AssertionError: expected { …(8) } to be undefined

The first test follows "should limit active jobs per group with simple groupConcurrency", with groupConcurrency: 1 and localConcurrency: 4. The second holds the first claim open in a transaction, which makes the race deterministic.

import pg from 'pg'

// inside describe('groupConcurrency')
  it('should limit active jobs per group when several workers fetch at once', async function () {
    ctx.boss = await helper.start({ ...ctx.bossConfig })

    const groupId = 'tenant-123'
    const groupConcurrency = 1
    const localConcurrency = 4
    const jobCount = 8

    let currentActive = 0
    let maxActive = 0

    for (let i = 0; i < jobCount; i++) {
      await ctx.boss.send(ctx.schema, { index: i }, { group: { id: groupId } })
    }

    await ctx.boss.work(ctx.schema, { localConcurrency, groupConcurrency, pollingIntervalSeconds: 0.5 }, async () => {
      currentActive++
      maxActive = Math.max(maxActive, currentActive)
      await delay(500)
      currentActive--
    })

    await delay(5000)

    expect(maxActive).toBeLessThanOrEqual(groupConcurrency)
  })

  it('should not claim another job of a group while a claim is uncommitted', async function () {
    ctx.boss = await helper.start({ ...ctx.bossConfig })

    const groupId = 'tenant-123'

    await ctx.boss.send(ctx.schema, { index: 1 }, { group: { id: groupId } })
    await ctx.boss.send(ctx.schema, { index: 2 }, { group: { id: groupId } })

    const client = new pg.Client({ connectionString: helper.getConnectionString() })
    await client.connect()

    try {
      await client.query('BEGIN')
      const db = { executeSql: (text: string, values?: unknown[]) => client.query(text, values) }

      const [first] = await ctx.boss.fetch(ctx.schema, { groupConcurrency: 1, db })
      assertTruthy(first)

      const [second] = await ctx.boss.fetch(ctx.schema, { groupConcurrency: 1 })
      await client.query('COMMIT')

      expect(second).toBeUndefined()
    } finally {
      await client.end()
    }
  })

Measurements

A standalone script against the published 12.38.0 package, with PostgreSQL 18.4 and Node.js 24.22.1. Five runs per setup. The last column is the highest number of active jobs in the group, sampled from pgboss.job every 20 ms. The handlers counted the same numbers.

// Usage: node repro.mjs [instances=1] [localConcurrency=4] [groupConcurrency=1]
// Sends 12 jobs in one group and works them on every instance.
import { PgBoss } from 'pg-boss'

const connectionString = process.env.DATABASE_URL ?? 'postgres://postgres:postgres@localhost:5432/postgres'
const instances = Number(process.argv[2] ?? 1)
const localConcurrency = Number(process.argv[3] ?? 4)
const groupConcurrency = Number(process.argv[4] ?? 1)
const queue = 'group-repro'
const groupId = 'group-a'
const jobCount = 12

const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms))

const bosses = []
for (let i = 0; i < instances; i++) {
  const boss = new PgBoss({ connectionString })
  boss.on('error', console.error)
  await boss.start()
  bosses.push(boss)
}

const [first] = bosses
await first.deleteQueue(queue)
await first.createQueue(queue)
for (let i = 0; i < jobCount; i++) {
  await first.send(queue, { i }, { group: { id: groupId } })
}

let running = 0
let maxRunning = 0
let completed = 0

for (const boss of bosses) {
  await boss.work(queue, { localConcurrency, groupConcurrency, pollingIntervalSeconds: 0.5 }, async () => {
    running++
    maxRunning = Math.max(maxRunning, running)
    await delay(500)
    running--
    completed++
  })
}

// Samples the number of active jobs in the group while the workers run.
let maxActive = 0
while (completed < jobCount) {
  const { rows } = await first.getDb().executeSql(
    "SELECT count(*)::int AS n FROM pgboss.job WHERE name = $1 AND group_id = $2 AND state = 'active'",
    [queue, groupId]
  )
  maxActive = Math.max(maxActive, rows[0].n)
  await delay(20)
}

console.log(`instances=${instances} localConcurrency=${localConcurrency} groupConcurrency=${groupConcurrency}: max active jobs in group = ${maxActive} (database), ${maxRunning} (handlers)`)

await Promise.all(bosses.map((boss) => boss.stop()))

Cause

For groupConcurrency: 1, fetchNextJob uses a NOT EXISTS probe to drop every candidate whose group has an active job, then claims the rest with FOR UPDATE OF j SKIP LOCKED. Under READ COMMITTED the probe reads the snapshot from the start of the fetch statement. A fetch that starts before another fetch commits sees no active job in the group. It skips the row the other fetch locked and claims the next job of the same group. Limits above 1 read active_group_count_map from the same snapshot and have the same gap, as the last row of the table shows.

The window is the runtime of one fetch statement. Workers that start together and poll on the same interval hit it, as the first test and the table show.

Possible fixes

Both ideas are untested.

• When groupConcurrency is set, take a transaction-level advisory lock for the queue in a statement before the fetch. The fetch then starts after the previous claim committed and sees it. This serializes fetches on that queue.
• After the claim, count the group again in a new statement and restore the excess, the way localGroupConcurrency already restores jobs past its limit.

If a strict limit is out of scope, the docs could say that each fetch checks the limit on its own, so N workers fetching together can claim up to N times a group's free slots.