Merge pull request IntersectMBO#3909 from input-output-hk/scp-2735-better-rollback-handling

silky · web-flow · commit 1f258993ca44 · 2021-09-15T09:57:08.000+01:00
make transaction status return an either
diff --git a/plutus-chain-index/src/Plutus/ChainIndex/TxIdState.hs b/plutus-chain-index/src/Plutus/ChainIndex/TxIdState.hs
@@ -12,9 +12,7 @@
 {-# LANGUAGE ViewPatterns          #-}
 
 module Plutus.ChainIndex.TxIdState(
-    TxIdState(..)
-    , TxConfirmedState(..)
-    , isConfirmed
+    isConfirmed
     , increaseDepth
     , initialStatus
     , transactionStatus
@@ -27,47 +25,16 @@ module Plutus.ChainIndex.TxIdState(
 import           Control.Lens                ((^.))
 import           Data.FingerTree             (Measured (..), (|>))
 import qualified Data.FingerTree             as FT
-import           Data.Map                    (Map)
 import qualified Data.Map                    as Map
 import           Data.Monoid                 (Last (..), Sum (..))
-import           Data.Semigroup.Generic      (GenericSemigroupMonoid (..))
-import           GHC.Generics                (Generic)
 import           Ledger                      (OnChainTx, TxId, eitherTx)
 import           Plutus.ChainIndex.Tx        (ChainIndexTx (..), ChainIndexTxOutputs (..), citxOutputs, citxTxId)
-import           Plutus.ChainIndex.Types     (BlockNumber (..), Depth (..), Point (..), Tip (..), TxStatus (..),
-                                              TxValidity (..), pointsToTip)
+import           Plutus.ChainIndex.Types     (BlockNumber (..), Depth (..), Point (..), Tip (..), TxConfirmedState (..),
+                                              TxIdState (..), TxStatus (..), TxStatusFailure (..), TxValidity (..),
+                                              pointsToTip)
 import           Plutus.ChainIndex.UtxoState (RollbackFailed (..), RollbackResult (..), UtxoIndex, UtxoState (..), tip,
                                               viewTip)
 
-data TxIdState = TxIdState
-  { txnsConfirmed :: Map TxId TxConfirmedState
-  -- ^ Number of times this transaction has been added as well as other
-  -- necessary metadata.
-  , txnsDeleted   :: Map TxId (Sum Int)
-  -- ^ Number of times this transaction has been deleted.
-  }
-  deriving stock (Eq, Generic, Show)
-
-instance Monoid TxIdState where
-    mappend = (<>)
-    mempty  = TxIdState { txnsConfirmed=mempty, txnsDeleted=mempty }
-
-data TxConfirmedState =
-  TxConfirmedState
-    { timesConfirmed :: Sum Int
-    , blockAdded     :: Last BlockNumber
-    , validity       :: Last TxValidity
-    }
-    deriving stock (Eq, Generic, Show)
-    deriving (Semigroup, Monoid) via (GenericSemigroupMonoid TxConfirmedState)
-
--- A semigroup instance that merges the two maps, instead of taking the
--- leftmost one.
-instance Semigroup TxIdState where
-  TxIdState{txnsConfirmed=c, txnsDeleted=d} <> TxIdState{txnsConfirmed=c', txnsDeleted=d'}
-    = TxIdState { txnsConfirmed = Map.unionWith (<>) c c'
-                , txnsDeleted   = Map.unionWith (<>) d d'
-                }
 
 -- | The 'TxStatus' of a transaction right after it was added to the chain
 initialStatus :: OnChainTx -> TxStatus
@@ -95,23 +62,22 @@ chainConstant = Depth 8
 
 -- | Given the current block, compute the status for the given transaction by
 -- checking to see if it has been deleted.
-transactionStatus :: BlockNumber -> TxIdState -> TxId -> TxStatus
+transactionStatus :: BlockNumber -> TxIdState -> TxId -> Either TxStatusFailure TxStatus
 transactionStatus currentBlock txIdState txId
   = case (confirmed, deleted) of
-       (Nothing, _)      -> Unknown
+       (Nothing, _)      -> Right Unknown
 
        (Just TxConfirmedState{blockAdded=Last (Just block'), validity=Last (Just validity')}, Nothing) ->
          if block' + (fromIntegral chainConstant) >= currentBlock
-            then newStatus block' validity'
-            else Committed validity'
+            then Right $ newStatus block' validity'
+            else Right $ Committed validity'
 
        (Just TxConfirmedState{timesConfirmed=confirms, blockAdded=Last (Just block'), validity=Last (Just validity')}, Just deletes) ->
          if confirms >= deletes
-            then newStatus block' validity'
-            else Unknown
+            then Right $ newStatus block' validity'
+            else Right $ Unknown
 
-       -- TODO: Proper error.
-       _ -> error $ "Unable to determine transactionStatus for TxId: " <> show txId <> " at block: " <> show currentBlock <> "."
+       _ -> Left $ TxIdStateInvalid currentBlock txId txIdState
     where
       newStatus block' validity' = TentativelyConfirmed (Depth $ fromIntegral $ currentBlock - block') validity'
       confirmed = Map.lookup txId (txnsConfirmed txIdState)
diff --git a/plutus-chain-index/src/Plutus/ChainIndex/Types.hs b/plutus-chain-index/src/Plutus/ChainIndex/Types.hs
@@ -20,6 +20,9 @@ module Plutus.ChainIndex.Types(
     , BlockNumber(..)
     , Depth(..)
     , Diagnostics(..)
+    , TxConfirmedState(..)
+    , TxStatusFailure(..)
+    , TxIdState(..)
     ) where
 
 import qualified Codec.Serialise                  as CBOR
@@ -28,6 +31,10 @@ import           Data.Aeson                       (FromJSON, ToJSON)
 import qualified Data.ByteArray                   as BA
 import qualified Data.ByteString.Lazy             as BSL
 import           Data.Default                     (Default (..))
+import           Data.Map                         (Map)
+import qualified Data.Map                         as Map
+import           Data.Monoid                      (Last (..), Sum (..))
+import           Data.Semigroup.Generic           (GenericSemigroupMonoid (..))
 import           Data.Set                         (Set)
 import qualified Data.Set                         as Set
 import           Data.Text.Prettyprint.Doc.Extras (PrettyShow (..))
@@ -212,3 +219,40 @@ data Diagnostics =
         }
         deriving stock (Eq, Ord, Show, Generic)
         deriving anyclass (ToJSON, FromJSON)
+
+data TxStatusFailure =
+      -- | We couldn't return the status because the 'TxIdState' was in a ...
+      -- state ... that we didn't know how to decode in
+      -- 'Plutus.ChainIndex.TxIdState.transactionStatus'.
+      TxIdStateInvalid BlockNumber TxId TxIdState
+      deriving (Show, Eq)
+
+data TxIdState = TxIdState
+  { txnsConfirmed :: Map TxId TxConfirmedState
+  -- ^ Number of times this transaction has been added as well as other
+  -- necessary metadata.
+  , txnsDeleted   :: Map TxId (Sum Int)
+  -- ^ Number of times this transaction has been deleted.
+  }
+  deriving stock (Eq, Generic, Show)
+
+instance Monoid TxIdState where
+    mappend = (<>)
+    mempty  = TxIdState { txnsConfirmed=mempty, txnsDeleted=mempty }
+
+data TxConfirmedState =
+  TxConfirmedState
+    { timesConfirmed :: Sum Int
+    , blockAdded     :: Last BlockNumber
+    , validity       :: Last TxValidity
+    }
+    deriving stock (Eq, Generic, Show)
+    deriving (Semigroup, Monoid) via (GenericSemigroupMonoid TxConfirmedState)
+
+-- A semigroup instance that merges the two maps, instead of taking the
+-- leftmost one.
+instance Semigroup TxIdState where
+  TxIdState{txnsConfirmed=c, txnsDeleted=d} <> TxIdState{txnsConfirmed=c', txnsDeleted=d'}
+    = TxIdState { txnsConfirmed = Map.unionWith (<>) c c'
+                , txnsDeleted   = Map.unionWith (<>) d d'
+                }
diff --git a/plutus-chain-index/test/Generators.hs b/plutus-chain-index/test/Generators.hs
@@ -47,9 +47,8 @@ import           Ledger.Tx                   (Address, TxIn (..), TxOut (..), Tx
 import           Ledger.TxId                 (TxId (..))
 import           Ledger.Value                (Value)
 import           Plutus.ChainIndex.Tx        (ChainIndexTx (..), ChainIndexTxOutputs (..))
-import           Plutus.ChainIndex.TxIdState (TxIdState)
 import qualified Plutus.ChainIndex.TxIdState as TxIdState
-import           Plutus.ChainIndex.Types     (BlockId (..), BlockNumber (..), Tip (..))
+import           Plutus.ChainIndex.Types     (BlockId (..), BlockNumber (..), Tip (..), TxIdState)
 import           Plutus.ChainIndex.UtxoState (TxUtxoBalance (..), fromTx)
 import qualified PlutusTx.Prelude            as PlutusTx
 
diff --git a/plutus-chain-index/test/Spec.hs b/plutus-chain-index/test/Spec.hs
@@ -101,9 +101,9 @@ rollbackTxIdState = property $ do
       status3 = transactionStatus (BlockNumber 2) (getState f3) (txB ^. citxTxId)
       status4 = transactionStatus (BlockNumber 3) (getState f4) (txB ^. citxTxId)
 
-  status2 === TentativelyConfirmed (Depth 0) TxValid
-  status3 === Unknown
-  status4 === TentativelyConfirmed (Depth 1) TxValid
+  status2 === (Right $ TentativelyConfirmed (Depth 0) TxValid)
+  status3 === (Right $ Unknown)
+  status4 === (Right $ TentativelyConfirmed (Depth 1) TxValid)
 
 transactionDepthIncreases :: Property
 transactionDepthIncreases = property $ do
@@ -121,8 +121,8 @@ transactionDepthIncreases = property $ do
       status2 = transactionStatus (BlockNumber 1) (UtxoState._usTxUtxoData (measure f2)) (txA ^. citxTxId)
       status3 = transactionStatus (BlockNumber (1 + d)) (UtxoState._usTxUtxoData (measure f2)) (txA ^. citxTxId)
 
-  status2 === increaseDepth status1
-  status3 === Committed TxValid
+  status2 === (increaseDepth <$> status1)
+  status3 === (Right $ Committed TxValid)
 
 uniqueTransactionIds :: Property
 uniqueTransactionIds = property $ do
diff --git a/plutus-pab/src/Plutus/PAB/Core/ContractInstance/BlockchainEnv.hs b/plutus-pab/src/Plutus/PAB/Core/ContractInstance/BlockchainEnv.hs
@@ -33,11 +33,12 @@ import           Plutus.V1.Ledger.Api                   (toBuiltin)
 import           Control.Concurrent.STM                 (STM)
 import qualified Control.Concurrent.STM                 as STM
 import           Control.Lens
-import           Control.Monad                          (forM_, unless, void, when)
+import           Control.Monad                          (forM_, void, when)
 import           Data.Foldable                          (foldl')
 import           Ledger.TimeSlot                        (SlotConfig)
 import           Plutus.ChainIndex                      (BlockNumber (..), ChainIndexTx (..), ChainIndexTxOutputs (..),
-                                                         InsertUtxoSuccess (..), RollbackResult (..), Tip (..),
+                                                         InsertUtxoFailed (..), InsertUtxoSuccess (..),
+                                                         RollbackFailed (..), RollbackResult (..), Tip (..),
                                                          TxConfirmedState (..), TxIdState (..), TxValidity (..),
                                                          UtxoState (..), blockId, citxTxId, fromOnChainTx, insert)
 import           Plutus.ChainIndex.Compatibility        (fromCardanoBlockHeader, fromCardanoPoint)
@@ -57,13 +58,18 @@ startNodeClient socket mode slotConfig networkId instancesState = do
     case mode of
       MockNode ->
         void $ MockClient.runChainSync socket slotConfig
-            (\block slot -> STM.atomically $ processMockBlock instancesState env block slot)
+            (\block slot -> handleSyncAction $ processMockBlock instancesState env block slot)
       AlonzoNode -> do
           let resumePoints = []
           void $ Client.runChainSync socket slotConfig networkId resumePoints
-            (\block slot -> STM.atomically $ processChainSyncEvent env block slot)
+            (\block slot -> handleSyncAction $ processChainSyncEvent env block slot)
     pure env
 
+-- | Deal with sync action failures from running this STM action. For now, we
+-- deal with them by simply calling `error`; i.e. the application exits.
+handleSyncAction :: STM (Either SyncActionFailure ()) -> IO ()
+handleSyncAction action = STM.atomically action >>= either (error . show) pure
+
 updateInstances :: IndexedBlock -> InstanceClientEnv -> STM ()
 updateInstances IndexedBlock{ibUtxoSpent, ibUtxoProduced} InstanceClientEnv{ceUtxoSpentRequests, ceUtxoProducedRequests} = do
   forM_ (Map.intersectionWith (,) ibUtxoSpent ceUtxoSpentRequests) $ \(onChainTx, requests) ->
@@ -72,23 +78,29 @@ updateInstances IndexedBlock{ibUtxoSpent, ibUtxoProduced} InstanceClientEnv{ceUt
     traverse (\OpenTxOutProducedRequest{otxProducingTxns} -> STM.tryPutTMVar otxProducingTxns txns) requests
 
 -- | Process a chain sync event that we receive from the alonzo node client
-processChainSyncEvent :: BlockchainEnv -> ChainSyncEvent -> Slot -> STM ()
-processChainSyncEvent blockchainEnv event _slot = case event of
-  Resume _                                                     -> pure () -- TODO: Handle resume
-  RollForward  (BlockInMode (C.Block header transactions) era) -> processBlock header blockchainEnv transactions era
-  RollBackward chainPoint                                      -> runRollback blockchainEnv chainPoint
+processChainSyncEvent :: BlockchainEnv -> ChainSyncEvent -> Slot -> STM (Either SyncActionFailure ())
+processChainSyncEvent blockchainEnv event _slot = do
+  case event of
+      Resume _                                                     -> pure $ Right () -- TODO: Handle resume
+      RollForward  (BlockInMode (C.Block header transactions) era) -> processBlock header blockchainEnv transactions era
+      RollBackward chainPoint                                      -> runRollback blockchainEnv chainPoint
+
+data SyncActionFailure
+  = RollbackFailure RollbackFailed
+  | InsertUtxoStateFailure InsertUtxoFailed
+  deriving (Show)
 
 -- | Roll back the chain to the given ChainPoint and slot.
-runRollback :: BlockchainEnv -> ChainPoint -> STM ()
+runRollback :: BlockchainEnv -> ChainPoint -> STM (Either SyncActionFailure ())
 runRollback BlockchainEnv{beTxChanges} chainPoint = do
   txIdStateIndex <- STM.readTVar beTxChanges
 
   let point = fromCardanoPoint chainPoint
       rs    = rollback point txIdStateIndex
 
   case rs of
-    Left e                                -> error $ "Rollback Failed: " <> show e
-    Right RollbackResult{rolledBackIndex} -> STM.writeTVar beTxChanges rolledBackIndex
+    Left e                                -> pure $ Left (RollbackFailure e)
+    Right RollbackResult{rolledBackIndex} -> Right <$> STM.writeTVar beTxChanges rolledBackIndex
 
 -- | Get transaction ID and validity from a cardano transaction in any era
 txEvent :: forall era. C.Tx era -> C.EraInMode era C.CardanoMode -> (TxId, TxValidity)
@@ -108,11 +120,17 @@ txMockEvent tx =
 
 -- | Update the blockchain env. with changes from a new block of cardano
 --   transactions in any era
-processBlock :: forall era. C.BlockHeader -> BlockchainEnv -> [C.Tx era] -> C.EraInMode era C.CardanoMode -> STM ()
+processBlock :: forall era. C.BlockHeader
+             -> BlockchainEnv
+             -> [C.Tx era]
+             -> C.EraInMode era C.CardanoMode
+             -> STM (Either SyncActionFailure ())
 processBlock header env transactions era =
-  unless (null transactions) $ do
-    let tip = fromCardanoBlockHeader header
-    updateTransactionState tip env (flip txEvent era <$> transactions)
+  if null transactions
+     then pure $ Right ()
+     else do
+        let tip = fromCardanoBlockHeader header
+        updateTransactionState tip env (flip txEvent era <$> transactions)
 
 -- | For the given transactions, perform the updates in the TxIdState, and
 -- also record that a new block has been processed.
@@ -121,19 +139,19 @@ updateTransactionState
   => Tip
   -> BlockchainEnv
   -> t (TxId, TxValidity)
-  -> STM ()
+  -> STM (Either SyncActionFailure ())
 updateTransactionState tip BlockchainEnv{beTxChanges, beCurrentBlock} xs = do
     txIdStateIndex <- STM.readTVar beTxChanges
     let txIdState = _usTxUtxoData $ measure $ txIdStateIndex
     blockNumber <- STM.readTVar beCurrentBlock
     let txIdState' = foldl' (insertNewTx blockNumber) txIdState xs
         is  = insert (UtxoState txIdState' tip) txIdStateIndex
     case is of
-      -- TODO: Proper error.
-      Left e                                         -> error $ "Insert of new TxIdState failed." <> show e
-      Right InsertUtxoSuccess{newIndex=newTxIdState} -> STM.writeTVar beTxChanges newTxIdState
-    STM.writeTVar beCurrentBlock (succ blockNumber)
-
+      Right InsertUtxoSuccess{newIndex=newTxIdState} -> do
+        STM.writeTVar beTxChanges newTxIdState
+        STM.writeTVar beCurrentBlock (succ blockNumber)
+        pure $ Right ()
+      Left e -> pure $ Left $ InsertUtxoStateFailure e
 
 insertNewTx :: BlockNumber -> TxIdState -> (TxId, TxValidity) -> TxIdState
 insertNewTx blockNumber TxIdState{txnsConfirmed, txnsDeleted} (txi, txValidity) =
@@ -150,26 +168,30 @@ insertNewTx blockNumber TxIdState{txnsConfirmed, txnsDeleted} (txi, txValidity)
 
 -- | Go through the transactions in a block, updating the 'BlockchainEnv'
 --   when any interesting addresses or transactions have changed.
-processMockBlock :: InstancesState -> BlockchainEnv -> Block -> Slot -> STM ()
+processMockBlock :: InstancesState -> BlockchainEnv -> Block -> Slot -> STM (Either SyncActionFailure ())
 processMockBlock instancesState env@BlockchainEnv{beAddressMap, beCurrentSlot, beCurrentBlock} transactions slot = do
   lastSlot <- STM.readTVar beCurrentSlot
   when (slot > lastSlot) $ do
     STM.writeTVar beCurrentSlot slot
-  unless (null transactions) $ do
-    addressMap <- STM.readTVar beAddressMap
-    let addressMap' = foldl' (processTx slot) addressMap transactions
-    STM.writeTVar beAddressMap addressMap'
-    blockNumber <- STM.readTVar beCurrentBlock
 
-    let tip = Tip { tipSlot = slot
-                  , tipBlockId = blockId transactions
-                  , tipBlockNo = blockNumber
-                  }
+  if null transactions
+     then pure $ Right ()
+     else do
+      addressMap <- STM.readTVar beAddressMap
+      let addressMap' = foldl' (processTx slot) addressMap transactions
+      STM.writeTVar beAddressMap addressMap'
+      blockNumber <- STM.readTVar beCurrentBlock
+
+      instEnv <- S.instancesClientEnv instancesState
+      updateInstances (indexBlock $ fmap fromOnChainTx transactions) instEnv
+
 
-    updateTransactionState tip env (txMockEvent <$> fmap fromOnChainTx transactions)
+      let tip = Tip { tipSlot = slot
+                    , tipBlockId = blockId transactions
+                    , tipBlockNo = blockNumber
+                    }
 
-    instEnv <- S.instancesClientEnv instancesState
-    updateInstances (indexBlock $ fmap fromOnChainTx transactions) instEnv
+      updateTransactionState tip env (txMockEvent <$> fmap fromOnChainTx transactions)
 
 processTx :: Slot -> AddressMap -> OnChainTx -> AddressMap
 processTx _ addressMap tx = addressMap' where
diff --git a/plutus-pab/src/Plutus/PAB/Core/ContractInstance/STM.hs b/plutus-pab/src/Plutus/PAB/Core/ContractInstance/STM.hs
@@ -424,8 +424,12 @@ waitForTxStatusChange oldStatus tx BlockchainEnv{beTxChanges, beCurrentBlock} =
     txIdState   <- _usTxUtxoData . measure <$> STM.readTVar beTxChanges
     blockNumber <- STM.readTVar beCurrentBlock
     let newStatus = transactionStatus blockNumber txIdState tx
-    guard $ oldStatus /= newStatus
-    pure newStatus
+    -- Succeed only if we _found_ a status and it was different; if
+    -- the status hasn't changed, _or_ there was an error computing
+    -- the status, keep retrying.
+    case newStatus of
+      Right s | s /= oldStatus -> pure s
+      _                        -> empty
 
 -- | The value at an address
 valueAt :: Address -> BlockchainEnv -> STM Value.Value
