Merge pull request #124 from coingaming/B1M2-5847-4

B1M-5847 Abort stuck channels at startup

Author: drownbes

btc-lsp/config/model

@@ -24,6 +24,7 @@ SwapIntoLn
   expiresAt UTCTime
   insertedAt UTCTime
   updatedAt UTCTime
+  psbtPendingChanId Lnd.PendingChannelId Maybe
   UniqueSwapIntoLnFundInvUuid uuid
   UniqueSwapIntoLnFundAddress fundAddress
   deriving Eq Show Generic

btc-lsp/src/BtcLsp/Psbt/PsbtOpener.hs

@@ -2,6 +2,7 @@
 
 module BtcLsp.Psbt.PsbtOpener
   ( openChannelPsbt,
+    abortChannelPsbt,
     OpenChannelPsbtResult (..),
     OpenUpdateEvt (..),
   )
@@ -141,18 +142,22 @@ data OpenChannelPsbtResult = OpenChannelPsbtResult
     fundAsync :: Async (Either Failure Lnd.ChannelPoint)
   }
 
+abortChannelPsbt :: Env m => Lnd.PendingChannelId -> m ()
+abortChannelPsbt p =
+  void $ runExceptT $ withLndT Lnd.fundingStateStep ($ shimCancelReq p)
+
 openChannelPsbt ::
   Env m =>
+  Lnd.PendingChannelId ->
   Lock ->
   [PsbtUtxo] ->
   NodePubKey ->
   OnChainAddress 'Gain ->
   Money 'Lsp 'OnChain 'Gain ->
   Privacy ->
   ExceptT Failure m OpenChannelPsbtResult
-openChannelPsbt lock utxos toPubKey changeAddress lspFee private = do
+openChannelPsbt pcid lock utxos toPubKey changeAddress lspFee private = do
   chan <- lift T.newTChanIO
-  pcid <- Lnd.newPendingChanId
   let openChannelRequest =
         openChannelReq pcid toPubKey (coerce (2 * amt)) (coerce amt) private
   let subUpdates u = void . T.atomically . T.writeTChan chan $ LndUpdate u
@@ -164,11 +169,11 @@ openChannelPsbt lock utxos toPubKey changeAddress lspFee private = do
   case res of
     Left e -> throwE . FailureInt . FailurePrivate $ inspect e
     Right _ -> do
-      fundA <- lift . spawnLink $ runExceptT $ fundStep pcid chan
+      fundA <- lift . spawnLink $ runExceptT $ fundStep chan
       pure $ OpenChannelPsbtResult chan fundA
   where
     amt = sumAmt utxos - coerce lspFee
-    fundStep pcid chan = do
+    fundStep chan = do
       upd <- T.atomically $ T.readTChan chan
       $(logTM) DebugS $ logStr $ "Got chan status update" <> inspect upd
       case upd of
@@ -179,15 +184,15 @@ openChannelPsbt lock utxos toPubKey changeAddress lspFee private = do
           sPsbtResp <- finalizePsbt psbt'
           $(logTM) DebugS $ logStr $ "Used psbt for funding:" <> inspect sPsbtResp
           void $ withLndT Lnd.fundingStateStep ($ psbtFinalizeReq pcid (Lnd.Psbt $ FNP.signedPsbt sPsbtResp))
-          fundStep pcid chan
+          fundStep chan
         LndUpdate (Lnd.OpenStatusUpdate _ (Just (Lnd.OpenStatusUpdateChanPending p))) -> do
           $(logTM) DebugS $ logStr $ "Chan is pending... mining..." <> inspect p
-          fundStep pcid chan
+          fundStep chan
         LndUpdate (Lnd.OpenStatusUpdate _ (Just (Lnd.OpenStatusUpdateChanOpen (Lnd.ChannelOpenUpdate cp)))) -> do
           $(logTM) DebugS $ logStr $ "Chan is open" <> inspect cp
           pure cp
         LndSubFail -> do
