char-slop/berry

git clone https://git.t4t.associates/char-slop/berry

Charlotte Sommcp: add streaming fetch on conversation-local and remote environmentse5ec3802026-10-10

import { assertEquals, assertRejects, assertStringIncludes } from "@std/assert";
import { BACKGROUND_CONTEXT as context } from "@earendil-works/chord/context";
import { until } from "@berry/util";
import { BOOTSTRAP, Connection, RemoteEnvironment } from "./connection.ts";

async function fixture(handler: (request: Request) => Response | Promise<Response>) {
  const server = Deno.serve({ hostname: "127.0.0.1", port: 0, onListen() {} }, handler);
  const connection = await Connection.open("fetch-test", [
    Deno.execPath(),
    "eval",
    "--quiet",
    BOOTSTRAP,
  ]);
  return {
    env: new RemoteEnvironment("fetch-test", connection, "/"),
    url: `http://127.0.0.1:${server.addr.port}`,
    connection,
    async [Symbol.asyncDispose]() {
      await connection.close();
      await server.shutdown();
    },
  };
}

Deno.test("remote fetch originates at the worker, preserving binary requests, metadata, and incremental bodies", async () => {
  const release = Promise.withResolvers<void>();
  const encoder = new TextEncoder();
  await using f = await fixture(async (request) => {
    const path = new URL(request.url).pathname;
    if (path === "/redirect") {
      return new Response(null, { status: 302, headers: { location: "/stream" } });
    }
    if (path === "/stream") {
      return new Response(
        new ReadableStream({
          start(controller) {
            controller.enqueue(encoder.encode("first"));
          },
          async pull(controller) {
            await release.promise;
            controller.enqueue(encoder.encode("second"));
            controller.close();
          },
        }),
      );
    }
    return new Response(await request.arrayBuffer(), {
      status: 201,
      statusText: "Created",
      headers: { "x-seen": request.headers.get("x-secret") ?? "" },
    });
  });
  const fetchHere = globalThis.fetch;
  globalThis.fetch = () => Promise.reject(new Error("fetch must run on the worker"));
  try {
    const echoed = await f.env.fetch(f.url + "/echo", {
      method: "POST",
      headers: new Headers({ "x-secret": "from-remote" }),
      body: new Uint8Array([0, 255, 10]),
      signal: AbortSignal.timeout(5000),
    });
    assertEquals(echoed.status, 201);
    assertEquals(echoed.statusText, "Created");
    assertEquals(echoed.headers.get("x-seen"), "from-remote");
    assertEquals(new Uint8Array(await echoed.arrayBuffer()), new Uint8Array([0, 255, 10]));

    const response = await f.env.fetch(f.url + "/redirect", { signal: AbortSignal.timeout(5000) });
    assertEquals(response.redirected, true);
    assertEquals(response.url, f.url + "/stream");
    const reader = response.body!.getReader();
    assertEquals(new TextDecoder().decode((await reader.read()).value), "first");
    release.resolve();
    assertEquals(new TextDecoder().decode((await reader.read()).value), "second");
    assertEquals((await reader.read()).done, true);
  } finally {
    release.resolve();
    globalThis.fetch = fetchHere;
  }
});

Deno.test("remote fetch aborts before headers and stops streaming bodies on abort, cancel, or environment cleanup", async () => {
  const entered = Promise.withResolvers<void>();
  const release = Promise.withResolvers<void>();
  let cancelled = 0;
  await using f = await fixture(async (request) => {
    if (new URL(request.url).pathname === "/headers") {
      entered.resolve();
      await release.promise;
      return new Response("late");
    }
    return new Response(
      new ReadableStream({
        start(controller) {
          controller.enqueue(new TextEncoder().encode("first"));
        },
        cancel() {
          cancelled++;
        },
      }),
    );
  });
  try {
    const beforeHeaders = new AbortController();
    const pending = f.env.fetch(f.url + "/headers", { signal: beforeHeaders.signal });
    const rejected = assertRejects(() => pending);
    await entered.promise;
    beforeHeaders.abort();
    await rejected;
    release.resolve();

    const abort = new AbortController();
    const response = await f.env.fetch(f.url + "/body", { signal: abort.signal });
    const reader = response.body!.getReader();
    await reader.read();
    const reading = assertRejects(() => reader.read());
    abort.abort();
    await reading;
    await until(() => cancelled === 1);

    const discard = await f.env.fetch(f.url + "/body");
    await discard.body!.cancel();
    await until(() => cancelled === 2);

    const orphan = await f.env.fetch(f.url + "/body");
    const orphanReader = orphan.body!.getReader();
    await orphanReader.read();
    const lost = assertRejects(() => orphanReader.read());
    await f.env.cleanup(context);
    await lost;
    await until(() => cancelled === 3);
  } finally {
    release.resolve();
  }
});

Deno.test("losing the worker fails a streaming response rather than silently ending it", async () => {
  await using f = await fixture(() =>
    new Response(
      new ReadableStream({
        start(controller) {
          controller.enqueue(new TextEncoder().encode("first"));
        },
      }),
    )
  );
  const response = await f.env.fetch(f.url);
  const reader = response.body!.getReader();
  await reader.read();
  const reading = assertRejects(() => reader.read(), Error, "lost");
  await f.connection.close();
  await reading;
});

Deno.test("remote fetch handles bodyless responses and reports network failures", async () => {
  await using f = await fixture(() => new Response(null, { status: 204 }));
  const response = await f.env.fetch(f.url);
  assertEquals(response.status, 204);
  assertEquals(response.body, null);
  const failed = await assertRejects(() => f.env.fetch("http://127.0.0.1:1"), TypeError);
  assertStringIncludes(failed.message, "fetch-test");
});