Introduce changes to increment observation

Signed-off-by: Sasha Bogicevic <[email protected]>

hydra-chain-observer/src/Hydra/ChainObserver.hs

@@ -92,6 +92,7 @@ data ChainObserverLog
   | HeadCommitTx {headId :: HeadId}
   | HeadCollectComTx {headId :: HeadId}
   | HeadDepositTx {headId :: HeadId}
+  | HeadIncrementTx {headId :: HeadId}
   | HeadDecrementTx {headId :: HeadId}
   | HeadCloseTx {headId :: HeadId}
   | HeadFanoutTx {headId :: HeadId}
@@ -204,6 +205,7 @@ chainSyncClient tracer networkId startingPoint observerHandler =
     OnInitTx{headId} -> HeadInitTx{headId}
     OnCommitTx{headId} -> HeadCommitTx{headId}
     OnCollectComTx{headId} -> HeadCollectComTx{headId}
+    OnIncrementTx{headId} -> HeadIncrementTx{headId}
     OnDepositTx{headId} -> HeadDepositTx{headId}
     OnDecrementTx{headId} -> HeadDecrementTx{headId}
     OnCloseTx{headId} -> HeadCloseTx{headId}

hydra-cluster/src/HydraNode.hs

@@ -380,14 +380,14 @@ withHydraNode' tracer chainConfig workDir hydraNodeId hydraSKey hydraVKeys allNo
                 }
           )
             { std_out = maybe CreatePipe UseHandle mGivenStdOut
-            , std_err = CreatePipe
+            , std_err = Inherit
             }
 
     traceWith tracer $ HydraNodeCommandSpec $ show $ cmdspec p
 
     withCreateProcess p $ \_stdin mCreatedStdOut mCreatedStdErr processHandle ->
       case (mCreatedStdOut <|> mGivenStdOut, mCreatedStdErr) of
-        (Just out, Just err) -> action out err processHandle
+        (Just out, Nothing) -> action out stderr processHandle
         (Nothing, _) -> error "Should not happen™"
         (_, Nothing) -> error "Should not happen™"
  where

hydra-explorer/src/Hydra/Explorer/ExplorerState.hs

@@ -226,6 +226,50 @@ aggregateCollectComObservation headId point blockNo currentHeads =
       , blockNo = blockNo
       }
 
+aggregateDepositObservation :: HeadId -> ChainPoint -> BlockNo -> [HeadState] -> [HeadState]
+aggregateDepositObservation headId point blockNo currentHeads =
+  case findHeadState headId currentHeads of
+    Just headState ->
+      let newHeadState = headState{status = Open}
+       in replaceHeadState newHeadState currentHeads
+    Nothing -> currentHeads <> [newUnknownHeadState]
+ where
+  newUnknownHeadState =
+    HeadState
+      { headId
+      , seedTxIn = Unknown
+      , status = Open
+      , contestationPeriod = Unknown
+      , members = Unknown
+      , contestations = Seen 0
+      , snapshotNumber = Seen 0
+      , contestationDeadline = Unknown
+      , point = point
+      , blockNo = blockNo
+      }
+
+aggregateIncrementObservation :: HeadId -> ChainPoint -> BlockNo -> [HeadState] -> [HeadState]
+aggregateIncrementObservation headId point blockNo currentHeads =
+  case findHeadState headId currentHeads of
+    Just headState ->
+      let newHeadState = headState{status = Open}
+       in replaceHeadState newHeadState currentHeads
+    Nothing -> currentHeads <> [newUnknownHeadState]
+ where
+  newUnknownHeadState =
+    HeadState
+      { headId
+      , seedTxIn = Unknown
+      , status = Open
+      , contestationPeriod = Unknown
+      , members = Unknown
+      , contestations = Seen 0
+      , snapshotNumber = Seen 0
+      , contestationDeadline = Unknown
+      , point = point
+      , blockNo = blockNo
+      }
+
 aggregateDecrementObservation :: HeadId -> ChainPoint -> BlockNo -> [HeadState] -> [HeadState]
 aggregateDecrementObservation headId point blockNo currentHeads =
   case findHeadState headId currentHeads of
@@ -364,6 +408,16 @@ aggregateHeadObservations observations explorerState =
           { heads = aggregateCollectComObservation headId point blockNo heads
           , tick = TickState point blockNo
           }