-          void $ withLndT Lnd.fundingStateStep ($ shimCancelReq pcid)
+          lift $ abortChannelPsbt pcid
           void $ lockUtxos (getOutPoint <$> utxos)
           throwE (FailureInt $ FailurePrivate "Lnd subscription failed. Trying to cancel psbt flow. Its ok if cancel fails")
         _ -> throwE (FailureInt $ FailurePrivate "Unexpected update")

btc-lsp/src/BtcLsp/Storage/Model.hs

@@ -21,6 +21,7 @@ import BtcLsp.Import.External
 import qualified BtcLsp.Import.Psql as Psql
 import Database.Persist.Quasi
 import Database.Persist.TH
+import qualified LndClient as Lnd
 import qualified LndClient.Data.ChannelBackup as Lnd
 
 -- You can define all of your database entities in the entities file.

btc-lsp/src/BtcLsp/Storage/Model/SwapIntoLn.hs

@@ -16,6 +16,7 @@ module BtcLsp.Storage.Model.SwapIntoLn
     getByUuidSql,
     getByFundAddressSql,
     withLockedRowSql,
+    getSwapsInPsbtThreadSql,
     UtxoInfo (..),
     SwapInfo (..),
   )
@@ -25,6 +26,7 @@ import BtcLsp.Import hiding (Storage (..))
 import qualified BtcLsp.Import.Psql as Psql
 import qualified BtcLsp.Math.Swap as Math
 import qualified BtcLsp.Storage.Util as Util
