tech-log-frontend/tests/unit/realtime/live-poll-handoff-coordinator.test.ts

import { describe, expect, it, vi } from "vitest";

import type { ClockPort } from "../../../src/application/ports/clock-port.ts";
import type { RealtimeResult } from "../../../src/application/ports/realtime/shared.ts";
import {
  createLivePollHandoffCoordinator,
  type LivePollHandoffCoordinatorDependencies,
  type LivePollHandoffLimits,
  type LiveProbeLease,
} from "../../../src/adapters/realtime/live-poll-handoff-coordinator.ts";
import {
  realtimeFailure,
  realtimeSuccess,
} from "../../../src/adapters/realtime/result.ts";

type Sleeper = Readonly<{
  dueAt: number;
  resolve(): void;
  reject(): void;
  signal?: AbortSignal;
  onAbort(): void;
}>;

class ManualClock implements ClockPort {
  current = 0;
  readonly sleepers: Sleeper[] = [];

  now(): number {
    return this.current;
  }

  sleep(milliseconds: number, signal?: AbortSignal): Promise<void> {
    return new Promise((resolve, reject) => {
      if (signal?.aborted) {
        reject(new DOMException("Aborted", "AbortError"));
        return;
      }
      let sleeper: Sleeper;
      const onAbort = () => {
        this.remove(sleeper);
        reject(new DOMException("Aborted", "AbortError"));
      };
      sleeper = {
        dueAt: this.current + milliseconds,
        resolve: () => {
          signal?.removeEventListener("abort", onAbort);
          resolve();
        },
        reject: () => {
          signal?.removeEventListener("abort", onAbort);
          reject(new DOMException("Aborted", "AbortError"));
        },
        signal,
        onAbort,
      };
      signal?.addEventListener("abort", onAbort, { once: true });
      this.sleepers.push(sleeper);
    });
  }

  advance(milliseconds: number): void {
    this.current += milliseconds;
    const ready = this.sleepers.filter(
      (sleeper) => sleeper.dueAt <= this.current,
    );
    for (const sleeper of ready) {
      this.remove(sleeper);
      sleeper.resolve();
    }
  }

  private remove(target: Sleeper): void {
    const index = this.sleepers.indexOf(target);
    if (index >= 0) this.sleepers.splice(index, 1);
  }
}

const limits: LivePollHandoffLimits = Object.freeze({
  quiescenceTimeoutMs: 100,
  maxActiveQueueCount: 3,
  maxActiveQueueBytes: 30,
  maxProbeBufferedEvents: 3,
  maxProbeBufferedBytes: 30,
  maxItemBytes: 10,
});

function createHarness(
  input: Readonly<{
    initial?: "LIVE" | "POLL";
    limits?: LivePollHandoffLimits;
    apply?: LivePollHandoffCoordinatorDependencies<string>["apply"];
    recover?: LivePollHandoffCoordinatorDependencies<string>["establishAuthoritativeCheckpoint"];
  }> = {},
) {
  const clock = new ManualClock();
  const effects: string[] = [];
  const recoveries: string[] = [];
  const apply =
    input.apply ??
    vi.fn(async ({ writer, value }) => {
      effects.push(`${writer}:${value}`);
      return realtimeSuccess(undefined);
    });
  const recover =
    input.recover ??
    vi.fn(async ({ from, to }) => {
      recoveries.push(`${from}->${to}`);
      return realtimeSuccess(undefined);
    });
  const coordinator = createLivePollHandoffCoordinator({
    initial: {
      writer: input.initial ?? "LIVE",
      authoritativeCheckpointEstablished: true,
    },
    limits: input.limits ?? limits,
    apply,
    establishAuthoritativeCheckpoint: recover,
    clock,
  });
  return { apply, clock, coordinator, effects, recover, recoveries };
}

async function flush(): Promise<void> {
  for (let turn = 0; turn < 12; turn += 1) {
    await Promise.resolve();
  }
}

