fix: stabilize stuck cleanup (wix#2721)

Author: noomorph

detox/src/client/AsyncWebSocket.js

@@ -4,6 +4,7 @@ const log = require('../utils/logger').child({ __filename });
 const Deferred = require('../utils/Deferred');
 const DetoxInternalError = require('../errors/DetoxInternalError');
 const DetoxRuntimeError = require('../errors/DetoxRuntimeError');
+const InflightRequest = require('./InflightRequest');
 
 const EVENTS = {
   OPEN: Object.freeze({ event: 'WS_OPEN' }),
@@ -13,6 +14,10 @@ const EVENTS = {
   LATE_RESPONSE: Object.freeze({ event: 'WS_LATE_RESPONSE' }),
 };
 
+const DEFAULT_SEND_OPTIONS = {
+  timeout: 0,
+};
+
 class AsyncWebSocket {
   constructor(url) {
     this._log = log.child({ url });
@@ -73,7 +78,7 @@ class AsyncWebSocket {
     return closing.promise;
   }
 
-  async send(message) {
+  async send(message, options = DEFAULT_SEND_OPTIONS) {
     if (!this.isOpen) {
       throw new DetoxRuntimeError({
         message: 'Cannot send a message over the closed web socket. See the payload below:',
@@ -87,9 +92,7 @@ class AsyncWebSocket {
     }
 
     const messageId = message.messageId;
-    const inFlight = this.inFlightPromises[messageId] = new Deferred();
-    inFlight.message = message;
-
+    const inFlight = this.inFlightPromises[messageId] = new InflightRequest(message).withTimeout(options.timeout);
     const messageAsString = JSON.stringify(message);
     this._log.trace(EVENTS.SEND, messageAsString);
     this._ws.send(messageAsString);
@@ -107,6 +110,8 @@ class AsyncWebSocket {
 
   resetInFlightPromises() {
     for (const messageId of _.keys(this.inFlightPromises)) {
+      const inFlight = this.inFlightPromises[messageId];
+      inFlight.clearTimeout();
       delete this.inFlightPromises[messageId];
       this._abortedMessageIds.add(+messageId);
     }
@@ -122,8 +127,8 @@ class AsyncWebSocket {
     const inFlightPromises = _.values(this.inFlightPromises);
 
     this.resetInFlightPromises();
-    for (const deferred of inFlightPromises) {
-      deferred.reject(error);
+    for (const inflight of inFlightPromises) {
+      inflight.reject(error);
     }
 
     if (!hasPendingActions) {
@@ -220,7 +225,7 @@ class AsyncWebSocket {
       }
     } catch (error) {
       this.rejectAll(new DetoxRuntimeError({
-        message: 'Unexpected error on an attempt to handle a message over the web socket.',
+        message: 'Unexpected error on an attempt to handle the response received over the web socket.',
         hint: 'Examine the inner error:\n\n' + DetoxRuntimeError.format(error) + '\n\nThe payload was:',
         debugInfo: data,
       }));

detox/src/client/AsyncWebSocket.test.js

@@ -1,3 +1,5 @@
+jest.useFakeTimers('modern');
+
 const _ = require('lodash');
 const permaproxy = require('funpermaproxy');
 const config = require('../configuration/configurations.mock').validSession;
@@ -42,6 +44,10 @@ describe('AsyncWebSocket', () => {
     log = require('../utils/logger');
   });
 
+  afterEach(() => {
+    jest.clearAllTimers();
+  });
+
   describe('.open()', () => {
     it(`should normally resolve`, async () => {
       await expect(connect()).resolves.not.toThrowError();
@@ -112,6 +118,16 @@ describe('AsyncWebSocket', () => {
         expect(await response1).toEqual(expected1);
       });
 
+      it(`should not set expiration timers by default`, async () => {
+        aws.send(generateRequest());
+        expect(jest.getTimerCount()).toBe(0);
+      });
+
+      it(`should set an expiration timer if sent with timeout > 0`, async () => {
+        aws.send(generateRequest(), { timeout: 100 });
+        expect(jest.getTimerCount()).toBe(1);
+      });
+
       it(`should reject all messages in the flight if there's an error`, async () => {
         const sendPromise1 = aws.send(generateRequest());
         const sendPromise2 = aws.send(generateRequest());
@@ -153,6 +169,34 @@ describe('AsyncWebSocket', () => {
         const error = log.error.mock.calls[0][1];
         expect(error).toMatchSnapshot();
       });
+
+      it(`should fail if the message timeout has expired`, async () => {
+        const sendPromise = aws.send(generateRequest(), { timeout: 5000 });
+        expect(jest.getTimerCount()).toBe(1);
+        jest.advanceTimersByTime(5000);
+        await expect(sendPromise).rejects.toThrowErrorMatchingSnapshot();
+        expect(jest.getTimerCount()).toBe(0);
+      });
+
+      it(`should cancel the expiration timer if the request has been answered`, async () => {
+        aws.send(generateRequest(100), { timeout: 5000 });
+        expect(jest.getTimerCount()).toBe(1);
+        socket.mockMessage({ messageId: 100, type: 'response' });
+        expect(jest.getTimerCount()).toBe(0);
+      });
+
+      it(`should cancel the expiration timer if all in-flights have been reset`, async () => {
+        aws.send(generateRequest(100), { timeout: 5000 });
+        await aws.resetInFlightPromises();
+        expect(jest.getTimerCount()).toBe(0);
+      });
+
+      it(`should cancel the expiration timer if all in-flights have been rejected`, async () => {
+        const promise = aws.send(generateRequest(100), { timeout: 5000 });
+        await aws.rejectAll(new Error('Just because'));
+        expect(jest.getTimerCount()).toBe(0);
+        await expect(promise).rejects.toThrow('Just because');
+      });
     });
   });
 
@@ -253,9 +297,9 @@ describe('AsyncWebSocket', () => {
       const message1 = aws.send(generateRequest());
       const message2 = aws.send(generateRequest());
 
-      aws.rejectAll(anError());
-      await expect(message1).rejects.toEqual(anError());
-      await expect(message2).rejects.toEqual(anError());
+      aws.rejectAll(anError('TestError'));
+      await expect(message1).rejects.toThrow(/TestError/);
+      await expect(message2).rejects.toThrow(/TestError/);
     });
   });
 
@@ -312,7 +356,7 @@ describe('AsyncWebSocket', () => {
   }
 
   function generateRequest(messageId) {
-    return {message: 'a message', messageId};
+    return {type: 'invoke', message: 'a message', messageId};
   }
 
   function generateResponse(message, messageId) {
@@ -323,8 +367,8 @@ describe('AsyncWebSocket', () => {
     };
   }
 
-  function anError() {
-    const err = new Error('TestError');
+  function anError(msg = 'TestError') {
+    const err = new Error(msg);
     delete err.stack;
     return err;
   }

detox/src/client/Client.js

@@ -23,6 +23,7 @@ class Client {
     this._onBeforeAppCrash = this._onBeforeAppCrash.bind(this);
     this._onAppDisconnected = this._onAppDisconnected.bind(this);
     this._onUnhandledServerError = this._onUnhandledServerError.bind(this);
+    this._logError = this._logError.bind(this);
 
     this._sessionId = sessionId;
     this._slowInvocationTimeout = debugSynchronization;
@@ -73,13 +74,13 @@ class Client {
 
     try {
       if (this.isConnected) {
-        await this.sendAction(new actions.Cleanup(this._successfulTestRun));
+        await this.sendAction(new actions.Cleanup(this._successfulTestRun)).catch(this._logError);
 
         this._whenAppIsConnected = this._invalidState('while cleaning up')
         this._whenAppIsReady = this._whenAppIsConnected;
       }
     } finally {
-      await this._asyncWebSocket.close();
+      await this._asyncWebSocket.close().catch(this._logError);
     }
 
     delete this.terminateApp; // property injection
@@ -127,38 +128,38 @@ class Client {
   }
 
   async sendAction(action) {
-    const options = this._inferSendOptions(action);
+    const { queryStatus, ...options } = this._inferSendOptions(action);
 
-    return await ((options.queryStatus)
-      ? this._sendMonitoredAction(action)
-      : this._doSendAction(action))
+    return await (queryStatus
+      ? this._sendMonitoredAction(action, options)
+      : this._doSendAction(action, options))
   }
 
   _inferSendOptions(action) {
     if ( action instanceof actions.CurrentStatus
       || action instanceof actions.Login
       || action instanceof actions.Cleanup
     ) {
-      return { queryStatus: false };
+      return { queryStatus: false, timeout: 5000 };
     }
 
-    return { queryStatus: true };
+    return { queryStatus: true, timeout: 0 };
   }
 
-  async _sendMonitoredAction(action) {
+  async _sendMonitoredAction(action, options) {
     try {
       this._scheduleSlowInvocationQuery();
-      return await this._doSendAction(action);
+      return await this._doSendAction(action, options);
     } finally {
       this._unscheduleSlowInvocationQuery();
     }
   }
 
-  async _doSendAction(action) {
+  async _doSendAction(action, options) {
     const errorWithUserStack = createErrorWithUserStack();
 
     try {
-      const parsedResponse = await this._asyncWebSocket.send(action);
+      const parsedResponse = await this._asyncWebSocket.send(action, options);
       if (parsedResponse && parsedResponse.type === 'serverError') {
         throw deserializeError(parsedResponse.params.error);
       }
@@ -355,6 +356,10 @@ class Client {
       new DetoxInternalError(`Detected an attempt to interact with Detox Client ${state}.`)
     );
   }
+
+  _logError(e) {
+    log.error({ event: 'ERROR' }, DetoxRuntimeError.format(e));
+  }
 }
 
 module.exports = Client;

detox/src/client/Client.test.js

@@ -39,6 +39,10 @@ describe('Client', () => {
       });
     });
 
+    mockAws.close.mockImplementation(async () => {
+
+    });
+
     mockAws.mockBusy = () => {
       const deferred = new Deferred();
       mockAws.send.mockImplementationOnce(() => deferred.promise);
@@ -130,7 +134,7 @@ describe('Client', () => {
       mockAws.mockResponse('loginSuccess', {});
       expect(mockAws.send).not.toHaveBeenCalled();
       await client.connect();
-      expect(mockAws.send).toHaveBeenCalledWith(new actions.Login(validSession.sessionId));
+      expect(mockAws.send).toHaveBeenCalledWith(new actions.Login(validSession.sessionId), SEND_OPTIONS.TIMED);
     });
 
     it('should not consider itself connected to the app if "loginSuccess" params.appConnected = false', async () => {
@@ -160,14 +164,14 @@ describe('Client', () => {
 
     it('should schedule "currentStatus" query when it takes too long', async () => {
       const { action } = await simulateInFlightAction();
-      expect(mockAws.send).toHaveBeenCalledWith(action);
+      expect(mockAws.send).toHaveBeenCalledWith(action, SEND_OPTIONS.DEFAULT);
       expect(mockAws.send).toHaveBeenCalledTimes(2); // action + login
 
       mockAws.mockBusy(); // for the current status query
       jest.advanceTimersByTime(validSession.debugSynchronization);
       await fastForwardAllPromises();
 
-      expect(mockAws.send).toHaveBeenCalledWith(new actions.CurrentStatus());
+      expect(mockAws.send).toHaveBeenCalledWith(new actions.CurrentStatus(), SEND_OPTIONS.TIMED);
       expect(jest.getTimerCount()).toBe(0); // should not spam with "currentStatus" queries
     });
 
@@ -258,7 +262,7 @@ describe('Client', () => {
       mockAws.mockResponse('whateverDone');
       const action = anAction();
       await client.sendAction(action);
-      expect(mockAws.send).toHaveBeenCalledWith(action);
+      expect(mockAws.send).toHaveBeenCalledWith(action, SEND_OPTIONS.DEFAULT);
     });
 
     it('should pass the parsed response to action.handle()', async () => {
@@ -297,7 +301,6 @@ describe('Client', () => {
       ['waitForActive', 'waitForActiveDone', actions.WaitForActive],
       ['waitUntilReady', 'ready', actions.Ready],
       ['currentStatus', 'currentStatusResult', actions.CurrentStatus, {}, { status: 'App is idle' }],
-      ['cleanup', 'cleanupDone', actions.Cleanup, true],
     ])('.%s', (methodName, expectedResponseType, Action, params, expectedResponseParams) => {
       beforeEach(async () => {
         await client.connect();
@@ -308,7 +311,7 @@ describe('Client', () => {
         await client[methodName](params);
 
         const action = new Action(params);
-        expect(mockAws.send).toHaveBeenCalledWith(action);
+        expect(mockAws.send).toHaveBeenCalledWith(action, { timeout: expect.any(Number) });
       });
 
       it(`should throw on a wrong response from device`, async () => {
@@ -362,7 +365,7 @@ describe('Client', () => {
       await client.connect();
       mockAws.mockResponse('cleanupDone');
       await client.cleanup();
-      expect(mockAws.send).toHaveBeenCalledWith(new actions.Cleanup(true));
+      expect(mockAws.send).toHaveBeenCalledWith(new actions.Cleanup(true), SEND_OPTIONS.TIMED);
     });
 
     it('should send cleanup action (stopRunner=false) to the app if there were failed invocations', async () => {
@@ -371,7 +374,7 @@ describe('Client', () => {
       await expect(client.execute(anInvocation)).rejects.toThrowError(/Test Failed.*SomeDetails/);
       mockAws.mockResponse('cleanupDone');
       await client.cleanup();
-      expect(mockAws.send).toHaveBeenCalledWith(new actions.Cleanup(false));
+      expect(mockAws.send).toHaveBeenCalledWith(new actions.Cleanup(false), SEND_OPTIONS.TIMED);
     });
 
     it('should close the websocket upon "cleanupDone" from the app', async () => {
@@ -381,11 +384,32 @@ describe('Client', () => {
       expect(mockAws.close).toHaveBeenCalled();
     });
 
-    it('should close the websocket even on error', async () => {
+    it('should close the websocket even if getting "cleanupDone" fails', async () => {
+      await client.connect();
+      mockAws.mockResponse('serverError');
+      await client.cleanup();
+      expect(mockAws.close).toHaveBeenCalled();
+    });
+
+    it('should close the websocket even on an inner error', async () => {
+      await client.connect();
+      mockAws.mockSyncError('MyError');
+      await client.cleanup();
+      expect(mockAws.close).toHaveBeenCalled();
+      expect(log.error).toHaveBeenCalledWith({ event: 'ERROR' }, expect.stringContaining('MyError'));
+    });
+
+    it('should not bail even if the world is crashing, instead it should log errors and exit calmly', async () => {
       await client.connect();
-      mockAws.mockSyncError('UnexpectedError');
-      await expect(client.cleanup()).rejects.toThrowError('UnexpectedError');
+
+      mockAws.send.mockRejectedValue('MyError1');
+      mockAws.close.mockRejectedValue('MyError2');
+
+      await client.cleanup();
+
       expect(mockAws.close).toHaveBeenCalled();
+      expect(log.error).toHaveBeenCalledWith({ event: 'ERROR' }, expect.stringContaining('MyError1'));
+      expect(log.error).toHaveBeenCalledWith({ event: 'ERROR' }, expect.stringContaining('MyError2'));
     });
 
     it('should delete the injected .terminateApp method', async () => {
@@ -543,7 +567,7 @@ describe('Client', () => {
       mockAws.mockEventCallback('appConnected');
       mockAws.mockResponse('ready');
       await fastForwardAllPromises();
-      expect(mockAws.send).toHaveBeenCalledWith(new actions.Ready());
+      expect(mockAws.send).toHaveBeenCalledWith(new actions.Ready(), SEND_OPTIONS.DEFAULT);
       expect(isReady).toBe(true);
     });
 
@@ -558,7 +582,7 @@ describe('Client', () => {
       await client.connect();
       await fastForwardAllPromises();
       expect(isReady).toBe(true);
-      expect(mockAws.send).not.toHaveBeenCalledWith(new actions.Ready());
+      expect(mockAws.send).not.toHaveBeenCalledWith(new actions.Ready(), expect.anything());
     });
   })
 
@@ -692,4 +716,9 @@ describe('Client', () => {
       await Promise.resolve();
     }
   }
+
+  const SEND_OPTIONS = {
+    DEFAULT: { timeout: 0 },
+    TIMED: { timeout: 5000 },
+  };
 });