Preserve Deno cancellation across streamed HTTP bodies

cli/proxy-main.ts

@@ -34,6 +34,7 @@ import "../extensions/ext-observability-opentelemetry/src/index.ts";
 import "../extensions/ext-observability-sentry/src/index.ts";
 import "../extensions/ext-redis/src/index.ts";
 import "../extensions/ext-schema-zod/src/index.ts";
+import "../extensions/ext-node-websocket-ws/src/index.ts";
 
 setLoggerPreset("cli");
 

deno.json

@@ -521,7 +521,7 @@
     "generate:force": "deno run -A scripts/build/run-generate.ts --force",
     "generate:dev-ui": "deno run -A extensions/ext-dev-ui-react/scripts/generate-styles.ts && deno run -A extensions/ext-dev-ui-react/scripts/prebundle.ts",
     "generate:dev-ui:check": "deno run -A extensions/ext-dev-ui-react/scripts/generate-styles.ts --check && deno run -A extensions/ext-dev-ui-react/scripts/prebundle.ts --check",
-    "generate:manifests:check": "deno run --frozen -A extensions/ext-yaml/scripts/prebundle.ts --check && deno run -A scripts/build/generate-embedded-npm-packages.ts --check && deno run -A scripts/build/generate-templates-manifest.ts --check && deno task generate:dev-ui:check && deno run -A scripts/build/prebundle-hydration-runtime.ts --check && deno run -A scripts/build/prebundle-client-scripts.ts --check && deno run -A scripts/build/prebundle-bridge.ts --check && deno run -A scripts/build/prebundle-rsc-scripts.ts --check && deno run -A scripts/build/generate-core-skills.ts --check",
+    "generate:manifests:check": "deno run --frozen -A extensions/ext-node-websocket-ws/scripts/prebundle.ts --check && deno run --frozen -A extensions/ext-yaml/scripts/prebundle.ts --check && deno run -A scripts/build/generate-embedded-npm-packages.ts --check && deno run -A scripts/build/generate-templates-manifest.ts --check && deno task generate:dev-ui:check && deno run -A scripts/build/prebundle-hydration-runtime.ts --check && deno run -A scripts/build/prebundle-client-scripts.ts --check && deno run -A scripts/build/prebundle-bridge.ts --check && deno run -A scripts/build/prebundle-rsc-scripts.ts --check && deno run -A scripts/build/generate-core-skills.ts --check",
     "start": "deno task generate && deno run --allow-read --allow-write --allow-net --allow-env --allow-run --allow-sys --unstable-worker-options --unstable-net cli/main.ts",
     "start:headless": "deno task generate && deno run --allow-read --allow-write --allow-net --allow-env --allow-run --allow-sys --unstable-worker-options --unstable-net cli/main.ts --headless",
     "proxy": "deno task generate && deno run --allow-read --allow-write --allow-net --allow-env --allow-run --allow-sys --unstable-worker-options --unstable-net cli/main.ts serve --mode=proxy",
@@ -554,7 +554,7 @@
     "coverage:gate": "deno coverage coverage --include=src/ --include=cli/ --exclude=/tests/ --exclude=/__tests__/ --lcov > coverage/lcov.info && deno run --allow-read scripts/lint/check-coverage.ts 80",
     "coverage:ci:shard": "deno run --config=scripts/test.deno.json --allow-read --allow-write --allow-run --allow-env scripts/test/coverage-ci.ts shard",
     "coverage:ci:merge": "deno run --config=scripts/test.deno.json --no-npm --allow-read --allow-write --allow-run --allow-env scripts/test/coverage-ci.ts merge",
-    "coverage:ci:integration-client": "rm -rf coverage/integration-client && VF_RUNS_CLI_COVERAGE_DIR=coverage/integration-client deno task test:file --coverage=coverage/integration-client tests/integration/integrations/ tests/integration/cli/integration/ tests/integration/cli/project/ tests/integration/runs/ && deno run --config=scripts/test.deno.json --no-npm --allow-read --allow-write --allow-run --allow-env scripts/test/coverage-ci.ts report coverage/integration-client",
+    "coverage:ci:integration-client": "rm -rf coverage/integration-client && VF_RUNS_CLI_COVERAGE_DIR=coverage/integration-client deno task test:file --coverage=coverage/integration-client tests/integration/integrations/ tests/integration/cli/integration/ tests/integration/cli/project/ tests/integration/runs/ tests/integration/core/deno-http-cancellation.test.ts tests/integration/core/deno-http2-transport.test.ts tests/integration/core/deno-node-http-upgrade.test.ts tests/integration/core/deno-websocket-dom.test.ts && deno run --config=scripts/test.deno.json --no-npm --allow-read --allow-write --allow-run --allow-env scripts/test/coverage-ci.ts report coverage/integration-client",
     "coverage:ci:node-executor": "node scripts/test/coverage-node-executor.mjs",
     "coverage:ci": "deno task test:coverage:unit && deno task coverage:gate",
     "coverage:html": "deno coverage coverage --include=src/ --include=cli/ --exclude=/tests/ --exclude=/__tests__/ --html",

