Observe distributed outputs from decrementTx

This allows us to align the HeadLogic part about clearing pending
decommits more faithfully.

Required to add a new method `outputsOfUTxO` to `IsTx` which feels to
become a bit "too big" for a type class.

hydra-node/src/Hydra/Chain.hs

@@ -24,7 +24,7 @@ import Hydra.ContestationPeriod (ContestationPeriod)
 import Hydra.Crypto (MultiSignature)
 import Hydra.Environment (Environment (..))
 import Hydra.HeadId (HeadId, HeadSeed)
-import Hydra.Ledger (ChainSlot, IsTx, UTxOType)
+import Hydra.Ledger (ChainSlot, IsTx (..), UTxOType)
 import Hydra.OnChainId (OnChainId)
 import Hydra.Party (Party)
 import Hydra.Snapshot (ConfirmedSnapshot, Snapshot, SnapshotNumber, SnapshotVersion)
@@ -123,7 +123,11 @@ data OnChainTx tx
       }
   | OnAbortTx {headId :: HeadId}
   | OnCollectComTx {headId :: HeadId}
-  | OnDecrementTx {headId :: HeadId, newVersion :: SnapshotVersion}
+  | OnDecrementTx
+      { headId :: HeadId
+      , newVersion :: SnapshotVersion
+      , distributedOutputs :: [TxOutType tx]
+      }
   | OnCloseTx
       { headId :: HeadId
       , snapshotNumber :: SnapshotNumber
@@ -142,7 +146,7 @@ deriving stock instance IsTx tx => Show (OnChainTx tx)
 deriving anyclass instance IsTx tx => ToJSON (OnChainTx tx)
 deriving anyclass instance IsTx tx => FromJSON (OnChainTx tx)
 
-instance (Arbitrary tx, Arbitrary (UTxOType tx)) => Arbitrary (OnChainTx tx) where
+instance (Arbitrary tx, Arbitrary (TxOutType tx), Arbitrary (UTxOType tx)) => Arbitrary (OnChainTx tx) where
   arbitrary = genericArbitrary
 
 -- | Exceptions thrown by 'postTx'.

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

@@ -319,8 +319,8 @@ convertObservation = \case
     pure OnCommitTx{headId, party, committed}
   CollectCom CollectComObservation{headId} ->
     pure OnCollectComTx{headId}
-  Decrement DecrementObservation{headId, newVersion} ->
-    pure OnDecrementTx{headId, newVersion}
+  Decrement DecrementObservation{headId, newVersion, distributedOutputs} ->
+    pure OnDecrementTx{headId, newVersion, distributedOutputs}
   Close CloseObservation{headId, snapshotNumber, threadOutput = ClosedThreadOutput{closedContestationDeadline}} ->
     pure
       OnCloseTx

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

@@ -1105,6 +1105,7 @@ observeCollectComTx utxo tx = do
 data DecrementObservation = DecrementObservation
   { headId :: HeadId
   , newVersion :: SnapshotVersion
+  , distributedOutputs :: [TxOut CtxUTxO]
   }
   deriving stock (Show, Eq, Generic)
 
@@ -1132,6 +1133,8 @@ observeDecrementTx utxo tx = do
             DecrementObservation
               { headId
               , newVersion = fromChainSnapshotVersion version
+              , -- NOTE: Head output must be in first position
+                distributedOutputs = drop 1 $ toUTxOContext <$> txOuts' tx
               }
         _ -> Nothing
     _ -> Nothing

hydra-node/src/Hydra/HeadLogic.hs

@@ -84,11 +84,12 @@ import Hydra.HeadLogic.State (
  )
 import Hydra.Ledger (
   ChainSlot,
-  IsTx,
+  IsTx (..),
   Ledger (..),
   TxIdType,
   UTxOType,
   applyTransactions,
+  outputsOfTx,
   txId,
   utxoFromTx,
   withoutUTxO,
@@ -832,10 +833,9 @@ onOpenNetworkReqDec env ledger ttl openState decommitTx =
 
 -- ** Decrementing funds from the Head
 
--- | Observe a decommit transaction. If the outputs of the pending decommit tx
--- are equal to the latest confirmed UTxO to decommit, then we consider the
--- decommit valid, and we take the funds out of the head and remove the decommit
--- tx in flight.
+-- | Observe a decrement transaction. If the outputs match the ones of the
+-- pending decommit tx, then we consider the decommit finalized, and remove the
+-- decommit tx in flight.
 --
 -- Finally, if the client observing happens to be the leader, then a new ReqSn
 -- is broadcasted.
@@ -847,19 +847,23 @@ onOpenChainDecrementTx ::
   OpenState tx ->
   -- | New open state version
   SnapshotVersion ->
+  -- | Outputs removed by the decrement
+  [TxOutType tx] ->
   Outcome tx
-onOpenChainDecrementTx Environment{party} openState newVersion
-  | -- Spec: if outputs(txω) = 𝑈ω
-    -- REVIEW: should we get Uω from observation instead of local state?
-    (utxoFromTx <$> decommitTx) == utxoToDecommit =
-      -- Spec: txω ← ⊥
-      --       vˆ  ← v
-      newState DecommitFinalized{newVersion}
-        <> cause (ClientEffect $ ServerOutput.DecommitFinalized{headId})
-        -- Spec: if ŝ = S⁻.s ∧ leader(S⁻.s + 1) = i
-        --         multicast (reqSn, vˆ, S⁻.s + 1, T̂ , txω )
-        & maybeEmitSnapshot
-  | otherwise = noop
+onOpenChainDecrementTx Environment{party} openState newVersion distributedTxOuts =
+  -- Spec: if outputs(txω) = 𝑈ω
+  case decommitTx of
+    Nothing -> noop -- TODO: what if decommit observed but none pending?
+    Just tx
+      | outputsOfTx tx == distributedTxOuts ->
+          -- Spec: txω ← ⊥
+          --       vˆ  ← v
+          newState DecommitFinalized{newVersion}
+            <> cause (ClientEffect $ ServerOutput.DecommitFinalized{headId})
+            -- Spec: if ŝ = S⁻.s ∧ leader(S⁻.s + 1) = i
+            --         multicast (reqSn, vˆ, S⁻.s + 1, T̂ , txω )
+            & maybeEmitSnapshot
+      | otherwise -> noop -- TODO: what if decrement not matching pending decommit?
  where
   partyIsLeader = isLeader parameters party nextSn && not (null localTxs)
 
@@ -877,7 +881,7 @@ onOpenChainDecrementTx Environment{party} openState newVersion
 
   seenSn = seenSnapshotNumber seenSnapshot
 
-  Snapshot{number = confirmedSn, utxoToDecommit} = getSnapshot confirmedSnapshot
+  Snapshot{number = confirmedSn} = getSnapshot confirmedSnapshot
 
   nextSn = confirmedSn + 1
 
@@ -1111,10 +1115,10 @@ update env ledger st ev = case (st, ev) of
     onOpenClientDecommit env headId ledger currentSlot coordinatedHeadState decommitTx
   (Open openState, NetworkInput ttl (ReceivedMessage{msg = ReqDec{transaction}})) ->
     onOpenNetworkReqDec env ledger ttl openState transaction
-  (Open openState@OpenState{headId = ourHeadId}, ChainInput Observation{observedTx = OnDecrementTx{headId, newVersion}})
+  (Open openState@OpenState{headId = ourHeadId}, ChainInput Observation{observedTx = OnDecrementTx{headId, newVersion, distributedOutputs}})
     -- TODO: What happens if observed decrement tx get's rolled back?
     | ourHeadId == headId ->
-        onOpenChainDecrementTx env openState newVersion
+        onOpenChainDecrementTx env openState newVersion distributedOutputs
     | otherwise ->
         Error NotOurHead{ourHeadId, otherHeadId = headId}
   -- Closed

hydra-node/src/Hydra/Ledger.hs

@@ -30,20 +30,27 @@ class
   , ToCBOR (TxIdType tx)
   , FromJSONKey (TxIdType tx)
   , ToJSONKey (TxIdType tx)
+  , --
+    Eq (TxOutType tx)
+  , Show (TxOutType tx)
+  , Arbitrary (TxOutType tx)
+  , ToJSON (TxOutType tx)
+  , FromJSON (TxOutType tx)
   , --
     Eq (UTxOType tx)
   , Show (UTxOType tx)
   , Arbitrary (UTxOType tx)
-  , FromJSON (UTxOType tx)
   , Monoid (UTxOType tx)
+  , FromJSON (UTxOType tx)
   , ToJSON (UTxOType tx)
   , FromCBOR (UTxOType tx)
   , ToCBOR (UTxOType tx)
   ) =>
   IsTx tx
   where
-  type UTxOType tx = utxo | utxo -> tx
   type TxIdType tx
+  type TxOutType tx = out | out -> tx
+  type UTxOType tx = utxo | utxo -> tx
   type ValueType tx
 
   -- XXX(SN): this name easily conflicts
@@ -58,9 +65,16 @@ class
   -- | Get the UTxO produced by given transaction.
   utxoFromTx :: tx -> UTxOType tx
 
+  -- | Get only the outputs in given UTxO.
+  outputsOfUTxO :: UTxOType tx -> [TxOutType tx]
+
   -- | Return the left-hand side without the right-hand side.
   withoutUTxO :: UTxOType tx -> UTxOType tx -> UTxOType tx
 
+-- | Get outputs of a transaction.
+outputsOfTx :: IsTx tx => tx -> [TxOutType tx]
+outputsOfTx = outputsOfUTxO . utxoFromTx
+
 -- | A generic description for a chain slot all implementions need to use.
 newtype ChainSlot = ChainSlot Natural
   deriving stock (Ord, Eq, Show, Generic)

hydra-node/src/Hydra/Ledger/Cardano.hs

@@ -109,6 +109,7 @@ fromChainSlot (ChainSlot s) = fromIntegral s
 
 instance IsTx Tx where
   type TxIdType Tx = TxId
+  type TxOutType Tx = TxOut CtxUTxO
   type UTxOType Tx = UTxO
   type ValueType Tx = Value
 
@@ -122,6 +123,8 @@ instance IsTx Tx where
 
   utxoFromTx = Api.utxoFromTx
 
+  outputsOfUTxO = toList
+
   withoutUTxO = UTxO.difference
 
 instance ToCBOR Tx where

hydra-node/src/Hydra/Ledger/Simple.hs

@@ -45,14 +45,16 @@ data SimpleTx = SimpleTx
 type SimpleId = Integer
 
 instance IsTx SimpleTx where
-  type UTxOType SimpleTx = Set SimpleTxIn
   type TxIdType SimpleTx = SimpleId
+  type TxOutType SimpleTx = SimpleTxOut
+  type UTxOType SimpleTx = Set SimpleTxOut
   type ValueType SimpleTx = Int
 
   txId (SimpleTx tid _ _) = tid
   balance = Set.size
-  hashUTxO = toStrict . foldMap (serialise . unSimpleTxIn)
+  hashUTxO = toStrict . foldMap (serialise . unSimpleTxOut)
   utxoFromTx = txOutputs
+  outputsOfUTxO = toList
   withoutUTxO = Set.difference
 
   txSpendingUTxO utxo =
@@ -110,20 +112,20 @@ instance IsChainState SimpleTx where
 -- MockTxIn
 --
 
--- | An identifier for a single output of a 'SimpleTx'.
-newtype SimpleTxIn = SimpleTxIn {unSimpleTxIn :: Integer}
+-- | A single output of a 'SimpleTx' having an integer identity and sole value.
+newtype SimpleTxOut = SimpleTxOut {unSimpleTxOut :: Integer}
   deriving stock (Generic)
   deriving newtype (Eq, Ord, Show, Num, ToJSON, FromJSON)
 
-instance Arbitrary SimpleTxIn where
+instance Arbitrary SimpleTxOut where
   shrink = genericShrink
   arbitrary = genericArbitrary
 
-instance ToCBOR SimpleTxIn where
-  toCBOR (SimpleTxIn inId) = toCBOR inId
+instance ToCBOR SimpleTxOut where
+  toCBOR (SimpleTxOut inId) = toCBOR inId
 
-instance FromCBOR SimpleTxIn where
-  fromCBOR = SimpleTxIn <$> fromCBOR
+instance FromCBOR SimpleTxOut where
+  fromCBOR = SimpleTxOut <$> fromCBOR
 
 simpleLedger :: Ledger SimpleTx
 simpleLedger =
@@ -141,10 +143,10 @@ simpleLedger =
 --
 
 utxoRef :: Integer -> UTxOType SimpleTx
-utxoRef = Set.singleton . SimpleTxIn
+utxoRef = Set.singleton . SimpleTxOut
 
 utxoRefs :: [Integer] -> UTxOType SimpleTx
-utxoRefs = Set.fromList . fmap SimpleTxIn
+utxoRefs = Set.fromList . fmap SimpleTxOut
 
 aValidTx :: Integer -> SimpleTx
 aValidTx n = SimpleTx n mempty (utxoRef n)
@@ -155,12 +157,12 @@ aValidTx n = SimpleTx n mempty (utxoRef n)
 
 listOfCommittedUTxOs :: Integer -> Gen [UTxOType SimpleTx]
 listOfCommittedUTxOs numCommits =
-  pure $ Set.singleton . SimpleTxIn <$> [1 .. numCommits]
+  pure $ Set.singleton . SimpleTxOut <$> [1 .. numCommits]
 
 genSequenceOfValidTransactions :: UTxOType SimpleTx -> Gen [SimpleTx]
 genSequenceOfValidTransactions initialUTxO = do
   n <- fromIntegral <$> getSize
-  let maxId = if Set.null initialUTxO then 0 else unSimpleTxIn (maximum initialUTxO)
+  let maxId = if Set.null initialUTxO then 0 else unSimpleTxOut (maximum initialUTxO)
   numTxs <- choose (1, n)
   foldlM newTx (maxId, initialUTxO, mempty) [1 .. numTxs] >>= \(_, _, txs) -> pure (reverse txs)
  where
@@ -172,9 +174,9 @@ genSequenceOfValidTransactions initialUTxO = do
     (newMax, ins, outs) <- genInputsAndOutputs maxId utxo
     pure (newMax, (utxo Set.\\ ins) `Set.union` outs, SimpleTx txid ins outs : txs)
 
-  genInputsAndOutputs :: Integer -> Set SimpleTxIn -> Gen (Integer, Set SimpleTxIn, Set SimpleTxIn)
+  genInputsAndOutputs :: Integer -> Set SimpleTxOut -> Gen (Integer, Set SimpleTxOut, Set SimpleTxOut)
   genInputsAndOutputs maxId utxo = do
     ins <- sublistOf (Set.toList utxo)
     numOuts <- choose (1, 10)
     let outs = fmap (+ maxId) [1 .. numOuts]
-    pure (maximum outs, Set.fromList ins, Set.fromList $ fmap SimpleTxIn outs)
+    pure (maximum outs, Set.fromList ins, Set.fromList $ fmap SimpleTxOut outs)

hydra-node/test/Hydra/BehaviorSpec.hs

@@ -792,7 +792,7 @@ createMockNetwork node nodes =
 -- | Derive an 'OnChainTx' from 'PostChainTx' to simulate a "perfect" chain.
 -- NOTE: This implementation announces hard-coded contestationDeadlines. Also,
 -- all heads will have the same 'headId' and 'headSeed'.
-toOnChainTx :: Monoid (UTxOType tx) => UTCTime -> PostChainTx tx -> OnChainTx tx
+toOnChainTx :: IsTx tx => UTCTime -> PostChainTx tx -> OnChainTx tx
 toOnChainTx now = \case
   InitTx{participants, headParameters} ->
     OnInitTx{headId = testHeadId, headSeed = testHeadSeed, headParameters, participants}
@@ -801,7 +801,11 @@ toOnChainTx now = \case
   CollectComTx{headId} ->
     OnCollectComTx{headId}
   DecrementTx{headId, snapshot} ->
-    OnDecrementTx{headId, newVersion = getField @"version" snapshot}
+    OnDecrementTx
+      { headId
+      , newVersion = getField @"version" snapshot
+      , distributedOutputs = maybe mempty outputsOfUTxO $ getField @"utxoToDecommit" snapshot
+      }
   CloseTx{confirmedSnapshot} ->
     OnCloseTx
       { headId = testHeadId

hydra-node/test/Hydra/HeadLogicSpec.hs

@@ -621,7 +621,7 @@ spec =
           `shouldBe` Error (NotOurHead{ourHeadId = testHeadId, otherHeadId})
 
       prop "ignores decrementTx of another head" $ \otherHeadId -> do
-        let decrementOtherHead = observeTx $ OnDecrementTx{headId = otherHeadId, newVersion = 1}
+        let decrementOtherHead = observeTx $ OnDecrementTx{headId = otherHeadId, newVersion = 1, distributedOutputs = mempty}
         update bobEnv ledger (inOpenState threeParties) decrementOtherHead
           `shouldBe` Error (NotOurHead{ourHeadId = testHeadId, otherHeadId})
 

hydra-node/test/Hydra/Model/Payment.hs

@@ -86,6 +86,7 @@ instance HasVariables Payment where
 -- | Making `Payment` an instance of `IsTx` allows us to use it with `HeadLogic'`s messages.
 instance IsTx Payment where
   type TxIdType Payment = Int
+  type TxOutType Payment = (CardanoSigningKey, Value)
   type UTxOType Payment = [(CardanoSigningKey, Value)]
   type ValueType Payment = Value
   txId = error "undefined"
@@ -96,6 +97,7 @@ instance IsTx Payment where
     [(from, value)] -> Payment{from, to = from, value}
     _ -> error "cant spend from multiple utxo in one payment"
   utxoFromTx Payment{to, value} = [(to, value)]
+  outputsOfUTxO = id
   withoutUTxO a b =
     let as = second valueToList <$> a
         bs = second valueToList <$> b