describe("live/poll authoritative writer handoff", () => {
  it("serializes effects through the one current writer lease", async () => {
    const first = deferred<RealtimeResult<void>>();
    const starts: string[] = [];
    let activeEffects = 0;
    let highWatermark = 0;
    const harness = createHarness({
      apply: vi.fn(async ({ value }) => {
        starts.push(value);
        activeEffects += 1;
        highWatermark = Math.max(highWatermark, activeEffects);
        if (value === "first") await first.promise;
        activeEffects -= 1;
        return realtimeSuccess(undefined);
      }),
    });
    const writer = harness.coordinator.currentWriter();
    expect(writer?.writer).toBe("LIVE");

    const firstWrite = writer!.write("first", 5);
    const secondWrite = writer!.write("second", 6);
    await flush();
    expect(starts).toEqual(["first"]);

    first.resolve(realtimeSuccess(undefined));
    await expect(firstWrite).resolves.toMatchObject({
      ok: true,
      value: { kind: "APPLIED", writer: "LIVE" },
    });
    await expect(secondWrite).resolves.toMatchObject({
      ok: true,
      value: { kind: "APPLIED", writer: "LIVE" },
    });
    expect(starts).toEqual(["first", "second"]);
    expect(highWatermark).toBe(1);
  });

  it("fails closed when a non-cooperative active writer fills the bounded tail", async () => {
    let firstSignal: AbortSignal | undefined;
    let firstIsCurrent: (() => boolean) | undefined;
    const starts: string[] = [];
    const harness = createHarness({
      limits: {
        ...limits,
        maxActiveQueueCount: 2,
        maxActiveQueueBytes: 10,
      },
      apply: vi.fn(async ({ value, signal, isCurrent }) => {
        starts.push(value);
        if (value === "first") {
          firstSignal = signal;
          firstIsCurrent = isCurrent;
          return await new Promise<RealtimeResult<void>>(() => {});
        }
        return realtimeSuccess(undefined);
      }),
    });
    const writer = harness.coordinator.currentWriter()!;
    const first = writer.write("first", 5);
    const second = writer.write("second", 5);
    await flush();

    expect(starts).toEqual(["first"]);
    expect(firstIsCurrent?.()).toBe(true);
    await expect(writer.write("overflow", 1)).resolves.toMatchObject({
      ok: false,
      error: {
        kind: "QUEUE_OVERFLOW",
        operation: "APPLY",
        retryable: false,
      },
    });
    expect(firstSignal?.aborted).toBe(true);
    expect(firstIsCurrent?.()).toBe(false);
    expect(writer.isCurrent()).toBe(false);
    expect(harness.coordinator.inspect()).toMatchObject({
      state: "CLOSED",
      activeWriter: null,
    });
    expect(starts).toEqual(["first"]);
    await expect(writer.write("after-close", 1)).resolves.toMatchObject({
      ok: false,
      error: { kind: "CLOSED" },
    });
    void first;
    void second;
  });

  it("tracks a retired active writer after handoff queue overflow", async () => {
    let release: ((result: RealtimeResult<void>) => void) | undefined;
    const harness = createHarness({
      limits: {
        ...limits,
        maxActiveQueueCount: 2,
        maxActiveQueueBytes: 10,
      },
      apply: vi.fn(async ({ value }) => {
        if (value === "first") {
          return await new Promise<RealtimeResult<void>>((resolve) => {
            release = resolve;
          });
        }
        return realtimeSuccess(undefined);
      }),
    });
    const writer = harness.coordinator.currentWriter()!;
    const first = writer.write("first", 5);
    const second = writer.write("second", 5);
    await flush();
    await expect(writer.write("overflow", 1)).resolves.toMatchObject({
      ok: false,
      error: { kind: "QUEUE_OVERFLOW" },
    });

    // R-03. The fail-close dropped the active reference, but the writer is
    // still running, so close() must not claim quiescence.
    const closing = harness.coordinator.close();
    await flush();
    harness.clock.advance(100);
    await expect(closing).resolves.toMatchObject({
      ok: false,
      error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
    });

    release?.(realtimeSuccess(undefined));
    void first;
    void second;
  });

  it("fences and aborts live, waits for quiescence, then recovers before activating poll", async () => {
    const liveEffect = deferred<RealtimeResult<void>>();
    const checkpoint = deferred<RealtimeResult<void>>();
    const order: string[] = [];
    let liveSignal: AbortSignal | undefined;
    let liveIsCurrent: (() => boolean) | undefined;
    const harness = createHarness({
      apply: vi.fn(async ({ writer, value, signal, isCurrent }) => {
        order.push(`apply:${writer}:${value}`);
        if (writer === "LIVE") {
          liveSignal = signal;
          liveIsCurrent = isCurrent;
          return await liveEffect.promise;
        }
        return realtimeSuccess(undefined);
      }),
      recover: vi.fn(async ({ from, to }) => {
        order.push(`recover:${from}->${to}`);
        return await checkpoint.promise;
      }),
    });
    const live = harness.coordinator.currentWriter()!;
    const pendingEffect = live.write("in-flight", 9);
    await flush();

    const transition = harness.coordinator.switchToPoll();
    expect(liveSignal?.aborted).toBe(true);
    expect(liveIsCurrent?.()).toBe(false);
    expect(live.isCurrent()).toBe(false);
    await expect(live.write("stale", 999)).resolves.toMatchObject({
      ok: false,
      error: { kind: "SCOPE_FENCED" },
    });
    expect(harness.recover).not.toHaveBeenCalled();
    await expect(
      harness.coordinator.switchToPoll(),
    ).resolves.toMatchObject({
      ok: false,
      error: { kind: "PROTOCOL_MISMATCH" },
    });

    liveEffect.resolve(realtimeSuccess(undefined));
    await expect(pendingEffect).resolves.toMatchObject({
      ok: false,
      error: { kind: "SCOPE_FENCED" },
    });
    await flush();
    expect(order).toEqual([
      "apply:LIVE:in-flight",
      "recover:LIVE->POLL",
    ]);
    expect(harness.coordinator.currentWriter()).toBeNull();

    checkpoint.resolve(realtimeSuccess(undefined));
    const result = await transition;
    expect(result).toMatchObject({
      ok: true,
      value: { writer: "POLL" },
    });
    if (!result.ok) throw new Error("expected poll writer");
    await result.value.write("polled", 6);
    expect(order.at(-1)).toBe("apply:POLL:polled");
    expect(result.value.generation).toBeGreaterThan(live.generation);
    expect(harness.coordinator.inspect().state).toBe("POLL_ACTIVE");
  });

  it("fails closed when the prior writer cannot quiesce before the bound", async () => {
    const harness = createHarness({
      apply: vi.fn(
        async () =>
          await new Promise<RealtimeResult<void>>(() => {}),
      ),
    });
    const live = harness.coordinator.currentWriter()!;
    void live.write("hung", 4);
    await flush();

    const transition = harness.coordinator.switchToPoll();
    await flush();
    expect(harness.clock.sleepers).toHaveLength(1);
    harness.clock.advance(100);

    await expect(transition).resolves.toMatchObject({
      ok: false,
      error: { kind: "IDLE_TIMEOUT", operation: "RECOVER" },
    });
    expect(harness.recover).not.toHaveBeenCalled();
    expect(harness.coordinator.inspect()).toMatchObject({
      state: "CLOSED",
      activeWriter: null,
    });
  });

  it("keeps poll authoritative while probing, then recovers and drains live values serially", async () => {
    const firstLive = deferred<RealtimeResult<void>>();
    const order: string[] = [];
    let activeEffects = 0;
    let highWatermark = 0;
    const harness = createHarness({
      initial: "POLL",
      apply: vi.fn(async ({ writer, value }) => {
        activeEffects += 1;
        highWatermark = Math.max(highWatermark, activeEffects);
        order.push(`apply:${writer}:${value}`);
        if (value === "live-1") await firstLive.promise;
        activeEffects -= 1;
        return realtimeSuccess(undefined);
      }),
      recover: vi.fn(async ({ from, to }) => {
        order.push(`recover:${from}->${to}`);
        return realtimeSuccess(undefined);
      }),
    });
    const poll = harness.coordinator.currentWriter()!;
    const opened = harness.coordinator.beginLiveProbe();
    expect(opened.ok).toBe(true);
    if (!opened.ok) throw new Error("expected live probe");
    const live = opened.value;

    await expect(live.write("live-1", 6)).resolves.toMatchObject({
      ok: true,
      value: { kind: "BUFFERED" },
    });
    await live.write("live-2", 6);
    expect(live.isCurrent()).toBe(false);
    expect(poll.isCurrent()).toBe(true);
    await poll.write("poll-during-probe", 8);
    expect(order).toEqual(["apply:POLL:poll-during-probe"]);

    const activation = live.activate();
    await vi.waitFor(() =>
      expect(order).toEqual([
        "apply:POLL:poll-during-probe",
        "recover:POLL->LIVE",
        "apply:LIVE:live-1",
      ]),
    );
    await live.write("live-3", 6);
    expect(harness.coordinator.inspect()).toMatchObject({
      state: "LIVE_PROBING",
      bufferedEvents: 2,
      transitioning: true,
    });

    firstLive.resolve(realtimeSuccess(undefined));
    const activated = await activation;
    expect(activated).toMatchObject({
      ok: true,
      value: { writer: "LIVE" },
    });
    expect(order).toEqual([
      "apply:POLL:poll-during-probe",
      "recover:POLL->LIVE",
      "apply:LIVE:live-1",
      "apply:LIVE:live-2",
      "apply:LIVE:live-3",
    ]);
    expect(highWatermark).toBe(1);
    expect(live.isCurrent()).toBe(true);
    expect(poll.isCurrent()).toBe(false);
    await expect(poll.write("stale-poll", 5)).resolves.toMatchObject({
      ok: false,
      error: { kind: "SCOPE_FENCED" },
    });
    expect(harness.coordinator.inspect()).toMatchObject({
      state: "LIVE_ACTIVE",
      bufferedEvents: 0,
      bufferedBytes: 0,
    });
  });

  it("drops an overflowing probe without displacing poll and never reuses its generation", async () => {
    const harness = createHarness({
      initial: "POLL",
      limits: {
        ...limits,
        maxProbeBufferedEvents: 2,
      },
    });
    const poll = harness.coordinator.currentWriter()!;
    const first = expectProbe(harness.coordinator.beginLiveProbe());
    await first.write("one", 3);
    await first.write("two", 3);
    await expect(first.write("overflow", 3)).resolves.toMatchObject({
      ok: false,
      error: { kind: "QUEUE_OVERFLOW" },
    });
    expect(harness.coordinator.inspect().state).toBe("POLL_ACTIVE");
    expect(poll.isCurrent()).toBe(true);
    await expect(first.write("stale", 999)).resolves.toMatchObject({
      ok: false,
      error: { kind: "SCOPE_FENCED" },
    });

    const second = expectProbe(harness.coordinator.beginLiveProbe());
    expect(second.generation).toBeGreaterThan(first.generation);
    const canceled = second.cancel();
    expect(canceled).toMatchObject({
      ok: true,
      value: { writer: "POLL", generation: poll.generation },
    });
    expect(second.signal.aborted).toBe(true);
    expect(harness.coordinator.inspect().state).toBe("POLL_ACTIVE");
  });

  it("closes if authoritative recovery fails and never activates the candidate", async () => {
    const harness = createHarness({
      recover: vi.fn(async () =>
        realtimeFailure("CURSOR_EXPIRED", "RECOVER"),
      ),
    });
    const live = harness.coordinator.currentWriter()!;

    await expect(
      harness.coordinator.switchToPoll(),
    ).resolves.toMatchObject({
      ok: false,
      error: { kind: "CURSOR_EXPIRED", operation: "RECOVER" },
    });
    expect(live.signal.aborted).toBe(true);
    expect(harness.coordinator.inspect()).toMatchObject({
      state: "CLOSED",
      activeWriter: null,
    });
  });

  it("bounds a non-cooperative authoritative checkpoint", async () => {
    let checkpointIsCurrent: (() => boolean) | undefined;
    const harness = createHarness({
      recover: vi.fn(async ({ isCurrent }) => {
        checkpointIsCurrent = isCurrent;
        return await new Promise<RealtimeResult<void>>(() => {});
      }),
    });

    const transition = harness.coordinator.switchToPoll();
    await flush();
    expect(checkpointIsCurrent?.()).toBe(true);
    expect(harness.clock.sleepers).toHaveLength(1);
    harness.clock.advance(100);

    await expect(transition).resolves.toMatchObject({
      ok: false,
      error: { kind: "IDLE_TIMEOUT", operation: "RECOVER" },
    });
    expect(checkpointIsCurrent?.()).toBe(false);
    expect(harness.coordinator.inspect().state).toBe("CLOSED");
  });

  it("aborts and bounds close quiescence when an effect ignores cancellation", async () => {
    let effectSignal: AbortSignal | undefined;
    const harness = createHarness({
      apply: vi.fn(
        async ({ signal }) => {
          effectSignal = signal;
          return await new Promise<RealtimeResult<void>>(() => {});
        },
      ),
    });
    const live = harness.coordinator.currentWriter()!;
    void live.write("hung-close", 8);
    await flush();

    const closing = harness.coordinator.close();
    expect(effectSignal?.aborted).toBe(true);
    expect(harness.coordinator.inspect().state).toBe("CLOSED");
    await flush();
    harness.clock.advance(100);
    await expect(closing).resolves.toMatchObject({
      ok: false,
      error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
    });
    // RT-RR-03. A second close re-runs rather than replaying a cached verdict.
    // The writer is still hung, so it still reports a timeout.
    const second = harness.coordinator.close();
    await flush();
    harness.clock.advance(100);
    await expect(second).resolves.toMatchObject({
      ok: false,
      error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
    });
  });

  /**
   * RT-RR-03. Caching the first timeout forever meant a writer that later
   * settled could never be proved quiescent: every subsequent close replayed
   * the stale failure and the retained registry could never be pruned.
   */
  it("converges to success once a retired writer finally settles", async () => {
    let release: ((value: RealtimeResult<void>) => void) | undefined;
    const harness = createHarness({
      apply: vi.fn(
        async () =>
          await new Promise<RealtimeResult<void>>((resolve) => {
            release = resolve;
          }),
      ),
    });
    const live = harness.coordinator.currentWriter()!;
    void live.write("late-settle", 8);
    await flush();

    const first = harness.coordinator.close();
    await flush();
    harness.clock.advance(100);
    await expect(first).resolves.toMatchObject({
      ok: false,
      error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
    });

    // The writer finishes after the first close gave up.
    release?.(realtimeSuccess(undefined));
    await flush();
    await flush();

    const second = harness.coordinator.close();
    await flush();
    harness.clock.advance(100);
    await expect(second).resolves.toMatchObject({ ok: true });
  });

  /**
   * RT-RR-04. Checkpoint work is an external authority call like a writer
   * tail. Racing it against a timeout bounded the public wait but left it out
   * of the retained registry, so `close()` could report quiescence while the
   * checkpoint was still running.
   */
  it("does not report quiescence while a checkpoint is still running", async () => {
    let checkpointSignal: AbortSignal | undefined;
    const harness = createHarness({
      recover: vi.fn(async ({ signal }) => {
        checkpointSignal = signal;
        return await new Promise<RealtimeResult<void>>(() => {});
      }),
    });

    const transition = harness.coordinator.switchToPoll();
    await flush();
    expect(checkpointSignal).toBeDefined();

    const closing = harness.coordinator.close();
    await flush();
    harness.clock.advance(100);
    await expect(closing).resolves.toMatchObject({
      ok: false,
      error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
    });

    harness.clock.advance(100);
    await flush();
    await transition;
  });

  it("rejects invalid initial authority and resource ceilings", () => {
    expect(() =>
      createLivePollHandoffCoordinator({
        initial: {
          writer: "LIVE",
          authoritativeCheckpointEstablished: false,
        } as never,
        limits,
        apply: async () => realtimeSuccess(undefined),
        establishAuthoritativeCheckpoint: async () =>
          realtimeSuccess(undefined),
      }),
    ).toThrow(/initial checkpoint/u);
    expect(() =>
      createLivePollHandoffCoordinator({
        initial: {
          writer: "POLL",
          authoritativeCheckpointEstablished: true,
        },
        limits: { ...limits, maxProbeBufferedEvents: 257 },
        apply: async () => realtimeSuccess(undefined),
        establishAuthoritativeCheckpoint: async () =>
          realtimeSuccess(undefined),
      }),
    ).toThrow(/limits/u);
    expect(() =>
      createLivePollHandoffCoordinator({
        initial: {
          writer: "POLL",
          authoritativeCheckpointEstablished: true,
        },
        limits: { ...limits, maxActiveQueueCount: 257 },
        apply: async () => realtimeSuccess(undefined),
        establishAuthoritativeCheckpoint: async () =>
          realtimeSuccess(undefined),
      }),
    ).toThrow(/limits/u);
  });
});

function expectProbe(
  result: RealtimeResult<LiveProbeLease<string>>,
): LiveProbeLease<string> {
  if (!result.ok) throw new Error("expected live probe");
  return result.value;
}

function deferred<Value>() {
  let resolve!: (value: Value) => void;
  const promise = new Promise<Value>((selectedResolve) => {
    resolve = selectedResolve;
  });
  return { promise, resolve };
}