extensions/ext-node-websocket-ws/deno.json

@@ -1,7 +1,10 @@
 {
   "name": "@veryfront/ext-node-websocket-ws",
   "version": "0.1.0",
-  "exports": "./src/index.ts",
+  "exports": {
+    ".": "./src/index.ts",
+    "./portable": "./src/portable.ts"
+  },
   "veryfront": {
     "extension": true,
     "activation": "auto",
@@ -29,6 +32,8 @@
   },
   "tasks": {
     "check": "deno check src/index.ts src/*.test.ts",
-    "test": "deno test --allow-env=WS_NO_BUFFER_UTIL,WS_NO_UTF_8_VALIDATE --allow-read=deno.json src/"
+    "test": "deno test --allow-env=WS_NO_BUFFER_UTIL,WS_NO_UTF_8_VALIDATE --allow-read=deno.json src/",
+    "generate": "deno run --frozen -A scripts/prebundle.ts",
+    "generate:check": "deno run --frozen -A scripts/prebundle.ts --check"
   }
 }

extensions/ext-node-websocket-ws/scripts/prebundle.ts

@@ -0,0 +1,152 @@
+#!/usr/bin/env -S deno run --frozen -A
+/** Bundle the pinned ws implementation with optional native accelerators disabled. */
+import { fileURLToPath } from "node:url";
+import { build, stop } from "npm:esbuild@0.28.1";
+import sourceIntegrity from "./ws-source-integrity.json" with { type: "json" };
+const checkOnly = Deno.args.length === 1 && Deno.args[0] === "--check";
+if (Deno.args.length !== (checkOnly ? 1 : 0)) throw new TypeError("Usage: prebundle.ts [--check]");
+const root = new URL("../../../", import.meta.url);
+const resolved = new URL(import.meta.resolve("ws"));
+if (resolved.protocol !== "file:") {
+  throw new TypeError("Expected a locally resolved frozen WebSocket package");
+}
+const packageRoot = new URL("./", resolved);
+async function captureSourceGraph(): Promise<
+  { digest: string; license: string; manifest: string }
+> {
+  const files = new Map<string, Uint8Array<ArrayBuffer>>();
+  const readFile = async (relative: string): Promise<void> => {
+    const url = new URL(relative, packageRoot);
+    if (!(await Deno.lstat(url)).isFile) {
+      throw new Error("WebSocket source inputs must be regular files");
+    }
+    files.set(relative, await Deno.readFile(url));
+  };
+  const walk = async (relative: string): Promise<void> => {
+    for await (const entry of Deno.readDir(new URL(relative, packageRoot))) {
+      const path = relative + entry.name;
+      if (entry.isDirectory) await walk(path + "/");
+      else if (entry.isFile) await readFile(path);
+      else throw new Error("WebSocket source inputs must not contain symbolic links");
+    }
+  };
+  await readFile("package.json");
+  await readFile("LICENSE");
+  await readFile("index.js");
+  await readFile("wrapper.mjs");
+  await walk("lib/");
+  const rows: string[] = [];
+  // Integrity rows use code-unit ordering, independent of the host locale.
+  const paths = [...files.keys()].sort((left, right) => left < right ? -1 : left > right ? 1 : 0);
+  for (const path of paths) {
+    const bytes = files.get(path)!;
+    const digest = new Uint8Array(await crypto.subtle.digest("SHA-256", bytes)).toHex();
+    rows.push(path + String.fromCharCode(0) + digest + String.fromCharCode(10));
+  }
+  if (files.size !== sourceIntegrity.sourceFileCount) {
+    throw new Error("WebSocket source file count differs from the qualified package");
+  }
+  const digest = new Uint8Array(
+    await crypto.subtle.digest("SHA-256", new TextEncoder().encode(rows.join(""))),
+  ).toHex();
+  return {
+    digest,
+    license: new TextDecoder("utf-8", { fatal: true }).decode(files.get("LICENSE")!),
+    manifest: new TextDecoder("utf-8", { fatal: true }).decode(files.get("package.json")!),
+  };
+}
+const source = await captureSourceGraph();
+if (source.digest !== sourceIntegrity.sourceGraphSha256) {
+  throw new Error("WebSocket source differs from the qualified published archive");
+}
+const manifest = JSON.parse(source.manifest) as {
+  name: string;
+  version: string;
+  main: string;
+};
+if (
+  manifest.name !== sourceIntegrity.package || manifest.version !== sourceIntegrity.version ||
+  manifest.main !== "index.js"
+) {
+  throw new Error("The WebSocket package does not expose the qualified entry");
+}
+const lock = JSON.parse(await Deno.readTextFile(new URL("deno.lock", root))) as {
+  npm: Record<string, { integrity: string }>;
+};
+const identity = `${manifest.name}@${manifest.version}`;
+const integrity = lock.npm[identity]?.integrity;
+if (integrity !== sourceIntegrity.integrity) {
+  throw new Error("WebSocket repository lock does not match the qualified package integrity");
+}
+const entry = new URL("wrapper.mjs", packageRoot);
+let bundle: string;
+try {
+  const result = await build({
+    entryPoints: [fileURLToPath(entry)],
+    bundle: true,
+    platform: "node",
+    format: "esm",
+    target: "es2022",
+    minify: true,
+    write: false,
+    legalComments: "inline",
+    define: {
+      "process.env.WS_NO_BUFFER_UTIL": "true",
+      "process.env.WS_NO_UTF_8_VALIDATE": "true",
+    },
+    banner: {
+      js:
+        'import { createRequire } from "node:module"; const require = createRequire(import.meta.url);',
+    },
+  });
+  if (result.outputFiles.length !== 1) throw new Error("Expected one WebSocket runtime bundle");
+  bundle = result.outputFiles[0]!.text;
+} finally {
+  await stop();
+}
+const bundleBytes = new TextEncoder().encode(bundle);
+if (bundle.trim().length === 0 || bundleBytes.length > 128 * 1024) {
+  throw new RangeError("WebSocket runtime must be nonempty and at most 128 KiB");
+}
+if (/process\.env|Deno\.env|bufferutil|utf-8-validate/.test(bundle)) {
+  throw new Error("WebSocket runtime must not access host env or optional native modules");
+}
+const digest = new Uint8Array(await crypto.subtle.digest("SHA-256", bundleBytes)).toHex();
+if ((await captureSourceGraph()).digest !== source.digest) {
+  throw new Error("WebSocket source changed during bundling");
+}
+const license = source.license;
+if (license.includes("*/")) throw new Error("Unexpected license comment delimiter");
+const generated =
+  `// deno-fmt-ignore-file\n// deno-lint-ignore-file -- Verified generated upstream WebSocket bundle.\n/**\n * Official ${identity}, with optional native accelerators disabled at build time.\n * Package integrity: ${integrity}\n * Upstream source SHA-256: ${source.digest}
+ * Bundle SHA-256: ${digest}\n * Generated by extensions/ext-node-websocket-ws/scripts/prebundle.ts.\n *\n${
+    license.split("\n").map((line) => line.length > 0 ? " * " + line.trimEnd() : " *").join("\n")
+  }\n */\n${bundle}`;
+const output = new URL("../src/portable-runtime.generated.js", import.meta.url);
+if (checkOnly) {
+  if (await Deno.readTextFile(output) !== generated) {
+    throw new Error("WebSocket portable bundle is stale; run its prebundle generator");
+  }
+} else {
+  const temporary = await Deno.makeTempFile({
+    dir: fileURLToPath(new URL("../src/", import.meta.url)),
+    prefix: ".websocket-runtime-",
+    suffix: ".tmp",
+  });
+  try {
+    await Deno.writeTextFile(temporary, generated);
+    await Deno.rename(temporary, output);
+  } catch (error) {
+    try {
+      await Deno.remove(temporary);
+    } catch (cleanupError) {
+      if (!(cleanupError instanceof Deno.errors.NotFound)) {
+        throw new AggregateError([error, cleanupError], "WebSocket generation and cleanup failed");
+      }
+    }
+    throw error;
+  }
+  console.log(
+    `WebSocket portable runtime ${identity}: ${bundleBytes.length} bytes, sha256:${digest}`,
+  );
+}

