How to check queue depth and consumer lag

Sample the queue every second instead of reading it once. node sample.mjs 28 prints depth, the age of the oldest pending job, arrivals and completions. In the run below, depth 30 at second 7 and depth 31 at second 20 are the same level with opposite futures. The slope separates them.

Why check this

A queue absorbs the difference between how fast work arrives and how fast it is done. Nothing looks broken while it absorbs, so the first symptom is an hour of backlog.

Run it after a change to the worker, after a release that adds a producer, and when a ticket says a job arrived "eventually". The failure it prevents: a consumer that keeps up on staging falls 4.27 jobs a second behind under real arrivals, reaching a 10 second wait in two minutes.

Depth is a level, the jobs waiting now. Lag is a rate, how fast that level moves. A depth of 500 draining needs nothing; a depth of 50 growing by 4 a second is an outage in twenty minutes. Neither is the per-process figure in How to measure event loop lag.

Prerequisites

• Node 22. The queue is a table in node:sqlite, experimental and warning on import. Step 1 shows it; later commands set NODE_NO_WARNINGS=1.
• The four programs below, in one directory. Nothing is specific to SQLite: depth counts pending rows and lag is the age of the oldest.
• The schema. Save as queue-init.mjs.

// Creates the queue table. One row per job. state: pending | running | done.
import { DatabaseSync } from 'node:sqlite';
import { rmSync } from 'node:fs';

for (const f of ['queue.db', 'queue.db-wal', 'queue.db-shm']) rmSync(f, { force: true });
const db = new DatabaseSync('queue.db');
db.exec(`
  PRAGMA journal_mode = WAL;
  CREATE TABLE jobs (
    id          INTEGER PRIMARY KEY,
    enqueued_at INTEGER NOT NULL,
    claimed_at  INTEGER,
    done_at     INTEGER,
    state       TEXT NOT NULL DEFAULT 'pending'
  );
  CREATE INDEX jobs_pending ON jobs (state, id);
`);
console.log(db.prepare('SELECT sql FROM sqlite_master WHERE name = ?').get('jobs').sql);
db.close();

• The producer. It reports what it sent, not what it asked for. Save as produce.mjs.

// produce.mjs <rate/s> <seconds>. Enqueues on an interval and reports what it
// actually sent, because a setInterval does not deliver its nominal rate.
import { DatabaseSync } from 'node:sqlite';

const rate = Number(process.argv[2]);
const secs = Number(process.argv[3]);
const db = new DatabaseSync('queue.db');
db.exec('PRAGMA busy_timeout = 5000');
const insert = db.prepare('INSERT INTO jobs (enqueued_at) VALUES (?)');

let sent = 0;
const t0 = Date.now();
const timer = setInterval(() => {
  insert.run(Date.now());
  sent += 1;
}, 1000 / rate);

setTimeout(() => {
  clearInterval(timer);
  const el = (Date.now() - t0) / 1000;
  console.log(
    `producer requested ${rate}/s for ${secs}s, sent ${sent} in ${el.toFixed(2)}s, ` +
      `measured ${(sent / el).toFixed(2)}/s`,
  );
  db.close();
}, secs * 1000);

• The consumer. It separates wall clock from busy time, so capacity is measured on jobs run, not on idle seconds. Save as worker.mjs.

// worker.mjs <service_ms> <seconds> <oldest|newest> <tag>
// Claims one job, holds it for service_ms, marks it done. Reports its own rate.
import { DatabaseSync } from 'node:sqlite';

const svc = Number(process.argv[2]);
const secs = Number(process.argv[3]);
const dir = process.argv[4] === 'newest' ? 'DESC' : 'ASC';
const tag = process.argv[5] ?? 'w1';

const db = new DatabaseSync('queue.db');
db.exec('PRAGMA busy_timeout = 5000');
const claim = db.prepare(
  `UPDATE jobs SET state = 'running', claimed_at = ?
     WHERE id = (SELECT id FROM jobs WHERE state = 'pending' ORDER BY id ${dir} LIMIT 1)
   RETURNING id`,
);
const finish = db.prepare("UPDATE jobs SET state = 'done', done_at = ? WHERE id = ?");
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

const t0 = Date.now();
let done = 0;
let busy = 0;
while (Date.now() - t0 < secs * 1000) {
  const row = claim.get(Date.now());
  if (!row) { await sleep(20); continue; }
  const c0 = Date.now();
  await sleep(svc);
  finish.run(Date.now(), row.id);
  busy += Date.now() - c0;
  done += 1;
}
const el = (Date.now() - t0) / 1000;
console.log(
  `${tag} claims ${dir === 'ASC' ? 'oldest' : 'newest'} first, service ${svc} ms nominal, ` +
    `finished ${done} in ${el.toFixed(2)}s, busy ${(busy / 1000).toFixed(2)}s, ` +
    `${(busy / Math.max(done, 1)).toFixed(1)} ms per job, capacity ${(done / (busy / 1000)).toFixed(2)}/s`,
);
db.close();

• The sampler. Save as sample.mjs.

