<commit_before>"use strict";
const util         = require('util');
const stream       = require('stream');
const co           = require('co');
const Q            = require('q');
const _            = require('underscore');
const moment       = require('moment');
const vucoin       = require('vucoin');
const hashf        = require('./ucp/hashf');
const dos2unix     = require('./system/dos2unix');
const logger       = require('./logger')('sync');
const rawer        = require('./ucp/rawer');
const constants    = require('../lib/constants');
const Peer         = require('../lib/entity/peer');
const multimeter   = require('multimeter');
const pulling      = require('../lib/pulling');
const makeQuerablePromise = require('../lib/querablep');

const CONST_BLOCKS_CHUNK = 250;
const EVAL_REMAINING_INTERVAL = 1000;
const CONST_MAX_SIMULTANEOUS_DOWNLOADS = 50;


module.exports = Synchroniser;

function Synchroniser (server, host, port, conf, interactive) {

  const that = this;

  let speed = 0, blocksApplied = 0;
  const baseWatcher = interactive ? new MultimeterWatcher() : new LoggerWatcher();

  // Wrapper to also push event stream
  const watcher = {
    writeStatus: baseWatcher.writeStatus,
    downloadPercent: (pct) => {
      if (pct !== undefined && baseWatcher.downloadPercent() < pct) {
        that.push({ download: pct });
      }
      return baseWatcher.downloadPercent(pct);
    },
    appliedPercent: (pct) => {
      if (pct !== undefined && baseWatcher.appliedPercent() < pct) {
        that.push({ applied: pct });
      }
      return baseWatcher.appliedPercent(pct);
    },
    end: baseWatcher.end
  };

  stream.Duplex.call(this, { objectMode: true });

  // Unused, but made mandatory by Duplex interface
  this._read = () => null;
  this._write = () => null;

  if (interactive) {
    logger.mute();
  }

  // Services
  const PeeringService     = server.PeeringService;
  const BlockchainService  = server.BlockchainService;

  const vucoinOptions = {
    timeout: constants.NETWORK.SYNC_LONG_TIMEOUT
  };

  const dal = server.dal;

  const logRemaining = (to) => co(function*() {
    const lCurrent = yield dal.getCurrentBlockOrNull();
    const localNumber = lCurrent ? lCurrent.number : -1;

    if (to > 1 && speed > 0) {
      const remain = (to - (localNumber + 1 + blocksApplied));
      const secondsLeft = remain / speed;
      const momDuration = moment.duration(secondsLeft*1000);
      watcher.writeStatus('Remaining ' + momDuration.humanize() + '');
    }
  });

  this.test = (to, chunkLen, askedCautious, nopeers) => co(function*() {
    const vucoin = yield getVucoin(host, port, vucoinOptions);
    const peering = yield Q.nfcall(vucoin.network.peering.get);
    const peer = new Peer(peering);
    const node = yield peer.connect();
    return Q.nfcall(node.blockchain.current);
  });

  this.sync = (to, chunkLen, askedCautious, nopeers) => co(function*() {

    try {

      const vucoin = yield getVucoin(host, port, vucoinOptions);
      const peering = yield Q.nfcall(vucoin.network.peering.get);

      let peer = new Peer(peering);
      logger.info("Try with %s %s", peer.getURL(), peer.pubkey.substr(0, 6));
      let node = yield peer.connect();
      node.pubkey = peer.pubkey;
      logger.info('Sync started.');

      //============
      // Blockchain headers
      //============
      logger.info('Getting remote blockchain info...');
      watcher.writeStatus('Connecting to ' + host + '...');
      const lCurrent = yield dal.getCurrentBlockOrNull();
      const localNumber = lCurrent ? lCurrent.number : -1;
      let rCurrent;
      if (isNaN(to)) {
        rCurrent = yield Q.nfcall(node.blockchain.current);
      } else {
        rCurrent = yield Q.nfcall(node.blockchain.block, to);
      }
      to = rCurrent.number;

      //=======
      // Peers (just for P2P download)
      //=======
      let peers = [];
      if (!nopeers) {
        watcher.writeStatus('Peers...');
        const merkle = yield dal.merkleForPeers();
        const getPeers = Q.nbind(node.network.peering.peers.get, node);
        const json2 = yield getPeers({});
        const rm = new NodesMerkle(json2);
        if(rm.root() != merkle.root()){
          const leavesToAdd = [];
          const json = yield getPeers({ leaves: true });
          _(json.leaves).forEach((leaf) => {
            if(merkle.leaves().indexOf(leaf) == -1){
              leavesToAdd.push(leaf);
            }
          });
          peers = yield leavesToAdd.map((leaf) => co(function*() {
            try {
              const json3 = yield getPeers({ "leaf": leaf });
              const jsonEntry = json3.leaf.value;
              const endpoint = jsonEntry.endpoints[0];
              watcher.writeStatus('Peer ' + endpoint);
              return jsonEntry;
            } catch (e) {
              logger.warn("Could not get peer of leaf %s, continue...", leaf);
              return null;
            }
          }));
        }
        else {
          watcher.writeStatus('Peers already known');
        }
      }

      peers = peers.filter((p) => p);

      //============
      // Blockchain
      //============
      logger.info('Downloading Blockchain...');

      // We use cautious mode if it is asked, or not particulary asked but blockchain has been started
      const cautious = (askedCautious === true || (askedCautious === undefined && localNumber >= 0));
      const downloader = new P2PDownloader(localNumber, to, CONST_MAX_SIMULTANEOUS_DOWNLOADS, peers, watcher);

      downloader.start();

      let lastPullBlock = null;
      let dao = pulling.abstractDao({

        // Get the local blockchain current block
        localCurrent: () => co(function*() {
          if (cautious) {
            return yield dal.getCurrentBlockOrNull();
          } else {
            if (lCurrent && !lastPullBlock) {
              lastPullBlock = lCurrent;
            }
            return lastPullBlock;
          }
        }),

        // Get the remote blockchain (bc) current block
        remoteCurrent: (peer) => Promise.resolve(rCurrent),

        // Get the remote peers to be pulled
        remotePeers: () => co(function*() {
          return [node];
        }),

        // Get block of given peer with given block number
        getLocalBlock: (number) => dal.getBlock(number),

        downloadBlocks: (thePeer, number) => co(function *() {
          // Note: we don't care about the particular peer asked by the method. We use the network instead.
          const numberOffseted = number - (localNumber + 1);
          const targetChunk = Math.floor(numberOffseted / CONST_BLOCKS_CHUNK);
          // Return the download promise! Simple.
          return downloader.getChunk(targetChunk);
        }),


        applyBranch: (blocks) => co(function *() {
          if (cautious) {
            for (const block of blocks) {
              yield dao.applyMainBranch(block);
            }
          } else {
            yield server.BlockchainService.saveBlocksInMainBranch(blocks);
          }
          lastPullBlock = blocks[blocks.length - 1];
          watcher.appliedPercent(Math.floor(blocks[blocks.length - 1].number / to * 100));
          return true;
        }),

        applyMainBranch: (block) => co(function *() {
          const addedBlock = yield server.BlockchainService.submitBlock(block, true, constants.FORK_ALLOWED);
          server.streamPush(addedBlock);
          watcher.appliedPercent(Math.floor(block.number / to * 100));
        }),

        // Eventually remove forks later on
        removeForks: () => co(function*() {}),

        // Tells wether given peer is a member peer
        isMemberPeer: (thePeer) => co(function *() {
          let idty = yield dal.getWrittenIdtyByPubkey(thePeer.pubkey);
          return (idty && idty.member) || false;
        })
      });

      const logInterval = setInterval(() => logRemaining(to), EVAL_REMAINING_INTERVAL);
      yield pulling.pull(conf, dao);

      // Finished blocks
      watcher.downloadPercent(100.0);
      watcher.appliedPercent(100.0);

      if (logInterval) {
        clearInterval(logInterval);
      }

      // Save currency parameters given by root block
      const rootBlock = yield server.dal.getBlock(0);
      yield BlockchainService.saveParametersForRootBlock(rootBlock);
      server.dal.blockDAL.cleanCache();

      //=======
      // Peers
      //=======
      if (!nopeers) {
        watcher.writeStatus('Peers...');
        yield syncPeer(node);
        const merkle = yield dal.merkleForPeers();
        const getPeers = Q.nbind(node.network.peering.peers.get, node);
        const json2 = yield getPeers({});
        const rm = new NodesMerkle(json2);
        if(rm.root() != merkle.root()){
          const leavesToAdd = [];
          const json = yield getPeers({ leaves: true });
          _(json.leaves).forEach((leaf) => {
            if(merkle.leaves().indexOf(leaf) == -1){
              leavesToAdd.push(leaf);
            }
          });
          for (const leaf of leavesToAdd) {
            const json3 = yield getPeers({ "leaf": leaf });
            const jsonEntry = json3.leaf.value;
            const sign = json3.leaf.value.signature;
            const entry = {};
            ["version", "currency", "pubkey", "endpoints", "block"].forEach((key) => {
              entry[key] = jsonEntry[key];
            });
            entry.signature = sign;
            watcher.writeStatus('Peer ' + entry.pubkey);
            yield PeeringService.submitP(entry, false, to === undefined);
          }
        }
        else {
          watcher.writeStatus('Peers already known');
        }
      }

      watcher.end();
      that.push({ sync: true });
      logger.info('Sync finished.');
    } catch (err) {
      that.push({ sync: false, msg: err });
      err && watcher.writeStatus(err.message || String(err));
      watcher.end();
      throw err;
    }
  });

  function getVucoin(theHost, thePort, options) {
    return new Promise(function (resolve, reject) {
      vucoin(theHost, thePort, function (err, node) {
        if (err) {
          return reject('Cannot sync: ' + err);
        }
        resolve(node);
      }, options);
    });
  }

  //============
  // Peer
  //============
  function syncPeer (node) {

    // Global sync vars
    const remotePeer = new Peer({});
    let remoteJsonPeer = {};

    return co(function *() {
      const json = yield Q.nfcall(node.network.peering.get);
      remotePeer.copyValuesFrom(json);
      const entry = remotePeer.getRaw();<commit_msg>Fix #620 Error during sync displayed [object Object]<commit_after>"use strict";
const util         = require('util');
const stream       = require('stream');
const co           = require('co');
const Q            = require('q');
const _            = require('underscore');
const moment       = require('moment');
const vucoin       = require('vucoin');
const hashf        = require('./ucp/hashf');
const dos2unix     = require('./system/dos2unix');
const logger       = require('./logger')('sync');
const rawer        = require('./ucp/rawer');
const constants    = require('../lib/constants');
const Peer         = require('../lib/entity/peer');
const multimeter   = require('multimeter');
const pulling      = require('../lib/pulling');
const makeQuerablePromise = require('../lib/querablep');

const CONST_BLOCKS_CHUNK = 250;
const EVAL_REMAINING_INTERVAL = 1000;
const CONST_MAX_SIMULTANEOUS_DOWNLOADS = 50;


module.exports = Synchroniser;

function Synchroniser (server, host, port, conf, interactive) {

  const that = this;

  let speed = 0, blocksApplied = 0;
  const baseWatcher = interactive ? new MultimeterWatcher() : new LoggerWatcher();

  // Wrapper to also push event stream
  const watcher = {
    writeStatus: baseWatcher.writeStatus,
    downloadPercent: (pct) => {
      if (pct !== undefined && baseWatcher.downloadPercent() < pct) {
        that.push({ download: pct });
      }
      return baseWatcher.downloadPercent(pct);
    },
    appliedPercent: (pct) => {
      if (pct !== undefined && baseWatcher.appliedPercent() < pct) {
        that.push({ applied: pct });
      }
      return baseWatcher.appliedPercent(pct);
    },
    end: baseWatcher.end
  };

  stream.Duplex.call(this, { objectMode: true });

  // Unused, but made mandatory by Duplex interface
  this._read = () => null;
  this._write = () => null;

  if (interactive) {
    logger.mute();
  }

  // Services
  const PeeringService     = server.PeeringService;
  const BlockchainService  = server.BlockchainService;

  const vucoinOptions = {
    timeout: constants.NETWORK.SYNC_LONG_TIMEOUT
  };

  const dal = server.dal;

  const logRemaining = (to) => co(function*() {
    const lCurrent = yield dal.getCurrentBlockOrNull();
    const localNumber = lCurrent ? lCurrent.number : -1;

    if (to > 1 && speed > 0) {
      const remain = (to - (localNumber + 1 + blocksApplied));
      const secondsLeft = remain / speed;
      const momDuration = moment.duration(secondsLeft*1000);
      watcher.writeStatus('Remaining ' + momDuration.humanize() + '');
    }
  });

  this.test = (to, chunkLen, askedCautious, nopeers) => co(function*() {
    const vucoin = yield getVucoin(host, port, vucoinOptions);
    const peering = yield Q.nfcall(vucoin.network.peering.get);
    const peer = new Peer(peering);
    const node = yield peer.connect();
    return Q.nfcall(node.blockchain.current);
  });

  this.sync = (to, chunkLen, askedCautious, nopeers) => co(function*() {

    try {

      const vucoin = yield getVucoin(host, port, vucoinOptions);
      const peering = yield Q.nfcall(vucoin.network.peering.get);

      let peer = new Peer(peering);
      logger.info("Try with %s %s", peer.getURL(), peer.pubkey.substr(0, 6));
      let node = yield peer.connect();
      node.pubkey = peer.pubkey;
      logger.info('Sync started.');

      //============
      // Blockchain headers
      //============
      logger.info('Getting remote blockchain info...');
      watcher.writeStatus('Connecting to ' + host + '...');
      const lCurrent = yield dal.getCurrentBlockOrNull();
      const localNumber = lCurrent ? lCurrent.number : -1;
      let rCurrent;
      if (isNaN(to)) {
        rCurrent = yield Q.nfcall(node.blockchain.current);
      } else {
        rCurrent = yield Q.nfcall(node.blockchain.block, to);
      }
      to = rCurrent.number;

      //=======
      // Peers (just for P2P download)
      //=======
      let peers = [];
      if (!nopeers) {
        watcher.writeStatus('Peers...');
        const merkle = yield dal.merkleForPeers();
        const getPeers = Q.nbind(node.network.peering.peers.get, node);
        const json2 = yield getPeers({});
        const rm = new NodesMerkle(json2);
        if(rm.root() != merkle.root()){
          const leavesToAdd = [];
          const json = yield getPeers({ leaves: true });
          _(json.leaves).forEach((leaf) => {
            if(merkle.leaves().indexOf(leaf) == -1){
              leavesToAdd.push(leaf);
            }
          });
          peers = yield leavesToAdd.map((leaf) => co(function*() {
            try {
              const json3 = yield getPeers({ "leaf": leaf });
              const jsonEntry = json3.leaf.value;
              const endpoint = jsonEntry.endpoints[0];
              watcher.writeStatus('Peer ' + endpoint);
              return jsonEntry;
            } catch (e) {
              logger.warn("Could not get peer of leaf %s, continue...", leaf);
              return null;
            }
          }));
        }
        else {
          watcher.writeStatus('Peers already known');
        }
      }

      peers = peers.filter((p) => p);

      //============
      // Blockchain
      //============
      logger.info('Downloading Blockchain...');

      // We use cautious mode if it is asked, or not particulary asked but blockchain has been started
      const cautious = (askedCautious === true || (askedCautious === undefined && localNumber >= 0));
      const downloader = new P2PDownloader(localNumber, to, CONST_MAX_SIMULTANEOUS_DOWNLOADS, peers, watcher);

      downloader.start();

      let lastPullBlock = null;
      let dao = pulling.abstractDao({

        // Get the local blockchain current block
        localCurrent: () => co(function*() {
          if (cautious) {
            return yield dal.getCurrentBlockOrNull();
          } else {
            if (lCurrent && !lastPullBlock) {
              lastPullBlock = lCurrent;
            }
            return lastPullBlock;
          }
        }),

        // Get the remote blockchain (bc) current block
        remoteCurrent: (peer) => Promise.resolve(rCurrent),

        // Get the remote peers to be pulled
        remotePeers: () => co(function*() {
          return [node];
        }),

        // Get block of given peer with given block number
        getLocalBlock: (number) => dal.getBlock(number),

        downloadBlocks: (thePeer, number) => co(function *() {
          // Note: we don't care about the particular peer asked by the method. We use the network instead.
          const numberOffseted = number - (localNumber + 1);
          const targetChunk = Math.floor(numberOffseted / CONST_BLOCKS_CHUNK);
          // Return the download promise! Simple.
          return downloader.getChunk(targetChunk);
        }),


        applyBranch: (blocks) => co(function *() {
          if (cautious) {
            for (const block of blocks) {
              yield dao.applyMainBranch(block);
            }
          } else {
            yield server.BlockchainService.saveBlocksInMainBranch(blocks);
          }
          lastPullBlock = blocks[blocks.length - 1];
          watcher.appliedPercent(Math.floor(blocks[blocks.length - 1].number / to * 100));
          return true;
        }),

        applyMainBranch: (block) => co(function *() {
          const addedBlock = yield server.BlockchainService.submitBlock(block, true, constants.FORK_ALLOWED);
          server.streamPush(addedBlock);
          watcher.appliedPercent(Math.floor(block.number / to * 100));
        }),

        // Eventually remove forks later on
        removeForks: () => co(function*() {}),

        // Tells wether given peer is a member peer
        isMemberPeer: (thePeer) => co(function *() {
          let idty = yield dal.getWrittenIdtyByPubkey(thePeer.pubkey);
          return (idty && idty.member) || false;
        })
      });

      const logInterval = setInterval(() => logRemaining(to), EVAL_REMAINING_INTERVAL);
      yield pulling.pull(conf, dao);

      // Finished blocks
      watcher.downloadPercent(100.0);
      watcher.appliedPercent(100.0);

      if (logInterval) {
        clearInterval(logInterval);
      }

      // Save currency parameters given by root block
      const rootBlock = yield server.dal.getBlock(0);
      yield BlockchainService.saveParametersForRootBlock(rootBlock);
      server.dal.blockDAL.cleanCache();

      //=======
      // Peers
      //=======
      if (!nopeers) {
        watcher.writeStatus('Peers...');
        yield syncPeer(node);
        const merkle = yield dal.merkleForPeers();
        const getPeers = Q.nbind(node.network.peering.peers.get, node);
        const json2 = yield getPeers({});
        const rm = new NodesMerkle(json2);
        if(rm.root() != merkle.root()){
          const leavesToAdd = [];
          const json = yield getPeers({ leaves: true });
          _(json.leaves).forEach((leaf) => {
            if(merkle.leaves().indexOf(leaf) == -1){
              leavesToAdd.push(leaf);
            }
          });
          for (const leaf of leavesToAdd) {
            try {
              const json3 = yield getPeers({ "leaf": leaf });
              const jsonEntry = json3.leaf.value;
              const sign = json3.leaf.value.signature;
              const entry = {};
              ["version", "currency", "pubkey", "endpoints", "block"].forEach((key) => {
                entry[key] = jsonEntry[key];
              });
              entry.signature = sign;
              watcher.writeStatus('Peer ' + entry.pubkey);
            } catch (e) {
              logger.warn(e);
            }
          }
        }
        else {
          watcher.writeStatus('Peers already known');
        }
      }

      watcher.end();
      that.push({ sync: true });
      logger.info('Sync finished.');
    } catch (err) {
      that.push({ sync: false, msg: err });
      err && watcher.writeStatus(err.message || (err.uerr && err.uerr.message) || String(err));
      watcher.end();
      throw err;
    }
  });

  function getVucoin(theHost, thePort, options) {
    return new Promise(function (resolve, reject) {
      vucoin(theHost, thePort, function (err, node) {
        if (err) {
          return reject('Cannot sync: ' + err);
        }
        resolve(node);
      }, options);
    });
  }

  //============
  // Peer
  //============
  function syncPeer (node) {

    // Global sync vars
    const remotePeer = new Peer({});
    let remoteJsonPeer = {};

    return co(function *() {
      const json = yield Q.nfcall(node.network.peering.get);
      remotePeer.copyValuesFrom(json);
      const entry = remotePeer.getRaw();
{'fraction_non_alphanumeric': 0.14121688391134535, 'fraction_numerical': 0.005505623264787539, 'mean_word_length': 2.218537028623353, 'pattern_counts': {'":': 4, '<': 7, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3775689', 'n_tokens_mistral': 6270, 'n_tokens_neox': 6098, 'n_words': 1451}