refactor(daemon): Client cancellation context
MODIFY
packages/daemon/index.js
packages/daemon/index.js
@@ -7,6 +7,7 @@ import fs from 'fs';
import path from 'path';
import { E } from '@endo/eventual-send';
+import { makePromiseKit } from '@endo/promise-kit';
import { whereEndoState, whereEndoSock, whereEndoCache } from '@endo/where';
import { makeEndoClient } from './src/client.js';
@@ -24,13 +25,16 @@ const endoDaemonPath = url.fileURLToPath(
);
export const terminate = async (locator = defaultLocator) => {
- const { getBootstrap, finalize } = await makeEndoClient(
+ const { resolve: cancel, promise: cancelled } = makePromiseKit();
+ const { getBootstrap, closed } = await makeEndoClient(
'harbinger',
locator.sockPath,
+ cancelled,
);
const bootstrap = getBootstrap();
await E(E.get(bootstrap).privateFacet).terminate();
- finalize();
+ cancel();
+ await closed;
};
export const start = async (locator = defaultLocator) => {

MODIFY
packages/daemon/src/client.js
packages/daemon/src/client.js
@@ -7,13 +7,14 @@ import { makeNodeNetstringCapTP } from './connection.js';
* @template TBootstrap
* @param {string} name
* @param {string} sockPath
+ * @param {Promise<void>} cancelled
* @param {TBootstrap=} bootstrap
*/
-export const makeEndoClient = async (name, sockPath, bootstrap) => {
+export const makeEndoClient = async (name, sockPath, cancelled, bootstrap) => {
const conn = net.connect(sockPath);
await new Promise((resolve, reject) => {
conn.on('connect', resolve);
conn.on('error', reject);
});
- return makeNodeNetstringCapTP(name, conn, conn, bootstrap);
+ return makeNodeNetstringCapTP(name, conn, conn, cancelled, bootstrap);
};

MODIFY
packages/daemon/src/connection.js
packages/daemon/src/connection.js
@@ -13,15 +13,15 @@ const textDecoder = new TextDecoder();
* @param {string} name
* @param {import('@endo/stream').Stream<unknown, any, unknown, unknown>} writer
* @param {import('@endo/stream').Stream<any, undefined, undefined, undefined>} reader
+ * @param {Promise<void>} cancelled
* @param {TBootstrap} bootstrap
*/
-const makeCapTPWithStreams = (name, writer, reader, bootstrap) => {
+const makeCapTPWithStreams = (name, writer, reader, cancelled, bootstrap) => {
/** @param {any} message */
const send = message => {
return writer.next(message);
};
- // TODO cancellation context
const { dispatch, getBootstrap, abort } = makeCapTP(name, send, bootstrap);
const drained = (async () => {
@@ -30,13 +30,14 @@ const makeCapTPWithStreams = (name, writer, reader, bootstrap) => {
}
})();
+ const closed = cancelled.finally(async () => {
+ abort();
+ await Promise.all([writer.return(undefined), drained]);
+ });
+
return {
getBootstrap,
- drained,
- finalize() {
- abort();
- return writer.return(undefined);
- },
+ closed,
};
};
@@ -59,12 +60,14 @@ const bytesToMessage = bytes => {
* @param {string} name
* @param {import('stream').Writable} nodeWriter
* @param {import('stream').Readable} nodeReader
+ * @param {Promise<void>} cancelled
* @param {TBootstrap} bootstrap
*/
export const makeNodeNetstringCapTP = (
name,
nodeWriter,
nodeReader,
+ cancelled,
bootstrap,
) => {
const writer = mapWriter(
@@ -75,5 +78,5 @@ export const makeNodeNetstringCapTP = (
makeNetstringReader(makeNodeReader(nodeReader)),
bytesToMessage,
);
- return makeCapTPWithStreams(name, writer, reader, bootstrap);
+ return makeCapTPWithStreams(name, writer, reader, cancelled, bootstrap);
};

MODIFY
packages/daemon/src/daemon.js
packages/daemon/src/daemon.js
@@ -53,10 +53,11 @@ const makeWorker = async locator => {
console.error(`Endo worker started PID ${child.pid} UUID ${uuid}`);
const stream = /** @type {import('stream').Duplex} */ (child.stdio[3]);
assert(stream);
- const { getBootstrap, drained, finalize } = makeNodeNetstringCapTP(
+ const { getBootstrap, closed } = makeNodeNetstringCapTP(
`Worker ${uuid}`,
stream,
stream,
+ cancelled,
undefined,
);
@@ -69,7 +70,7 @@ const makeWorker = async locator => {
});
});
- const terminated = Promise.all([exited, drained]);
+ const terminated = Promise.all([exited, closed]);
const { resolve: cancelWorker, promise: workerCancelled } = makePromiseKit();
@@ -168,8 +169,14 @@ export const main = async () => {
process.exit(-1);
});
server.on('connection', conn => {
- const { drained } = makeNodeNetstringCapTP('Endo', conn, conn, endoFacets);
- drained.catch(sinkError);
+ const { closed } = makeNodeNetstringCapTP(
+ 'Endo',
+ conn,
+ conn,
+ cancelled,
+ endoFacets,
+ );
+ closed.catch(sinkError);
});
cancelled.catch(() => {

MODIFY
packages/daemon/src/worker.js
packages/daemon/src/worker.js
@@ -53,19 +53,15 @@ export const main = async () => {
const workerFacet = makeWorkerFacet();
- const { drained, finalize } = makeNodeNetstringCapTP(
+ const { closed } = makeNodeNetstringCapTP(
'Endo',
writer,
reader,
+ cancelled,
workerFacet,
);
- cancelled.catch(async () => {
- finalize();
- writer.close();
- });
-
- drained.catch(sinkError);
+ closed.catch(sinkError);
};
main().catch(sinkError);

MODIFY
packages/daemon/test/test-endo.js
packages/daemon/test/test-endo.js
@@ -11,6 +11,7 @@ import test from 'ava';
import url from 'url';
import path from 'path';
import { E } from '@endo/far';
+import { makePromiseKit } from '@endo/promise-kit';
import { start, stop, restart, clean, makeEndoClient } from '../index.js';
const { raw } = String;
@@ -27,20 +28,23 @@ const locator = {
};
test.serial('lifecycle', async t => {
+ const { resolve: cancel, promise: cancelled } = makePromiseKit();
+
await clean(locator);
await start(locator);
await stop(locator);
await restart(locator);
- const { getBootstrap, finalize, drained } = await makeEndoClient(
+ const { getBootstrap, closed } = await makeEndoClient(
'client',
locator.sockPath,
+ cancelled,
);
const bootstrap = getBootstrap();
const worker = await E(E.get(bootstrap).privateFacet).makeWorker();
await E(E.get(worker).actions).terminate();
- finalize();
- await drained;
+ cancel();
+ await closed;
await stop(locator);
t.pass();

refactor(daemon): Client cancellation context