fix(client): properly handle unhandled disconnect

# Conflicts:
#	commons/Queue.ts

Author: blakebyrnes

client/connections/ConnectionToCore.ts

@@ -182,10 +182,15 @@ export default abstract class ConnectionToCore extends TypedEventEmitter<{
   }
 
   public async createSession(options: ICreateSessionOptions): Promise<CoreSession> {
-    const sessionMeta = await this.commandQueue.run<ISessionMeta>('Session.create', options);
-    const session = new CoreSession({ ...sessionMeta, sessionName: options.sessionName }, this);
-    this.coreSessions.track(session);
-    return session;
+    try {
+      const sessionMeta = await this.commandQueue.run<ISessionMeta>('Session.create', options);
+      const session = new CoreSession({ ...sessionMeta, sessionName: options.sessionName }, this);
+      this.coreSessions.track(session);
+      return session;
+    } catch (error) {
+      if (error instanceof DisconnectedFromCoreError && this.isDisconnecting) return null;
+      throw error;
+    }
   }
 
   public getSession(sessionId: string): CoreSession {
@@ -294,8 +299,6 @@ export default abstract class ConnectionToCore extends TypedEventEmitter<{
 
   protected cancelPendingRequests(): void {
     const host = String(this.resolvedHost);
-    this.coreSessions.stop(new DisconnectedFromCoreError(host));
-    this.commandQueue.stop(new DisconnectedFromCoreError(host));
     for (const entry of this.pendingRequestsById.values()) {
       const id = entry.id;
       if (this.connectRequestId === id || this.disconnectRequestId === id) {
@@ -304,6 +307,8 @@ export default abstract class ConnectionToCore extends TypedEventEmitter<{
       this.pendingRequestsById.delete(id);
       this.rejectPendingRequest(entry, new DisconnectedFromCoreError(host));
     }
+    this.coreSessions.stop(new DisconnectedFromCoreError(host));
+    this.commandQueue.stop(new DisconnectedFromCoreError(host));
   }
 
   private createPendingResult(): IResolvablePromiseWithId {

client/lib/CoreCommandQueue.ts

@@ -29,6 +29,9 @@ export default class CoreCommandQueue {
   }
 
   public run<T>(command: string, ...args: any[]): Promise<T> {
+    if (this.connection.isDisconnecting) {
+      return Promise.resolve(null);
+    }
     return this.internalQueue.run<T>(this.runRequest.bind(this, command, args)).catch(error => {
       error.stack += `${this.sessionMarker}`;
       throw error;

client/lib/Handler.ts

@@ -1,11 +1,13 @@
 import { createPromise, pickRandom } from '@secret-agent/commons/utils';
 import ShutdownHandler from '@secret-agent/commons/ShutdownHandler';
-import Log from '@secret-agent/commons/Logger';
+import Log, { hasBeenLoggedSymbol } from '@secret-agent/commons/Logger';
+import { CanceledPromiseError } from '@secret-agent/commons/interfaces/IPendingWaitEvent';
 import IAgentCreateOptions from '../interfaces/IAgentCreateOptions';
 import IConnectionToCoreOptions from '../interfaces/IConnectionToCoreOptions';
 import Agent from './Agent';
 import ConnectionToCore from '../connections/ConnectionToCore';
 import ConnectionFactory from '../connections/ConnectionFactory';
+import DisconnectedFromCoreError from '../connections/DisconnectedFromCoreError';
 
 type SettledDispatchesBySessionId = { [sessionId: string]: { args: any; error?: Error } };
 type PendingDispatch = { resolution: Promise<Error | void>; sessionId?: string; args: any };
@@ -118,10 +120,15 @@ export default class Handler {
     const dispatches = [...this.dispatches];
     // clear out dispatches everytime you check it
     this.dispatches.length = 0;
+    const startStack = new Error('').stack.split(/\r?\n/).slice(1).join('\n');
     await Promise.all(
       dispatches.map(async dispatch => {
         const err = await dispatch.resolution;
-        if (err) throw err;
+        if (err) {
+          const marker = `------WAIT FOR ALL DISPATCHES`.padEnd(50, '-');
+          err.stack += `\n${marker}\n${startStack}`;
+          throw err;
+        }
       }),
     );
     // keep going if there are new things queued
@@ -166,12 +173,25 @@ export default class Handler {
   }
 
   private async logUnhandledError(error: Error): Promise<void> {
+    if (error instanceof DisconnectedFromCoreError) return;
+    if (!error || error[hasBeenLoggedSymbol]) return;
     // if error and there are remote connections, log error here
-    if (error && this.connections.some(x => !!x.options.host)) {
-      log.error('UnhandledRejection', { error, sessionId: null });
+    if (this.connections.some(x => !!x.options.host)) {
+      log.error('UnhandledRejection (Client)', { error, sessionId: null });
     }
     // eslint-disable-next-line promise/no-promise-in-callback
-    await Promise.all(this.connections.map(x => x.logUnhandledError(error)));
+    await Promise.all(
+      this.connections.map(x => {
+        return x.logUnhandledError(error).catch(logError => {
+          if (logError instanceof CanceledPromiseError) return;
+          log.error('UnhandledRejection.CouldNotSendToCore', {
+            error: logError,
+            connectionHost: x.hostOrError,
+            sessionId: null,
+          });
+        });
+      }),
+    );
   }
 
   private onDisconnected(connection: ConnectionToCore): void {

commons/Queue.ts

@@ -2,6 +2,7 @@ import IResolvablePromise from '@secret-agent/core-interfaces/IResolvablePromise
 import { createPromise } from './utils';
 import { CanceledPromiseError } from './interfaces/IPendingWaitEvent';
 import Resolvable from './Resolvable';
+import getPrototypeOf = Reflect.getPrototypeOf;
 
 type AsyncCallback<T> = (value?: any) => Promise<T>;
 
@@ -24,7 +25,7 @@ export default class Queue {
     this.queue.push({
       promise,
       cb,
-      startStack: new Error('').stack.split('\n').slice(1).join('\n'),
+      startStack: new Error('').stack.split(/\r?\n/).slice(1).join('\n'),
     });
 
     this.next().catch(() => null);
@@ -68,7 +69,9 @@ export default class Queue {
     this.activeCount += 1;
     try {
       const res = await Promise.race([next.cb(), this.abortPromise.promise]);
-      if (this.abortPromise.isResolved) throw await this.abortPromise.promise;
+      if (this.abortPromise.isResolved) {
+        return this.reject(next, await this.abortPromise.promise);
+      }
 
       next.promise.resolve(res);
     } catch (error) {
@@ -77,13 +80,16 @@ export default class Queue {
       this.activeCount -= 1;
     }
 
-    setImmediate(() => this.next());
+    setImmediate(() => this.next().catch(() => null));
   }
 
-  private reject(entry: IQueueEntry, error: Error): void {
-    const marker = `------${this.stacktraceMarker}`.padEnd(50, '-');
+  private reject(entry: IQueueEntry, sourceError: Error): void {
+    const error = <Error>Object.create(getPrototypeOf(sourceError));
+    error.message = sourceError.message;
+    Object.assign(error, sourceError);
 
-    error.stack = `${error.stack}\n${marker}\n${entry.startStack}`;
+    const marker = `------${this.stacktraceMarker}`.padEnd(50, '-');
+    error.stack = `${sourceError.stack}\n${marker}\n${entry.startStack}`;
     entry.promise.reject(error);
   }
 }

core/index.ts

@@ -124,6 +124,7 @@ export default class Core {
   }
 
   public static logUnhandledError(clientError: Error, fatalError = false): void {
+    if (!clientError || clientError[hasBeenLoggedSymbol]) return;
     if (fatalError) {
       log.error('UnhandledError(fatal)', { clientError, sessionId: null });
     } else if (!clientError[hasBeenLoggedSymbol]) {

full-client/test/handler.test.ts

@@ -224,6 +224,42 @@ describe('connectionToCore', () => {
     await expect(handler.waitForAllDispatches()).rejects.toThrowError(DisconnectedFromCoreError);
   });
 
+  it('handles disconnects from client', async () => {
+    const coreHost = await CoreProcess.spawn({});
+    Helpers.onClose(() => CoreProcess.kill('SIGINT'));
+    const connection = new RemoteConnectionToCore({
+      maxConcurrency: 2,
+      host: coreHost,
+    });
+    await connection.connect();
+
+    const handler = new Handler(connection);
+    Helpers.needsClosing.push(handler);
+
+    const waitForGoto = createPromise();
+    const dispatchErrorPromise = createPromise<Error>();
+    handler.dispatchAgent(async agent => {
+      try {
+        await agent.goto(koaServer.baseUrl);
+        const promise = agent.waitForMillis(10e3);
+        await new Promise(resolve => setTimeout(resolve, 50));
+        waitForGoto.resolve();
+        await promise;
+      } catch (error) {
+        dispatchErrorPromise.resolve(error);
+        throw error;
+      }
+    });
+    await waitForGoto.promise;
+    await connection.disconnect();
+    await new Promise(setImmediate);
+    await expect(dispatchErrorPromise).resolves.toBeTruthy();
+    const dispatchError = await dispatchErrorPromise;
+    expect(dispatchError).toBeInstanceOf(DisconnectedFromCoreError);
+    expect((dispatchError as DisconnectedFromCoreError).coreHost).toBe(coreHost);
+    await expect(handler.waitForAllDispatches()).rejects.toThrowError(DisconnectedFromCoreError);
+  });
+
   it('handles core server ending websocket (econnreset)', async () => {
     const coreHost = await Core.server.address;
     // @ts-ignore