Plutip server fixes

plutip-server/Api/Handlers.hs

@@ -6,15 +6,19 @@ module Api.Handlers (
 import Cardano.Api (serialiseToCBOR)
 import Cardano.Launcher.Node (nodeSocketFile)
 import Cardano.Wallet.Shelley.Launch.Cluster (RunningNode (RunningNode))
-import Control.Concurrent.MVar (isEmptyMVar, putMVar, takeMVar)
+import Control.Concurrent (threadDelay)
+import Control.Concurrent.Async (race)
+import Control.Concurrent.MVar (isEmptyMVar, putMVar, tryTakeMVar)
 import Control.Monad (unless)
 import Control.Monad.Except (runExceptT, throwError)
 import Control.Monad.Extra (unlessM)
 import Control.Monad.IO.Class (liftIO)
 import Control.Monad.Reader (ReaderT, ask, asks)
 import Data.ByteString.Base16 qualified as Base16
 import Data.Default (def)
+import Data.Either (fromRight)
 import Data.Foldable (for_)
+import Data.List.Extra (firstJust)
 import Data.Text.Encoding qualified as Text
 import Data.Traversable (for)
 import System.Directory (doesFileExist)
@@ -24,7 +28,8 @@ import Test.Plutip.Internal.BotPlutusInterface.Setup (keysDir)
 import Test.Plutip.Internal.BotPlutusInterface.Wallet (BpiWallet (signKey), addSomeWallet)
 import Test.Plutip.Internal.LocalCluster (startCluster, stopCluster)
 import Test.Plutip.Internal.Types (ClusterEnv (runningNode))
-import Test.Plutip.LocalCluster (waitSeconds)
+import Test.Plutip.LocalCluster (cardanoMainnetAddress)
+import Test.Plutip.Tools.CardanoApi (AwaitWalletFundedError (AwaitingCapiError, AwaitingTimeoutError), awaitWalletFunded)
 import Types (
   AppM,
   ClusterStartupFailureReason (
@@ -51,6 +56,7 @@ import Types (
   StopClusterRequest (StopClusterRequest),
   StopClusterResponse (StopClusterFailure, StopClusterSuccess),
  )
+import UnliftIO.Exception (throwString)
 
 startClusterHandler :: ServerOptions -> StartClusterRequest -> AppM StartClusterResponse
 startClusterHandler
@@ -65,43 +71,60 @@ startClusterHandler
     isClusterDown <- liftIO $ isEmptyMVar statusMVar
     unless isClusterDown $ throwError ClusterIsRunningAlready
     let cfg = def {relayNodeLogs = nodeLogs, chainIndexPort = Nothing}
-    (statusTVar, res@(clusterEnv, _)) <- liftIO $ startCluster cfg setup
+    (statusTVar, (clusterEnv, wallets)) <- liftIO $ startCluster cfg setup
     liftIO $ putMVar statusMVar statusTVar
+    res <- liftIO $ race (threadDelay 2_000_000) $ waitForFundingTxs clusterEnv wallets
+    -- throw Exception for cardano-cli errors.
+    -- Ignore wait timeout error - return from this handler doesn't guarantee funded wallets immedietely.
+    maybe (return ()) throwString $ fromRight Nothing res
     let nodeConfigPath = getNodeConfigFile clusterEnv
     -- safeguard against directory tree structure changes
     unlessM (liftIO $ doesFileExist nodeConfigPath) $ throwError NodeConfigNotFound
     pure $
       ClusterStartupSuccess $
         ClusterStartupParameters
-          { privateKeys = getWalletPrivateKey <$> snd res
+          { privateKeys = getWalletPrivateKey <$> wallets
           , nodeSocketPath = getNodeSocketFile clusterEnv
           , nodeConfigPath = nodeConfigPath
           , keysDirectory = keysDir clusterEnv
           }
     where
       setup :: ReaderT ClusterEnv IO (ClusterEnv, [BpiWallet])
       setup = do
-        env <- ask
         wallets <- do
           for keysToGenerate $ \lovelaceAmounts -> do
             addSomeWallet (fromInteger . unLovelace <$> lovelaceAmounts)
-        waitSeconds 2 -- wait for transactions to submit
-        pure (env, wallets)
+        env <- ask
+        return (env, wallets)
+
+      -- wait for confirmation of funding txs, throw the first error if there's any
+      waitForFundingTxs clusterEnv wallets = do
+        res <- for wallets $ \w ->
+          awaitWalletFunded clusterEnv (cardanoMainnetAddress w)
+        return $
+          firstJust
+            ( \case
+                Left (AwaitingCapiError e) -> Just $ show e
+                Left AwaitingTimeoutError -> Nothing
+                Right () -> Nothing
+            )
+            res
+
       getNodeSocketFile (runningNode -> RunningNode conn _ _ _) = nodeSocketFile conn
       getNodeConfigFile =
         -- assumption is that node.config lies in the same directory as node.socket
         flip replaceFileName "node.config" . getNodeSocketFile
+
       getWalletPrivateKey :: BpiWallet -> PrivateKey
       getWalletPrivateKey = Text.decodeUtf8 . Base16.encode . serialiseToCBOR . signKey
       interpret = fmap (either ClusterStartupFailure id) . runExceptT
 
 stopClusterHandler :: StopClusterRequest -> AppM StopClusterResponse
 stopClusterHandler StopClusterRequest = do
   statusMVar <- asks status
-  isClusterDown <- liftIO $ isEmptyMVar statusMVar
-  if isClusterDown
-    then pure $ StopClusterFailure "Cluster is not running"
-    else do
-      statusTVar <- liftIO $ takeMVar statusMVar
+  maybeClusterStatus <- liftIO $ tryTakeMVar statusMVar
+  case maybeClusterStatus of
+    Nothing -> pure $ StopClusterFailure "Cluster is not running"
+    Just statusTVar -> do
       liftIO $ stopCluster statusTVar
       pure $ StopClusterSuccess

plutip-server/Types.hs

@@ -37,14 +37,7 @@ import Data.Text (Text)
 import GHC.Generics (Generic)
 import Network.Wai.Handler.Warp (Port)
 import Test.Plutip.Internal.BotPlutusInterface.Wallet (BpiWallet)
-import Test.Plutip.Internal.LocalCluster (
-  ClusterStatus (
-    ClusterClosed,
-    ClusterClosing,
-    ClusterStarted,
-    ClusterStarting
-  ),
- )
+import Test.Plutip.Internal.LocalCluster (ClusterStatus)
 import Test.Plutip.Internal.Types (ClusterEnv)
 import UnliftIO.STM (TVar)
 

plutip.cabal

@@ -203,6 +203,7 @@ executable plutip-server
   default-language: Haskell2010
   build-depends:
     , aeson
+    , async
     , base
     , base16-bytestring
     , bytestring

src/Test/Plutip/Internal/LocalCluster.hs

@@ -11,22 +11,24 @@ module Test.Plutip.Internal.LocalCluster (
 ) where
 
 import Cardano.Api qualified as CAPI
+import Cardano.BM.Configuration.Model qualified as CM
 import Cardano.BM.Data.Severity qualified as Severity
 import Cardano.BM.Data.Tracer (HasPrivacyAnnotation, HasSeverityAnnotation (getSeverityAnnotation))
-import Cardano.BM.Configuration.Model qualified as CM
 import Cardano.CLI (LogOutput (LogToFile), withLoggingNamed)
+import Cardano.Launcher (ProcessHasExited (ProcessHasExited))
 import Cardano.Launcher.Node (nodeSocketFile)
 import Cardano.Startup (installSignalHandlers, setDefaultFilePermissions, withUtf8Encoding)
 import Cardano.Wallet.Logging (stdoutTextTracer, trMessageText)
 import Cardano.Wallet.Shelley.Launch (TempDirLog, withSystemTempDir)
 import Cardano.Wallet.Shelley.Launch.Cluster (ClusterLog, localClusterConfigFromEnv, testMinSeverityFromEnv, walletMinSeverityFromEnv, withCluster)
-import Control.Concurrent.Async (async)
+import Control.Concurrent.Async (Async, async, cancel)
 import Control.Monad (unless, void, when)
 import Control.Monad.IO.Class (liftIO)
 import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Control.Monad.Reader (ReaderT (runReaderT))
-import Control.Retry (constantDelay, limitRetries, recoverAll)
+import Control.Retry (constantDelay, limitRetries, logRetries, recoverAll, recovering)
 import Control.Tracer (Tracer, contramap, traceWith)
+import Data.ByteString.Char8 qualified as B
 import Data.Foldable (for_)
 import Data.Kind (Type)
 import Data.Maybe (catMaybes, fromMaybe, isJust)
@@ -44,7 +46,7 @@ import System.Directory (canonicalizePath, copyFile, createDirectoryIfMissing, d
 import System.Environment (setEnv)
 import System.Exit (die)
 import System.FilePath ((</>))
-import System.IO (IOMode (WriteMode), hClose, openFile, stdout)
+import System.IO (IOMode (WriteMode), hClose, openFile, stderr, stdout)
 import Test.Plutip.Config (
   PlutipConfig (
     chainIndexPort,
@@ -115,26 +117,26 @@ withPlutusInterface :: forall (a :: Type). PlutipConfig -> (ClusterEnv -> IO a)
 withPlutusInterface conf action = do
   -- current setup requires `cardano-node` and `cardano-cli` as external processes
   checkProcessesAvailable ["cardano-node", "cardano-cli"]
-
   withLocalClusterSetup conf $ \dir clusterLogs _walletLogs nodeConfigLogHdl -> do
     result <- withLoggingNamed "cluster" clusterLogs $ \(_, (_, trCluster)) -> do
       let tr' = contramap MsgCluster $ trMessageText trCluster
       clusterCfg <- localClusterConfigFromEnv
       withRedirectedStdoutHdl nodeConfigLogHdl $ \restoreStdout ->
-        withCluster
-          tr'
-          dir
-          clusterCfg
-          []
-          (\rn -> restoreStdout $ runActionWthSetup rn dir trCluster action)
+        retryClusterFailedStartup $
+          withCluster
+            tr'
+            dir
+            clusterCfg
+            []
+            (\rn -> restoreStdout $ runActionWthSetup rn dir trCluster action)
     handleLogs dir conf
     return result
   where
-    runActionWthSetup rn dir trCluster userActon = do
+    runActionWthSetup rn dir trCluster userAction = do
       let tracer' = trMessageText trCluster
       waitForRelayNode tracer' rn
       -- launch chain index in seperate thread, logs to stdout
-      ciPort <- launchChainIndex conf rn dir
+      (ciPort, runningChainIndex) <- launchChainIndex conf rn dir
       traceWith tracer' (ChaiIndexStartedAt ciPort)
       let cEnv =
             ClusterEnv
@@ -147,7 +149,21 @@ withPlutusInterface conf action = do
               }
 
       BotSetup.runSetup cEnv -- run preparations to use `bot-plutus-interface`
-      userActon cEnv -- executing user action on cluster
+      userAction cEnv `finally` cancel runningChainIndex -- executing user action on cluster
+
+    -- withCluster has a race condition between checking for available ports and claiming the ports.
+    -- This may cause failure at the cluster startup (the "resource busy (Address already in use)" error)
+    -- Given this is rare and the problem root sits in cardano-wallet, lets simply retry the startup few times full of hope.
+    retryClusterFailedStartup =
+      let msg err = B.pack $ "Retrying cluster startup due to: " <> show err <> "\n"
+          shouldRetry =
+            pure . \case
+              ProcessHasExited _ _ -> True
+              _ -> False
+       in recovering
+            (limitRetries 5)
+            [logRetries shouldRetry (\_ y _ -> B.hPutStr stderr $ msg y)]
+            . const
 
 -- Redirect stdout to a provided handle providing mask to temporarily revert back to initial stdout.
 withRedirectedStdoutHdl :: Handle -> ((forall b. IO b -> IO b) -> IO a) -> IO a
@@ -240,7 +256,7 @@ waitForRelayNode trCluster rn = do
     trace = traceWith trCluster WaitingRelayNode
 
 -- | Launch the chain index in a separate thread.
-launchChainIndex :: PlutipConfig -> RunningNode -> FilePath -> IO Int
+launchChainIndex :: PlutipConfig -> RunningNode -> FilePath -> IO (Int, Async ())
 launchChainIndex conf (RunningNode sp _block0 (_gp, _vData) _) dir = do
   config <- defaultConfig
   CM.setMinSeverity config Severity.Notice
@@ -256,8 +272,8 @@ launchChainIndex conf (RunningNode sp _block0 (_gp, _vData) _) dir = do
                 fromEnum
                 (chainIndexPort conf)
           }
-  void . async $ void $ ChainIndex.runMainWithLog (const $ return ()) config chainIndexConfig
-  return $ cicPort chainIndexConfig
+  running <- async $ ChainIndex.runMainWithLog (const $ return ()) config chainIndexConfig
+  return (cicPort chainIndexConfig, running)
 
 handleLogs :: HasCallStack => FilePath -> PlutipConfig -> IO ()
 handleLogs clusterDir conf =

src/Test/Plutip/Tools/CardanoApi.hs

@@ -4,14 +4,20 @@ module Test.Plutip.Tools.CardanoApi (
   utxosAtAddress,
   queryProtocolParams,
   queryTip,
+  awaitWalletFunded,
+  AwaitWalletFundedError (AwaitingCapiError, AwaitingTimeoutError),
 ) where
 
 import Cardano.Api qualified as C
 import Cardano.Api.Shelley (ProtocolParameters)
 import Cardano.Launcher.Node (nodeSocketFile)
 import Cardano.Slotting.Slot (WithOrigin)
 import Cardano.Wallet.Shelley.Launch.Cluster (RunningNode (RunningNode))
+import Control.Arrow (right)
 import Control.Exception (Exception)
+import Control.Retry (constantDelay, limitRetries, retrying)
+import Data.Either (fromRight)
+import Data.Map qualified as M
 import Data.Set qualified as Set
 import GHC.Generics (Generic)
 import Ouroboros.Consensus.HardFork.Combinator.AcrossEras (EraMismatch)
@@ -71,3 +77,36 @@ flattenQueryResult ::
 flattenQueryResult = \case
   Right (Right res) -> Right res
   err -> Left $ SomeError (show err)
+
+data AwaitWalletFundedError
+  = AwaitingCapiError CardanoApiError
+  | AwaitingTimeoutError
+
+instance Show AwaitWalletFundedError where
+  show (AwaitingCapiError (SomeError e)) = e
+  show AwaitingTimeoutError = "Awaiting funding transaction timed out."
+
+-- | Waits till specified address is funded using cardano-node query.
+-- Performs 20 tries with 0.2 seconds between tries, which should be a sane default.
+-- Waits till there's any utxos at an address - works for us as funds will be send with tx per address.
+awaitWalletFunded ::
+  ClusterEnv ->
+  C.AddressAny ->
+  IO (Either AwaitWalletFundedError ())
+awaitWalletFunded cenv addr = toErrorMsg <$> retrying policy checkResponse action
+  where
+    -- With current defaults the slot length is 0.2s and block gets produced about every second slot.
+    -- We are expected to wait 0.4s, waiting 4s we are almost guaranteed (p>0.9999)
+    delay = 200_000 -- in microseconds, 0.2s.
+    policy = constantDelay delay <> limitRetries 20
+
+    action _ = right (M.null . C.unUTxO) <$> utxosAtAddress cenv addr
+
+    checkResponse _ = return . fromRight False
+
+    toErrorMsg = \case
+      Left e -> Left $ AwaitingCapiError e
+      Right noUtxos ->
+        if noUtxos
+          then Left AwaitingTimeoutError
+          else Right ()

test/Spec/Integration.hs

@@ -188,7 +188,7 @@ test =
           (initAda [100])
           (withContract $ const failingTimeContract)
           [shouldFail] -- FIXME: add check that "OutsideValidityIntervalUTxO" is in error message
-          -- [shouldSucceed] 
+          -- [shouldSucceed]
       , assertExecution
           "Passes validation with exact time range checks"
           (initAda [100])