quixos-nix-helpers/portable-service.mjs

/** 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 rejectPorts = (id, reason) => {
    for (const [key, call] of pending)
      if (call.invocation === id) {
        pending.delete(key);
        call.reject(reason);
      }
  };
  const cancel = (id, reason) => {
    const token = active.get(id);
    if (token) {
      token.cancelled = true;
      token.controller.abort(reason);
    }
    rejectPorts(id, 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") {
      const token = active.get(message.invocation);
      if (token && token.frame !== message.frame) throw Error("INVALID_CANCELLATION_FRAME");
      cancel(message.invocation, Error("INVOCATION_CANCELLED"));
      // A completed result may already be in flight. Acknowledge without resurrecting its context.
      if (!token) send({ abi, kind: "cancelled", invocation: message.invocation, frame: message.frame });
      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) || active.get(message.invocation).cancelled)) return;
      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 = { frame: message.frame, controller: new AbortController(), cancelled: false };
    active.set(message.invocation, token);
    let sequence = 0;
    const channel = (request) => {
      if (active.get(message.invocation) !== token || token.cancelled)
        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,
        Object.freeze({ signal: token.controller.signal, effect: message.effect }),
      )
      .then(
        (output) => {
          if (active.get(message.invocation) === token && !token.cancelled)
            send({ abi, invocation: message.invocation, status: "returned", output });
        },
        (error) => {
          if (active.get(message.invocation) === token && !token.cancelled)
            send({ abi, invocation: message.invocation, status: "failed", error: errorBody(error) });
        },
      )
      .catch(fatal)
      .finally(() => {
        if (active.get(message.invocation) === token) {
          active.delete(message.invocation);
          token.controller.abort(Error("INVOCATION_ENDED"));
          rejectPorts(message.invocation, Error("INVOCATION_ENDED"));
          if (token.cancelled && !stopped)
            send({ abi, kind: "cancelled", invocation: message.invocation, frame: token.frame });
        }
      })
      .catch(fatal);
  };
  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 });
}