extensions/ext-node-websocket-ws/scripts/ws-source-integrity.json

@@ -0,0 +1,7 @@
+{
+  "package": "ws",
+  "version": "8.21.1",
+  "integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==",
+  "sourceGraphSha256": "42934414004ba40e5be9f6b82aed4870b363e493bbbf5655d057c08bb02ac94e",
+  "sourceFileCount": 17
+}

extensions/ext-node-websocket-ws/src/index.ts

@@ -1,17 +1,12 @@
 /** Default `ws` implementation of Veryfront's Node WebSocket contract. */
 
 import type { ExtensionFactory } from "veryfront/extensions";
-import {
-  captureNodeWebSocketServer,
-  createNodeWebSocketServerProvider,
-  NodeWebSocketServerProviderName,
-} from "veryfront/extensions/websocket";
+import { NodeWebSocketServerProviderName } from "veryfront/extensions/websocket";
 // @ts-types="@types/ws"
-import { WebSocketServer } from "ws";
+import { WebSocket, WebSocketServer } from "ws";
+import { createWsProvider } from "./provider.ts";
 
-export const WsNodeWebSocketServerProvider = createNodeWebSocketServerProvider(
-  (options) => captureNodeWebSocketServer(new WebSocketServer(options)),
-);
+export const WsNodeWebSocketServerProvider = createWsProvider(WebSocketServer, WebSocket);
 
 /** Create the standard Node.js `ws` transport extension. */
 export const extNodeWebSocketWs: ExtensionFactory = () => {

extensions/ext-node-websocket-ws/src/package-boundary.test.ts

@@ -18,6 +18,10 @@ Deno.test("ws extension is auto-activated and owns every third-party dependency"
   }]);
   assertEquals(manifest.imports.ws, "npm:ws@8.21.1");
   assertEquals(manifest.imports["@types/ws"], "npm:@types/ws@8.18.1");
