fix: Bitcoin Core work queue exhaustion when fetching transactions

BoltzExchange · Feb 9, 2024 · 029fb78 · 029fb78
1 parent d82bc20
commit 029fb78
Show file tree

Hide file tree

Showing 2 changed files with 75 additions and 21 deletions.
diff --git a/lib/swap/UtxoNursery.ts b/lib/swap/UtxoNursery.ts
@@ -59,6 +59,8 @@ class UtxoNursery extends TypedEventEmitter<{
     preimage: Buffer;
   };
 }> {
+  private static readonly maxParallelRequests = 6;
+
   // Locks
   private lock = new AsyncLock();
 
@@ -455,27 +457,11 @@ class UtxoNursery extends TypedEventEmitter<{
       }
     }
 
-    const inputTxs = (
-      await Promise.all(
-        Array.from(
-          new Set<string>(
-            transaction.ins.map((input) =>
-              getHexString(reverseBuffer(input.hash)),
-            ),
-          ).values(),
-        ).map((id) => chainClient.getRawTransaction(id)),
-      )
-    ).map((txHex) => parseTransaction(wallet.type, txHex));
-
-    const prevAddreses = inputTxs
-      .map((tx) => tx.outs)
-      .flat()
-      .map((output) =>
-        wallet.type === CurrencyType.Liquid
-          ? (wallet as WalletLiquid).encodeAddress(output.script, false)
-          : wallet.encodeAddress(output.script),
-      );
-
+    const prevAddreses = await this.getPreviousAddresses(
+      transaction,
+      chainClient,
+      wallet,
+    );
     if (prevAddreses.some(this.blocks.isBlocked)) {
       this.emit('swap.lockup.failed', {
         swap: updatedSwap,
@@ -575,6 +561,60 @@ class UtxoNursery extends TypedEventEmitter<{
 
     return false;
   };
+
+  private getPreviousAddresses = async (
+    transaction: Transaction | LiquidTransaction,
+    chainClient: ChainClient,
+    wallet: Wallet,
+  ) => {
+    const inputTxsIds = this.chunkArray(
+      Array.from(
+        new Set<string>(
+          transaction.ins.map((input) =>
+            getHexString(reverseBuffer(input.hash)),
+          ),
+        ).values(),
+      ),
+      UtxoNursery.maxParallelRequests,
+    );
+
+    const inputTxs = (
+      await Promise.all(
+        inputTxsIds.map((ids) => this.getTransactions(chainClient, ids)),
+      )
+    )
+      .flat()
+      .map((txHex) => parseTransaction(wallet.type, txHex));
+
+    return inputTxs
+      .map((tx) => tx.outs)
+      .flat()
+      .map((output) =>
+        wallet.type === CurrencyType.Liquid
+          ? (wallet as WalletLiquid).encodeAddress(output.script, false)
+          : wallet.encodeAddress(output.script),
+      );
+  };
+
+  private getTransactions = async (chainClient: ChainClient, ids: string[]) => {
+    const txs: string[] = [];
+
+    for (const id of ids) {
+      txs.push(await chainClient.getRawTransaction(id));
+    }
+
+    return txs;
+  };
+
+  private chunkArray = <T>(array: T[], size: number) => {
+    const chunks: T[][] = Array.from({ length: size }, () => []);
+
+    for (let i = 0; i < array.length; i++) {
+      chunks[i % size].push(array[i]);
+    }
+
+    return chunks.filter((chunk) => chunk.length !== 0);
+  };
 }
 
 export default UtxoNursery;
diff --git a/test/unit/swap/UtxoNursery.spec.ts b/test/unit/swap/UtxoNursery.spec.ts
@@ -970,4 +970,18 @@ describe('UtxoNursery', () => {
       false,
     );
   });
+
+  test.each`
+    data                              | size  | chunks
+    ${[0, 1]}                         | ${1}  | ${[[0, 1]]}
+    ${[0, 1]}                         | ${2}  | ${[[0], [1]]}
+    ${[0, 1]}                         | ${3}  | ${[[0], [1]]}
+    ${[0, 1]}                         | ${15} | ${[[0], [1]]}
+    ${[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]} | ${2}  | ${[[0, 2, 4, 6, 8], [1, 3, 5, 7, 9]]}
+    ${[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]} | ${3}  | ${[[0, 3, 6, 9], [1, 4, 7], [2, 5, 8]]}
+  `('should chunk arrays', ({ data, size, chunks }) => {
+    const res = nursery['chunkArray'](data, size);
+    expect(res.length).toEqual(chunks.length);
+    expect(res).toEqual(chunks);
+  });
 });