// sample.mjs <seconds>. One line per second: depth, age of the oldest pending
// job, total arrived, total done. Depth alone is a level; the column beside it
// is what turns it into a rate.
import { DatabaseSync } from 'node:sqlite';

const secs = Number(process.argv[2]);
const db = new DatabaseSync('queue.db');
db.exec('PRAGMA busy_timeout = 5000');
const q = db.prepare(`
  SELECT (SELECT COUNT(*) FROM jobs WHERE state = 'pending')    AS depth,
         (SELECT MIN(enqueued_at) FROM jobs WHERE state = 'pending') AS oldest,
         (SELECT COUNT(*) FROM jobs)                            AS arrived,
         (SELECT COUNT(*) FROM jobs WHERE state = 'done')       AS done
`);

const pad = (v, n) => String(v).padStart(n);
console.log('  t  depth  oldest_ms  arrived  done');
let t = 0;
const timer = setInterval(() => {
  t += 1;
  const r = q.get();
  const age = r.oldest === null ? 0 : Date.now() - r.oldest;
  console.log(`${pad(t, 3)}${pad(r.depth, 7)}${pad(age, 11)}${pad(r.arrived, 9)}${pad(r.done, 6)}`);
  if (t >= secs) { clearInterval(timer); db.close(); }
}, 1000);

Steps

1. 1

   Step 1.

   Create the queue table.

   node queue-init.mjs
   

   CREATE TABLE jobs (
     id          INTEGER PRIMARY KEY,
     enqueued_at INTEGER NOT NULL,
     claimed_at  INTEGER,
     done_at     INTEGER,
     state       TEXT NOT NULL DEFAULT 'pending'
   )
   (node:42132) ExperimentalWarning: SQLite is an experimental feature and might change at any time
   (Use `node --trace-warnings ...` to show where the warning was created)

   Two columns carry the check: state gives depth, enqueued_at gives age. Run export NODE_NO_WARNINGS=1 before the rest.

2. 2

   Step 2.

   Measure the arrival rate with no consumer running.

   node produce.mjs 12 10
   

   producer requested 12/s for 10s, sent 108 in 10.01s, measured 10.79/s

   The interval was asked for 83.3 ms and fired at 92.7 ms. Every number below is built from measured, never requested.

3. 3

   Step 3.

   Sample the queue while nothing is consuming it.

   node sample.mjs 3
   

     t  depth  oldest_ms  arrived  done
   1    108      10997      108     0
   2    108      11997      108     0
   3    108      13012      108     0

   Depth is frozen at 108 while the oldest job ages by 1000 ms a second. Flat depth is not health: nothing has been claimed.

4. 4

   Step 4.

   Start one consumer, overload it for 15 seconds, and keep sampling 13 seconds past the producer.

   node queue-init.mjs > /dev/null
   node worker.mjs 150 30 oldest w1 > worker.log 2>&1 &
   node produce.mjs 12 15 > produce.log 2>&1 &
   node sample.mjs 28
   wait
   cat produce.log worker.log
   

     t  depth  oldest_ms  arrived  done
   1      4        371       10     5
   2      8        728       21    12
   3     13       1182       32    18
   4     17       1536       43    25
   5     22       1983       54    31
   6     26       2336       65    38
   7     30       2782       75    44
   8     34       3138       86    51
   9     39       3594       97    57
   10     43       3957      108    64
   11     48       4401      119    70
   12     51       4763      129    77
   13     56       5220      140    83
   14     60       5578      151    90
   15     64       6028      161    96
   16     57       6382      161   103
   17     51       6831      161   109
   18     44       7173      161   116
   19     38       7626      161   122
   20     31       7980      161   129
   21     25       8432      161   135
   22     18       8783      161   142
   23     12       9236      161   148
   24      5       9589      161   155
   25      0          0      161   161
   26      0          0      161   161
   27      0          0      161   161
   28      0          0      161   161
   producer requested 12/s for 15s, sent 161 in 15.01s, measured 10.73/s
   w1 claims oldest first, service 150 ms nominal, finished 161 in 30.02s, busy 24.91s, 154.7 ms per job, capacity 6.46/s

   Depth climbs from 4 to 64 over 14 seconds, a slope of 4.29 a second. Arrival 10.73 minus capacity 6.46 is 4.27, so the slope is that subtraction. With arrivals stopped it falls at 6.4, the service rate, and empties at second 25 against a predicted 9.9.

5. 5

   Step 5.

   Run a near balanced load, arrivals slightly above capacity, and read the oldest age.

   node queue-init.mjs > /dev/null
   node worker.mjs 150 27 oldest w1 > worker.log 2>&1 &
   node produce.mjs 8 25 > produce.log 2>&1 &
   node sample.mjs 25
   wait
   cat produce.log worker.log
   

     t  depth  oldest_ms  arrived  done
   1      1         46        7     5
   …
   12     13       1622       90    76
   …
   24     26       3433      180   153
   25     25       3509      186   160
   producer requested 8/s for 25s, sent 186 in 25.00s, measured 7.44/s
   w1 claims oldest first, service 150 ms nominal, finished 172 in 27.10s, busy 26.82s, 155.9 ms per job, capacity 6.41/s

   A deficit of 1.03 a second gives depth 25 after 25 seconds. The oldest job waited 3509 ms, near the depth over the capacity, 3.9 s.

