Merge PR #883 from 'nodech/wallet-interactive-rescan'

CHANGELOG.md

@@ -31,7 +31,7 @@ process and allows parallel rescans.
       - `compactInterval` - what is the current compaction interval config.
       - `nextCompaction` - when will the next compaction trigger after restart.
       - `lastCompaction` - when was the last compaction run.
-  - Introduce `scan interactive` hook (start, filter)
+  - Introduce `scan interactive` hook (start, filter, fullLock)
 
 ### Node HTTP Client:
   - Introduce `scanInteractive` method that starts interactive rescan.
@@ -43,6 +43,9 @@ process and allows parallel rescans.
   - Add `getFee`, an HTTP alternative to estimateFee socket call.
 
 ### Wallet Changes
+- Add migration that recalculates txdb balances to fix any inconsistencies.
+- Wallet will now use `interactive scan` for initial sync(on open) and rescan.
+
 #### Configuration
 - Wallet now has option `wallet-migrate-no-rescan`/`migrate-no-rescan` if you
   want to disable rescan when migration recommends it. It may result in the
@@ -54,21 +57,22 @@ process and allows parallel rescans.
 
 #### Wallet API
 
-- Add migration that recalculates txdb balances to fix any inconsistencies.
+- WalletNode now emits `open` and `close` events.
 - WalletDB Now emits events for: `open`, `close`, `connect`, `disconnect`.
 - WalletDB
   - `open()` no longer calls `connect` and needs separate call `connect`.
   - `open()` no longer calls scan, instead only rollbacks and waits for
     sync to do the rescan.
   - emits events for: `open`, `close`, `connect`, `disconnect`, `sync done`.
 
-### Wallet HTTP Client
+#### Wallet HTTP
   - All transaction creating endpoints now accept `hardFee` for specifying the
     exact fee.
   - All transaction sending endpoints now fundlock/queue tx creation. (no more
     conflicting transactions)
   - Add options to `getNames` for passing `own`.
 
+
 ## v6.0.0
 
 ### Node and Wallet HTTP API

lib/blockchain/chain.js

@@ -2266,16 +2266,55 @@ class Chain extends AsyncEmitter {
     }
   }
 
+  /** @typedef {import('./common').ScanAction} ScanAction */
+
+  /**
+   * @callback ScanInteractiveIterCB
+   * @param {ChainEntry} entry
+   * @param {TX[]} txs
+   * @returns {Promise<ScanAction>}
+   */
+
   /**
    * Interactive scan the blockchain for transactions containing specified
    * address hashes. Allows repeat and abort.
    * @param {Hash|Number} start - Block hash or height to start at.
    * @param {BloomFilter} filter - Starting bloom filter containing tx,
    * address and name hashes.
-   * @param {Function} iter - Iterator.
+   * @param {ScanInteractiveIterCB} iter - Iterator.
+   * @param {Boolean} [fullLock=false]
+   * @returns {Promise}
    */
 
