fix(engine): report worker crashes correctly (#43)

* fix: update typescript for Promise related errors

* test: add failing test case for #43

* fix(engine): report worker crashes correctly

- Caused by previous result streaming feature
- Reported by latest Typescript

lib/engine/engine.ts

@@ -1,7 +1,6 @@
 import { Worker, isMainThread } from 'worker_threads';
 
 import workerTask, { WorkerMessage, createErrorMessage } from './worker-task';
-import { LintMessage } from './types';
 import { resolveConfigurationLocation } from '@config';
 
 type WorkerCallback<T> = (worker: Worker) => T;
@@ -20,7 +19,7 @@ function scanRepository(
     repository: string,
     onMessage: (message: WorkerMessage) => void,
     workerCallback: EffectCallback
-): Promise<LintMessage[]> {
+): Promise<void> {
     return new Promise(resolve => {
         // Notify about worker starting. It can take a while to get worker starting up
         // Prevents showing blank screen between worker start and repository reading
@@ -55,35 +54,39 @@ function scanRepository(
         });
 
         worker.on('error', (error: Error & { code?: string }) => {
-            onMessage({
-                type: 'WORKER_ERROR',
-                payload: error.code,
-            });
-
-            const message = [error.code, error.message]
-                .filter(Boolean)
-                .join(' ');
-
-            resolve([
-                createErrorMessage({ message, path: repository, ruleId: '' }),
-            ]);
+            const messages = [
+                createErrorMessage({
+                    path: repository,
+                    ruleId: '',
+                    message: [error.code, error.message]
+                        .filter(Boolean)
+                        .join(' '),
+                }),
+            ];
+
+            onMessage({ type: 'WORKER_ERROR', payload: error.code });
+            onMessage({ type: 'ON_RESULT', payload: { messages } });
+            resolve();
         });
 
         worker.on('exit', code => {
             cleanup();
 
             // 0 = success, 1 = termination
             if (code === 0 || code === 1) {
-                return resolve([]);
+                return resolve();
             }
 
-            resolve([
+            const messages = [
                 createErrorMessage({
                     message: `Worker exited with code ${code}`,
                     path: repository,
                     ruleId: '',
                 }),
-            ]);
+            ];
+
+            onMessage({ type: 'ON_RESULT', payload: { messages } });
+            resolve();
         });
     });
 }

lib/engine/worker-task.ts

@@ -11,12 +11,10 @@ export type WorkerMessage =
     | { type: 'READ' }
     | { type: 'CLONE' }
     | { type: 'PULL' }
+    | { type: 'ON_RESULT'; payload: { messages: LintMessage[] } }
     | { type: 'LINT_START'; payload: number }
     | { type: 'LINT_END' }
-    | {
-          type: 'FILE_LINT_END';
-          payload: { messages: LintMessage[]; fileIndex: number };
-      }
+    | { type: 'FILE_LINT_END'; payload: { fileIndex: number } }
     | { type: 'FILE_LINT_SLOW'; payload: { path: string; lintTime: number } }
     | { type: 'LINTER_CRASH'; payload: string }
     | { type: 'WORKER_ERROR'; payload?: string }
