SCP-2204: Add an HTTP route for stopping instances (#3087)

* SCP-2204: Add an HTTP route for stopping instances

plutus-pab/src/Plutus/PAB/Core.hs

@@ -54,6 +54,8 @@ module Plutus.PAB.Core
     -- * Agent threads
     , ContractInstanceEffects
     , handleAgentThread
+    , stopInstance
+    , instanceActivity
     -- * Querying the state
     , instanceState
     , observableState
@@ -72,6 +74,7 @@ module Plutus.PAB.Core
     , askUserEnv
     , askBlockchainEnv
     , askInstancesState
+    , runningInstances
     -- * Run PAB effects in separate threads
     , PABRunner(..)
     , pabRunner
@@ -83,18 +86,20 @@ module Plutus.PAB.Core
     , timed
     ) where
 
+import           Control.Applicative                     (Alternative (..))
 import           Control.Concurrent.STM                  (STM)
 import qualified Control.Concurrent.STM                  as STM
 import           Control.Monad                           (forM, guard, void)
 import           Control.Monad.Freer                     (Eff, LastMember, Member, interpret, reinterpret, runM, send,
                                                           subsume, type (~>))
-import           Control.Monad.Freer.Error               (Error, runError)
+import           Control.Monad.Freer.Error               (Error, runError, throwError)
 import           Control.Monad.Freer.Extras.Log          (LogMessage, LogMsg (..), LogObserve, handleObserveLog,
                                                           logInfo, mapLog)
 import qualified Control.Monad.Freer.Extras.Modify       as Modify
 import           Control.Monad.Freer.Reader              (Reader (..), ask, asks, runReader)
 import           Control.Monad.IO.Class                  (MonadIO (..))
 import qualified Data.Aeson                              as JSON
+import           Data.Foldable                           (traverse_)
 import qualified Data.Map                                as Map
 import           Data.Proxy                              (Proxy (..))
 import           Data.Set                                (Set)
@@ -107,7 +112,8 @@ import           Ledger.Value                            (Value)
 import           Plutus.Contract.Effects.ExposeEndpoint  (ActiveEndpoint (..))
 import           Plutus.PAB.Core.ContractInstance        (ContractInstanceMsg)
 import qualified Plutus.PAB.Core.ContractInstance        as ContractInstance
-import           Plutus.PAB.Core.ContractInstance.STM    (BlockchainEnv, InstancesState, OpenEndpoint (..))
+import           Plutus.PAB.Core.ContractInstance.STM    (Activity (Active), BlockchainEnv, InstancesState,
+                                                          OpenEndpoint (..))
 import qualified Plutus.PAB.Core.ContractInstance.STM    as Instances
 import           Plutus.PAB.Effects.Contract             (ContractDefinitionStore, ContractEffect, ContractStore,
                                                           PABContract (..), addDefinition, getState)
