Remove race condition in postTransactionOld where pending (spent) UTxO could be selected as inputs

lib/core/cardano-wallet-core.cabal

@@ -209,6 +209,7 @@ library
       Cardano.Wallet.Unsafe
       Cardano.Wallet.Version
       Cardano.Wallet.Version.TH
+      Control.Concurrent.Concierge
       Crypto.Hash.Utils
       Data.Function.Utils
       Data.Time.Text
@@ -299,6 +300,7 @@ test-suite unit
     , http-types
     , iohk-monitoring
     , io-classes
+    , io-sim
     , lattices
     , lens
     , memory
@@ -407,6 +409,7 @@ test-suite unit
       Cardano.Wallet.RegistrySpec
       Cardano.Wallet.TransactionSpec
       Cardano.WalletSpec
+      Control.Concurrent.ConciergeSpec
       Data.Function.UtilsSpec
       Data.QuantitySpec
       Data.Time.TextSpec

lib/core/src/Cardano/Wallet/Api.hs

@@ -11,6 +11,8 @@
 {-# LANGUAGE TypeFamilies #-}
 {-# LANGUAGE TypeOperators #-}
 
+{- HLINT ignore "Use newtype instead of data" -}
+
 module Cardano.Wallet.Api
     ( -- * API
       Api
@@ -146,6 +148,8 @@ module Cardano.Wallet.Api
     , ApiLayer (..)
     , HasWorkerRegistry
     , workerRegistry
+    , WalletLock (..)
+    , walletLocks
     , HasDBFactory
     , dbFactory
     , tokenMetadataClient
@@ -247,6 +251,8 @@ import Cardano.Wallet.TokenMetadata
     ( TokenMetadataClient )
 import Cardano.Wallet.Transaction
     ( TransactionLayer )
+import Control.Concurrent.Concierge
+    ( Concierge )
 import Control.Tracer
     ( Tracer, contramap )
 import Data.ByteString
@@ -1073,14 +1079,21 @@ data ApiLayer s (k :: Depth -> Type -> Type)
         (TransactionLayer k)
         (DBFactory IO s k)
         (WorkerRegistry WalletId (DBLayer IO s k))
+        (Concierge IO WalletLock)
         (TokenMetadataClient IO)
     deriving (Generic)
 
+-- | Locks that are held by the wallet in order to enforce
+-- sequential executation of some API actions.
+-- Used with "Control.Concurrent.Concierge".
+data WalletLock = PostTransactionOld WalletId
+    deriving (Eq, Ord, Show)
+
 instance HasWorkerCtx (DBLayer IO s k) (ApiLayer s k) where
     type WorkerCtx (ApiLayer s k) = WalletLayer IO s k
     type WorkerMsg (ApiLayer s k) = WalletWorkerLog
     type WorkerKey (ApiLayer s k) = WalletId
-    hoistResource db transform (ApiLayer _ tr gp nw tl _ _ _) =
+    hoistResource db transform (ApiLayer _ tr gp nw tl _ _ _ _) =
         WalletLayer (contramap transform tr) gp nw tl db
 
 {-------------------------------------------------------------------------------
@@ -1115,6 +1128,12 @@ tokenMetadataClient
 tokenMetadataClient =
     typed @(TokenMetadataClient IO)
 
+walletLocks
+    :: forall ctx. (HasType (Concierge IO WalletLock) ctx)
+    => Lens' ctx (Concierge IO WalletLock)
+walletLocks =
+    typed @(Concierge IO WalletLock)
+
 {-------------------------------------------------------------------------------
                               Type Families
 -------------------------------------------------------------------------------}

lib/core/src/Cardano/Wallet/Api/Server.hs

@@ -178,8 +178,10 @@ import Cardano.Wallet.Api
     , HasDBFactory
     , HasTokenMetadataClient
     , HasWorkerRegistry
+    , WalletLock (..)
     , dbFactory
     , tokenMetadataClient
+    , walletLocks
     , workerRegistry
     )
 import Cardano.Wallet.Api.Server.Tls
@@ -578,6 +580,7 @@ import qualified Cardano.Wallet.Primitive.Types.TokenBundle as TokenBundle
 import qualified Cardano.Wallet.Primitive.Types.Tx as W
 import qualified Cardano.Wallet.Primitive.Types.UTxO as UTxO
 import qualified Cardano.Wallet.Registry as Registry
+import qualified Control.Concurrent.Concierge as Concierge
 import qualified Data.Aeson as Aeson
 import qualified Data.ByteString as BS
 import qualified Data.ByteString.Lazy as BL
@@ -1850,7 +1853,8 @@ postTransactionOld ctx genChange (ApiT wid) body = do
             , txTimeToLive = ttl
             }
 
-    (sel, tx, txMeta, txTime) <- withWorkerCtx ctx wid liftE liftE $ \wrk -> do
+    (sel, tx, txMeta, txTime) <- withWorkerCtx ctx wid liftE liftE $ \wrk ->
+      atomicallyWithHandler (ctx ^. walletLocks) (PostTransactionOld wid) $ do
         w <- liftHandler $ W.readWalletUTxOIndex @_ @s @k wrk wid
         sel <- liftHandler
             $ W.selectAssets @_ @s @k wrk w txCtx outs (const Prelude.id)
@@ -3093,7 +3097,8 @@ newApiLayer tr g0 nw tl df tokenMeta coworker = do
     re <- Registry.empty
     let trTx = contramap MsgSubmitSealedTx tr
     let trW = contramap MsgWalletWorker tr
-    let ctx = ApiLayer trTx trW g0 nw tl df re tokenMeta
+    locks <- Concierge.newConcierge
+    let ctx = ApiLayer trTx trW g0 nw tl df re locks tokenMeta
     listDatabases df >>= mapM_ (startWalletWorker ctx coworker)
     return ctx
 
@@ -3216,6 +3221,14 @@ withWorkerCtx ctx wid onMissing onNotResponding action =
     re = ctx ^. workerRegistry @s @k
     df = ctx ^. dbFactory @s @k
 
+{-------------------------------------------------------------------------------
+    Atomic handler operations
+-------------------------------------------------------------------------------}
+atomicallyWithHandler
+    :: Ord lock
+    => Concierge.Concierge IO lock -> lock -> Handler a -> Handler a
+atomicallyWithHandler c l = Handler . Concierge.atomicallyWith c l . runHandler'
+
 {-------------------------------------------------------------------------------
                                 Error Handling
 -------------------------------------------------------------------------------}

lib/core/src/Control/Concurrent/Concierge.hs

@@ -0,0 +1,88 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE RankNTypes #-}
+
+{- HLINT ignore "Use newtype instead of data" -}
+
+-- |
+-- Copyright: © 2021 IOHK
+-- License: Apache-2.0
+--
+-- This module provides a utility for ordering concurrent actions
+-- via locks.
+module Control.Concurrent.Concierge
+    ( Concierge
+    , newConcierge
+    , atomicallyWith
+    , atomicallyWithLifted
+    )
+    where
+
+import Prelude
+
+import Control.Monad.Class.MonadFork
+    ( MonadThread, ThreadId, myThreadId )
+import Control.Monad.Class.MonadSTM
+    ( MonadSTM
+    , TVar
+    , atomically
+    , modifyTVar
+    , newTVarIO
+    , readTVar
+    , retry
+    , writeTVar
+    )
+import Control.Monad.Class.MonadThrow
+    ( MonadThrow, bracket )
+import Control.Monad.IO.Class
+    ( MonadIO, liftIO )
+import Data.Map.Strict
+    ( Map )
+
+import qualified Data.Map.Strict as Map
+
+{-------------------------------------------------------------------------------
+    Concierge
+-------------------------------------------------------------------------------}
+-- | At a 'Concierge', you can obtain a lock and
+-- enforce sequential execution of concurrent 'IO' actions.
+--
+-- Back in the old days, hotel concierges used to give out keys.
+-- But after the cryptocurrency revolution, they give out locks. :)
+-- (The term /lock/ is standard terminology in concurrent programming.)
+data Concierge m lock = Concierge
+    { locks :: TVar m (Map lock (ThreadId m))
+    }
+
+-- | Create a new 'Concierge' that keeps track of locks.
+newConcierge :: MonadSTM m => m (Concierge m lock)
+newConcierge = Concierge <$> newTVarIO Map.empty
+
+-- | Obtain a lock from a 'Concierge' and run an 'IO' action.
+--
+-- If the same (equal) lock is already taken at this 'Concierge',
+-- the thread will be blocked until the lock becomes available.
+--
+-- The action may throw a synchronous or asynchronous exception.
+-- In both cases, the lock is returned to the concierge.
+atomicallyWith
+    :: (Ord lock, MonadIO m, MonadThrow m)
+    => Concierge IO lock -> lock -> m a -> m a
+atomicallyWith = atomicallyWithLifted liftIO
+
+-- | More polymorphic version of 'atomicallyWith'.
+atomicallyWithLifted
+    :: (Ord lock, MonadSTM m, MonadThread m, MonadThrow n)
+    => (forall b. m b -> n b)
+    -> Concierge m lock -> lock -> n a -> n a
+atomicallyWithLifted lift Concierge{locks} lock action =
+    bracket acquire (const release) (const action)
+  where
+    acquire = lift $ do
+        tid <- myThreadId
+        atomically $ do
+            ls <- readTVar locks
+            case Map.lookup lock ls of
+                Just _  -> retry
+                Nothing -> writeTVar locks $ Map.insert lock tid ls
+    release = lift $
+        atomically $ modifyTVar locks $ Map.delete lock

lib/core/test/unit/Control/Concurrent/ConciergeSpec.hs

@@ -0,0 +1,83 @@
+{-# LANGUAGE RankNTypes #-}
+module Control.Concurrent.ConciergeSpec
+    ( spec
+    ) where
+
+import Prelude
+
+import Control.Concurrent.Concierge
+    ( atomicallyWithLifted, newConcierge )
+import Control.Monad.Class.MonadFork
+    ( forkIO )
+import Control.Monad.Class.MonadSay
+    ( say )
+import Control.Monad.Class.MonadTimer
+    ( threadDelay )
+import Control.Monad.IOSim
+    ( IOSim, runSimTrace, selectTraceEventsSay )
+import Control.Monad.Trans.Class
+    ( lift )
+import Control.Monad.Trans.Except
+    ( ExceptT, catchE, runExceptT, throwE )
+import Test.Hspec
+    ( Spec, describe, it, parallel )
+import Test.QuickCheck
+    ( Property, (===) )
+
+spec :: Spec
+spec = do
+    parallel $ describe "Control.Concurrent.Concierge" $ do
+        it "Atomic operations do not interleave"
+            unit_atomic
+
+        it "throwE in ExceptT releases lock"
+            unit_exceptT_release_lock
+
+{-------------------------------------------------------------------------------
+    Properties
+-------------------------------------------------------------------------------}
+-- | Deterministic test for atomicity.
+-- We have to compare a program run that interleaves
+-- against one that is atomic.
+unit_atomic :: Bool
+unit_atomic =
+    ("ABAB" == sayings testInterleave) && ("AABB" == sayings testAtomic)
+  where
+    sayings :: (forall s. IOSim s a) -> String
+    sayings x = concat . selectTraceEventsSay $ runSimTrace x
+
+    testAtomic = do
+        concierge <- newConcierge
+        test $ atomicallyWithLifted id concierge ()
+    testInterleave = test id
+
+    test :: (forall a. IOSim s a -> IOSim s a) -> IOSim s ()
+    test atomically = do
+        _ <- forkIO $ atomically (delay 1 >> action "B")
+        atomically $ action "A"
+        delay 4
+
+    action :: String -> IOSim s ()
+    action s = say s >> delay 2 >> say s
+
+    delay :: Int -> IOSim s ()
+    delay n = threadDelay (fromIntegral n*0.1)
+
+-- | Check that using 'throwE' in the 'ExceptE' monad releases the lock
+unit_exceptT_release_lock :: Property
+unit_exceptT_release_lock =
+    ["A"] === selectTraceEventsSay (runSimTrace $ runExceptT test)
+  where
+    liftE :: IOSim s a -> ExceptT String (IOSim s) a
+    liftE = lift
+
+    test :: ExceptT String (IOSim s) ()
+    test = do
+        concierge <- liftE newConcierge
+        let atomically = atomicallyWithLifted liftE concierge ()
+        _ <- tryE $ atomically $ throwE "X"
+        atomically $ liftE $ say "A"
+
+-- not exported in  transformers <= 0.5.6
+tryE :: Monad m => ExceptT e m a -> ExceptT e m (Either e a)
+tryE action = (Right <$> action) `catchE` (pure . Left)