feat: 0-conf circuit breaker

lib/db/Database.ts

@@ -11,6 +11,7 @@ import KeyProvider from './models/KeyProvider';
 import MarkedSwap from './models/MarkedSwap';
 import Pair from './models/Pair';
 import PendingEthereumTransaction from './models/PendingEthereumTransaction';
+import PendingLockupTransaction from './models/PendingLockupTransaction';
 import Referral from './models/Referral';
 import ReverseRoutingHint from './models/ReverseRoutingHint';
 import ReverseSwap from './models/ReverseSwap';
@@ -113,9 +114,10 @@ class Database {
 
     await Promise.all([Swap.sync(), ReverseSwap.sync()]);
     await Promise.all([
+      MarkedSwap.sync(),
       ChannelCreation.sync(),
       ReverseRoutingHint.sync(),
-      MarkedSwap.sync(),
+      PendingLockupTransaction.sync(),
     ]);
   };
 
@@ -138,6 +140,7 @@ class Database {
     ChannelCreation.load(Database.sequelize);
     DatabaseVersion.load(Database.sequelize);
     ReverseRoutingHint.load(Database.sequelize);
+    PendingLockupTransaction.load(Database.sequelize);
     PendingEthereumTransaction.load(Database.sequelize);
   };
 }

lib/db/models/PendingLockupTransaction.ts

@@ -0,0 +1,47 @@
+import { DataTypes, Model, Sequelize } from 'sequelize';
+import Swap from './Swap';
+
+export type PendingLockupTransactionType = {
+  swapId: string;
+  chain: string;
+};
+
+class PendingLockupTransaction
+  extends Model
+  implements PendingLockupTransactionType
+{
+  public swapId!: string;
+  public chain!: string;
+
+  public static load = (sequelize: Sequelize) => {
+    PendingLockupTransaction.init(
+      {
+        swapId: {
+          type: new DataTypes.STRING(255),
+          primaryKey: true,
+          allowNull: false,
+        },
+        chain: {
+          type: new DataTypes.STRING(255),
+          allowNull: false,
+        },
+      },
+      {
+        sequelize,
+        modelName: 'pendingLockupTransaction',
+        indexes: [
+          {
+            unique: false,
+            fields: ['chain'],
+          },
+        ],
+      },
+    );
+
+    PendingLockupTransaction.belongsTo(Swap, {
+      foreignKey: 'swapId',
+    });
+  };
+}
+
+export default PendingLockupTransaction;

lib/db/repositories/PendingLockupTransactionRepository.ts

@@ -0,0 +1,22 @@
+import PendingLockupTransaction from '../models/PendingLockupTransaction';
+
+class PendingLockupTransactionRepository {
+  public static create = (swapId: string, chain: string) =>
+    PendingLockupTransaction.create({ swapId, chain });
+
+  public static destroy = async (swapId: string) =>
+    PendingLockupTransaction.destroy({
+      where: {
+        swapId,
+      },
+    });
+
+  public static getForChain = (chain: string) =>
+    PendingLockupTransaction.findAll({
+      where: {
+        chain,
+      },
+    });
+}
+
+export default PendingLockupTransactionRepository;

lib/rates/LockupTransactionTracker.ts

