Paralell processing of TXs within a block

.env-example

@@ -8,3 +8,5 @@ TX_REST_API_PORT=5455
 TX_REST_API_IP=localhost
 START_BLOCK_HEIGHT=525000
 SEEN_TX_MAX=100000
+FILTER_CONCURRENCY=20
+MEMO_TX_CONCURRENCY=20

README.md

@@ -61,6 +61,8 @@ See `.env-example`. Key variables:
 | `RPC_IP` / `RPC_PORT` | `172.17.0.1` / `8332` | Full node RPC |
 | `ZMQ_PORT` | `28332` | Full node ZMQ |
 | `TX_REST_API_PORT` | `5455` | TX indexer control API |
+| `FILTER_CONCURRENCY` | `20` | Parallel Memo tx detection per block |
+| `MEMO_TX_CONCURRENCY` | `20` | Parallel Memo tx processing per block |
 
 ## Tests
 

config/index.js

@@ -16,6 +16,13 @@ export default {
   zmqBlockQueueMax: process.env.ZMQ_BLOCK_QUEUE_MAX ? parseInt(process.env.ZMQ_BLOCK_QUEUE_MAX) : 1000,
   txCacheMax: process.env.TX_CACHE_MAX ? parseInt(process.env.TX_CACHE_MAX) : 100000,
 
+  filterConcurrency: process.env.FILTER_CONCURRENCY
+    ? parseInt(process.env.FILTER_CONCURRENCY)
+    : 20,
+  memoTxConcurrency: process.env.MEMO_TX_CONCURRENCY
+    ? parseInt(process.env.MEMO_TX_CONCURRENCY)
+    : 20,
+
   startBlockHeight: process.env.START_BLOCK_HEIGHT
     ? parseInt(process.env.START_BLOCK_HEIGHT)
     : 525000,

dev-docs/architecture.md

@@ -60,8 +60,8 @@ Controllers
         │
         ▼
 Use cases
-  index-blocks.js      — processBlock, processMemoTx
-  filter-block.js      — filterMemoTxs (parallel pre-check)
+  index-blocks.js      — processBlock, processMemoTx (parallel per block)
+  filter-block.js      — filterMemoTxs (parallel pre-check, block order preserved)
   state.js             — synced block height
   action-types/*.js    — per Memo action handlers
   utils.js
@@ -157,7 +157,7 @@ Each store is a separate LevelDB with JSON values. Keys are txids, addresses, or
 | `zeromq` | Indexer | Subscribe to full node |
 | `@chris.troutner/retry-queue` | Indexer | Retry RPC on transient failure |
 | `axios` | Indexer | psf-memo-db REST client |
-| `p-queue` / `p-retry` | Indexer | Parallel block filter with retries |
+| `p-queue` / `p-retry` | Indexer | Parallel block filter and parallel per-tx processing within a block |
 | `express` | Indexer | TX indexer control API |
 | `level` | psf-memo-db | Embedded JSON LevelDB |
 | `koa` + `koa-router` | psf-memo-db | REST server |

dev-docs/design-decisions-and-tradeoffs.md

@@ -51,15 +51,21 @@ This document records **why** the Memo indexer stack looks the way it does, incl
 
 **Risk:** Indexer state diverges from memo.sv for excluded actions until handlers are added.
 
-## 4. No DAG sort within blocks
+## 4. Parallel processing within blocks (no DAG sort)
 
-**Decision:** Process filtered Memo txids in block order, sequentially.
+**Decision:** Filter and process Memo txs concurrently within a block using `p-queue`. Default concurrency is 20 for both phases (`FILTER_CONCURRENCY`, `MEMO_TX_CONCURRENCY`). Filter results are reordered to match block tx order before processing starts.
 
-**SLP context:** SEND transactions may spend token outputs created earlier in the same block; DAG sort orders them correctly.
+**SLP context:** SEND transactions may spend token outputs created earlier in the same block; DAG sort orders them correctly and writes must stay serial when UTXO state overlaps.
 
-**Memo context:** Social actions do not consume each other’s UTXOs in a token graph. Replies reference parent txids by hash but do not require reordering spends within the block.
+**Memo context:** Social actions do not consume each other’s UTXOs in a token graph. Most handler writes are keyed by `txid` and are independent across transactions. Parallel `processMemoTx` is safe for typical blocks.
 
-**Tradeoff:** If future Memo actions introduce intra-block dependencies, DAG or topological sort may be needed—unlikely for current social ops.
+**Soft ordering:** Likes that reference a post in the same block may compute `tip = 0` if the post handler has not finished yet. Replies still persist parent/child links and post bodies regardless of completion order. Profile/follow handlers keyed by address can race if the same signer emits multiple updates in one block (rare).
+
+**Within a single tx:** Multiple Memo `OP_RETURN` outputs in one transaction are still dispatched sequentially inside `processMemoTx`.
+
+**Tradeoff:** Higher throughput vs. HTTP contention on `psf-memo-db`. Lower `MEMO_TX_CONCURRENCY` if the DB service becomes the bottleneck.
+
+**Alternative rejected:** Full serial processing—simple but leaves RPC/DB latency on the table during IBD when blocks contain many unrelated Memo txs.
 
 ## 5. Scan all transaction outputs for OP_RETURN
 

dev-docs/theory-of-operation.md

@@ -63,7 +63,7 @@ sequenceDiagram
   loop each candidate txid
     FB->>TX: isMemoTx(txid)
   end
-  loop each memo txid
+  loop each memo txid (parallel, up to MEMO_TX_CONCURRENCY)
     BI->>BI: processMemoTx(txid)
     BI->>DB: reads/writes via handlers
   end
@@ -71,11 +71,11 @@ sequenceDiagram
 
 **Step 1 — Fetch block:** `getblock` returns the list of transaction ids in the block (verbosity 1).
 
-**Step 2 — Filter:** `filterMemoTxs` runs up to 20 concurrent `isMemoTx` checks (`p-queue`). Each check loads the transaction (with in-memory cache in `transaction.js`) and scans **all outputs** for a Memo `OP_RETURN`.
+**Step 2 — Filter:** `filterMemoTxs` runs up to `FILTER_CONCURRENCY` (default 20) concurrent `isMemoTx` checks (`p-queue`). Each check loads the transaction (with in-memory cache in `transaction.js`) and scans **all outputs** for a Memo `OP_RETURN`. Results are returned in **block tx order** (parallel checks use a set, then the original block list is filtered).
 
 **Why scan all outputs?** Unlike SLP, which conventionally places token data in `vout[0]`, Memo actions may appear in any output’s `scriptPubKey`. The Go reference iterates every `TxOut`.
 
-**Step 3 — Process:** For each Memo txid, `processMemoTx` runs sequentially within the block (no DAG sort—Memo social actions do not form token-like dependency chains within a block).
+**Step 3 — Process:** `processMemoTxs` runs up to `MEMO_TX_CONCURRENCY` (default 20) concurrent `processMemoTx` calls via `p-queue`. There is no DAG sort—Memo social actions do not form token-like dependency chains within a block. Multiple OP_RETURN outputs in a **single** transaction are still handled sequentially inside `processMemoTx`.
 
 ## Phase 3: Processing a single Memo transaction
 

production/docker/block-indexer/.env-example

@@ -14,3 +14,5 @@ TX_REST_API_IP=172.17.0.1
 
 START_BLOCK_HEIGHT=525000
 EXIT_ON_MISSING_BACKUP=false
+FILTER_CONCURRENCY=20
+MEMO_TX_CONCURRENCY=20

psf-memo-block-indexer.js

@@ -69,7 +69,7 @@ async function start () {
     console.log('TX indexer started.')
 
     let loopCnt = 0
-    let liveStatus = status
+    const liveStatus = status
     do {
       let blockHeight = await queue.addToQueue(adapters.rpc.getBlockCount, {})
       const block = adapters.zmq.getBlock()

src/use-cases/filter-block.js

@@ -4,6 +4,7 @@
 
 import PQueue from 'p-queue'
 import pRetry from 'p-retry'
+import config from '../../config/index.js'
 
 class FilterBlock {
   constructor (localConfig = {}) {
@@ -27,17 +28,17 @@ class FilterBlock {
   }
 
   async filterMemoTxs (txids) {
-    const memoTxs = []
+    const memoTxSet = new Set()
     const tasks = txids.map((txid) => async () => {
       const isMemo = await this.retryWrapper(
         this.adapters.transaction.isMemoTx.bind(this.adapters.transaction),
         txid
       )
-      if (isMemo) memoTxs.push(txid)
+      if (isMemo) memoTxSet.add(txid)
     })
 
     await this.pQueue.addAll(tasks)
-    return memoTxs
+    return txids.filter((txid) => memoTxSet.has(txid))
   }
 }
 

src/use-cases/index-blocks.js

@@ -3,6 +3,8 @@
 */
 
 import RetryQueue from '@chris.troutner/retry-queue'
+import PQueue from 'p-queue'
+import config from '../../config/index.js'
 import FilterBlock from './filter-block.js'
 import { findMemoOutputs, getSignerAddress } from '../lib/memo-parser.js'
 import { dispatchMemoAction } from './action-types/index.js'
@@ -15,6 +17,7 @@ class IndexBlocks {
     this.adapters = localConfig.adapters
     this.filterBlock = new FilterBlock({ adapters: this.adapters })
     this.retryQueue = new RetryQueue()
+    this.pQueue = new PQueue({ concurrency: config.memoTxConcurrency })
     this.processBlock = this.processBlock.bind(this)
     this.processMemoTx = this.processMemoTx.bind(this)
     this.processMemoTxs = this.processMemoTxs.bind(this)
@@ -63,9 +66,10 @@ class IndexBlocks {
   }
 
   async processMemoTxs (txids, blockHeight) {
-    for (let i = 0; i < txids.length; i++) {
-      await this.processMemoTx(txids[i], blockHeight)
-    }
+    const tasks = txids.map((txid) => async () => {
+      await this.processMemoTx(txid, blockHeight)
+    })
+    await this.pQueue.addAll(tasks)
     return true
   }
 

test/unit/use-cases/filter-block.unit.js

@@ -5,10 +5,11 @@ import FilterBlock from '../../../src/use-cases/filter-block.js'
 describe('#FilterBlock', () => {
   let uut
   let sandbox
+  let adapters
 
   beforeEach(() => {
     sandbox = sinon.createSandbox()
-    const adapters = {
+    adapters = {
       transaction: {
         isMemoTx: sandbox.stub()
       }
@@ -26,4 +27,15 @@ describe('#FilterBlock', () => {
     const result = await uut.filterMemoTxs(['tx1', 'tx2', 'tx3'])
     assert.deepEqual(result, ['tx1', 'tx3'])
   })
+
+  it('should preserve block tx order when filtering in parallel', async () => {
+    adapters.transaction.isMemoTx.reset()
+    adapters.transaction.isMemoTx.callsFake(async (txid) => {
+      return txid === 'tx-a' || txid === 'tx-c' || txid === 'tx-e'
+    })
+
+    const blockOrder = ['tx-a', 'tx-b', 'tx-c', 'tx-d', 'tx-e']
+    const result = await uut.filterMemoTxs(blockOrder)
+    assert.deepEqual(result, ['tx-a', 'tx-c', 'tx-e'])
+  })
 })

test/unit/use-cases/index-blocks.unit.js

@@ -0,0 +1,66 @@
+import { assert } from 'chai'
+import sinon from 'sinon'
+import IndexBlocks from '../../../src/use-cases/index-blocks.js'
+
+describe('#IndexBlocks', () => {
+  let uut
+  let sandbox
+  let adapters
+
+  beforeEach(() => {
+    sandbox = sinon.createSandbox()
+    adapters = {
+      ptxDb: {
+        get: sandbox.stub().rejects(new Error('not found')),
+        create: sandbox.stub().resolves({ success: true })
+      },
+      processErrorDb: { create: sandbox.stub().resolves({ success: true }) },
+      transaction: {
+        get: sandbox.stub().resolves({
+          txid: 'tx1',
+          vin: [{ txid: 'prev', vout: 0, scriptSig: { hex: '471044...' } }],
+          vout: [{
+            scriptPubKey: {
+              hex: '6a046d020102',
+              addresses: ['bitcoincash:qptest']
+            }
+          }]
+        })
+      }
+    }
+    uut = new IndexBlocks({ adapters })
+  })
+
+  afterEach(() => sandbox.restore())
+
+  describe('#processMemoTxs', () => {
+    it('should process all memo txs in parallel', async () => {
+      const processMemoTx = sandbox.stub(uut, 'processMemoTx').resolves(true)
+
+      await uut.processMemoTxs(['tx-a', 'tx-b', 'tx-c'], 600000)
+
+      assert.equal(processMemoTx.callCount, 3)
+      assert.deepEqual(
+        processMemoTx.getCalls().map((call) => call.args[0]).sort(),
+        ['tx-a', 'tx-b', 'tx-c']
+      )
+      processMemoTx.getCalls().forEach((call) => {
+        assert.equal(call.args[1], 600000)
+      })
+    })
+
+    it('should fail the block if any memo tx fails', async () => {
+      sandbox.stub(uut, 'processMemoTx')
+        .onFirstCall().resolves(true)
+        .onSecondCall().rejects(new Error('db error'))
+        .onThirdCall().resolves(true)
+
+      try {
+        await uut.processMemoTxs(['tx-a', 'tx-b', 'tx-c'], 600000)
+        assert.fail('Expected processMemoTxs to throw')
+      } catch (err) {
+        assert.include(err.message, 'db error')
+      }
+    })
+  })
+})