Merge pull request #1080 from 0xProject/upgradeBlockstream

Fix dropped events issue in Order-watcher and Contract-wrappers subscriptions

3 files changed, 53 insertions, 21 deletions

diff --git a/packages/order-watcher/CHANGELOG.json b/packages/order-watcher/CHANGELOG.json
index d4b0e6607..9ab0905fe 100644
--- a/packages/order-watcher/CHANGELOG.json
+++ b/packages/order-watcher/CHANGELOG.json
@@ -1,21 +1,36 @@
 [
     {
-        "timestamp": 1537875740,
+        "version": "2.0.0",
+        "changes": [
+            {
+                "note":
+                    "Fixes dropped events issue by fetching logs by blockHash instead of blockNumber. Support for fetching by blockHash was added in Geth > v1.8.13 and Parity > v2.1.0. Infura works too.",
+                "pr": 1080
+            },
+            {
+                "note":
+                    "Fix misunderstanding about blockstream interface callbacks and pass the raw JSON RPC responses to it",
+                "pr": 1080
+            }
+        ]
+    },
+    {
         "version": "1.0.5",
         "changes": [
             {
                 "note": "Dependencies updated"
             }
-        ]
+        ],
+        "timestamp": 1537875740
     },
     {
-        "timestamp": 1537541580,
         "version": "1.0.4",
         "changes": [
             {
                 "note": "Dependencies updated"
             }
-        ]
+        ],
+        "timestamp": 1537541580
     },
     {
         "version": "1.0.3",
diff --git a/packages/order-watcher/package.json b/packages/order-watcher/package.json
index d383211a8..b00a76e55 100644
--- a/packages/order-watcher/package.json
+++ b/packages/order-watcher/package.json
@@ -82,7 +82,7 @@
         "@0xproject/web3-wrapper": "^3.0.0",
         "bintrees": "^1.0.2",
         "ethereum-types": "^1.0.7",
-        "ethereumjs-blockstream": "5.0.0",
+        "ethereumjs-blockstream": "6.0.0",
         "ethers": "3.0.22",
         "lodash": "^4.17.5"
     },
diff --git a/packages/order-watcher/src/order_watcher/event_watcher.ts b/packages/order-watcher/src/order_watcher/event_watcher.ts
index eca235e26..9ea301815 100644
--- a/packages/order-watcher/src/order_watcher/event_watcher.ts
+++ b/packages/order-watcher/src/order_watcher/event_watcher.ts
@@ -1,6 +1,6 @@
 import { intervalUtils, logUtils } from '@0xproject/utils';
-import { Web3Wrapper } from '@0xproject/web3-wrapper';
-import { BlockParamLiteral, BlockWithoutTransactionData, LogEntry, Provider } from 'ethereum-types';
+import { marshaller, Web3Wrapper } from '@0xproject/web3-wrapper';
+import { BlockParamLiteral, FilterObject, LogEntry, Provider, RawLogEntry } from 'ethereum-types';
 import { Block, BlockAndLogStreamer, Log } from 'ethereumjs-blockstream';
 import * as _ from 'lodash';
 
