disclaw/tests/unit/channel-queue.test.ts

import { describe, it, expect, vi } from "vitest";
import { ChannelQueue } from "../../src/runtime/channel-queue";

/**
 * Returns a task that resolves after `ms` milliseconds and pushes its label
 * into the shared `order` array.
 */
function makeTask(order: string[], label: string, ms = 0): () => Promise<void> {
  return () =>
    new Promise<void>((resolve) => {
      setTimeout(() => {
        order.push(label);
        resolve();
      }, ms);
    });
}

describe("ChannelQueue", () => {
  it("(a) executes tasks in FIFO order within the same channel", async () => {
    const queue = new ChannelQueue();
    const order: string[] = [];

    const p1 = queue.enqueue("ch1", makeTask(order, "first"));
    const p2 = queue.enqueue("ch1", makeTask(order, "second"));
    const p3 = queue.enqueue("ch1", makeTask(order, "third"));

    await Promise.all([p1, p2, p3]);

    expect(order).toEqual(["first", "second", "third"]);
  });

  it("(b) a failing task does not block the next task", async () => {
    const queue = new ChannelQueue();
    const order: string[] = [];

    const failing: () => Promise<void> = () =>
      Promise.reject(new Error("intentional failure"));

    // Suppress the console.error output from ChannelQueue internals
    const spy = vi.spyOn(console, "error").mockImplementation(() => {});

    const p1 = queue.enqueue("ch1", failing);
    const p2 = queue.enqueue("ch1", makeTask(order, "after-error"));

    await Promise.all([p1, p2]);

    spy.mockRestore();

    expect(order).toEqual(["after-error"]);
  });

  it("(c) map entry is removed after all tasks complete", async () => {
    const queue = new ChannelQueue();

    const p1 = queue.enqueue("ch1", () => Promise.resolve());
    const p2 = queue.enqueue("ch1", () => Promise.resolve());

    await Promise.all([p1, p2]);

    // Allow the finally-cleanup microtask to run
    await Promise.resolve();

    expect((queue as unknown as { queues: Map<string, unknown> }).queues.size).toBe(0);
  });

  it("(d) different channels run in parallel, not serialized", async () => {
    const queue = new ChannelQueue();
    const completionOrder: string[] = [];

    // ch-slow gets a 50 ms task; ch-fast gets a 10 ms task.
    // If channels were serialized (ch-slow first), ch-fast would finish after
    // ch-slow. With true parallelism ch-fast finishes first.
    const pSlow = queue.enqueue(
      "ch-slow",
      () =>
        new Promise<void>((resolve) => {
          setTimeout(() => {
            completionOrder.push("slow");
            resolve();
          }, 50);
        })
    );

    const pFast = queue.enqueue(
      "ch-fast",
      () =>
        new Promise<void>((resolve) => {
          setTimeout(() => {
            completionOrder.push("fast");
            resolve();
          }, 10);
        })
    );

    await Promise.all([pSlow, pFast]);

    // fast must finish before slow
    expect(completionOrder).toEqual(["fast", "slow"]);
  });
});