@@ -211,9 +209,13 @@ export default async function workerTask(): Promise<void> {
             // Catch crashing linter
             const crashMessage = parseErrorStack(error, file);
 
+            postMessage({
+                type: 'ON_RESULT',
+                payload: { messages: [crashMessage] },
+            });
             postMessage({
                 type: 'FILE_LINT_END',
-                payload: { messages: [crashMessage], fileIndex },
+                payload: { fileIndex },
             });
             postMessage({
                 type: 'LINTER_CRASH',
@@ -228,10 +230,8 @@ export default async function workerTask(): Promise<void> {
             .filter(Boolean)
             .map(message => ({ ...message, path }));
 
-        postMessage({
-            type: 'FILE_LINT_END',
-            payload: { messages, fileIndex },
-        });
+        postMessage({ type: 'ON_RESULT', payload: { messages } });
+        postMessage({ type: 'FILE_LINT_END', payload: { fileIndex } });
     }
 
     if (!config.cache) {

lib/index.ts

@@ -62,11 +62,13 @@ async function scanRepo(repository: string) {
                 case 'PULL':
                     return logger.onRepositoryPull(repository);
 
+                case 'ON_RESULT':
+                    return results.push(...message.payload.messages);
+
                 case 'LINT_START':
                     return logger.onLintStart(repository, message.payload);
 
                 case 'FILE_LINT_END':
-                    results.push(...message.payload.messages);
                     return logger.onFileLintEnd(
                         repository,
                         message.payload.fileIndex

package.json

@@ -76,7 +76,7 @@
         "strip-ansi": "^6.0.0",
         "ts-jest": "^26.4.4",
         "tscpaths": "^0.0.9",
-        "typescript": "^4.0.2"
+        "typescript": "^4.2.4"
     },
     "peerDependencies": {
         "eslint": ">=7"

test/unit/__mocks__/@config.ts

@@ -14,3 +14,6 @@ export default new Proxy(mockConfig, {
 });
 
 export const validateConfig = jest.fn().mockResolvedValue(undefined);
+export const resolveConfigurationLocation = jest
+    .fn()
+    .mockReturnValue('mock-configuration-location');

test/unit/__mocks__/worker_threads.ts

@@ -0,0 +1,33 @@
+export const isMainThread = true;
+
+type Listener = (...args: any[]) => void;
+
+interface Listeners {
+    message: Listener[];
+    error: Listener[];
+    exit: Listener[];
+}
+
+const listeners: Listeners = {
+    message: [],
+    error: [],
+    exit: [],
+};
+
+export const Emitter = {
+    emit(event: keyof Listeners, data: unknown): void {
+        listeners[event].forEach(listener => listener(data));
+    },
+
+    clear(): void {
+        listeners.message.splice(0);
+        listeners.error.splice(0);
+        listeners.exit.splice(0);
+    },
+};
+
+export class Worker {
+    on(event: keyof Listeners, listener: Listener): void {
+        listeners[event].push(listener);
+    }
+}

test/unit/engine.test.ts

@@ -0,0 +1,78 @@
+import Engine from '@engine';
+import { Emitter } from '__mocks__/worker_threads';
+
+jest.mock('worker_threads', () => require('./__mocks__/worker_threads'));
+jest.unmock('@engine');
+
+const REPOSITORY = 'test-repository';
+const cleanup = jest.fn();
+const workerCallback = jest.fn().mockReturnValue(cleanup);
+
+describe('engine', () => {
+    afterEach(() => {
+        Emitter.clear();
+    });
+
+    test('crashing worker is reported', async () => {
+        const onMessage = jest.fn();
+        const promise = Engine.scanRepository(
+            REPOSITORY,
+            onMessage,
+            workerCallback
+        );
+
+        Emitter.emit('error', { code: 'mock-error' });
+
+        expect(onMessage).toHaveBeenCalledWith({
+            type: 'WORKER_ERROR',
+            payload: 'mock-error',
+        });
+
+        expect(onMessage).toHaveBeenLastCalledWith({
+            type: 'ON_RESULT',
+            payload: {
+                messages: [
+                    {
+                        message: 'mock-error',
+                        path: 'test-repository',
+                        ruleId: '',
+                        column: 0,
+                        line: 0,
+                        severity: 0,
+                    },
+                ],
+            },
+        });
+
+        await promise;
+    });
+
+    test('unexpected worker exits are reported', async () => {
+        const onMessage = jest.fn();
+        const promise = Engine.scanRepository(
+            REPOSITORY,
+            onMessage,
+            workerCallback
+        );
+
+        Emitter.emit('exit', 987);
+
+        expect(onMessage).toHaveBeenCalledWith({
+            payload: {
+                messages: [
+                    {
+                        message: 'Worker exited with code 987',
+                        path: 'test-repository',
+                        ruleId: '',
+                        column: 0,
+                        line: 0,
+                        severity: 0,
+                    },
+                ],
+            },
+            type: 'ON_RESULT',
+        });
+
+        await promise;
+    });
+});

yarn.lock

@@ -5770,10 +5770,10 @@ typedarray-to-buffer@^3.1.5:
   dependencies:
     is-typedarray "^1.0.0"
 
-typescript@^4.0.2:
-  version "4.0.5"
-  resolved "https://registry.yarnpkg.com/typescript/-/typescript-4.0.5.tgz#ae9dddfd1069f1cb5beb3ef3b2170dd7c1332389"
-  integrity sha512-ywmr/VrTVCmNTJ6iV2LwIrfG1P+lv6luD8sUJs+2eI9NLGigaN+nUQc13iHqisq7bra9lnmUSYqbJvegraBOPQ==
+typescript@^4.2.4:
+  version "4.2.4"
+  resolved "https://registry.yarnpkg.com/typescript/-/typescript-4.2.4.tgz#8610b59747de028fda898a8aef0e103f156d0961"
+  integrity sha512-V+evlYHZnQkaz8TRBuxTA92yZBPotr5H+WhQ7bD3hZUndx5tGOa1fuCgeSjxAzM1RiN5IzvadIXTVefuuwZCRg==
 
 undefsafe@^2.0.3:
   version "2.0.3"