@@ -20,7 +20,6 @@ enum LogEventState {
  */
 export class EventWatcher {
     private readonly _web3Wrapper: Web3Wrapper;
-    private readonly _stateLayer: BlockParamLiteral;
     private readonly _isVerbose: boolean;
     private _blockAndLogStreamerIfExists: BlockAndLogStreamer<Block, Log> | undefined;
     private _blockAndLogStreamIntervalIfExists?: NodeJS.Timer;
@@ -35,7 +34,6 @@ export class EventWatcher {
     ) {
         this._isVerbose = isVerbose;
         this._web3Wrapper = new Web3Wrapper(provider);
-        this._stateLayer = stateLayer;
         this._pollingIntervalMs = _.isUndefined(pollingIntervalIfExistsMs)
             ? DEFAULT_EVENT_POLLING_INTERVAL_MS
             : pollingIntervalIfExistsMs;
@@ -62,8 +60,8 @@ export class EventWatcher {
             throw new Error(OrderWatcherError.SubscriptionAlreadyPresent);
         }
         this._blockAndLogStreamerIfExists = new BlockAndLogStreamer(
-            this._getBlockOrNullAsync.bind(this),
-            this._web3Wrapper.getLogsAsync.bind(this._web3Wrapper),
+            this._blockstreamGetBlockOrNullAsync.bind(this),
+            this._blockstreamGetLogsAsync.bind(this),
             this._onBlockAndLogStreamerError.bind(this),
         );
         const catchAllLogFilter = {};
@@ -83,12 +81,30 @@ export class EventWatcher {
         );
     }
     // This method only exists in order to comply with the expected interface of Blockstream's constructor
-    private async _getBlockOrNullAsync(): Promise<BlockWithoutTransactionData | null> {
-        const blockIfExists = await this._web3Wrapper.getBlockIfExistsAsync.bind(this._web3Wrapper);
-        if (_.isUndefined(blockIfExists)) {
-            return null;
-        }
-        return blockIfExists;
+    private async _blockstreamGetBlockOrNullAsync(hash: string): Promise<Block | null> {
+        const shouldIncludeTransactionData = false;
+        const blockOrNull = await this._web3Wrapper.sendRawPayloadAsync<Block | null>({
+            method: 'eth_getBlockByHash',
+            params: [hash, shouldIncludeTransactionData],
+        });
+        return blockOrNull;
+    }
+    // This method only exists in order to comply with the expected interface of Blockstream's constructor
+    private async _blockstreamGetLatestBlockOrNullAsync(): Promise<Block | null> {
+        const shouldIncludeTransactionData = false;
+        const blockOrNull = await this._web3Wrapper.sendRawPayloadAsync<Block | null>({
+            method: 'eth_getBlockByNumber',
+            params: [BlockParamLiteral.Latest, shouldIncludeTransactionData],
+        });
+        return blockOrNull;
+    }
+    // This method only exists in order to comply with the expected interface of Blockstream's constructor
+    private async _blockstreamGetLogsAsync(filterOptions: FilterObject): Promise<RawLogEntry[]> {
+        const logs = await this._web3Wrapper.sendRawPayloadAsync<RawLogEntry[]>({
+            method: 'eth_getLogs',
+            params: [filterOptions],
+        });
+        return logs as RawLogEntry[];
     }
     private _stopBlockAndLogStream(): void {
         if (_.isUndefined(this._blockAndLogStreamerIfExists)) {
@@ -103,19 +119,20 @@ export class EventWatcher {
     private async _onLogStateChangedAsync(
         callback: EventWatcherCallback,
         isRemoved: boolean,
-        log: LogEntry,
+        rawLog: RawLogEntry,
     ): Promise<void> {
+        const log: LogEntry = marshaller.unmarshalLog(rawLog);
         await this._emitDifferencesAsync(log, isRemoved ? LogEventState.Removed : LogEventState.Added, callback);
     }
     private async _reconcileBlockAsync(): Promise<void> {
-        const latestBlockIfExists = await this._web3Wrapper.getBlockIfExistsAsync(this._stateLayer);
-        if (_.isUndefined(latestBlockIfExists)) {
+        const latestBlockOrNull = await this._blockstreamGetLatestBlockOrNullAsync();
+        if (_.isNull(latestBlockOrNull)) {
             return; // noop
         }
         // We need to coerce to Block type cause Web3.Block includes types for mempool blocks
         if (!_.isUndefined(this._blockAndLogStreamerIfExists)) {
             // If we clear the interval while fetching the block - this._blockAndLogStreamer will be undefined
-            await this._blockAndLogStreamerIfExists.reconcileNewBlock((latestBlockIfExists as any) as Block);
+            await this._blockAndLogStreamerIfExists.reconcileNewBlock(latestBlockOrNull);
         }
     }
     private async _emitDifferencesAsync(