fix: Force terminate the deno subprocess and handle scenarios when trying to terminate the worker before it has been setup

Author: KallynGowdy

package-lock.json

@@ -53,7 +53,8 @@
         "tslib": "2.0.0",
         "gulp": "^4.0.0",
         "del": "^3.0.0",
-        "typedoc": "^0.17.8"
+        "typedoc": "^0.17.8",
+        "ps-list": "7.2.0"
     },
     "dependencies": {
         "ws": "^7.3.1",

package.json

@@ -3,6 +3,7 @@ import { readFileSync } from 'fs';
 import path from 'path';
 import { URL } from 'url';
 import { MessageChannel, MessagePort } from './MessageChannel';
+import psList from 'ps-list';
 
 console.log = jest.fn();
 jest.setTimeout(10000);
@@ -13,6 +14,8 @@ describe('DenoWorker', () => {
     const echoScript = readFileSync(echoFile, { encoding: 'utf-8' });
     const pingFile = path.resolve(__dirname, './test/ping.js');
     const pingScript = readFileSync(pingFile, { encoding: 'utf-8' });
+    const infiniteFile = path.resolve(__dirname, './test/infinite.js');
+    const infiniteScript = readFileSync(infiniteFile, { encoding: 'utf-8' });
 
     afterEach(() => {
         if (worker) {
@@ -285,4 +288,158 @@ describe('DenoWorker', () => {
             expect(final).toEqual('pong');
         });
     });
+
+    describe('terminate()', () => {
+        it('should kill the deno process when terminated immediately', async () => {
+            let denoProcesses = await getDenoProcesses();
+            expect(denoProcesses).toEqual([]);
+
+            worker = new DenoWorker(echoScript, {
+                permissions: {
+                    allowNet: [`https://google.com`],
+                },
+            });
+            worker.terminate();
+
+            denoProcesses = await getDenoProcesses();
+            expect(denoProcesses).toEqual([]);
+        });
+
+        it('should kill the deno process when terminated after the initial connection', async () => {
+            let denoProcesses = await getDenoProcesses();
+            expect(denoProcesses).toEqual([]);
+
+            worker = new DenoWorker(echoScript, {
+                permissions: {
+                    allowNet: [`https://google.com`],
+                },
+            });
+
+            let ret: any;
+            let resolve: any;
+            let promise = new Promise((res, rej) => {
+                resolve = res;
+            });
+            worker.onmessage = (e) => {
+                ret = e.data;
+                resolve();
+            };
+
+            worker.postMessage({
+                type: 'echo',
+                message: 'Hello',
+            });
+
+            await promise;
+
+            worker.terminate();
+
+            denoProcesses = await getDenoProcesses();
+            expect(denoProcesses).toEqual([]);
+        });
+
+        it('should kill the deno process when terminated while sending data', async () => {
+            let denoProcesses = await getDenoProcesses();
+            expect(denoProcesses).toEqual([]);
+
+            worker = new DenoWorker(echoScript, {
+                permissions: {
+                    allowNet: [`https://google.com`],
+                },
+            });
+
+            let ret: any;
+            let resolve: any;
+            let promise = new Promise((res, rej) => {
+                resolve = res;
+            });
+            worker.onmessage = (e) => {
+                ret = e.data;
+                resolve();
+            };
+
+            worker.terminate();
+
+            worker.postMessage({
+                type: 'echo',
+                message: 'Hello',
+            });
+
+            denoProcesses = await getDenoProcesses();
+            expect(denoProcesses).toEqual([]);
+        });
+
+        it('should kill the deno process when terminated while recieving data', async () => {
+            let denoProcesses = await getDenoProcesses();
+            expect(denoProcesses).toEqual([]);
+
+            worker = new DenoWorker(echoScript, {
+                permissions: {
+                    allowNet: [`https://google.com`],
+                },
+            });
+
+            let ret: any;
+            let resolve: any;
+            let promise = new Promise((res, rej) => {
+                resolve = res;
+            });
+            worker.onmessage = (e) => {
+                ret = e.data;
+                console.log('Message');
+                resolve();
+            };
+
+            worker.postMessage({
+                type: 'echo',
+                message: 'Hello',
+            });
+
+            await Promise.resolve();
+
+            worker.terminate();
+
+            denoProcesses = await getDenoProcesses();
+            expect(denoProcesses).toEqual([]);
+        });
+
+        it('should kill the deno process when terminated while the script is in an infinite loop', async () => {
+            let denoProcesses = await getDenoProcesses();
+            expect(denoProcesses).toEqual([]);
+
+            worker = new DenoWorker(infiniteScript, {
+                permissions: {
+                    allowNet: [`https://google.com`],
+                },
+            });
+
+            let ret: any;
+            let resolve: any;
+            let promise = new Promise((res, rej) => {
+                resolve = res;
+            });
+            worker.onmessage = (e) => {
+                ret = e.data;
+                resolve();
+            };
+
+            await promise;
+
+            worker.postMessage({
+                type: 'echo',
+                message: 'Hello',
+            });
+
+            worker.terminate();
+
+            denoProcesses = await getDenoProcesses();
+            expect(denoProcesses).toEqual([]);
+        });
+    });
 });
+
+async function getDenoProcesses() {
+    const list = await psList();
+    const denoProcesses = list.filter((p) => /^deno/.test(p.name));
+    return denoProcesses;
+}

src/DenoWorker.spec.ts

@@ -16,6 +16,7 @@ import {
 } from './MessageTarget';
 import { MessagePort } from './MessageChannel';
 import { Stream, Readable, Duplex } from 'stream';
+import { forceKill } from './Utils';
 
 const DEFAULT_DENO_BOOTSTRAP_SCRIPT_PATH = __dirname.endsWith('src')
     ? resolve(__dirname, '../deno/index.ts')
@@ -217,6 +218,7 @@ export class DenoWorker {
 
         this._httpServer.listen({ host: '127.0.0.1', port: 0 }, () => {
             if (this._terminated) {
+                this._httpServer.close();
                 return;
             }
             const addr = this._httpServer.address();
@@ -240,6 +242,7 @@ export class DenoWorker {
             let runArgs = [] as string[];
 
             addOption(runArgs, '--reload', this._options.reload);
+            // addOption(runArgs, '--unstable', true);
 
             if (this._options.permissions) {
                 addOption(
@@ -342,12 +345,12 @@ export class DenoWorker {
     terminate() {
         this._terminated = true;
         if (this._process) {
-            this._process.kill();
+            // this._process.kill();
+            forceKill(this._process.pid);
             this._process = null;
         }
         if (this._httpServer) {
             this._httpServer.close();
-            this._httpServer = null;
         }
         if (this._server) {
             this._server.close();
@@ -387,6 +390,9 @@ export class DenoWorker {
         data: any,
         transfer?: Transferrable[]
     ) {
+        if (this._terminated) {
+            return;
+        }
         this._handleTransferrables(transfer);
         const structuredData = serializeStructure(data, transfer);
         if (channel !== null) {

src/DenoWorker.ts

@@ -1,4 +1,5 @@
 import { MessageChannel, MessagePort } from './MessageChannel';
+import { execSync } from 'child_process';
 
 export function polyfillMessageChannel() {
     const anyGlobalThis = globalThis as any;
@@ -7,3 +8,32 @@ export function polyfillMessageChannel() {
         anyGlobalThis.MessagePort = MessagePort;
     }
 }
+
+/**
+ * Forcefully kills the process with the given ID.
+ * On Linux/Unix, this means sending the process the SIGKILL signal.
+ * On Windows, this means using the taskkill executable to kill the process.
+ * @param pid The ID of the process to kill.
+ */
+export function forceKill(pid: number) {
+    const isWindows = /^win/.test(process.platform);
+    if (isWindows) {
+        return killWindows(pid);
+    } else {
+        return killUnix(pid);
+    }
+}
+
+function killWindows(pid: number) {
+    execSync(`taskkill /PID ${pid} /T /F`);
+}
+
+function killUnix(pid: number) {
+    const signal = 'SIGKILL';
+    process.kill(pid, signal);
+}
+
+export interface ExecResult {
+    stdout: string;
+    stdin: string;
+}

src/Utils.ts

@@ -0,0 +1,7 @@
+self.postMessage('test');
+
+self.onmessage = (e) => {
+    while (true) {
+        console.log('Running...');
+    }
+};