@@ -0,0 +1,83 @@
+import AsyncLock from 'async-lock';
+import { Op } from 'sequelize';
+import Logger from '../Logger';
+import { formatError, getChainCurrency, splitPairId } from '../Utils';
+import { IChainClient } from '../chain/ChainClient';
+import Swap from '../db/models/Swap';
+import PendingLockupTransactionRepository from '../db/repositories/PendingLockupTransactionRepository';
+import SwapRepository from '../db/repositories/SwapRepository';
+import { Currency } from '../wallet/WalletManager';
+import RateProvider from './RateProvider';
+
+class LockupTransactionTracker {
+  private readonly lock = new AsyncLock();
+  private readonly zeroConfAcceptedMap = new Map<string, boolean>();
+
+  constructor(
+    private readonly logger: Logger,
+    currencies: Map<string, Currency>,
+    private readonly rateProvider: RateProvider,
+  ) {
+    for (const currency of currencies.values()) {
+      if (currency.chainClient === undefined) {
+        continue;
+      }
+
+      this.listenToBlocks(currency.chainClient);
+      this.zeroConfAcceptedMap.set(currency.chainClient.symbol, true);
+    }
+  }
+
+  public zeroConfAccepted = (symbol: string): boolean =>
+    this.zeroConfAcceptedMap.get(symbol) || false;
+
+  public addPendingTransactionToTrack = async (swap: Swap) => {
+    const { base, quote } = splitPairId(swap.pair);
+    await PendingLockupTransactionRepository.create(
+      swap.id,
+      getChainCurrency(base, quote, swap.orderSide, false),
+    );
+  };
+
+  private listenToBlocks = (chainClient: IChainClient) => {
+    chainClient.on('block', async () => {
+      await this.lock.acquire(chainClient.symbol, () =>
+        this.checkPendingLockupsForChain(chainClient),
+      );
+    });
+  };
+
+  private checkPendingLockupsForChain = async (chainClient: IChainClient) => {
+    const pendingLockups = await PendingLockupTransactionRepository.getForChain(
+      chainClient.symbol,
+    );
+    const swaps = await SwapRepository.getSwaps({
+      id: {
+        [Op.in]: pendingLockups.map((p) => p.swapId),
+      },
+    });
+
+    for (const swap of swaps) {
+      try {
+        const info = await chainClient.getRawTransactionVerbose(
+          swap.lockupTransactionId!,
+        );
+        if (info.confirmations !== undefined && info.confirmations > 0) {
+          this.logger.debug(
+            `Pending lockup transaction of Submarine Swap ${swap.id} (${swap.lockupTransactionId}) confirmed`,
+          );
+          await PendingLockupTransactionRepository.destroy(swap.id);
+        }
+      } catch (e) {
+        this.logger.warn(
+          `Could find pending lockup transaction of Submarine Swap ${swap.id} (${swap.lockupTransactionId}): ${formatError(e)}`,
+        );
+        this.logger.warn(`Disabling 0-conf for ${chainClient.symbol}`);
+        this.zeroConfAcceptedMap.set(chainClient.symbol, false);
+        await this.rateProvider.setZeroConfAmount(chainClient.symbol, 0);
+      }
+    }
+  };
+}
+
+export default LockupTransactionTracker;

lib/rates/RateProvider.ts

