kongruity/backend/tests/queue.test.ts

import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import { enqueue, size, drain, configureQueue } from '../lib/queue.js';

type Deferred = { promise: Promise<void>; resolve: () => void };

const deferred = (): Deferred => {
  let resolve!: () => void;
  const promise = new Promise<void>((res) => {
    resolve = res;
  });
  return { promise, resolve };
};

// A macrotask boundary: enough for the queue to pick up work it deferred.
const flush = (): Promise<void> =>
  new Promise((resolve) => {
    setTimeout(resolve, 0);
  });

describe('queue', () => {
  let errors: ReturnType<typeof vi.spyOn>;

  beforeEach(() => {
    configureQueue({ maxAttempts: 3, baseDelayMs: 0 });
    errors = vi.spyOn(console, 'error').mockImplementation(() => {});
  });

  afterEach(async () => {
    await drain();
    vi.useRealTimers();
    vi.restoreAllMocks();
  });

  it('should run jobs one at a time in FIFO order', async () => {
    const order: string[] = [];
    const record = (id: string) => async (): Promise<void> => {
      order.push(`${id}:start`);
      await Promise.resolve();
      order.push(`${id}:end`);
    };

    enqueue('a', record('a'));
    enqueue('b', record('b'));
    enqueue('c', record('c'));

    await drain();

    expect(order).toEqual([
      'a:start', 'a:end',
      'b:start', 'b:end',
      'c:start', 'c:end',
    ]);
  });

  it('should not throw synchronously when a job throws synchronously', async () => {
    const thrower = (): Promise<void> => {
      throw new Error('sync boom');
    };

    expect(() => enqueue('sync-thrower', thrower)).not.toThrow();

    await drain();

    expect(errors).toHaveBeenCalled();
  });

  it('should retry a failing job up to the configured cap and then give up', async () => {
    configureQueue({ maxAttempts: 4 });
    let attempts = 0;

    enqueue('always-failing', async () => {
      attempts += 1;
      throw new Error('boom');
    });

    await drain();

    expect(attempts).toBe(4);
    expect(errors).toHaveBeenCalledOnce();
    expect(String(errors.mock.calls[0]?.[0])).toContain('always-failing');
  });

  it('should stop retrying as soon as an attempt succeeds', async () => {
    let attempts = 0;

    enqueue('flaky', async () => {
      attempts += 1;
      if (attempts < 2) throw new Error('transient');
    });

    await drain();

    expect(attempts).toBe(2);
    expect(errors).not.toHaveBeenCalled();
  });

  it('should keep running later jobs after one fails permanently', async () => {
    const completed: string[] = [];

    enqueue('doomed', async () => {
      throw new Error('boom');
    });
    enqueue('survivor', async () => {
      completed.push('survivor');
    });

    await drain();

    expect(completed).toEqual(['survivor']);
  });

  it('should grow the backoff delay between attempts', async () => {
    vi.useFakeTimers();
    vi.spyOn(Math, 'random').mockReturnValue(0);
    configureQueue({ maxAttempts: 4, baseDelayMs: 100 });

    let attempts = 0;
    enqueue('retrying', async () => {
      attempts += 1;
      throw new Error('boom');
    });

    await vi.advanceTimersByTimeAsync(0);
    expect(attempts).toBe(1);

    await vi.advanceTimersByTimeAsync(99);
    expect(attempts).toBe(1);
    await vi.advanceTimersByTimeAsync(2);
    expect(attempts).toBe(2);

    await vi.advanceTimersByTimeAsync(198);
    expect(attempts).toBe(2);
    await vi.advanceTimersByTimeAsync(2);
    expect(attempts).toBe(3);

    await vi.advanceTimersByTimeAsync(398);
    expect(attempts).toBe(3);
    await vi.advanceTimersByTimeAsync(2);
    expect(attempts).toBe(4);
  });

  it('should apply jitter within the backoff window', async () => {
    vi.useFakeTimers();
    vi.spyOn(Math, 'random').mockReturnValue(0.75);
    configureQueue({ maxAttempts: 2, baseDelayMs: 100 });

    let attempts = 0;
    enqueue('jittered', async () => {
      attempts += 1;
      throw new Error('boom');
    });

    await vi.advanceTimersByTimeAsync(174);
    expect(attempts).toBe(1);
    await vi.advanceTimersByTimeAsync(2);
    expect(attempts).toBe(2);
  });

  it('should resolve drain() only once in-flight work has finished', async () => {
    const gate = deferred();
    let finished = false;

    enqueue('blocked', async () => {
      await gate.promise;
      finished = true;
    });

    await flush();
    expect(finished).toBe(false);

    gate.resolve();
    await drain();

    expect(finished).toBe(true);
  });

  it('should resolve every concurrent drain() caller', async () => {
    const gate = deferred();
    enqueue('blocked', () => gate.promise);

    const waiters = Promise.all([drain(), drain(), drain()]);
    gate.resolve();

    await expect(waiters).resolves.toEqual([undefined, undefined, undefined]);
  });

  it('should resolve drain() immediately when the queue is idle', async () => {
    const winner = await Promise.race([
      drain().then(() => 'drained'),
      new Promise<string>((resolve) => {
        setTimeout(() => resolve('timer'), 0);
      }),
    ]);

    expect(winner).toBe('drained');
  });

  it('should report the pending count excluding the job in flight', async () => {
    const gate = deferred();

    enqueue('blocked', () => gate.promise);
    enqueue('second', async () => {});
    enqueue('third', async () => {});

    expect(size()).toBe(3);

    await flush();
    expect(size()).toBe(2);

    gate.resolve();
    await drain();

    expect(size()).toBe(0);
  });
});