+import qualified LndClient as Lnd
 
 createIgnoreSql ::
   ( MonadIO m
@@ -58,6 +60,7 @@ createIgnoreSql userEnt fundAddr feeAndChangeAddr refundAddr expAt chanPrivacy =
         swapIntoLnFeeMiner = Money 0,
         swapIntoLnStatus = SwapWaitingFundChain,
         swapIntoLnPrivacy = chanPrivacy,
+        swapIntoLnPsbtPendingChanId = Nothing,
         swapIntoLnExpiresAt = expAt,
         swapIntoLnInsertedAt = ct,
         swapIntoLnUpdatedAt = ct
@@ -126,16 +129,19 @@ updateInPsbtThreadSql ::
   ( MonadIO m
   ) =>
   SwapIntoLnId ->
+  Lnd.PendingChannelId ->
   ReaderT Psql.SqlBackend m ()
-updateInPsbtThreadSql id0 = do
+updateInPsbtThreadSql id0 pcid = do
   ct <- getCurrentTime
   Psql.update $ \row -> do
     Psql.set
       row
       [ SwapIntoLnStatus
           Psql.=. Psql.val SwapInPsbtThread,
         SwapIntoLnUpdatedAt
-          Psql.=. Psql.val ct
+          Psql.=. Psql.val ct,
+        SwapIntoLnPsbtPendingChanId
+          Psql.=. Psql.val (Just pcid)
       ]
     Psql.where_ $
       ( row Psql.^. SwapIntoLnId
@@ -158,7 +164,9 @@ updateRevertInPsbtThreadSql id0 = do
       [ SwapIntoLnStatus
           Psql.=. Psql.val SwapWaitingPeer,
         SwapIntoLnUpdatedAt
-          Psql.=. Psql.val ct
+          Psql.=. Psql.val ct,
+        SwapIntoLnPsbtPendingChanId
+          Psql.=. Psql.val Nothing
       ]
     Psql.where_ $
       ( row Psql.^. SwapIntoLnId
@@ -180,13 +188,32 @@ updateRevertAllInPsbtThreadSql = do
       [ SwapIntoLnStatus
           Psql.=. Psql.val SwapWaitingPeer,
         SwapIntoLnUpdatedAt
-          Psql.=. Psql.val ct
+          Psql.=. Psql.val ct,
+        SwapIntoLnPsbtPendingChanId
+          Psql.=. Psql.val Nothing
       ]
     Psql.where_
       ( row Psql.^. SwapIntoLnStatus
           Psql.==. Psql.val SwapInPsbtThread
       )
 
+getSwapsInPsbtThreadSql ::
+  ( MonadIO m
+  ) =>
+  ReaderT
+    Psql.SqlBackend
+    m
+    [Entity SwapIntoLn]
+getSwapsInPsbtThreadSql =
+  Psql.select $
+    Psql.from $ \swap -> do
+      Psql.locking Psql.ForUpdate
+      Psql.where_
+        ( swap Psql.^. SwapIntoLnStatus
+            Psql.==. Psql.val SwapInPsbtThread
+        )
+      pure swap
+
 updateExpiredSql ::
   ( MonadIO m,
     KatipContext m

btc-lsp/src/BtcLsp/Thread/LnChanOpener.hs

@@ -2,6 +2,7 @@
 
 module BtcLsp.Thread.LnChanOpener
   ( apply,
+    cleanupInPsbtThreadChannels,
   )
 where
 
@@ -14,6 +15,7 @@ import qualified BtcLsp.Storage.Model.SwapIntoLn as SwapIntoLn
 import qualified BtcLsp.Storage.Model.SwapUtxo as SwapUtxo
 import Control.Concurrent.Extra
 import qualified Data.Set as Set
+import qualified LndClient as Lnd
 import qualified LndClient.Data.ChannelPoint as ChannelPoint
 import qualified LndClient.Data.Peer as Peer
 import qualified LndClient.RPC.Silent as LndSilent
@@ -43,15 +45,27 @@ apply = do
           <$> SwapIntoLn.getSwapsWaitingPeerSql
     mapM_
       ( \(swp, usr) -> do
-          void $ runSql $ SwapIntoLn.updateInPsbtThreadSql $ entityKey swp
+          pcid <- Lnd.newPendingChanId
+          void $ runSql $ SwapIntoLn.updateInPsbtThreadSql (entityKey swp) pcid
           spawnLink $
             runSql $ do
-              r <- openChanSql lock swp usr
+              r <- openChanSql pcid lock swp usr
               whenLeft r $ pure $ SwapIntoLn.updateRevertInPsbtThreadSql $ entityKey swp
       )
       swaps
     sleep300ms
 
+cleanupInPsbtThreadChannels :: Env m => m ()
+cleanupInPsbtThreadChannels = runSql $ do
+  swaps <- SwapIntoLn.getSwapsInPsbtThreadSql
+  mapM_ abortChan swaps
+  SwapIntoLn.updateRevertAllInPsbtThreadSql
+  where
+    abortChan (Entity _ swp) = do
+      case swapIntoLnPsbtPendingChanId swp of
+        Just pcid -> lift $ PO.abortChannelPsbt pcid
+        Nothing -> pure ()
+
 --
 -- TODO : Do not open channel in case where
 -- there not is enough liquidity to perform swap.
@@ -61,18 +75,20 @@ apply = do
 openChanSql ::
   ( Env m
   ) =>
+  Lnd.PendingChannelId ->
   Lock ->
   Entity SwapIntoLn ->
   Entity User ->
   ReaderT Psql.SqlBackend m (Either (Entity SwapIntoLn) ())
-openChanSql lock (Entity swapKey _) userEnt = do
+openChanSql pcid lock (Entity swapKey _) userEnt = do
   res <-
     SwapIntoLn.withLockedRowSql swapKey (== SwapInPsbtThread) $
       \swapVal -> do
         utxos <- SwapUtxo.getSpendableUtxosBySwapIdSql swapKey
         cpEither <- lift . runExceptT $ do
           r <-
             PO.openChannelPsbt
+              pcid
               lock
               (swapUtxoToPsbtUtxo . entityVal <$> utxos)
               (userNodePubKey $ entityVal userEnt)

btc-lsp/src/BtcLsp/Thread/Main.hs

@@ -11,7 +11,6 @@ import BtcLsp.Data.AppM (runApp)
 import qualified BtcLsp.Data.Env as Env
 import BtcLsp.Import
 import qualified BtcLsp.Storage.Migration as Storage
-import qualified BtcLsp.Storage.Model.SwapIntoLn as Storage
 import qualified BtcLsp.Thread.BlockScanner as BlockScanner
 import qualified BtcLsp.Thread.Expirer as Expirer
 import qualified BtcLsp.Thread.LnChanOpener as LnChanOpener
@@ -72,7 +71,7 @@ apply = do
       $(logTM) InfoS "Running postgres migrations..."
       Storage.migrateAll
       $(logTM) InfoS "Reverting SwapInPsbtThread -> SwapWaitingPeer"
-      void $ runSql Storage.updateRevertAllInPsbtThreadSql
+      void LnChanOpener.cleanupInPsbtThreadChannels
       log <- getYesodLog
       pool <- getSqlPool
       $(logTM) InfoS "Spawning lsp threads..."

btc-lsp/test/PsbtOpenerSpec.hs

@@ -11,6 +11,7 @@ import BtcLsp.Psbt.Utils (swapUtxoToPsbtUtxo)
 import qualified BtcLsp.Storage.Model.SwapUtxo as SwapUtxo
 import qualified BtcLsp.Thread.BlockScanner as BlockScanner
 import Control.Concurrent.Extra
+import qualified LndClient as Lnd
 import qualified LndClient.Data.Channel as CH
 import qualified LndClient.Data.GetInfo as Lnd
 import qualified LndClient.Data.ListChannels as ListChannels
@@ -55,7 +56,8 @@ spec = do
     profitAddr <- genAddress LndLsp
     Lnd.GetInfoResponse alicePubKey _ _ <- withLndTestT LndAlice Lnd.getInfo id
     lock <- liftIO newLock
-    openChanRes <- PO.openChannelPsbt lock psbtUtxos alicePubKey (unsafeNewOnChainAddress $ Lnd.address profitAddr) (coerce lspFee) Public
+    pcid <- Lnd.newPendingChanId
+    openChanRes <- PO.openChannelPsbt pcid lock psbtUtxos alicePubKey (unsafeNewOnChainAddress $ Lnd.address profitAddr) (coerce lspFee) Public
     void . lift . spawnLink $ do
       sleep1s
       mine 1 LndLsp
@@ -100,8 +102,9 @@ spec = do
     profitAddr <- genAddress LndLsp
     Lnd.GetInfoResponse alicePubKey _ _ <- withLndTestT LndAlice Lnd.getInfo id
     lock <- liftIO newLock
+    pcid <- Lnd.newPendingChanId
     openChanRes <-
-      PO.openChannelPsbt lock psbtUtxos alicePubKey (unsafeNewOnChainAddress $ Lnd.address profitAddr) (coerce lspFee) Public
+      PO.openChannelPsbt pcid lock psbtUtxos alicePubKey (unsafeNewOnChainAddress $ Lnd.address profitAddr) (coerce lspFee) Public
     void . lift . spawnLink $ do
       sleep1s
       mine 1 LndLsp
@@ -151,13 +154,15 @@ spec = do
     void $ lift $ runSql $ SwapUtxo.updateRefundedSql (entityKey <$> utxos0) (TxId "dummy refund tx")
     let psbtUtxos0 = swapUtxoToPsbtUtxo . entityVal <$> utxos0
 
-    openChanRes0 <- PO.openChannelPsbt psbtFlowLock psbtUtxos0 alicePubKey (unsafeNewOnChainAddress $ Lnd.address profitAddr) (coerce lspFee) Public
+    pcid0 <- Lnd.newPendingChanId
+    openChanRes0 <- PO.openChannelPsbt pcid0 psbtFlowLock psbtUtxos0 alicePubKey (unsafeNewOnChainAddress $ Lnd.address profitAddr) (coerce lspFee) Public
 
     utxos1 <- lift $ runSql $ SwapUtxo.getSpendableUtxosBySwapIdSql swp1Id
     void $ lift $ runSql $ SwapUtxo.updateRefundedSql (entityKey <$> utxos1) (TxId "dummy refund tx")
     let psbtUtxos1 = swapUtxoToPsbtUtxo . entityVal <$> utxos1
 
-    openChanRes1 <- PO.openChannelPsbt psbtFlowLock psbtUtxos1 alicePubKey (unsafeNewOnChainAddress $ Lnd.address profitAddr) (coerce lspFee) Public
+    pcid1 <- Lnd.newPendingChanId
+    openChanRes1 <- PO.openChannelPsbt pcid1 psbtFlowLock psbtUtxos1 alicePubKey (unsafeNewOnChainAddress $ Lnd.address profitAddr) (coerce lspFee) Public
 
     sleep5s