Build checked portable package artifacts and shared runtime assets

2026-09-20 00:25:53 -07:00
parent 0c3c9c6ad8
commit f352ac0c3b
5 changed files with 474 additions and 0 deletions
@@ -0,0 +1,151 @@
+/** Generated-service transport. The supervisor owns process isolation, deadlines and termination.
+ * stdout is reserved for this protocol; application diagnostics belong on stderr. */
+export function servePortableRegistry(
+  registry,
+  { maxBytes = 4 * 1024 * 1024, maxInvocations = 64, maxPortCalls = 10000 } = {},
+) {
+  const abi = "quixos-transaction-ports-v1";
+  const active = new Map();
+  const pending = new Map();
+  let buffer = Buffer.alloc(0),
+    stopped = false;
+  const send = (message) => {
+    const encoded = JSON.stringify(message);
+    if (Buffer.byteLength(encoded) > maxBytes) throw Error("SERVICE_MESSAGE_LIMIT");
+    // A slow or absent coordinator must not cause an unbounded stdout queue.
+    if (process.stdout.writableLength + Buffer.byteLength(encoded) > maxBytes * 2) throw Error("SERVICE_BACKPRESSURE");
+    process.stdout.write(encoded + "\n");
+  };
+  const errorBody = (error) => ({
+    code: typeof error?.code === "string" ? error.code.slice(0, 1024) : "PACKAGE_ERROR",
+    message: String(error?.message ?? error).slice(0, 4096),
+  });
+  const cancel = (id, reason) => {
+    active.delete(id);
+    for (const [key, call] of pending)
+      if (call.invocation === id) {
+        pending.delete(key);
+        call.reject(reason);
+      }
+  };
+  const fatal = (error) => {
+    if (stopped) return;
+    stopped = true;
+    for (const id of active.keys()) cancel(id, error);
+    process.stderr.write(String(error?.message ?? error) + "\n");
+    process.exitCode = 1;
+    process.stdin.destroy();
+  };
+  const receive = (message) => {
+    if (!message || message.abi !== abi) throw Error("SERVICE_ABI_MISMATCH");
+    if (message.kind === "shutdown") {
+      for (const id of active.keys()) cancel(id, Error("SERVICE_SHUTDOWN"));
+      stopped = true;
+      process.stdin.destroy();
+      return;
+    }
+    if (message.kind === "cancel") {
+      cancel(message.invocation, Error("INVOCATION_CANCELLED"));
+      return;
+    }
+    if (message.kind === "port-result") {
+      const key = JSON.stringify([message.invocation, message.sequence]);
+      const call = pending.get(key);
+      if (!call && !active.has(message.invocation)) return; // A canceled call may already have a reply in flight.
+      if (!call || message.frame !== call.frame) throw Error("UNEXPECTED_PORT_RESULT");
+      pending.delete(key);
+      if (message.status === "returned") call.resolve(message.output);
+      else if (message.status === "failed")
+        call.reject(
+          Object.assign(Error(String(message.error?.message ?? "Port failed")), {
+            code: String(message.error?.code ?? "PORT_FAILED"),
+          }),
+        );
+      else throw Error("INVALID_PORT_RESULT");
+      return;
+    }
+    if (
+      message.kind !== "invoke" ||
+      typeof message.invocation !== "string" ||
+      typeof message.frame !== "string" ||
+      !message.invocation ||
+      !message.frame ||
+      message.invocation.length > 4096 ||
+      message.frame.length > 4096
+    )
+      throw Error("INVALID_INVOCATION");
+    if (active.has(message.invocation) || active.size >= maxInvocations) throw Error("INVOCATION_LIMIT_OR_DUPLICATE");
+    const token = {};
+    active.set(message.invocation, token);
+    let sequence = 0;
+    const channel = (request) => {
+      if (active.get(message.invocation) !== token) return Promise.reject(Error("EXPIRED_INVOCATION"));
+      if (++sequence > maxPortCalls || pending.size >= maxPortCalls) return Promise.reject(Error("PORT_CALL_LIMIT"));
+      const key = JSON.stringify([message.invocation, sequence]);
+      return new Promise((resolve, reject) => {
+        pending.set(key, { invocation: message.invocation, frame: message.frame, resolve, reject });
+        try {
+          send({
+            abi,
+            kind: "port",
+            frame: message.frame,
+            invocation: message.invocation,
+            sequence,
+            portId: request.portId,
+            action: { ...request.action, ...(request.key === undefined ? {} : { key: request.key }) },
+            input: request.input,
+          });
+        } catch (error) {
+          pending.delete(key);
+          reject(error);
+        }
+      });
+    };
+    void registry
+      .invoke(
+        { exportId: message.exportId, workspace: message.workspace, receiver: message.receiver, input: message.input },
+        channel,
+      )
+      .then(
+        (output) => {
+          if (active.get(message.invocation) === token)
+            send({ abi, invocation: message.invocation, status: "returned", output });
+        },
+        (error) => {
+          if (active.get(message.invocation) === token)
+            send({ abi, invocation: message.invocation, status: "failed", error: errorBody(error) });
+        },
+      )
+      .catch(fatal)
+      .finally(() => {
+        if (active.get(message.invocation) === token) cancel(message.invocation, Error("INVOCATION_ENDED"));
+      });
+  };
+  process.stdin.on("data", (bytes) => {
+    if (stopped) return;
+    try {
+      // Consume each line before retaining the remainder; combined transport chunks may contain many frames.
+      let offset = 0;
+      while (offset < bytes.length) {
+        const newline = bytes.indexOf(10, offset);
+        const end = newline < 0 ? bytes.length : newline;
+        if (buffer.length + end - offset > maxBytes) throw Error("SERVICE_MESSAGE_LIMIT");
+        buffer = Buffer.concat([buffer, bytes.subarray(offset, end)]);
+        if (newline < 0) break;
+        receive(JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(buffer)));
+        buffer = Buffer.alloc(0);
+        offset = newline + 1;
+        if (stopped) break;
+      }
+    } catch (error) {
+      fatal(error);
+    }
+  });
+  process.stdin.on("end", () => {
+    if (buffer.length) fatal(Error("TRUNCATED_SERVICE_MESSAGE"));
+    for (const id of active.keys()) cancel(id, Error("COORDINATOR_DISCONNECTED"));
+  });
+  process.stdin.on("error", fatal);
+  process.stdout.on("error", fatal);
+  send({ abi, kind: "ready", exports: registry.exports });
+}