refactor: deterministic queue execution

CHANGELOG.md

@@ -8,7 +8,7 @@ All notable changes to this project will be documented in this file.
 - Fixed the issue that cannot export huge logs - [#561](https://github.com/chrisleekr/binance-trading-bot/pull/561), [#567](https://github.com/chrisleekr/binance-trading-bot/pull/567)
 - Fixed the balance calculation to include dust balances by [@uhliksk](https://github.com/uhliksk) - [#571](https://github.com/chrisleekr/binance-trading-bot/pull/571)
 - Fixed the open orders to be cancelled when the current price is higher/lower than the order price by [@uhliksk](https://github.com/uhliksk) - [#569](https://github.com/chrisleekr/binance-trading-bot/pull/569)
-- Improved queue processing by replacing Bull queue to customised queue system by [@uhliksk](https://github.com/uhliksk) - [#562](https://github.com/chrisleekr/binance-trading-bot/pull/562), [#581](https://github.com/chrisleekr/binance-trading-bot/pull/581)
+- Improved queue processing by replacing Bull queue to customised queue system by [@uhliksk](https://github.com/uhliksk) - [#562](https://github.com/chrisleekr/binance-trading-bot/pull/562), [#581](https://github.com/chrisleekr/binance-trading-bot/pull/581), [#588](https://github.com/chrisleekr/binance-trading-bot/pull/588)
 - Added conservative sell strategy, which can reduce the sell trigger price as the grid gets deeper by [@rando128](https://github.com/rando128) - [#585](https://github.com/chrisleekr/binance-trading-bot/pull/585)
 
 Thanks [@uhliksk](https://github.com/uhliksk) and [@rando128](https://github.com/rando128) for your great contributions. 💯 :heart:

app/__tests__/server-binance.test.js

@@ -11,8 +11,6 @@ describe('server-binance', () => {
   let mockGetGlobalConfiguration;
 
   let mockGetAccountInfoFromAPI;
-  let mockLockSymbol;
-  let mockUnlockSymbol;
   let mockCacheExchangeSymbols;
 
   let mockSetupUserWebsocket;
@@ -57,8 +55,9 @@ describe('server-binance', () => {
       deleteAll: jest.fn().mockResolvedValue(true)
     };
     mockQueue = {
-      init: jest.fn().mockResolvedValue(true),
-      execute: jest.fn().mockResolvedValue(true)
+      prepareJob: jest.fn().mockResolvedValue(true),
+      execute: jest.fn().mockResolvedValue(true),
+      completeJob: jest.fn().mockResolvedValue(true)
     };
     mockSlack = {
       sendMessage: jest.fn().mockResolvedValue(true)
@@ -97,9 +96,6 @@ describe('server-binance', () => {
           }
         });
 
-        mockLockSymbol = jest.fn().mockResolvedValue(true);
-        mockUnlockSymbol = jest.fn().mockResolvedValue(true);
-
         mockSetupUserWebsocket = jest.fn().mockResolvedValue(true);
 
         mockSyncCandles = jest.fn().mockResolvedValue(true);
@@ -133,8 +129,6 @@ describe('server-binance', () => {
 
         jest.mock('../cronjob/trailingTradeHelper/common', () => ({
           getAccountInfoFromAPI: mockGetAccountInfoFromAPI,
-          lockSymbol: mockLockSymbol,
-          unlockSymbol: mockUnlockSymbol,
           cacheExchangeSymbols: mockCacheExchangeSymbols
         }));
 
@@ -237,8 +231,12 @@ describe('server-binance', () => {
         );
       });
 
-      it('triggers queue.init', () => {
-        expect(mockQueue.init).toHaveBeenCalled();
+      it('triggers queue.prepareJob', () => {
+        expect(mockQueue.prepareJob).toHaveBeenCalled();
+      });
+
+      it('triggers queue.completeJob', () => {
+        expect(mockQueue.completeJob).toHaveBeenCalled();
       });
     });
 
@@ -253,9 +251,6 @@ describe('server-binance', () => {
           }
         });
 
-        mockLockSymbol = jest.fn().mockResolvedValue(true);
-        mockUnlockSymbol = jest.fn().mockResolvedValue(true);
-
         mockSetupUserWebsocket = jest.fn().mockResolvedValue(true);
 
         mockSyncCandles = jest.fn().mockResolvedValue(true);
@@ -297,8 +292,6 @@ describe('server-binance', () => {
 
         jest.mock('../cronjob/trailingTradeHelper/common', () => ({
           getAccountInfoFromAPI: mockGetAccountInfoFromAPI,
-          lockSymbol: mockLockSymbol,
-          unlockSymbol: mockUnlockSymbol,
           cacheExchangeSymbols: mockCacheExchangeSymbols
         }));
 
@@ -370,9 +363,6 @@ describe('server-binance', () => {
           }
         });
 
-        mockLockSymbol = jest.fn().mockResolvedValue(true);
-        mockUnlockSymbol = jest.fn().mockResolvedValue(true);
-
         mockSetupUserWebsocket = jest.fn().mockResolvedValue(true);
 
         mockSyncCandles = jest.fn().mockResolvedValue(true);
@@ -414,8 +404,6 @@ describe('server-binance', () => {
 
         jest.mock('../cronjob/trailingTradeHelper/common', () => ({
           getAccountInfoFromAPI: mockGetAccountInfoFromAPI,
-          lockSymbol: mockLockSymbol,
-          unlockSymbol: mockUnlockSymbol,
           cacheExchangeSymbols: mockCacheExchangeSymbols
         }));
 
@@ -482,11 +470,15 @@ describe('server-binance', () => {
         });
 
         it('triggers queue.execute for BTCUSDT', () => {
-          expect(mockQueue.execute).toHaveBeenCalledWith(logger, 'BTCUSDT');
+          expect(mockQueue.execute).toHaveBeenCalledWith(logger, 'BTCUSDT', {
+            processFn: expect.any(Function)
+          });
         });
 
         it('triggers queue.execute for LTCUSDT', () => {
-          expect(mockQueue.execute).toHaveBeenCalledWith(logger, 'LTCUSDT');
+          expect(mockQueue.execute).toHaveBeenCalledWith(logger, 'LTCUSDT', {
+            processFn: expect.any(Function)
+          });
         });
       });
     });
@@ -502,9 +494,6 @@ describe('server-binance', () => {
           }
         });
 
-        mockLockSymbol = jest.fn().mockResolvedValue(true);
-        mockUnlockSymbol = jest.fn().mockResolvedValue(true);
-
         mockSetupUserWebsocket = jest.fn().mockResolvedValue(true);
 
         mockSyncCandles = jest.fn().mockResolvedValue(true);
@@ -546,8 +535,6 @@ describe('server-binance', () => {
 
         jest.mock('../cronjob/trailingTradeHelper/common', () => ({
           getAccountInfoFromAPI: mockGetAccountInfoFromAPI,
-          lockSymbol: mockLockSymbol,
-          unlockSymbol: mockUnlockSymbol,
           cacheExchangeSymbols: mockCacheExchangeSymbols
         }));
 
@@ -607,9 +594,6 @@ describe('server-binance', () => {
         }
       });
 
-      mockLockSymbol = jest.fn().mockResolvedValue(true);
-      mockUnlockSymbol = jest.fn().mockResolvedValue(true);
-
       mockSetupUserWebsocket = jest.fn().mockResolvedValue(true);
 
       mockSyncCandles = jest.fn().mockResolvedValue(true);
@@ -643,8 +627,6 @@ describe('server-binance', () => {
 
       jest.mock('../cronjob/trailingTradeHelper/common', () => ({
         getAccountInfoFromAPI: mockGetAccountInfoFromAPI,
-        lockSymbol: mockLockSymbol,
-        unlockSymbol: mockUnlockSymbol,
         cacheExchangeSymbols: mockCacheExchangeSymbols
       }));
 
@@ -770,9 +752,6 @@ describe('server-binance', () => {
         }
       });
 
-      mockLockSymbol = jest.fn().mockResolvedValue(true);
-      mockUnlockSymbol = jest.fn().mockResolvedValue(true);
-
       mockSetupUserWebsocket = jest.fn().mockResolvedValue(true);
 
       mockSyncCandles = jest.fn().mockResolvedValue(true);
@@ -806,8 +785,6 @@ describe('server-binance', () => {
 
       jest.mock('../cronjob/trailingTradeHelper/common', () => ({
         getAccountInfoFromAPI: mockGetAccountInfoFromAPI,
-        lockSymbol: mockLockSymbol,
-        unlockSymbol: mockUnlockSymbol,
         cacheExchangeSymbols: mockCacheExchangeSymbols
       }));
 

app/binance/__tests__/candles.test.js

@@ -241,7 +241,8 @@ describe('candles.js', () => {
     it('triggers queue.execute for ETHBTC', () => {
       expect(mockExecute).toHaveBeenCalledWith(loggerMock, 'ETHBTC', {
         correlationId: expect.any(String),
-        preprocessFn: expect.any(Function)
+        preprocessFn: expect.any(Function),
+        processFn: expect.any(Function)
       });
     });
   });

app/binance/__tests__/tickers.test.js

@@ -115,14 +115,16 @@ describe('tickers.js', () => {
     it('triggers queue.execute for BTCUSDT', () => {
       expect(mockExecute).toHaveBeenCalledWith(loggerMock, 'BTCUSDT', {
         correlationId: expect.any(String),
-        preprocessFn: expect.any(Function)
+        preprocessFn: expect.any(Function),
+        processFn: expect.any(Function)
       });
     });
 
     it('triggers queue.execute for BNBUSDT', () => {
       expect(mockExecute).toHaveBeenCalledWith(loggerMock, 'BNBUSDT', {
         correlationId: expect.any(String),
-        preprocessFn: expect.any(Function)
+        preprocessFn: expect.any(Function),
+        processFn: expect.any(Function)
       });
     });
 
@@ -152,7 +154,8 @@ describe('tickers.js', () => {
       it('triggers queue.execute for BTCUSDT', () => {
         expect(mockExecute).toHaveBeenCalledWith(loggerMock, 'BTCUSDT', {
           correlationId: expect.any(String),
-          preprocessFn: expect.any(Function)
+          preprocessFn: expect.any(Function),
+          processFn: expect.any(Function)
         });
       });
 

app/binance/__tests__/user.test.js

@@ -362,7 +362,8 @@ describe('user.js', () => {
           it('triggers queue.execute', () => {
             expect(mockExecute).toHaveBeenCalledWith(loggerMock, 'ETHUSDT', {
               correlationId: expect.any(String),
-              preprocessFn: expect.any(Function)
+              preprocessFn: expect.any(Function),
+              processFn: expect.any(Function)
             });
           });
 
@@ -680,7 +681,8 @@ describe('user.js', () => {
         it('triggers queue.execute', () => {
           expect(mockExecute).toHaveBeenCalledWith(loggerMock, 'ETHUSDT', {
             correlationId: expect.any(String),
-            preprocessFn: expect.any(Function)
+            preprocessFn: expect.any(Function),
+            processFn: expect.any(Function)
           });
         });
 

app/binance/candles.js

@@ -1,6 +1,7 @@
 const { v4: uuidv4 } = require('uuid');
 const _ = require('lodash');
 const queue = require('../cronjob/trailingTradeHelper/queue');
+const { executeTrailingTrade } = require('../cronjob/index');
 const { binance, mongo } = require('../helpers');
 const {
   getConfiguration
@@ -116,7 +117,8 @@ const syncCandles = async (logger, symbols) => {
 
       queue.execute(logger, symbol, {
         correlationId: uuidv4(),
-        preprocessFn: getCandles
+        preprocessFn: getCandles,
+        processFn: executeTrailingTrade
       });
     })
   );

app/binance/tickers.js

@@ -2,6 +2,7 @@ const { v4: uuidv4 } = require('uuid');
 const _ = require('lodash');
 const { binance, cache } = require('../helpers');
 const queue = require('../cronjob/trailingTradeHelper/queue');
+const { executeTrailingTrade } = require('../cronjob/index');
 
 const {
   getAccountInfo,
@@ -78,7 +79,8 @@ const setupTickersWebsocket = async (logger, symbols) => {
           if (canExecuteTrailingTrade) {
             queue.execute(symbolLogger, monitoringSymbol, {
               correlationId,
-              preprocessFn: saveCandle
+              preprocessFn: saveCandle,
+              processFn: executeTrailingTrade
             });
           } else {
             saveCandle();

app/binance/user.js

@@ -2,6 +2,7 @@ const { v4: uuidv4 } = require('uuid');
 const _ = require('lodash');
 const { binance } = require('../helpers');
 const queue = require('../cronjob/trailingTradeHelper/queue');
+const { executeTrailingTrade } = require('../cronjob/index');
 
 const {
   updateAccountInfo,
@@ -121,7 +122,8 @@ const setupUserWebsocket = async logger => {
 
       queue.execute(symbolLogger, symbol, {
         correlationId,
-        preprocessFn: checkLastOrder
+        preprocessFn: checkLastOrder,
+        processFn: executeTrailingTrade
       });
 
       const checkManualOrder = async () => {
@@ -155,7 +157,8 @@ const setupUserWebsocket = async logger => {
 
       queue.execute(symbolLogger, symbol, {
         correlationId,
-        preprocessFn: checkManualOrder
+        preprocessFn: checkManualOrder,
+        processFn: executeTrailingTrade
       });
     }
   });

app/cronjob/__tests__/trailingTrade.test.js

@@ -12,9 +12,6 @@ describe('trailingTrade', () => {
   let mockConfigGet;
 
   let mockGetAccountInfo;
-  let mockLockSymbol;
-  let mockIsSymbolLocked;
-  let mockUnlockSymbol;
 
   let mockGetSymbolConfiguration;
   let mockGetSymbolInfo;
@@ -40,9 +37,6 @@ describe('trailingTrade', () => {
   beforeEach(() => {
     jest.clearAllMocks().resetModules();
 
-    mockLockSymbol = jest.fn().mockResolvedValue(true);
-    mockUnlockSymbol = jest.fn().mockResolvedValue(true);
-
     mockLoggerInfo = jest.fn();
     mockSlackSendMessage = jest.fn().mockResolvedValue(true);
 
@@ -89,8 +83,6 @@ describe('trailingTrade', () => {
         get: mockConfigGet
       }));
 
-      mockIsSymbolLocked = jest.fn().mockResolvedValue(false);
-
       mockGetAccountInfo = jest.fn().mockResolvedValue({
         account: 'info'
       });
@@ -268,10 +260,7 @@ describe('trailingTrade', () => {
         }));
 
       jest.mock('../trailingTradeHelper/common', () => ({
-        getAccountInfo: mockGetAccountInfo,
-        lockSymbol: mockLockSymbol,
-        isSymbolLocked: mockIsSymbolLocked,
-        unlockSymbol: mockUnlockSymbol
+        getAccountInfo: mockGetAccountInfo
       }));
 
       jest.mock('../trailingTrade/steps', () => ({
@@ -301,16 +290,11 @@ describe('trailingTrade', () => {
         await trailingTradeExecute(logger, 'BTCUSDT');
       });
 
-      it(`triggers isSymbolLocked - BTCUSDT`, () => {
-        expect(mockIsSymbolLocked).toHaveBeenCalledWith(logger, 'BTCUSDT');
-      });
-
-      it('returns expected result for BTCUSDT', () => {
+      it('returns expected result for BTCUSDT finish', () => {
         expect(mockLoggerInfo).toHaveBeenCalledWith(
           {
             data: {
               symbol: 'BTCUSDT',
-              isLocked: false,
               featureToggle: { feature1Enabled: true },
               lastCandle: { got: 'lowest value' },
               accountInfo: { account: 'info' },
@@ -348,16 +332,11 @@ describe('trailingTrade', () => {
         await trailingTradeExecute(logger, 'ETHUSDT');
       });
 
-      it(`triggers isSymbolLocked - ETHUSDT`, () => {
-        expect(mockIsSymbolLocked).toHaveBeenCalledWith(logger, 'ETHUSDT');
-      });
-
       it('returns expected result for ETHUSDT', () => {
         expect(mockLoggerInfo).toHaveBeenCalledWith(
           {
             data: {
               symbol: 'ETHUSDT',
-              isLocked: false,
               featureToggle: { feature1Enabled: true },
               lastCandle: { got: 'lowest value' },
               accountInfo: { account: 'info' },
@@ -395,16 +374,11 @@ describe('trailingTrade', () => {
         await trailingTradeExecute(logger, 'LTCUSDT');
       });
 
-      it(`triggers isSymbolLocked - LTCUSDT`, () => {
-        expect(mockIsSymbolLocked).toHaveBeenCalledWith(logger, 'LTCUSDT');
-      });
-
       it('returns expected result for LTCUSDT', async () => {
         expect(mockLoggerInfo).toHaveBeenCalledWith(
           {
             data: {
               symbol: 'LTCUSDT',
-              isLocked: false,
               featureToggle: { feature1Enabled: true },
               lastCandle: { got: 'lowest value' },
               accountInfo: { account: 'info' },