[#135] Recover out of sync depth streams (keith1024)

Author: Jon Eyrick

node-binance-api.js

@@ -447,10 +447,17 @@ LIMIT_MAKER
         let context = _depthCacheContext[symbol];
         // This now conforms 100% to the Binance docs constraints on managing a local order book
         if ( !context.lastEventUpdateId && (depth.U > context.snapshotUpdateId + 1 || depth.u < context.snapshotUpdateId + 1 )) {
-            options.log('depthHandler :'+symbol+': Unexpected first depth event. Skipping. ' +
-                '!! IF this persists, the "'+symbol+'" cache is OUT OF SYNC !! ');
+            // I think if the count exceeded 1 we could deem the cache out of sync. But we'll
+            // be lenient and give the cache up to a count of 3 before calling it out of sync.
+            if ( ++context.skipCount > 2 ) {
+                const msg = 'depthHandler: ['+symbol+'] Skip count exceeded. The depth cache is out of sync.';
+                if ( options.verbose ) options.log(msg);
+                throw new Error(msg);
+            }
         } else if ( context.lastEventUpdateId && depth.U !== context.lastEventUpdateId + 1 ) {
-            options.log('depthHandler :'+symbol+': !! DEPTH CACHE OUT OF SYNC !!');
+            const msg = 'depthHandler: ['+symbol+'] Incorrect update ID. The depth cache is out of sync.';
+            if ( options.verbose ) options.log(msg);
+            throw new Error(msg);
         } else {
             for ( obj of depth.b ) { //bids
                 depthCache[symbol].bids[obj[0]] = parseFloat(obj[1]);
@@ -464,6 +471,7 @@ LIMIT_MAKER
                     delete depthCache[symbol].asks[obj[0]];
                 }
             }
+            context.skipCount = 0;
             context.lastEventUpdateId = depth.u;
         }
     };
@@ -873,6 +881,7 @@ LIMIT_MAKER
                     let context = _depthCacheContext[symbol];
                     context.snapshotUpdateId = undefined;
                     context.lastEventUpdateId = undefined;
+                    context.skipCount = 0;
                     context.messageQueue = [];
 
                     depthCache[symbol] = { bids: {}, asks: {} };
@@ -884,8 +893,12 @@ LIMIT_MAKER
                     if (context.messageQueue && !context.snapshotUpdateId ) {
                         context.messageQueue.push(depth);
                     } else {
-                        depthHandler(depth);
-                        if ( callback ) callback(symbol, depthCache[symbol]);
+                        try {
+                            depthHandler(depth);
+                            if ( callback ) callback(symbol, depthCache[symbol]);
+                        } catch (err) {
+                            reconnect();
+                        }
                     }
                 };
 
@@ -898,8 +911,12 @@ LIMIT_MAKER
                         context.snapshotUpdateId = json.lastUpdateId;
                         context.messageQueue = context.messageQueue.filter(depth => depth.u > context.snapshotUpdateId);
                         // Process any pending depth messages
-                        for ( let depth of context.messageQueue )
-                            depthHandler(depth, json.lastUpdateId);
+                        for ( let depth of context.messageQueue ) {
+                            // Although sync errors shouldn't ever happen here, we catch and swallow them anyway
+                            // just in case. The stream handler function above will deal with broken caches.
+                            try { depthHandler(depth); }
+                            catch (err) {}
+                        }
                         delete context.messageQueue;
                         if ( callback ) callback(symbol, depthCache[symbol]);
                     });