+  assertEquals(manifest.exports, {
+    ".": "./src/index.ts",
+    "./portable": "./src/portable.ts",
+  });
 });
 
 Deno.test("ws contracts stay behind the dedicated Node-only package subpath", () => {

extensions/ext-node-websocket-ws/src/portable.ts

@@ -0,0 +1,6 @@
+/** Pinned pure-JavaScript ws transport with no host environment or native addon reads. */
+// @ts-types="@types/ws"
+import { WebSocket, WebSocketServer } from "./portable-runtime.generated.js";
+import { createWsProvider } from "./provider.ts";
+
+export const PortableWsNodeWebSocketServerProvider = createWsProvider(WebSocketServer, WebSocket);

extensions/ext-node-websocket-ws/src/provider.ts

@@ -0,0 +1,23 @@
+// @ts-types="@types/ws"
+import type { WebSocket, WebSocketServer } from "ws";
+import {
+  captureNodeWebSocketServer,
+  createNodeWebSocketServerProvider,
+} from "veryfront/extensions/websocket";
+
+/** Preserve the transport contract for both native-optimized and portable ws. */
+export function createWsProvider(
+  Server: typeof WebSocketServer,
+  Connection: typeof WebSocket,
+) {
+  return createNodeWebSocketServerProvider(
+    (options) => captureNodeWebSocketServer(new Server(options)),
+    {
+      supportsIdleTimeout: true,
+      getBufferedAmount(connection) {
+        if (!(connection instanceof Connection)) throw new TypeError("Unknown ws connection");
+        return connection.bufferedAmount;
+      },
+    },
+  );
+}

scripts/build/run-generate.test.ts

@@ -86,6 +86,7 @@ describe("UNITS", () => {
       "dev-ui",
       "embedded-npm-packages",
       "hydration-runtime",
+      "portable-websocket-runtime",
       "rsc-scripts",
       "templates-manifest",
       "yaml-runtime",

scripts/build/run-generate.ts

@@ -44,6 +44,13 @@ export interface GeneratorUnit {
 }
 
 export const UNITS: GeneratorUnit[] = [
+  {
+    name: "portable-websocket-runtime",
+    commands: [["deno", "run", "--frozen", "-A", "extensions/ext-node-websocket-ws/scripts/prebundle.ts"]],
+    inputRoots: ["extensions/ext-node-websocket-ws"],
+    inputFiles: ["deno.lock", "deno.json"],
+    outputs: ["extensions/ext-node-websocket-ws/src/portable-runtime.generated.js"],
+  },
   {
     name: "yaml-runtime",
     commands: [["deno", "run", "--frozen", "-A", "extensions/ext-yaml/scripts/prebundle.ts"]],