jobqueue/tests/WorkerPool.test.ts

import { WorkerPool } from '../src/index.js';
import { waitFor } from './helpers.js';

describe('WorkerPool', () => {
  it('respects concurrency limits', async () => {
    const pool = new WorkerPool(2);
    let active = 0;
    let maxActive = 0;

    const tasks = Array.from({ length: 5 }, (_, index) =>
      pool.run(async () => {
        active += 1;
        maxActive = Math.max(maxActive, active);
        await new Promise((resolve) => {
          setTimeout(resolve, 20 + index);
        });
        active -= 1;
      }),
    );

    await Promise.all(tasks);
    await waitFor(() => pool.activeCount === 0 && pool.pendingCount === 0);

    expect(maxActive).toBe(2);
  });

  it('times out while draining slow tasks', async () => {
    const pool = new WorkerPool(1);
    let release!: () => void;
    const blocked = new Promise<void>((resolve) => {
      release = resolve;
    });

    void pool.run(async () => {
      await blocked;
    });

    await waitFor(() => pool.activeCount === 1);
    await expect(pool.drain(10)).rejects.toThrow(/Timed out waiting for workers to drain/);

    release();
    await pool.drain();
  });
});