quixos/quixos-nix-helpers
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add cooperative cancellation for asynchronous package invocations

Browse Source
This commit is contained in:
Timothy J. Aveni
2026-09-20 19:58:05 -07:00
parent f352ac0c3b
commit 9257198eec
1 changed files with 29 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
portable-service.mjs
+29 -9
View File
@@ -20,14 +20,21 @@ export function servePortableRegistry(
code: typeof error?.code === "string" ? error.code.slice(0, 1024) : "PACKAGE_ERROR", code: typeof error?.code === "string" ? error.code.slice(0, 1024) : "PACKAGE_ERROR",
message: String(error?.message ?? error).slice(0, 4096), message: String(error?.message ?? error).slice(0, 4096),
}); });
const cancel = (id, reason) => { const rejectPorts = (id, reason) => {
active.delete(id);
for (const [key, call] of pending) for (const [key, call] of pending)
if (call.invocation === id) { if (call.invocation === id) {
pending.delete(key); pending.delete(key);
call.reject(reason); 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) => { const fatal = (error) => {
if (stopped) return; if (stopped) return;
stopped = true; stopped = true;
@@ -45,13 +52,17 @@ export function servePortableRegistry(
return; return;
} }
if (message.kind === "cancel") { 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")); 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; return;
} }
if (message.kind === "port-result") { if (message.kind === "port-result") {
const key = JSON.stringify([message.invocation, message.sequence]); const key = JSON.stringify([message.invocation, message.sequence]);
const call = pending.get(key); const call = pending.get(key);
if (!call && !active.has(message.invocation)) return; // A canceled call may already have a reply in flight. if (!call && (!active.has(message.invocation) || active.get(message.invocation).cancelled)) return;
if (!call || message.frame !== call.frame) throw Error("UNEXPECTED_PORT_RESULT"); if (!call || message.frame !== call.frame) throw Error("UNEXPECTED_PORT_RESULT");
pending.delete(key); pending.delete(key);
if (message.status === "returned") call.resolve(message.output); if (message.status === "returned") call.resolve(message.output);
@@ -75,11 +86,12 @@ export function servePortableRegistry(
) )
throw Error("INVALID_INVOCATION"); throw Error("INVALID_INVOCATION");
if (active.has(message.invocation) || active.size >= maxInvocations) throw Error("INVOCATION_LIMIT_OR_DUPLICATE"); if (active.has(message.invocation) || active.size >= maxInvocations) throw Error("INVOCATION_LIMIT_OR_DUPLICATE");
const token = {}; const token = { frame: message.frame, controller: new AbortController(), cancelled: false };
active.set(message.invocation, token); active.set(message.invocation, token);
let sequence = 0; let sequence = 0;
const channel = (request) => { const channel = (request) => {
if (active.get(message.invocation) !== token) return Promise.reject(Error("EXPIRED_INVOCATION")); 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")); if (++sequence > maxPortCalls || pending.size >= maxPortCalls) return Promise.reject(Error("PORT_CALL_LIMIT"));
const key = JSON.stringify([message.invocation, sequence]); const key = JSON.stringify([message.invocation, sequence]);
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
@@ -105,21 +117,29 @@ export function servePortableRegistry(
.invoke( .invoke(
{ exportId: message.exportId, workspace: message.workspace, receiver: message.receiver, input: message.input }, { exportId: message.exportId, workspace: message.workspace, receiver: message.receiver, input: message.input },
channel, channel,
Object.freeze({ signal: token.controller.signal, effect: message.effect }),
) )
.then( .then(
(output) => { (output) => {
if (active.get(message.invocation) === token) if (active.get(message.invocation) === token && !token.cancelled)
send({ abi, invocation: message.invocation, status: "returned", output }); send({ abi, invocation: message.invocation, status: "returned", output });
}, },
(error) => { (error) => {
if (active.get(message.invocation) === token) if (active.get(message.invocation) === token && !token.cancelled)
send({ abi, invocation: message.invocation, status: "failed", error: errorBody(error) }); send({ abi, invocation: message.invocation, status: "failed", error: errorBody(error) });
}, },
) )
.catch(fatal) .catch(fatal)
.finally(() => { .finally(() => {
if (active.get(message.invocation) === token) cancel(message.invocation, Error("INVOCATION_ENDED")); 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) => { process.stdin.on("data", (bytes) => {
if (stopped) return; if (stopped) return;