fixes and fixes.

docker-compose.yml

@@ -53,6 +53,7 @@ services:
       PORT: 3000
       MAIN_DB_CONNECTION_URI: postgres://someuser:sosecret@psql:5432/rundb_regtest
       BLOB_DB_CONNECTION_URI: postgres://someuser:sosecret@psql:5432/blobs_regtest
+      BITCOIND_REST_URL: http://bitcoind:18332/rest
       RABBITMQ_URI: amqp://guest:guest@rabbitmq:5672
 
   psql:

package.json

@@ -63,7 +63,7 @@
     "stream-json": "^1.7.4"
   },
   "optionalDependencies": {
-    "zeromq": "^5.2.8"
+    "zeromq": "^6.0.0-beta.6"
   },
   "devDependencies": {
     "@runonbitcoin/nimble": "^1.0.13",

src/bin/start-api.js

@@ -20,6 +20,7 @@ const {
   RabbitQueue,
   buildMainServer
 } = require('../index')
+const { ExecutingSet } = require('../executing-set')
 
 // ------------------------------------------------------------------------------------------------
 // Globals
@@ -69,7 +70,8 @@ async function main () {
   await rabbitChannel.prefetch(20)
   execQueue = new RabbitQueue(rabbitChannel, 'exectx')
   trustQueue = new RabbitQueue(rabbitChannel, 'trusttx')
-  indexManager = new ExecutionManager(blobs, execQueue, trustQueue)
+  const execSet = new ExecutingSet(ds)
+  indexManager = new ExecutionManager(blobs, execQueue, trustQueue, execSet)
   server = buildMainServer(ds, blobs, indexManager, logger)
   await execQueue.setUp()
   await trustQueue.setUp()

src/bin/start-crawler.js

@@ -32,6 +32,7 @@ const {
 } = require('../index')
 
 const path = require('path')
+const { ExecutingSet } = require('../executing-set')
 
 // const { Crawler, BitcoinNodeConnection, BitcoinZmq, BitcoinRpc } = require('../index')
 // const { KnexBlobStorage } = require('../data-sources/knex-blob-storage')
@@ -86,10 +87,12 @@ const knexBlob = knex({
 })
 const blobs = new KnexBlobStorage(knexBlob, undefined, api)
 const ds = new KnexDatasource(knexInstance)
+const execSet = new ExecutingSet(ds)
 
 let indexManager
 let crawler = null
 let execQueue = null
+let trustQueue = null
 let rabbitConnection = null
 
 // ------------------------------------------------------------------------------------------------
@@ -101,8 +104,10 @@ async function main () {
   const rabbitChannel = await rabbitConnection.createChannel()
   await rabbitChannel.prefetch(20)
   execQueue = new RabbitQueue(rabbitChannel, 'exectx')
+  trustQueue = new RabbitQueue(rabbitChannel, 'trusttx')
   await execQueue.setUp()
-  indexManager = new ExecutionManager(blobs, execQueue)
+  await trustQueue.setUp()
+  indexManager = new ExecutionManager(blobs, execQueue, trustQueue, execSet)
   await indexManager.setUp()
   crawler = new Crawler(indexManager, api, ds, logger, { initialBlockConcurrency: INITIAL_BLOCK_CONCURRENCY })
   await ds.setUp()
@@ -128,6 +133,9 @@ async function shutdown () {
   if (execQueue !== null) {
     await execQueue.tearDown()
   }
+  if (trustQueue !== null) {
+    await trustQueue.tearDown()
+  }
   if (rabbitConnection !== null) {
     await rabbitConnection.close()
   }

src/bin/start-execution-worker.js

@@ -19,9 +19,9 @@ const {
   knex,
   KnexBlobStorage,
   TrustAllTrustList,
-  Executor,
   RabbitQueue,
-  ExecutionWorker
+  ExecutionWorker,
+  execution: { Executor }
 } = require('../index')
 const { ExecutingSet } = require('../executing-set')
 
@@ -77,7 +77,8 @@ async function main () {
   await rabbitChannel.prefetch(1)
   execQueue = new RabbitQueue(rabbitChannel, 'exectx')
   trustQueue = new RabbitQueue(rabbitChannel, 'trusttx')
-  worker = new ExecutionWorker(indexer, execQueue, trustQueue)
+  const execSet = new ExecutingSet(ds)
+  worker = new ExecutionWorker(indexer, execSet, execQueue, trustQueue)
 
   await execQueue.setUp()
   await trustQueue.setUp()

src/blockchain-api/bitcoin-node-connection.js

@@ -31,6 +31,9 @@ class BitcoinNodeConnection extends BlockchainApi {
 
   async fetch (txid) {
     const response = await fetch(`${this.url}/tx/${txid}.bin`)
+    if (response.status !== 200) {
+      throw new Error('error fetching tx')
+    }
     return response.buffer()
   }
 
@@ -44,7 +47,7 @@ class BitcoinNodeConnection extends BlockchainApi {
   }
 
   async onMempoolTx (fn) {
-    this.zmq.subscribe('rawtx', fn)
+    this.zmq.subscribe('hashtx', fn)
   }
 
   async onNewBlock (fn) {

src/blockchain-api/bitcoin-zmq.js

@@ -5,19 +5,20 @@ try {
 
 class BitcoinZmq {
   constructor (url) {
-    this.sock = zmq.socket('sub')
+    this.sock = new zmq.Subscriber()
     this.url = url
     this.handlers = new Map()
   }
 
   async connect () {
-    this.sock.connect(this.url)
-    this.sock.on('message', (topic, message) => {
+    await this.sock.connect(this.url)
+    for await (const [topic, msg] of this.sock) {
+      const txid = msg.toString('hex')
       const handler = this.handlers.get(topic.toString())
       if (handler) {
-        handler(message)
+        handler(txid)
       }
-    })
+    }
   }
 
   async subscribe (topic, handler) {

src/crawler.js

@@ -37,7 +37,8 @@ class Crawler {
     await this.api.setUp()
   }
 
-  async _receiveTransaction (rawTx, blockHeight = null) {
+  async _receiveTransaction (txid, blockHeight = null) {
+    const rawTx = await this.api.fetch(txid)
     return this.execManager.indexTxNow(rawTx, blockHeight)
   }
 

src/data-sources/knex-datasource.js

@@ -160,9 +160,11 @@ class KnexDatasource {
     return !!(result && result.indexed)
   }
 
-  async searchNonExecutedTxs () {
+  async searchNonExecutedTxs (limit) {
     return this.knex(TX.NAME)
       .where(TX.executed, false)
+      .orderBy(TX.time, 'ASC')
+      .limit(limit)
       .pluck(TX.txid)
   }
 

src/execution-manager.js

@@ -19,7 +19,7 @@ class ExecutionManager {
     const txid = await this.blobs.pushTx(null, txBuff)
     const rQueue = await this._execReplyQueue()
     this.executingSet.add(txid)
-    const result = await rQueue.publishAndAwaitResponse({ txid })
+    const result = await rQueue.publishAndAwaitResponse({ txid, blockHeight: blockHeight })
     this.executingSet.remove(txid)
     return result
   }
@@ -31,7 +31,6 @@ class ExecutionManager {
   // }
 
   async trustTxLater (txid, trust) {
-    if (this.executingSet.check(txid)) { return }
     this.executingSet.add(txid)
     await this.trustQueue.publish({ txid, trust })
   }

src/execution-worker.js

@@ -1,15 +1,15 @@
 class ExecutionWorker {
-  constructor (indexer, execQueue, trustQueue) {
+  constructor (indexer, execSet, execQueue, trustQueue) {
     this.indexer = indexer
     this.execQueue = execQueue
     this.trustQueue = trustQueue
+    this.execSet = execSet
     this.execSubscription = null
     this.trustSubscription = null
   }
 
   async setUp () {
     this.execSubscription = await this.execQueue.subscribe(async ({ txid, blockHeight }) => {
-      // console.log('starting: ', txid)
       try {
         const result = await this.indexer.indexTxid(txid, blockHeight)
         await this._handleIndexResult(result)
@@ -40,16 +40,18 @@ class ExecutionWorker {
   }
 
   async _handleIndexResult (result) {
-    const enableProms = result.enables.map(async txid => {
-      return this.execQueue.publish({ txid })
-    })
-    const unknownDepsProms = result.unknownDeps.map(async txid => {
-      return this.execQueue.publish({ txid })
-    })
-    const missingDeps = result.missingDeps.map(async txid => {
+    const newTxidsToIndex = await Promise.all([...result.unknownDeps, ...result.missingDeps].map(async txid => {
+      return this.execSet.check(txid) ? null : txid // We only want the ones that are not there.
+    }))
+      .then(list => list.filter(txid => txid))
+      .then(list => [...list, ...result.enables]) // the enable ones are always included to avoid race conditions.
+
+    const promises = newTxidsToIndex.map(async txid => {
+      await this.execSet.add(txid)
       return this.execQueue.publish({ txid })
     })
-    await Promise.all([...enableProms, ...unknownDepsProms, ...missingDeps])
+
+    await Promise.all(promises)
   }
 }
 

src/execution/api-executor.js

@@ -35,7 +35,6 @@ class ApiExecutor {
 
   async execute (txid) {
     if (this.executing.has(txid)) return
-    this.logger.debug('Enqueueing', txid, 'for execution')
     const token = await this.pool.acquire()
     try {
       this.executing.add(txid)

src/execution/executor.js

@@ -68,9 +68,6 @@ class Executor {
 
   async execute (txid, trustList) {
     if (this.executing.has(txid)) return new ExecutionResult(false, [], null)
-
-    this.logger.debug('Enqueueing', txid, 'for execution')
-
     const worker = await this.pool.acquire()
     const txBuf = await this.blobs.pullTx(txid, () => null)
 
@@ -79,10 +76,7 @@ class Executor {
     this.executing.add(txid)
     try {
       const result = await worker.send('execute', { txid, hex, trustList })
-      if (result.missingDeps) {
-        return new ExecutionResult(false, result.missingDeps, null, null)
-      }
-      return new ExecutionResult(true, [], result)
+      return new ExecutionResult(result.success, result.missingDeps, result, result.error)
     } catch (e) {
       return new ExecutionResult(false, [], null, e)
     } finally {