Route new PTS producers through Stream V6

src/client/index.test.ts

@@ -0,0 +1,275 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import {
+  PersistentTextStreaming,
+  type StreamId,
+  type StreamWriter,
+} from "./index.js";
+
+const streamId = "stream" as StreamId;
+
+type Call =
+  | { kind: "claim" }
+  | { kind: "append"; text: string; final: boolean }
+  | { kind: "status"; status: string };
+
+function setup(claims: boolean[] = [true]) {
+  const refs = {
+    lib: {
+      addChunk: {},
+      claimStream: {},
+      createStream: {},
+      deleteStream: {},
+      getStreamStatus: {},
+      getStreamText: {},
+      setStreamStatus: {},
+    },
+  };
+  const calls: Call[] = [];
+  const ctx = {
+    runMutation: vi.fn(async (ref: object, args: Record<string, unknown>) => {
+      if (ref === refs.lib.claimStream) {
+        calls.push({ kind: "claim" });
+        return claims.shift() ?? false;
+      }
+      if (ref === refs.lib.addChunk) {
+        calls.push({
+          kind: "append",
+          text: args.text as string,
+          final: args.final as boolean,
+        });
+        return null;
+      }
+      if (ref === refs.lib.setStreamStatus) {
+        calls.push({ kind: "status", status: args.status as string });
+        return null;
+      }
+      throw new Error("Unexpected mutation.");
+    }),
+  };
+  return {
+    calls,
+    ctx,
+    refs,
+    streaming: new PersistentTextStreaming(refs as never),
+  };
+}
+
+function request(): Request {
+  return new Request("https://example.com/stream", { method: "POST" });
+}
+
+afterEach(() => vi.useRealTimers());
+
+describe("PersistentTextStreaming producer", () => {
+  it("preserves one-winner claim behavior", async () => {
+    const { calls, ctx, streaming } = setup([true, false]);
+    const writer = vi.fn<StreamWriter<never>>(
+      async (_ctx, _request, _id, append) => {
+        await append("winner");
+      },
+    );
+
+    const winner = await streaming.stream(
+      ctx as never,
+      request(),
+      streamId,
+      writer,
+    );
+    const loser = await streaming.stream(
+      ctx as never,
+      request(),
+      streamId,
+      writer,
+    );
+
+    expect(loser.status).toBe(205);
+    expect(await winner.text()).toBe("winner");
+    expect(writer).toHaveBeenCalledTimes(1);
+    expect(calls.filter((call) => call.kind === "claim")).toHaveLength(2);
+  });
+
+  it("flushes on the 100ms timer before final durable completion", async () => {
+    vi.useFakeTimers();
+    const { calls, ctx, streaming } = setup();
+    let release: (() => void) | undefined;
+    const gate = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    const writer: StreamWriter<never> = async (_ctx, _request, _id, append) => {
+      await append("timed");
+      await gate;
+    };
+
+    const response = await streaming.stream(
+      ctx as never,
+      request(),
+      streamId,
+      writer,
+    );
+    await vi.advanceTimersByTimeAsync(99);
+    expect(calls.some((call) => call.kind === "append")).toBe(false);
+    await vi.advanceTimersByTimeAsync(1);
+    expect(calls).toContainEqual({
+      kind: "append",
+      text: "timed",
+      final: false,
+    });
+
+    release?.();
+    expect(await response.text()).toBe("timed");
+    expect(calls.at(-1)).toEqual({ kind: "status", status: "done" });
+  });
+
+  it("splits at the 16KiB byte bound and serializes append order", async () => {
+    const { calls, ctx, streaming } = setup();
+    const first = "a".repeat(12 * 1024);
+    const second = "🙂".repeat(2 * 1024);
+    const writer: StreamWriter<never> = async (_ctx, _request, _id, append) => {
+      await Promise.all([append(first), append(second)]);
+    };
+
+    const response = await streaming.stream(
+      ctx as never,
+      request(),
+      streamId,
+      writer,
+    );
+    expect(await response.text()).toBe(first + second);
+
+    const appends = calls.filter(
+      (call): call is Extract<Call, { kind: "append" }> =>
+        call.kind === "append",
+    );
+    expect(appends.map((call) => call.text).join("")).toBe(first + second);
+    expect(
+      appends.every(
+        (call) => new TextEncoder().encode(call.text).byteLength <= 16 * 1024,
+      ),
+    ).toBe(true);
+    expect(appends.at(-1)?.final).toBe(true);
+  });
+
+  it("persists the failure prefix before recording the producer error", async () => {
+    const { calls, ctx, streaming } = setup();
+    const failure = new Error("producer failed");
+    const writer: StreamWriter<never> = async (_ctx, _request, _id, append) => {
+      await append("durable prefix");
+      throw failure;
+    };
+
+    const response = await streaming.stream(
+      ctx as never,
+      request(),
+      streamId,
+      writer,
+    );
+    await expect(response.text()).rejects.toThrow("producer failed");
+    expect(calls.slice(-2)).toEqual([
+      { kind: "append", text: "durable prefix", final: false },
+      { kind: "status", status: "error" },
+    ]);
+  });
+
+  it("finishes durable writes when the raw body is never consumed", async () => {
+    const { calls, ctx, refs, streaming } = setup();
+    let durableDone: (() => void) | undefined;
+    const done = new Promise<void>((resolve) => {
+      durableDone = resolve;
+    });
+    ctx.runMutation.mockImplementation(
+      async (ref: object, args: Record<string, unknown>) => {
+        if (ref === refs.lib.claimStream) {
+          calls.push({ kind: "claim" });
+          return true;
+        }
+        if (ref === refs.lib.addChunk) {
+          calls.push({
+            kind: "append",
+            text: args.text as string,
+            final: args.final as boolean,
+          });
+          return null;
+        }
+        calls.push({ kind: "status", status: args.status as string });
+        if (args.status === "done") durableDone?.();
+        return null;
+      },
+    );
+    const text = "🙂".repeat(20 * 1024);
+    const writer: StreamWriter<never> = async (_ctx, _request, _id, append) => {
+      await append(text);
+    };
+
+    const response = await streaming.stream(
+      ctx as never,
+      request(),
+      streamId,
+      writer,
+    );
+
+    await done;
+    const appends = calls.filter(
+      (call): call is Extract<Call, { kind: "append" }> =>
+        call.kind === "append",
+    );
+    expect(appends.map((call) => call.text).join("")).toBe(text);
+    expect(calls.at(-1)).toEqual({ kind: "status", status: "done" });
+    await expect(response.text()).rejects.toThrow(
+      "Raw stream consumer fell behind durable replay.",
+    );
+  });
+
+  it("continues durable completion after the raw reader cancels", async () => {
+    const { calls, ctx, refs, streaming } = setup();
+    let release: (() => void) | undefined;
+    let persisted: (() => void) | undefined;
+    const gate = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    const done = new Promise<void>((resolve) => {
+      persisted = resolve;
+    });
+    ctx.runMutation.mockImplementation(
+      async (ref: object, args: Record<string, unknown>) => {
+        if (ref === refs.lib.claimStream) {
+          calls.push({ kind: "claim" });
+          return true;
+        }
+        if (ref === refs.lib.addChunk) {
+          calls.push({
+            kind: "append",
+            text: args.text as string,
+            final: args.final as boolean,
+          });
+          if (args.final === true) persisted?.();
+          return null;
+        }
+        calls.push({ kind: "status", status: args.status as string });
+        return null;
+      },
+    );
+    const writer: StreamWriter<never> = async (_ctx, _request, _id, append) => {
+      await append("first.");
+      await gate;
+      await append("second");
+    };
+
+    const response = await streaming.stream(
+      ctx as never,
+      request(),
+      streamId,
+      writer,
+    );
+    const reader = response.body!.getReader();
+    await reader.read();
+    await reader.cancel();
+    release?.();
+    await done;
+
+    expect(calls.filter((call) => call.kind === "append")).toEqual([
+      { kind: "append", text: "first.", final: false },
+      { kind: "append", text: "second", final: true },
+    ]);
+  });
+});