-  async scanInteractive(start, filter, iter) {
+  async scanInteractive(start, filter, iter, fullLock = false) {
+    if (fullLock) {
+      const unlock = await this.locker.lock();
+      try {
+        // We lock the whole chain, no longer lock per block scan.
+        return await this._scanInteractive(start, filter, iter, false);
+      } catch (e) {
+        this.logger.debug('Scan(interactive) errored. Error: %s', e.message);
+        throw e;
+      } finally {
+        unlock();
+      }
+    }
+
+    return this._scanInteractive(start, filter, iter, true);
+  }
+
+  /**
+   * Interactive scan the blockchain for transactions containing specified
+   * address hashes. Allows repeat and abort.
+   * @param {Hash|Number} start - Block hash or height to start at.
+   * @param {BloomFilter} filter - Starting bloom filter containing tx,
+   * address and name hashes.
+   * @param {ScanInteractiveIterCB} iter - Iterator.
+   * @param {Boolean} [lockPerScan=true] - if we should lock per block scan.
+   * @returns {Promise}
+   */
+
+  async _scanInteractive(start, filter, iter, lockPerScan = true) {
     if (start == null)
       start = this.network.genesis.hash;
 
@@ -2287,7 +2326,10 @@ class Chain extends AsyncEmitter {
     let hash = start;
 
     while (hash != null) {
-      const unlock = await this.locker.lock();
+      let unlock;
+
+      if (lockPerScan)
+        unlock = await this.locker.lock();
 
       try {
         const {entry, txs} = await this.db.scanBlock(hash, filter);
@@ -2333,7 +2375,8 @@ class Chain extends AsyncEmitter {
         this.logger.debug('Scan(interactive) errored. Error: %s', e.message);
         throw e;
       } finally {
-        unlock();
+        if (lockPerScan)
+          unlock();
       }
     }
   }

lib/blockchain/chaindb.js

@@ -1612,12 +1612,18 @@ class ChainDB {
     this.logger.info('Finished scanning %d blocks.', total);
   }
 
+  /**
+   * @typedef {Object} ScanBlockResult
+   * @property {ChainEntry} entry
+   * @property {TX[]} txs
+   */
+
   /**
    * Interactive scans block checks.
    * @param {Hash|Number} blockID - Block hash or height to start at.
    * @param {BloomFilter} [filter] - Starting bloom filter containing tx,
    * address and name hashes.
-   * @returns {Promise}
+   * @returns {Promise<ScanBlockResult>}
    */
 
   async scanBlock(blockID, filter) {

lib/blockchain/common.js

@@ -84,3 +84,39 @@ exports.scanActions = {
   REPEAT_ADD: 4,
   REPEAT: 5
 };
+
+/**
+ * @typedef {Object} ActionAbort
+ * @property {exports.scanActions} type - ABORT
+ */
+
+/**
+ * @typedef {Object} ActionNext
+ * @property {exports.scanActions} type - NEXT
+ */
+
+/**
+ * @typedef {Object} ActionRepeat
+ * @property {exports.ScanAction} type - REPEAT
+ */
+
+/**
+ * @typedef {Object} ActionRepeatAdd
+ * @property {exports.scanActions} type - REPEAT_ADD
+ * @property {Buffer[]} chunks
+ */
+
+/**
+ * @typedef {Object} ActionRepeatSet
+ * @property {exports.scanActions} type - REPEAT_SET
+ * @property {BloomFilter} filter
+ */
+
+/**
+ * @typedef {ActionAbort
+ *  | ActionNext
+ *  | ActionRepeat
+ *  | ActionRepeatAdd
+ *  | ActionRepeatSet
+ * } ScanAction
+ */

lib/client/node.js

@@ -370,16 +370,17 @@ class NodeClient extends Client {
    * Rescan for any missed transactions. (Interactive)
    * @param {Number|Hash} start - Start block.
    * @param {BloomFilter} [filter]
+   * @param {Boolean} [fullLock=false]
    * @returns {Promise}
    */
 
-  rescanInteractive(start, filter = null) {
+  rescanInteractive(start, filter = null, fullLock = false) {
     if (start == null)
       start = 0;
 
     assert(typeof start === 'number' || Buffer.isBuffer(start));
 
-    return this.call('rescan interactive', start, filter);
+    return this.call('rescan interactive', start, filter, fullLock);
   }
 }
 

lib/client/wallet.js

@@ -1600,4 +1600,6 @@ class Wallet extends EventEmitter {
  * Expose
  */
 
+WalletClient.Wallet = Wallet;
+
 module.exports = WalletClient;

lib/node/fullnode.js

@@ -369,11 +369,13 @@ class FullNode extends Node {
    * @param {Number|Hash} start - Start block.
    * @param {BloomFilter} filter
    * @param {Function} iter - Iterator.
+   * @param {Boolean} [fullLock=false] - lock the whole chain instead of per
+   * scan.
    * @returns {Promise}
    */
 
-  scanInteractive(start, filter, iter) {
-    return this.chain.scanInteractive(start, filter, iter);
+  scanInteractive(start, filter, iter, fullLock = false) {
+    return this.chain.scanInteractive(start, filter, iter, fullLock);
   }
 
   /**

lib/node/http.js

@@ -712,6 +712,7 @@ class HTTP extends Server {
       const valid = new Validator(args);
       const start = valid.uintbhash(0);
       const rawFilter = valid.buf(1);
+      const fullLock = valid.bool(2, false);
       let filter = socket.filter;
 
       if (start == null)
@@ -720,7 +721,7 @@ class HTTP extends Server {
       if (rawFilter)
         filter = BloomFilter.fromRaw(rawFilter);
 
-      return this.scanInteractive(socket, start, filter);
+      return this.scanInteractive(socket, start, filter, fullLock);
     });
   }
 
@@ -859,10 +860,11 @@ class HTTP extends Server {
    * @param {WebSocket} socket
    * @param {Hash} start
    * @param {BloomFilter} filter
+   * @param {Boolean} [fullLock=false]
    * @returns {Promise}
    */
 
-  async scanInteractive(socket, start, filter) {
+  async scanInteractive(socket, start, filter, fullLock = false) {
     const iter = async (entry, txs) => {
       const block = entry.encode();
       const raw = [];
@@ -921,12 +923,11 @@ class HTTP extends Server {
     };
 
     try {
-      await this.node.scanInteractive(start, filter, iter);
+      await this.node.scanInteractive(start, filter, iter, fullLock);
     } catch (err) {
-      return socket.call('block rescan interactive abort', err.message);
+      await socket.call('block rescan interactive abort', err.message);
+      throw err;
     }
-
-    return null;
   }
 }
 

lib/wallet/account.js

@@ -512,7 +512,7 @@ class Account extends bio.Struct {
    * Allocate new lookahead addresses if necessary.
    * @param {Number} receiveDepth
    * @param {Number} changeDepth
-   * @returns {Promise} - Returns {@link WalletKey}.
+   * @returns {Promise<WalletKey?>}
    */
 
   async syncDepth(b, receive, change) {

lib/wallet/client.js

@@ -11,11 +11,13 @@ const NodeClient = require('../client/node');
 const TX = require('../primitives/tx');
 const Coin = require('../primitives/coin');
 const NameState = require('../covenants/namestate');
+const {encoding} = require('bufio');
 
 const parsers = {
   'block connect': (entry, txs) => parseBlock(entry, txs),
   'block disconnect': entry => [parseEntry(entry)],
   'block rescan': (entry, txs) => parseBlock(entry, txs),
+  'block rescan interactive': (entry, txs) => parseBlock(entry, txs),
   'chain reset': entry => [parseEntry(entry)],
   'tx': tx => [TX.decode(tx)]
 };
@@ -74,10 +76,27 @@ class WalletClient extends NodeClient {
     return super.setFilter(filter.encode());
   }
 
+  /**
+   * Rescan for any missed transactions.
+   * @param {Number|Hash} start - Start block.
+   * @returns {Promise}
+   */
+
   async rescan(start) {
     return super.rescan(start);
   }
 
+  /**
+   * Rescan interactive for any missed transactions.
+   * @param {Number|Hash} start - Start block.
+   * @param {Boolean} [fullLock=false]
+   * @returns {Promise}
+   */
+
+  async rescanInteractive(start, fullLock) {
+    return super.rescanInteractive(start, null, fullLock);
+  }
+
   async getNameStatus(nameHash) {
     const json = await super.getNameStatus(nameHash);
     return NameState.fromJSON(json);
@@ -94,6 +113,9 @@ class WalletClient extends NodeClient {
  */
 
 function parseEntry(data) {
+  if (!data)
+    return null;
+
   // 32  hash
   // 4   height
   // 4   nonce
@@ -112,17 +134,24 @@ function parseEntry(data) {
 
   assert(Buffer.isBuffer(data));
   // Just enough to read the three data below
-  assert(data.length >= 44);
+  assert(data.length >= 80);
+
+  const hash = data.slice(0, 32);
+  const height = encoding.readU32(data, 32);
+  const time = encoding.readU64(data, 40);
+  const prevBlock = data.slice(48, 80);
 
   return {
-    hash: data.slice(0, 32),
-    height: data.readUInt32LE(32),
-    time: data.readUInt32LE(40)
+    hash,
+    height,
+    time,
+    prevBlock
   };
 }
 
 function parseBlock(entry, txs) {
   const block = parseEntry(entry);
+  assert(block);
   const out = [];
 
   for (const tx of txs)

lib/wallet/node.js

@@ -113,6 +113,7 @@ class WalletNode extends Node {
     await this.handleOpen();
 
     this.logger.info('Wallet node is loaded.');
+    this.emit('open');
   }
 
   /**
@@ -134,6 +135,7 @@ class WalletNode extends Node {
     await this.wdb.disconnect();
     await this.wdb.close();
     await this.handleClose();
+    this.emit('close');
   }
 }