[WIP] Introduce shared on-chain state to ease constructing txs

This commit only adds the needed machinery but tests aren't passing
and state is not really used yet.

hydra-node/src/Hydra/Chain/Direct.hs

@@ -14,7 +14,13 @@ import Hydra.Prelude
 
 import Cardano.Ledger.Alonzo.Tx (ValidatedTx)
 import Cardano.Ledger.Alonzo.TxSeq (txSeqTxns)
-import Control.Monad.Class.MonadSTM (newTQueueIO, readTQueue, writeTQueue)
+import Control.Monad.Class.MonadSTM (
+  newTQueueIO,
+  newTVarIO,
+  readTQueue,
+  readTVarIO,
+  writeTQueue,
+ )
 import Control.Tracer (nullTracer)
 import Data.Sequence.Strict (StrictSeq)
 import Hydra.Chain (
@@ -23,8 +29,8 @@ import Hydra.Chain (
   ChainComponent,
   PostChainTx (..),
  )
-import Hydra.Chain.Direct.Tx (OnChainHeadState (Ready), constructTx, observeTx)
-import Hydra.Ledger.Cardano (generateWith)
+import Hydra.Chain.Direct.Tx (OnChainHeadState (Closed), constructTx, runOnChainTxs)
+import Hydra.Chain.Direct.Util (Block, Era, defaultCodecs, nullConnectTracers, versions)
 import Hydra.Logging (Tracer)
 import Ouroboros.Consensus.Cardano.Block (GenTx (..), HardForkBlock (BlockAlonzo))
 import Ouroboros.Consensus.Ledger.SupportsMempool (ApplyTxErr)
@@ -76,33 +82,35 @@ withDirectChain ::
   ChainComponent tx IO ()
 withDirectChain _tracer magic iocp addr callback action = do
   queue <- newTQueueIO
+  headState <- newTVarIO Closed
   race_
     (action $ Chain{postTx = atomically . writeTQueue queue})
     ( connectTo
         (localSnocket iocp addr)
         nullConnectTracers
-        (versions magic (client queue callback))
+        (versions magic (client queue headState callback))
         addr
     )
 
 client ::
   (MonadST m, MonadTimer m) =>
   TQueue m (PostChainTx tx) ->
+  TVar m OnChainHeadState ->
   ChainCallback tx m ->
   NodeToClientVersion ->
   OuroborosApplication 'InitiatorMode LocalAddress LByteString m () Void
-client queue callback nodeToClientV =
+client queue headState callback nodeToClientV =
   nodeToClientProtocols
     ( const $
         pure $
           NodeToClientProtocols
             { localChainSyncProtocol =
                 InitiatorProtocolOnly $
-                  let peer = chainSyncClientPeer $ chainSyncClient callback
+                  let peer = chainSyncClientPeer $ chainSyncClient callback headState
                    in MuxPeer nullTracer cChainSyncCodec peer
             , localTxSubmissionProtocol =
                 InitiatorProtocolOnly $
-                  let peer = localTxSubmissionClientPeer $ txSubmissionClient queue
+                  let peer = localTxSubmissionClientPeer $ txSubmissionClient queue headState
                    in MuxPeer nullTracer cTxSubmissionCodec peer
             , localStateQueryProtocol =
                 InitiatorProtocolOnly $
@@ -120,10 +128,11 @@ client queue callback nodeToClientV =
 
 chainSyncClient ::
   forall m tx.
-  Monad m =>
+  (MonadSTM m) =>
   ChainCallback tx m ->
+  TVar m OnChainHeadState ->
   ChainSyncClient Block (Point Block) (Tip Block) m ()
-chainSyncClient callback =
+chainSyncClient callback headState =
   ChainSyncClient (pure clientStIdle)
  where
   -- FIXME: This won't work well with real client. Without acquiring any point
@@ -156,7 +165,9 @@ chainSyncClient callback =
           ChainSyncClient $ do
             -- REVIEW(SN): There seems to be no 'toList' for StrictSeq? That's
             -- why I resorted to foldMap using the list monoid ('pure')
-            mapM_ callback . catMaybes . foldMap (pure . observeTx) $ getAlonzoTxs blk
+            let txs = toList $ getAlonzoTxs blk
+            onChainTxs <- runOnChainTxs headState txs
+            mapM_ callback onChainTxs
             pure clientStIdle
       , recvMsgRollBackward =
           error "Rolled backward!"
@@ -166,14 +177,16 @@ txSubmissionClient ::
   forall m tx.
   MonadSTM m =>
   TQueue m (PostChainTx tx) ->
+  TVar m OnChainHeadState ->
   LocalTxSubmissionClient (GenTx Block) (ApplyTxErr Block) m ()
-txSubmissionClient queue =
+txSubmissionClient queue headState =
   LocalTxSubmissionClient clientStIdle
  where
   clientStIdle :: m (LocalTxClientStIdle (GenTx Block) (ApplyTxErr Block) m ())
   clientStIdle = do
     tx <- atomically $ readTQueue queue
-    pure $ SendMsgSubmitTx (fromPostChainTx tx) (const clientStIdle)
+    st <- readTVarIO headState
+    pure $ SendMsgSubmitTx (fromPostChainTx st tx) (const clientStIdle)
 
   -- FIXME
   -- This is where we need signatures and client credentials. Ideally, we would
@@ -182,11 +195,12 @@ txSubmissionClient queue =
   -- client submit a signed transaction.
   --
   -- For now, it simply does not sign..
-  fromPostChainTx :: PostChainTx tx -> GenTx Block
-  fromPostChainTx postChainTx = do
-    let txIn = generateWith arbitrary 42
-        unsignedTx = constructTx (Ready txIn) postChainTx
-    GenTxAlonzo $ mkShelleyTx unsignedTx
+  --
+  -- TODO inline constructTx to be able to decide here which side effect to do
+  -- eg. we need to ask the wallet for a TxIn to produce the transaction
+  fromPostChainTx :: OnChainHeadState -> PostChainTx tx -> GenTx Block
+  fromPostChainTx st postChainTx =
+    GenTxAlonzo $ mkShelleyTx $ constructTx st postChainTx
 
 --
 -- Helpers

hydra-node/src/Hydra/Chain/Direct/Tx.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE AllowAmbiguousTypes #-}
 {-# LANGUAGE TypeApplications #-}
 
 -- | Smart constructors for creating Hydra protocol transactions to be used in
@@ -22,6 +23,8 @@ import Cardano.Ledger.Crypto (StandardCrypto)
 import Cardano.Ledger.Era (hashScript)
 import Cardano.Ledger.ShelleyMA.Timelocks (ValidityInterval (..))
 import Cardano.Ledger.Val (inject)
+import Control.Monad (foldM)
+import Control.Monad.Class.MonadSTM (stateTVar)
 import qualified Data.Map as Map
 import qualified Data.Sequence.Strict as StrictSeq
 import qualified Data.Set as Set
@@ -55,19 +58,24 @@ type Era = AlonzoEra StandardCrypto
 -- | Maintains information needed to construct on-chain transactions
 -- depending on the current state of the head.
 data OnChainHeadState
-  = Ready {seedTx :: TxIn StandardCrypto}
-  | Initial {commits :: (TxIn StandardCrypto, PubKeyHash)}
+  = Closed
+  | Initial
+      { -- TODO add the output containing the SM token
+        -- TODO initials should be a list of inputs/PubKeyHas
+        -- TODO add commits
+        initials :: (TxIn StandardCrypto, PubKeyHash)
+      }
   deriving (Eq, Show, Generic)
 
 -- | Construct the Head protocol transactions as Alonzo 'Tx'. Note that
 -- 'ValidatedTx' this produces an unbalanced, unsigned transaction and this type
 -- was used (in contrast to 'TxBody') to be able to express included datums,
 -- onto which at least the 'initTx' relies on.
 constructTx :: OnChainHeadState -> PostChainTx tx -> ValidatedTx Era
-constructTx txIn tx =
-  case (txIn, tx) of
-    (Ready{seedTx}, InitTx p) -> initTx p seedTx
-    (Initial{commits}, AbortTx _utxo) -> abortTx commits
+constructTx st tx =
+  case (st, tx) of
+    (Closed, InitTx p) -> initTx p (error "undefined")
+    (Initial{initials}, AbortTx _utxo) -> uncurry abortTx initials
     _ -> error "not implemented"
 
 -- | Create the init transaction from some 'HeadParameters' and a single TxIn
@@ -119,8 +127,14 @@ initTx HeadParameters{contestationPeriod, parties} txIn =
 -- only possible if this is governed by the initial script and only for a single
 -- input. Of course, the Head protocol specifies we need to spend ALL the Utxo
 -- containing PTs.
-abortTx :: (TxIn StandardCrypto, PubKeyHash) -> ValidatedTx Era
-abortTx (txIn, pkh) =
+abortTx ::
+  -- | The input to be consumed by the abort transaction, locked by the validator
+  -- script
+  TxIn StandardCrypto ->
+  -- | The datum to provide to the validator script
+  PubKeyHash ->
+  ValidatedTx Era
+abortTx txIn pkh =
   mkUnsignedTx body dats redeemers scripts
  where
   body =
@@ -162,27 +176,40 @@ abortTx (txIn, pkh) =
 
 -- * Observe Hydra Head transactions
 
-observeTx :: ValidatedTx Era -> Maybe (OnChainTx tx)
-observeTx tx =
-  observeInitTx tx
-    <|> observeAbortTx tx
-
-observeInitTx :: ValidatedTx Era -> Maybe (OnChainTx tx)
-observeInitTx ValidatedTx{wits} = do
-  (Data d) <- firstDatum
-  fromData d >>= \case
-    Head.Initial cp ps ->
-      pure $ OnInitTx (contestationPeriodToDiffTime cp) (map convertParty ps)
-    _ -> Nothing
+-- | Update observable on-chain head state from on-chain transactions.
+-- NOTE(AB): I tried to separate the 2 functions, the one working on list of txs and the one
+-- working on single tx but I keep getting failed unification between `m` and `m0` which is
+-- puzzling...
+runOnChainTxs :: forall m tx. MonadSTM m => TVar m OnChainHeadState -> [ValidatedTx Era] -> m [OnChainTx tx]
+runOnChainTxs headState = atomically . foldM runOnChainTx []
  where
+  runOnChainTx :: [OnChainTx tx] -> ValidatedTx Era -> STM m [OnChainTx tx]
+  runOnChainTx observed tx = do
+    newObserved <- catMaybes <$> mapM (stateTVar headState) [observeInitTx tx, observeAbortTx tx]
+    pure $ observed <> newObserved
+
+observeInitTx :: ValidatedTx Era -> OnChainHeadState -> (Maybe (OnChainTx tx), OnChainHeadState)
+observeInitTx ValidatedTx{wits} st =
+  case extractParameters of
+    Just (Head.Initial cp ps) ->
+      ( Just $ OnInitTx (contestationPeriodToDiffTime cp) (map convertParty ps)
+      , st
+      )
+    _ -> (Nothing, st)
+ where
+  extractParameters = do
+    Data d <- firstDatum
+    fromData d
   firstDatum = snd . head <$> nonEmpty datums
 
   datums = Map.toList . unTxDats $ txdats wits
 
   convertParty = anonymousParty . partyToVerKey
 
-observeAbortTx :: ValidatedTx Era -> Maybe (OnChainTx tx)
-observeAbortTx _ = Just OnAbortTx
+  _firstInput = error "undefined"
+
+observeAbortTx :: ValidatedTx Era -> OnChainHeadState -> (Maybe (OnChainTx tx), OnChainHeadState)
+observeAbortTx _ st = (Just OnAbortTx, st)
 --
 
 -- * Helpers

hydra-node/test/Hydra/Chain/Direct/TxSpec.hs

@@ -28,8 +28,8 @@ import qualified Data.ByteString.Lazy as LBS
 import qualified Data.Map as Map
 import Data.Maybe (fromJust)
 import Data.Time.Clock.POSIX (posixSecondsToUTCTime)
-import Hydra.Chain (HeadParameters (..), PostChainTx (InitTx), toOnChainTx)
-import Hydra.Chain.Direct.Tx (constructTx, initTx, observeTx)
+import Hydra.Chain (HeadParameters (..), PostChainTx (..), toOnChainTx)
+import Hydra.Chain.Direct.Tx (OnChainHeadState (..), abortTx, constructTx, initTx, observeInitTx, plutusScript, scriptAddr)
 import Hydra.Chain.Direct.Util (Era)
 import qualified Hydra.Contract.Head as Head
 import qualified Hydra.Contract.Initial as Initial
@@ -53,7 +53,7 @@ spec =
   parallel $ do
     prop "observeTx . constructTx roundtrip" $ \postTx txIn time ->
       isImplemented postTx txIn -- TODO(SN): test all constructors
-        ==> observeTx (constructTx txIn postTx) === Just (toOnChainTx @SimpleTx time postTx)
+        ==> fst (observeInitTx (constructTx txIn postTx) Closed) === Just (toOnChainTx @SimpleTx time postTx)
 
     prop "transaction size below limit" $ \postTx txIn ->
       isImplemented postTx txIn -- TODO(SN): test all constructors
@@ -79,7 +79,7 @@ spec =
 
     describe "abortTx" $ do
       prop "transaction size below limit" $ \txIn bytes ->
-        let tx = abortTx (txIn, pkh)
+        let tx = abortTx txIn pkh
             pkh = PubKeyHash $ toBuiltin (bytes :: ByteString)
             cbor = serialize tx
             len = LBS.length cbor
@@ -90,7 +90,7 @@ spec =
       -- TODO(SN): this requires the abortTx to include a redeemer, for a TxIn,
       -- spending an Initial-validated output
       prop "validates against 'initial' script in haskell (unlimited budget)" $ \txIn bytes ->
-        let tx = abortTx (txIn, pkh)
+        let tx = abortTx txIn pkh
             pkh = PubKeyHash $ toBuiltin (bytes :: ByteString)
             -- input governed by initial script and a 'Plutus.PubKeyHash' datum
             utxo = UTxO $ Map.singleton txIn txOut
@@ -107,7 +107,7 @@ spec =
 isImplemented :: PostChainTx tx -> OnChainHeadState -> Bool
 isImplemented tx st =
   case (tx, st) of
-    (InitTx{}, Ready{}) -> True
+    (InitTx{}, Closed) -> True
     (AbortTx{}, Initial{}) -> True
     _ -> False
 
@@ -143,6 +143,6 @@ txOutNFT (TxOut _ value _) =
   unitQuantity (_name, q) = q == 1
 
 instance Arbitrary OnChainHeadState where
-  arbitrary = oneof [Ready <$> arbitrary, Initial <$> ((,) <$> arbitrary <*> genPubKeyHash)]
+  arbitrary = oneof [pure Closed, Initial <$> ((,) <$> arbitrary <*> genPubKeyHash)]
    where
     genPubKeyHash = PubKeyHash . toBuiltin <$> (arbitrary :: Gen ByteString)

hydra-node/test/Hydra/Chain/DirectSpec.hs

@@ -12,8 +12,8 @@ import Control.Concurrent (newEmptyMVar, putMVar, takeMVar)
 import Hydra.Chain (
   Chain (..),
   HeadParameters (HeadParameters),
-  OnChainTx (OnInitTx),
-  PostChainTx (InitTx),
+  OnChainTx (OnAbortTx, OnInitTx),
+  PostChainTx (AbortTx, InitTx),
  )
 import Hydra.Chain.Direct (withDirectChain)
 import Hydra.Chain.Direct.MockServer (withMockServer)
@@ -24,10 +24,10 @@ import Hydra.Party (Party, deriveParty, generateKey)
 spec :: Spec
 spec = parallel $ do
   it "publishes init tx and observes it also" $ do
+    calledBackAlice <- newEmptyMVar
+    calledBackBob <- newEmptyMVar
     withMockServer $ \networkMagic iocp socket _ -> do
-      calledBackAlice <- newEmptyMVar
       withDirectChain nullTracer networkMagic iocp socket (putMVar calledBackAlice) $ \Chain{postTx} -> do
-        calledBackBob <- newEmptyMVar
         withDirectChain nullTracer networkMagic iocp socket (putMVar calledBackBob) $ \_ -> do
           let parameters = HeadParameters 100 [alice, bob, carol]
           postTx $ InitTx @SimpleTx parameters
@@ -36,6 +36,22 @@ spec = parallel $ do
           failAfter 5 $
             takeMVar calledBackBob `shouldReturn` OnInitTx @SimpleTx 100 [alice, bob, carol]
 
+  it "can init and abort a head given nothing has been committed" $ do
+    calledBackAlice <- newEmptyMVar
+    calledBackBob <- newEmptyMVar
+    withMockServer $ \networkMagic iocp socket _ -> do
+      withDirectChain nullTracer networkMagic iocp socket (putMVar calledBackAlice) $ \Chain{postTx} -> do
+        withDirectChain nullTracer networkMagic iocp socket (putMVar calledBackBob) $ \_ -> do
+          let parameters = HeadParameters 100 [alice, bob, carol]
+          postTx $ InitTx @SimpleTx parameters
+          failAfter 5 $
+            takeMVar calledBackAlice `shouldReturn` OnInitTx 100 [alice, bob, carol]
+
+          postTx $ AbortTx mempty
+
+          failAfter 5 $
+            takeMVar calledBackBob `shouldReturn` OnAbortTx @SimpleTx
+
 alice, bob, carol :: Party
 alice = deriveParty $ generateKey 10
 bob = deriveParty $ generateKey 20