+      HeadObservation{point, blockNo, onChainTx = OnDepositTx{headId}} ->
+        ExplorerState
+          { heads = aggregateDepositObservation headId point blockNo heads
+          , tick = TickState point blockNo
+          }
+      HeadObservation{point, blockNo, onChainTx = OnIncrementTx{headId}} ->
+        ExplorerState
+          { heads = aggregateIncrementObservation headId point blockNo heads
+          , tick = TickState point blockNo
+          }
       HeadObservation{point, blockNo, onChainTx = OnDecrementTx{headId}} ->
         ExplorerState
           { heads = aggregateDecrementObservation headId point blockNo heads

hydra-node/src/Hydra/Chain.hs

@@ -57,6 +57,11 @@ data PostChainTx tx
   = InitTx {participants :: [OnChainId], headParameters :: HeadParameters}
   | AbortTx {utxo :: UTxOType tx, headSeed :: HeadSeed}
   | CollectComTx {utxo :: UTxOType tx, headId :: HeadId, headParameters :: HeadParameters}
+  | IncrementTx
+      { headId :: HeadId
+      , headParameters :: HeadParameters
+      , incrementingSnapshot :: ConfirmedSnapshot tx
+      }
   | DecrementTx
       { headId :: HeadId
       , headParameters :: HeadParameters
@@ -88,6 +93,7 @@ instance ArbitraryIsTx tx => Arbitrary (PostChainTx tx) where
     InitTx{participants, headParameters} -> InitTx <$> shrink participants <*> shrink headParameters
     AbortTx{utxo, headSeed} -> AbortTx <$> shrink utxo <*> shrink headSeed
     CollectComTx{utxo, headId, headParameters} -> CollectComTx <$> shrink utxo <*> shrink headId <*> shrink headParameters
+    IncrementTx{headId, headParameters, incrementingSnapshot} -> IncrementTx <$> shrink headId <*> shrink headParameters <*> shrink incrementingSnapshot
     DecrementTx{headId, headParameters, decrementingSnapshot} -> DecrementTx <$> shrink headId <*> shrink headParameters <*> shrink decrementingSnapshot
     CloseTx{headId, headParameters, openVersion, closingSnapshot} -> CloseTx <$> shrink headId <*> shrink headParameters <*> shrink openVersion <*> shrink closingSnapshot
     ContestTx{headId, headParameters, openVersion, contestingSnapshot} -> ContestTx <$> shrink headId <*> shrink headParameters <*> shrink openVersion <*> shrink contestingSnapshot
@@ -113,6 +119,11 @@ data OnChainTx tx
       { headId :: HeadId
       , utxo :: UTxOType tx
       }
+  | OnIncrementTx
+      { headId :: HeadId
+      , newVersion :: SnapshotVersion
+      , utxo :: UTxOType tx
+      }
   | OnDecrementTx
       { headId :: HeadId
       , newVersion :: SnapshotVersion
@@ -175,6 +186,7 @@ data PostTxError tx
   | FailedToConstructCollectTx
   | FailedToConstructDepositTx
   | FailedToConstructRecoverTx
+  | FailedToConstructIncrementTx
   | FailedToConstructDecrementTx
   | FailedToConstructFanoutTx
   deriving stock (Generic)

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

@@ -52,6 +52,7 @@ import Hydra.Chain.Direct.State (
   decrement,
   fanout,
   getKnownUTxO,
+  increment,
   initialize,
  )
 import Hydra.Chain.Direct.TimeHandle (TimeHandle (..))
@@ -393,6 +394,10 @@ prepareTxToPost timeHandle wallet ctx spendableUTxO tx =
       case collect ctx headId headParameters utxo spendableUTxO of
         Left _ -> throwIO (FailedToConstructCollectTx @Tx)
         Right collectTx -> pure collectTx
+    IncrementTx{headId, headParameters, incrementingSnapshot} ->
+      case increment ctx spendableUTxO headId headParameters incrementingSnapshot of
+        Left _ -> throwIO (FailedToConstructIncrementTx @Tx)
+        Right incrementTx' -> pure incrementTx'
     DecrementTx{headId, headParameters, decrementingSnapshot} ->
       case decrement ctx spendableUTxO headId headParameters decrementingSnapshot of
         Left _ -> throwIO (FailedToConstructDecrementTx @Tx)

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

@@ -109,6 +109,7 @@ import Hydra.Tx.Crypto (HydraKey)
 import Hydra.Tx.Decrement (decrementTx)
 import Hydra.Tx.Deposit (depositTx)
 import Hydra.Tx.Fanout (fanoutTx)
+import Hydra.Tx.Increment (incrementTx)
 import Hydra.Tx.Init (initTx)
 import Hydra.Tx.OnChainId (OnChainId)
 import Hydra.Tx.Snapshot (genConfirmedSnapshot)
@@ -477,6 +478,47 @@ collect ctx headId headParameters utxoToCollect spendableUTxO = do
 
   ChainContext{networkId, ownVerificationKey, scriptRegistry} = ctx
 
+data IncrementTxError
+  = InvalidHeadIdInIncrement {headId :: HeadId}
+  | CannotFindHeadOutputInIncrement
+  | SnapshotMissingIncrementUTxO
+  | SnapshotIncrementUTxOIsNull
+  deriving stock (Show)
+
+-- | Construct a increment transaction spending the head and deposit outputs in given 'UTxO',
+-- and producing single head output for pending 'utxoToCommit' of given 'Snapshot'.
+increment ::
+  ChainContext ->
+  -- | Spendable UTxO containing head and deposit outputs
+  UTxO ->
+  HeadId ->
+  HeadParameters ->
+  -- | Snapshot to increment with.
+  ConfirmedSnapshot Tx ->
+  Either IncrementTxError Tx
+increment ctx spendableUTxO headId headParameters incrementingSnapshot = do
+  pid <- headIdToPolicyId headId ?> InvalidHeadIdInIncrement{headId}
+  let utxoOfThisHead' = utxoOfThisHead pid spendableUTxO
+  headUTxO <- UTxO.find (isScriptTxOut headScript) utxoOfThisHead' ?> CannotFindHeadOutputInIncrement
+  case utxoToCommit of
+    Nothing ->
+      Left SnapshotMissingIncrementUTxO
+    Just depositUTxO
+      | null depositUTxO ->
+          Left SnapshotIncrementUTxOIsNull
+      | otherwise -> Right $ incrementTx scriptRegistry ownVerificationKey headId headParameters headUTxO sn sigs depositUTxO
+ where
+  headScript = fromPlutusScript @PlutusScriptV2 Head.validatorScript
+
+  Snapshot{utxoToCommit} = sn
+
+  (sn, sigs) =
+    case incrementingSnapshot of
+      ConfirmedSnapshot{snapshot, signatures} -> (snapshot, signatures)
+      _ -> (getSnapshot incrementingSnapshot, mempty)
+
+  ChainContext{ownVerificationKey, scriptRegistry} = ctx
+
 -- | Possible errors when trying to construct decrement tx
 data DecrementTxError
   = InvalidHeadIdInDecrement {headId :: HeadId}

hydra-node/src/Hydra/HeadLogic.hs

@@ -620,6 +620,7 @@ onOpenNetworkAckSn Environment{party} openState otherParty snapshotSignature sn
                 <> cause (ClientEffect $ ServerOutput.SnapshotConfirmed headId snapshot multisig)
                 -- Spec: if txω ≠ ⊥
                 --         postTx (decrement, v̂, ŝ, η, ηω)
+                & maybePostIncrementTx snapshot multisig
                 & maybePostDecrementTx snapshot multisig
                 -- Spec: if leader(s + 1) = i ∧ T̂ ≠ ∅
                 --         multicast (reqSn, v, ̅S.s + 1, T̂, txω)
@@ -675,6 +676,29 @@ onOpenNetworkAckSn Environment{party} openState otherParty snapshotSignature sn
           <> cause (NetworkEffect $ ReqSn version nextSn (txId <$> localTxs) decommitTx Nothing)
       else outcome
 
+  maybePostIncrementTx snapshot@Snapshot{utxoToCommit} signatures outcome =
+    case (commitUTxO, utxoToCommit) of
+      (Just commitUTxOFromState, Just commitUTxOFromSnapshot) ->
+        if commitUTxOFromState == commitUTxOFromSnapshot
+          then
+            outcome
+              <> causes
+                [ ClientEffect $
+                    ServerOutput.CommitApproved
+                      { headId
+                      , utxoToCommit = commitUTxOFromSnapshot
+                      }
+                , OnChainEffect
+                    { postChainTx =
+                        IncrementTx
+                          { headId
+                          , headParameters = parameters
+                          , incrementingSnapshot = ConfirmedSnapshot{snapshot, signatures}
+                          }
+                    }
+                ]
+          else outcome -- TODO: output some error here?
+      _ -> outcome
   maybePostDecrementTx snapshot@Snapshot{utxoToDecommit} signatures outcome =
     case (decommitTx, utxoToDecommit) of
       (Just tx, Just utxo) ->
@@ -705,7 +729,7 @@ onOpenNetworkAckSn Environment{party} openState otherParty snapshotSignature sn
     , headId
     } = openState
 
-  CoordinatedHeadState{seenSnapshot, localTxs, decommitTx, version} = coordinatedHeadState
+  CoordinatedHeadState{seenSnapshot, localTxs, decommitTx, commitUTxO, version} = coordinatedHeadState
 
 -- | Client request to decommit UTxO from the head.
 --

hydra-node/test/Hydra/BehaviorSpec.hs

@@ -847,6 +847,14 @@ toOnChainTx now = \case
     OnAbortTx{headId = testHeadId}
   CollectComTx{headId} ->
     OnCollectComTx{headId}
+  IncrementTx{headId, incrementingSnapshot} ->
+    OnIncrementTx
+      { headId
+      , newVersion = version + 1
+      , utxo = fromMaybe mempty utxoToCommit
+      }
+   where
+    Snapshot{version, utxoToCommit} = getSnapshot incrementingSnapshot
   DecrementTx{headId, decrementingSnapshot} ->
     OnDecrementTx
       { headId