@@ -120,7 +126,7 @@ import           Plutus.PAB.Events.ContractInstanceState (PartiallyDecodedRespon
 import           Plutus.PAB.Monitoring.PABLogMsg         (PABMultiAgentMsg (..))
 import           Plutus.PAB.Timeout                      (Timeout)
 import qualified Plutus.PAB.Timeout                      as Timeout
-import           Plutus.PAB.Types                        (PABError)
+import           Plutus.PAB.Types                        (PABError (ContractInstanceNotFound, InstanceAlreadyStopped))
 import           Plutus.PAB.Webserver.Types              (ContractActivationArgs (..))
 import           Wallet.API                              (PubKey, Slot)
 import qualified Wallet.API                              as WAPI
@@ -226,6 +232,32 @@ callEndpointOnInstance instanceID ep value = do
         $ STM.atomically
         $ Instances.callEndpointOnInstanceTimeout timeoutVar state (EndpointDescription ep) (JSON.toJSON value) instanceID
 
+-- | The 'InstanceState' for the instance. Throws a 'ContractInstanceNotFound' error if the instance does not exist.
+instanceStateInternal :: forall t env. ContractInstanceId -> PABAction t env Instances.InstanceState
+instanceStateInternal instanceId = do
+    instancesState <- asks @(PABEnvironment t env) instancesState
+    r <- liftIO $ STM.atomically $ (Left <$> Instances.instanceState instanceId instancesState) <|> (pure $ Right $ ContractInstanceNotFound instanceId)
+    case r of
+        Right err -> throwError err
+        Left s    -> pure s
+
+-- | Stop the instance.
+stopInstance :: forall t env. ContractInstanceId -> PABAction t env ()
+stopInstance instanceId = do
+    Instances.InstanceState{Instances.issStatus, Instances.issStop} <- instanceStateInternal instanceId
+    r' <- liftIO $ STM.atomically $ do
+            status <- STM.readTVar issStatus
+            case status of
+                Active -> STM.putTMVar issStop () >> pure Nothing
+                _      -> pure (Just $ InstanceAlreadyStopped instanceId)
+    traverse_ throwError r'
+
+-- | The 'Activity' of the instance.
+instanceActivity :: forall t env. ContractInstanceId -> PABAction t env Activity
+instanceActivity instanceId = do
+    Instances.InstanceState{Instances.issStatus} <- instanceStateInternal instanceId
+    liftIO $ STM.readTVarIO issStatus
+
 -- | Call a named endpoint on a contract instance. Fails immediately if the
 --   endpoint is not available.
 callEndpointOnInstance' ::
@@ -503,6 +535,9 @@ valueAt address = valueAtSTM address >>= liftIO . STM.atomically
 waitUntilFinished :: forall t env. ContractInstanceId -> PABAction t env (Maybe JSON.Value)
 waitUntilFinished i = finalResult i >>= liftIO . STM.atomically
 
+runningInstances :: forall t env. PABAction t env (Set ContractInstanceId)
+runningInstances = askInstancesState @t @env >>= liftIO . STM.atomically . Instances.runningInstances
+
 -- | Read the 'env' from the environment
 askUserEnv :: forall t env effs. Member (Reader (PABEnvironment t env)) effs => Eff effs env
 askUserEnv = interpret (handleUserEnvReader @t @env) ask

plutus-pab/src/Plutus/PAB/Core/ContractInstance.hs

@@ -29,6 +29,7 @@ module Plutus.PAB.Core.ContractInstance(
     , callEndpointOnInstance
     ) where
 
+import           Control.Applicative                              (Alternative (..))
 import           Control.Arrow                                    ((>>>))
 import           Control.Concurrent                               (forkIO)
 import           Control.Concurrent.STM                           (STM)
@@ -58,7 +59,7 @@ import           Wallet.Effects                                   (ChainIndexEff
 import           Wallet.Emulator.LogMessages                      (TxBalanceMsg)
 
 import           Plutus.Contract                                  (AddressChangeRequest (..))
-import           Plutus.PAB.Core.ContractInstance.STM             (Activity (Done), BlockchainEnv (..),
+import           Plutus.PAB.Core.ContractInstance.STM             (Activity (Done, Stopped), BlockchainEnv (..),
                                                                    InstanceState (..), InstancesState,
                                                                    callEndpointOnInstance, emptyInstanceState)
 import qualified Plutus.PAB.Core.ContractInstance.STM             as InstanceState
@@ -214,6 +215,7 @@ stmInstanceLoop ::
     -> Eff effs ()
 stmInstanceLoop def instanceId = do
     (currentState :: Contract.State t) <- Contract.getState @t instanceId
+    InstanceState{issStop} <- ask
     let resp = serialisableState (Proxy @t) currentState
     updateState resp
     case Contract.requests @t currentState of
@@ -222,10 +224,16 @@ stmInstanceLoop def instanceId = do
             ask >>= liftIO . STM.atomically . InstanceState.setActivity (Done err)
         _ -> do
             response <- respondToRequestsSTM @t instanceId currentState
-            event <- liftIO $ STM.atomically response
-            (newState :: Contract.State t) <- Contract.updateContract @t instanceId (caID def) currentState event
-            Contract.putState @t def instanceId newState
-            stmInstanceLoop @t def instanceId
+            let rsp' = Right <$> response
+                stop = Left <$> STM.takeTMVar issStop
+            event <- liftIO $ STM.atomically (stop <|> rsp')
+            case event of
+                Left () -> do
+                    ask >>= liftIO . STM.atomically . InstanceState.setActivity Stopped
+                Right event' -> do
+                    (newState :: Contract.State t) <- Contract.updateContract @t instanceId (caID def) currentState event'
+                    Contract.putState @t def instanceId newState
+                    stmInstanceLoop @t def instanceId
 
 -- | Update the TVars in the 'InstanceState' with data from the list
 --   of requests.

plutus-pab/src/Plutus/PAB/Core/ContractInstance/STM.hs

@@ -39,6 +39,7 @@ module Plutus.PAB.Core.ContractInstance.STM(
     , obervableContractState
     , instanceState
     , instanceIDs
+    , runningInstances
     ) where
 
 import           Control.Applicative                      (Alternative (..))
@@ -189,7 +190,9 @@ awaitEndpointResponse Request{rqID, itID} InstanceState{issEndpoints} = do
 -- | Whether the contract instance is still waiting for an event.
 data Activity =
         Active
+        | Stopped -- ^ Instance was stopped before all requests were handled
         | Done (Maybe Value) -- ^ Instance finished, possibly with an error
+        deriving (Eq, Show)
 
 -- | The state of an active contract instance.
 data InstanceState =
@@ -199,6 +202,7 @@ data InstanceState =
         , issTransactions    :: TVar (Set TxId) -- ^ Transactions whose status the contract is interested in
         , issStatus          :: TVar Activity -- ^ Whether the instance is still running.
         , issObservableState :: TVar (Maybe Value) -- ^ Serialised observable state of the contract instance (if available)
+        , issStop            :: TMVar () -- ^ Stop the instance if a value is written into the TMVar.
         }
 
 -- | An 'InstanceState' value with empty fields
@@ -210,6 +214,7 @@ emptyInstanceState =
         <*> STM.newTVar mempty
         <*> STM.newTVar Active
         <*> STM.newTVar Nothing
+        <*> STM.newEmptyTMVar
 
 -- | Add an address to the set of addresses that the instance is watching
 addAddress :: Address -> InstanceState -> STM ()
@@ -321,8 +326,9 @@ finalResult instanceId m = do
     InstanceState{issStatus} <- instanceState instanceId m
     v <- STM.readTVar issStatus
     case v of
-        Done r -> pure r
-        _      -> empty
+        Done r  -> pure r
+        Stopped -> pure Nothing
+        _       -> empty
 
 -- | Insert an 'InstanceState' value into the 'InstancesState'
 insertInstance :: ContractInstanceId -> InstanceState -> InstancesState -> STM ()
@@ -359,3 +365,15 @@ valueAt addr BlockchainEnv{beAddressMap} = do
 -- | The current slot number
 currentSlot :: BlockchainEnv -> STM Slot
 currentSlot BlockchainEnv{beCurrentSlot} = STM.readTVar beCurrentSlot
+
+-- | The IDs of contract instances that are currently running
+runningInstances :: InstancesState -> STM (Set ContractInstanceId)
+runningInstances (InstancesState m) = do
+    let flt :: InstanceState -> STM (Maybe InstanceState)
+        flt s@InstanceState{issStatus} = do
+            status <- STM.readTVar issStatus
+            case status of
+                Active -> pure (Just s)
+                _      -> pure Nothing
+    mp <- STM.readTVar m
+    Map.keysSet . Map.mapMaybe id <$> traverse flt mp

plutus-pab/src/Plutus/PAB/Simulator.hs

@@ -27,13 +27,17 @@ module Plutus.PAB.Simulator(
     , mkSimulatorHandlers
     , addWallet
     -- * Simulator actions
+    -- ** Logging
     , logString
     , logPretty
     -- ** Agent actions
     , payToWallet
     , activateContract
     , callEndpointOnInstance
     , handleAgentThread
+    , Activity(..)
+    , stopInstance
+    , instanceActivity
     -- ** Control actions
     , makeBlock
     -- * Querying the state
@@ -98,7 +102,8 @@ import           Ledger.Value                                   (Value, flattenV
 import           Plutus.PAB.Core                                (EffectHandlers (..))
 import qualified Plutus.PAB.Core                                as Core
 import qualified Plutus.PAB.Core.ContractInstance.BlockchainEnv as BlockchainEnv
-import           Plutus.PAB.Core.ContractInstance.STM           (BlockchainEnv, InstancesState, OpenEndpoint)
+import           Plutus.PAB.Core.ContractInstance.STM           (Activity (..), BlockchainEnv, InstancesState,
+                                                                 OpenEndpoint)
 import qualified Plutus.PAB.Core.ContractInstance.STM           as Instances
 import           Plutus.PAB.Effects.Contract                    (ContractStore)
 import qualified Plutus.PAB.Effects.Contract                    as Contract
@@ -695,6 +700,14 @@ handleAgentThread ::
     -> Simulation t a
 handleAgentThread = Core.handleAgentThread
 
+-- | Stop the instance.
+stopInstance :: forall t. ContractInstanceId -> Simulation t ()
+stopInstance = Core.stopInstance
+
+-- | The 'Activity' state of the instance
+instanceActivity :: forall t. ContractInstanceId -> Simulation t Activity
+instanceActivity = Core.instanceActivity
+
 -- | Create a new wallet with a random key and add it to the list of simulated wallets
 addWallet :: forall t. Simulation t (Wallet, PubKey)
 addWallet = do

plutus-pab/src/Plutus/PAB/Types.hs

@@ -49,6 +49,7 @@ data PABError
     | InvalidUUIDError  Text
     | OtherError Text -- ?
     | EndpointCallError NotificationError
+    | InstanceAlreadyStopped ContractInstanceId -- ^ Attempt to stop the instance failed because it was not running
     | WalletNotFound Wallet
     deriving stock (Show, Eq, Generic)
     deriving anyclass (ToJSON, FromJSON)
@@ -69,6 +70,7 @@ instance Pretty PABError where
         InvalidUUIDError t         -> "Invalid UUID:" <+> pretty t
         OtherError t               -> "Other error:" <+> pretty t
         EndpointCallError n        -> "Endpoint call failed:" <+> pretty n
+        InstanceAlreadyStopped i   -> "Instance already stopped:" <+> pretty i
         WalletNotFound w           -> "Wallet not found:" <+> pretty w
 
 data DbConfig =

plutus-pab/src/Plutus/PAB/Webserver/API.hs

@@ -17,7 +17,7 @@ import qualified Data.Aeson                 as JSON
 import           Data.Text                  (Text)
 import           Plutus.PAB.Webserver.Types (ContractActivationArgs, ContractInstanceClientState,
                                              ContractSignatureResponse, FullReport)
-import           Servant.API                (Capture, Get, JSON, Post, ReqBody, (:<|>), (:>))
+import           Servant.API                (Capture, Get, JSON, Post, Put, ReqBody, (:<|>), (:>))
 import           Servant.API.WebSocket      (WebSocketPending)
 import           Wallet.Types               (ContractInstanceId, NotificationError)
 
@@ -46,6 +46,7 @@ type NewAPI t walletId -- see note [WalletID type in wallet API]
                     (Capture "contract-instance-id" Text :>
                         ( "status" :> Get '[JSON] (ContractInstanceClientState t) -- Current status of contract instance
                         :<|> "endpoint" :> Capture "endpoint-name" String :> ReqBody '[JSON] JSON.Value :> Post '[JSON] () -- Call an endpoint. Make
+                        :<|> "stop" :> Put '[JSON] () -- Terminate the instance.
                         )
                     )
             :<|> "instances" :> "wallet" :> Capture "wallet-id" walletId :> Get '[JSON] [ContractInstanceClientState t]

plutus-pab/src/Plutus/PAB/Webserver/Handler.hs

@@ -37,6 +37,7 @@ import           Data.Foldable                           (traverse_)
 import qualified Data.Map                                as Map
 import           Data.Maybe                              (mapMaybe)
 import           Data.Proxy                              (Proxy (..))
+import qualified Data.Set                                as Set
 import           Data.Text                               (Text)
 import qualified Data.UUID                               as UUID
 import           Ledger                                  (Slot, Value, pubKeyHash)
@@ -114,13 +115,15 @@ handlerNew ::
        Contract.PABContract t =>
        (ContractActivationArgs (Contract.ContractDef t) -> PABAction t env ContractInstanceId)
             :<|> (Text -> PABAction t env (ContractInstanceClientState (Contract.ContractDef t))
-                                        :<|> (String -> JSON.Value -> PABAction t env ()))
+                                        :<|> (String -> JSON.Value -> PABAction t env ())
+                                        :<|> PABAction t env ()
+                                        )
             :<|> (Integer -> PABAction t env [ContractInstanceClientState (Contract.ContractDef t)])
             :<|> PABAction t env [ContractInstanceClientState (Contract.ContractDef t)]
             :<|> PABAction t env [ContractSignatureResponse (Contract.ContractDef t)]
 handlerNew =
         (activateContract
-            :<|> (\x -> (parseContractId x >>= contractInstanceState) :<|> (\y z -> parseContractId x >>= \x' -> callEndpoint x' y z))
+            :<|> (\x -> (parseContractId x >>= contractInstanceState) :<|> (\y z -> parseContractId x >>= \x' -> callEndpoint x' y z) :<|> (parseContractId x >>= shutdown))
             :<|> instancesForWallets
             :<|> allInstanceStates
             :<|> availableContracts)
@@ -163,15 +166,20 @@ instancesForWallets wallet = filter ((==) (Wallet wallet) . cicWallet) <$> allIn
 allInstanceStates :: forall t env. Contract.PABContract t => PABAction t env [ContractInstanceClientState (Contract.ContractDef t)]
 allInstanceStates = do
     mp <- Contract.getActiveContracts @t
+    inst <- Core.runningInstances
+    let isRunning i = Set.member i inst
     let get (i, ContractActivationArgs{caWallet, caID}) = fromInternalState caID i caWallet . Contract.serialisableState (Proxy @t) <$> Contract.getState @t i
-    traverse get $ Map.toList mp
+    filter (isRunning . cicContract) <$> traverse get (Map.toList mp)
 
 availableContracts :: forall t env. Contract.PABContract t => PABAction t env [ContractSignatureResponse (Contract.ContractDef t)]
 availableContracts = do
     def <- Contract.getDefinitions @t
     let mkSchema s = ContractSignatureResponse s <$> Contract.exportSchema @t s
     traverse mkSchema def
 
+shutdown :: forall t env. ContractInstanceId -> PABAction t env ()
+shutdown = Core.stopInstance
+
 -- | Proxy for the wallet API
 walletProxyClientEnv ::
     forall t env.

plutus-pab/test/Plutus/PAB/CoreSpec.hs

@@ -9,7 +9,7 @@
 
 module Plutus.PAB.CoreSpec
     ( tests
-    , pingPongSpec
+    , stopContractInstanceTest
     ) where
 
 import           Control.Lens                             ((&), (+~))
@@ -62,9 +62,6 @@ import           Wallet.Rollup                            (doAnnotateBlockchain)
 import           Wallet.Rollup.Types                      (DereferencedInput, dereferencedInputs, isFound)
 import           Wallet.Types                             (ContractInstanceId)
 
-pingPongSpec :: IO ()
-pingPongSpec = waitForUpdateTest
-
 tests :: TestTree
 tests = testGroup "Plutus.PAB.Core" [installContractTests, executionTests]
 
@@ -102,6 +99,7 @@ executionTests =
         , currencyTest
         , rpcTest
         , testCase "wait for update" waitForUpdateTest
+        , testCase "stop contract instance" stopContractInstanceTest
         ]
 
 waitForUpdateTest :: IO ()
@@ -124,6 +122,15 @@ waitForUpdateTest =
         _ <- Simulator.waitForState (is Pinged) p1
         pure ()
 
+stopContractInstanceTest :: IO ()
+stopContractInstanceTest = do
+    runScenario $ do
+        p1 <- Simulator.activateContract defaultWallet PingPong
+        _ <- Simulator.stopInstance p1
+        _ <- Simulator.waitUntilFinished p1
+        st <- Simulator.instanceActivity p1
+        assertEqual "Instance should be 'Stopped'" st Simulator.Stopped
+
 currencyTest :: TestTree
 currencyTest =
     let mps = SimpleMPS{tokenName="my token", amount = 10000}