6. 6

   Step 6.

   Repeat that run with one change: the consumer claims the newest job first.

   node queue-init.mjs > /dev/null
   node worker.mjs 150 27 newest w1 > worker.log 2>&1 &
   node produce.mjs 8 25 > produce.log 2>&1 &
   node sample.mjs 25
   wait
   cat produce.log worker.log
   

     t  depth  oldest_ms  arrived  done
   1      1         62        7     5
   …
   12     12      11033       89    76
   …
   24     23      23128      177   153
   25     23      24128      183   159
   producer requested 8/s for 25s, sent 183 in 25.01s, measured 7.32/s
   w1 claims newest first, service 150 ms nominal, finished 172 in 27.13s, busy 26.82s, 156.0 ms per job, capacity 6.41/s

   Same rates, same throughput, depth 23 against 25. The oldest job has waited 24128 ms against 3509 ms and ages by 1008 ms a second, because it is never the one claimed.

7. 7

   Step 7.

   Add the consumer the arithmetic asks for. 10.73 divided by 6.46 is 1.66, so two.

   node queue-init.mjs > /dev/null
   node worker.mjs 150 20 oldest w1 > w1.log 2>&1 &
   node worker.mjs 150 20 oldest w2 > w2.log 2>&1 &
   node produce.mjs 12 15 > produce.log 2>&1 &
   node sample.mjs 18
   wait
   cat produce.log w1.log w2.log
   

     t  depth  oldest_ms  arrived  done
   1      0          0       10     8
   …
   5      1         15       54    52
   …
   15      0          0      160   160
   16      0          0      160   160
   producer requested 12/s for 15s, sent 160 in 15.01s, measured 10.66/s
   w1 claims oldest first, service 150 ms nominal, finished 80 in 20.01s, busy 12.44s, 155.5 ms per job, capacity 6.43/s
   w2 claims oldest first, service 150 ms nominal, finished 80 in 20.01s, busy 12.46s, 155.8 ms per job, capacity 6.42/s

   The arrivals that built a 64 job backlog now peak at depth 1 and an oldest age of 15 ms. Each consumer was busy 12.4 of its 20 seconds, the headroom figure.

How to read the result

| What you see | What it means | What to do | | --- | --- | --- | | Depth flat, oldest age climbing 1000 ms a second, done unchanged | Nothing is claiming. The queue is a bucket, not a pipeline | Check the consumer process before anything else. Step 3 is what a dead worker looks like | | Depth on a straight rising line | Arrivals exceed service. The slope is the deficit in jobs a second | Divide arrival rate by one consumer's measured capacity and round up. Step 7 ran the answer | | Depth falling, oldest age still rising | The backlog is draining and the head is still late. Both were true from second 16 to second 24 | Alert on oldest age, not on depth. Depth turned down 9 seconds before latency did | | Depth low and steady, oldest age growing at wall clock rate | The head is never claimed. Priority or newest first ordering starves it | Read the claim query's ORDER BY. Step 6 is this failure in 25 seconds | | Depth near zero, oldest age tens of ms | Service rate is above arrival rate | Record the busy fraction as the headroom figure. Two consumers at 62 percent busy here | | Depth rising while the consumer's busy time is far below its wall clock | The consumer is waiting, not working. Claim contention or a poll interval | Measure capacity over busy time only, as worker.mjs does |

Common mistakes

What to check next

• How to test a stuck background job: a job stuck in running holds depth up while throughput looks fine.
• How to check if a queue worker is running: the answer to step 3.
• How to check job order in a queue: what step 7's second consumer does to ordering.
• How to test job timeout and cancellation: an unbounded service time makes capacity unknowable.
• How to measure event loop lag: the other lag, inside one process.

FAQ

What is queue depth?

The number of jobs waiting to be claimed at one instant. It is a level and carries no direction: depth 30 was rising at second 7 and depth 31 falling at second 20 of one run.

What is consumer lag?

How far behind the consumer is. As a rate, arrival minus service: 10.73 minus 6.46 predicted the depth slope 4.29. As a time, the age of the oldest pending job: 9589 ms.

How many consumers does a queue need?

Arrival rate over one consumer's measured capacity, rounded up. Step 4 measured 10.73 over 6.46, so two, and step 7's depth peaked at 1. Measure capacity over busy time or an idle consumer halves its figure.

What queue depth is too high?

No fixed number. Divide depth by the service rate for a wait and compare it to what the job promises. Depth 64 at 6.46 a second is 9.9 seconds: fine nightly, an incident for a password reset.

How often should the queue be sampled?

Often enough to see the slope. One sample a second named the 4.29 deficit in four rows; one a minute shows the 15 second overload as a point.

Verified