@@ -153,6 +153,11 @@ class RateProvider {
 
   public has = (pairId: string) => this.configPairs.has(pairId);
 
+  public setZeroConfAmount = async (symbol: string, amount: number) => {
+    this.zeroConfAmounts.set(symbol, amount);
+    await this.updateRates();
+  };
+
   private updateRates = async () => {
     // The fees for the ERC20 tokens depend on the rates
     // Updating rates and fees at the same time would result in a race condition

lib/service/Service.ts

@@ -66,6 +66,7 @@ import {
   GetInfoResponse,
   LightningInfo,
 } from '../proto/boltzrpc_pb';
+import LockupTransactionTracker from '../rates/LockupTransactionTracker';
 import RateProvider from '../rates/RateProvider';
 import { PairTypeLegacy } from '../rates/providers/RateProviderLegacy';
 import ErrorsSwap from '../swap/Errors';
@@ -185,6 +186,11 @@ class Service {
       config.retryInterval,
       blocks,
       config.swap,
+      new LockupTransactionTracker(
+        this.logger,
+        this.currencies,
+        this.rateProvider,
+      ),
     );
 
     this.eventHandler = new EventHandler(

lib/swap/SwapManager.ts

@@ -49,6 +49,7 @@ import ChannelCreationRepository from '../db/repositories/ChannelCreationReposit
 import ReverseRoutingHintRepository from '../db/repositories/ReverseRoutingHintRepository';
 import ReverseSwapRepository from '../db/repositories/ReverseSwapRepository';
 import SwapRepository from '../db/repositories/SwapRepository';
+import LockupTransactionTracker from '../rates/LockupTransactionTracker';
 import RateProvider from '../rates/RateProvider';
 import Blocks from '../service/Blocks';
 import InvoiceExpiryHelper from '../service/InvoiceExpiryHelper';
@@ -147,6 +148,7 @@ class SwapManager {
     retryInterval: number,
     private readonly blocks: Blocks,
     swapConfig: SwapConfig,
+    lockupTransactionTracker: LockupTransactionTracker,
   ) {
     this.deferredClaimer = new DeferredClaimer(
       this.logger,
@@ -166,6 +168,7 @@ class SwapManager {
       retryInterval,
       this.blocks,
       this.deferredClaimer,
+      lockupTransactionTracker,
     );
 
     this.reverseRoutingHints = new ReverseRoutingHints(

lib/swap/SwapNursery.ts

@@ -44,6 +44,7 @@ import {
   LightningClient,
 } from '../lightning/LightningClient';
 import FeeProvider from '../rates/FeeProvider';
+import LockupTransactionTracker from '../rates/LockupTransactionTracker';
 import RateProvider from '../rates/RateProvider';
 import Blocks from '../service/Blocks';
 import TimeoutDeltaProvider from '../service/TimeoutDeltaProvider';
@@ -101,12 +102,18 @@ class SwapNursery extends TypedEventEmitter<SwapNurseryEvents> {
     private retryInterval: number,
     blocks: Blocks,
     private readonly claimer: DeferredClaimer,
+    lockupTransactionTracker: LockupTransactionTracker,
   ) {
     super();
 
     this.logger.info(`Setting Swap retry interval to ${retryInterval} seconds`);
 
-    this.utxoNursery = new UtxoNursery(this.logger, this.walletManager, blocks);
+    this.utxoNursery = new UtxoNursery(
+      this.logger,
+      this.walletManager,
+      blocks,
+      lockupTransactionTracker,
+    );
     this.lightningNursery = new LightningNursery(this.logger);
     this.invoiceNursery = new InvoiceNursery(this.logger);
     this.channelNursery = new ChannelNursery(

lib/swap/UtxoNursery.ts

@@ -27,6 +27,7 @@ import ReverseSwap from '../db/models/ReverseSwap';
 import Swap from '../db/models/Swap';
 import ReverseSwapRepository from '../db/repositories/ReverseSwapRepository';
 import SwapRepository from '../db/repositories/SwapRepository';
+import LockupTransactionTracker from '../rates/LockupTransactionTracker';
 import Blocks from '../service/Blocks';
 import Wallet from '../wallet/Wallet';
 import WalletLiquid from '../wallet/WalletLiquid';
@@ -72,6 +73,7 @@ class UtxoNursery extends TypedEventEmitter<{
     private readonly logger: Logger,
     private readonly walletManager: WalletManager,
     private readonly blocks: Blocks,
+    private readonly lockupTransactionTracker: LockupTransactionTracker,
   ) {
     super();
   }
@@ -472,7 +474,10 @@ class UtxoNursery extends TypedEventEmitter<{
 
     // Confirmed transactions do not have to be checked for 0-conf criteria
     if (!confirmed) {
-      if (updatedSwap.acceptZeroConf !== true) {
+      if (
+        updatedSwap.acceptZeroConf !== true ||
+        !this.lockupTransactionTracker.zeroConfAccepted(chainClient.symbol)
+      ) {
         this.emit('swap.lockup.zeroconf.rejected', {
           transaction,
           swap: updatedSwap,
@@ -481,12 +486,12 @@ class UtxoNursery extends TypedEventEmitter<{
         return;
       }
 
-      const signalsRBF = await this.transactionSignalsRbf(
+      const signalsRbf = await this.transactionSignalsRbf(
         chainClient,
         transaction,
       );
 
-      if (signalsRBF) {
+      if (signalsRbf) {
         this.emit('swap.lockup.zeroconf.rejected', {
           transaction,
           swap: updatedSwap,
@@ -522,6 +527,7 @@ class UtxoNursery extends TypedEventEmitter<{
           updatedSwap.id
         }: ${transaction.getId()}:${swapOutput.vout}`,
       );
+      await this.lockupTransactionTracker.addPendingTransactionToTrack(swap);
     }
 
     chainClient.removeOutputFilter(swapOutput.script);