No processing of transactions during a decommit

hydra-node/json-schemas/logs.yaml

@@ -1055,6 +1055,16 @@ definitions:
             $ref: "api.yaml#/components/schemas/SnapshotNumber"
           leader:
             $ref: "api.yaml#/components/schemas/Party"
+      - title: "ReqSnDecommitNotSettled"
+        description: >-
+          Received a ReqSn message with specified new decommit but the previous one was not settled.
+        additionalProperties: false
+        required:
+          - tag
+        properties:
+          tag:
+            type: string
+            enum: ["ReqSnDecommitNotSettled"]
       - title: "InvalidMultisignature"
         description: >-
           Multisignature computed for a snapshot from individual parties signature is invalid.
@@ -2202,21 +2212,33 @@ definitions:
               Description of the cause of the validation failure.
       - title: WaitOnSnapshotNumber
         description: >-
-          Current observed snapshot is not the right one, waiting for some other
+          Current observed snapshot number is not the right one, waiting for some other
           number.
         type: object
         additionalProperties: false
         required:
           - tag
-          - waitingFor
+          - waitingForNumber
         properties:
           tag:
             type: string
             enum: ["WaitOnSnapshotNumber"]
-          waitingFor:
+          waitingForNumber:
             "$ref": "api.yaml#/components/schemas/SnapshotNumber"
-            description: >-
-              The expected number.
+      - title: WaitOnSnapshotVersion
+        description: >-
+          Requested snapshot version is not up to date, waiting for the next version number.
+        type: object
+        additionalProperties: false
+        required:
+          - tag
+          - waitingForVersion
+        properties:
+          tag:
+            type: string
+            enum: ["WaitOnSnapshotVersion"]
+          waitingForVersion:
+            "$ref": "api.yaml#/components/schemas/SnapshotVersion"
       - title: WaitOnSeenSnapshot
         description: >-
           No current snapshot is available, waiting for some snapshot to start.

hydra-node/src/Hydra/HeadLogic.hs

@@ -405,50 +405,49 @@ onOpenNetworkReqSn ::
   Maybe tx ->
   Outcome tx
 onOpenNetworkReqSn env ledger st otherParty sv sn requestedTxIds mDecommitTx =
-  -- Spec: require v = v ∧ s = ŝ + 1 ∧ leader(s) = j
+  -- Spec: require s = ŝ + 1 ∧ leader(s) = j
   requireReqSn $
     -- Spec: wait ŝ = ̅S.s
     waitNoSnapshotInFlight $
-      -- Spec: require ̅S.𝑈 ◦ txω ≠ ⊥
-      --       ηω ← combine(outputs(txω))
-      --       𝑈_active ← ̅S.𝑈 ◦ txω \ outputs(txω)
-      requireApplicableDecommitTx $ \(activeUTxO, mUtxoToDecommit) ->
-        -- Resolve transactions by-id
-        waitResolvableTxs $ \requestedTxs -> do
-          -- Spec: require 𝑈_active ◦ Treq ≠ ⊥
-          --       𝑈 ← 𝑈_active ◦ Treq
-          requireApplyTxs activeUTxO requestedTxs $ \u -> do
-            -- Spec: ŝ ← ̅S.s + 1
-            -- NOTE: confSn == seenSn == sn here
-            let nextSnapshot =
-                  Snapshot
-                    { headId
-                    , version = version
-                    , number = sn
-                    , confirmed = requestedTxIds
-                    , utxo = u
-                    , utxoToDecommit = mUtxoToDecommit
+      -- Spec: wait v = v̂
+      waitOnSnapshotVersion $
+        requireApplicableDecommitTx $ \(activeUTxO, mUtxoToDecommit) ->
+          -- Resolve transactions by-id
+          waitResolvableTxs $ \requestedTxs -> do
+            -- Spec: require 𝑈_active ◦ Treq ≠ ⊥
+            --       𝑈 ← 𝑈_active ◦ Treq
+            requireApplyTxs activeUTxO requestedTxs $ \u -> do
+              -- Spec: ŝ ← ̅S.s + 1
+              -- NOTE: confSn == seenSn == sn here
+              let nextSnapshot =
+                    Snapshot
+                      { headId
+                      , version = version
+                      , number = sn
+                      , confirmed = requestedTxIds
+                      , utxo = u
+                      , utxoToDecommit = mUtxoToDecommit
+                      }
+              -- Spec: η ← combine(𝑈)
+              --       σᵢ ← MS-Sign(kₕˢⁱᵍ, (cid‖v‖ŝ‖η‖ηω))
+              let snapshotSignature = sign signingKey nextSnapshot
+              -- Spec: multicast (ackSn, ŝ, σᵢ)
+              (cause (NetworkEffect $ AckSn snapshotSignature sn) <>) $ do
+                -- Spec: ̂Σ ← ∅
+                --       L̂ ← 𝑈
+                --       𝑋 ← T
+                --       T̂ ← ∅
+                --       for tx ∈ 𝑋 : L̂ ◦ tx ≠ ⊥
+                --         T̂ ← T̂ ⋃ {tx}
+                --         L̂ ← L̂ ◦ tx
+                let (newLocalTxs, newLocalUTxO) = pruneTransactions u
+                newState
+                  SnapshotRequested
+                    { snapshot = nextSnapshot
+                    , requestedTxIds
+                    , newLocalUTxO
+                    , newLocalTxs
                     }
-            -- Spec: η ← combine(𝑈)
-            --       σᵢ ← MS-Sign(kₕˢⁱᵍ, (cid‖v‖ŝ‖η‖ηω))
-            let snapshotSignature = sign signingKey nextSnapshot
-            -- Spec: multicast (ackSn, ŝ, σᵢ)
-            (cause (NetworkEffect $ AckSn snapshotSignature sn) <>) $ do
-              -- Spec: ̂Σ ← ∅
-              --       L̂ ← 𝑈
-              --       𝑋 ← T
-              --       T̂ ← ∅
-              --       for tx ∈ 𝑋 : L̂ ◦ tx ≠ ⊥
-              --         T̂ ← T̂ ⋃ {tx}
-              --         L̂ ← L̂ ◦ tx
-              let (newLocalTxs, newLocalUTxO) = pruneTransactions u
-              newState
-                SnapshotRequested
-                  { snapshot = nextSnapshot
-                  , requestedTxIds
-                  , newLocalUTxO
-                  , newLocalTxs
-                  }
  where
   requireReqSn continue
     | sv /= version =
@@ -466,6 +465,12 @@ onOpenNetworkReqSn env ledger st otherParty sv sn requestedTxIds mDecommitTx =
     | otherwise =
         wait $ WaitOnSnapshotNumber seenSn
 
+  waitOnSnapshotVersion continue
+    | version == sv =
+        continue
+    | otherwise =
+        wait $ WaitOnSnapshotVersion sv
+
   waitResolvableTxs continue =
     case toList (fromList requestedTxIds \\ Map.keysSet allTxs) of
       [] -> continue $ mapMaybe (`Map.lookup` allTxs) requestedTxIds
@@ -475,15 +480,27 @@ onOpenNetworkReqSn env ledger st otherParty sv sn requestedTxIds mDecommitTx =
     case mDecommitTx of
       Nothing -> cont (confirmedUTxO, Nothing)
       Just decommitTx ->
-        -- Spec: require ̅S.𝑈 ◦ txω /= ⊥
-        case applyTransactions ledger currentSlot confirmedUTxO [decommitTx] of
-          Left (_, err) ->
-            Error $ RequireFailed $ SnapshotDoesNotApply sn (txId decommitTx) err
-          Right newConfirmedUTxO -> do
-            -- Spec: 𝑈_active ← ̅S.𝑈 ◦ txω \ outputs(txω)
-            let utxoToDecommit = utxoFromTx decommitTx
-            let activeUTxO = newConfirmedUTxO `withoutUTxO` utxoToDecommit
-            cont (activeUTxO, Just utxoToDecommit)
+        -- Spec:
+        -- if v = S̄.v ∧ S̄.txω ̸= ⊥
+        --   require S̄.txω = txω
+        --   Uactive ← S̄.U
+        --   Uω ← S̄.Uω
+        -- else
+        --   require S̄.U ◦ txω ̸= ⊥
+        --   Uactive ← S̄.U ◦ txω \ outputs(txω )
+        --   Uω ← outputs(txω )
+        if sv == confVersion && isJust confUTxOToDecommit
+          then
+            if confUTxOToDecommit == Just (utxoFromTx decommitTx)
+              then cont (confirmedUTxO, confUTxOToDecommit)
+              else Error $ RequireFailed ReqSnDecommitNotSettled
+          else case applyTransactions ledger currentSlot confirmedUTxO [decommitTx] of
+            Left (_, err) ->
+              Error $ RequireFailed $ SnapshotDoesNotApply sn (txId decommitTx) err
+            Right newConfirmedUTxO -> do
+              let utxoToDecommit = utxoFromTx decommitTx
+              let activeUTxO = newConfirmedUTxO `withoutUTxO` utxoToDecommit
+              cont (activeUTxO, Just utxoToDecommit)
 
   -- NOTE: at this point we know those transactions apply on the localUTxO because they
   -- are part of the localTxs. The snapshot can contain less transactions than the ones
@@ -512,6 +529,12 @@ onOpenNetworkReqSn env ledger st otherParty sv sn requestedTxIds mDecommitTx =
     InitialSnapshot{} -> 0
     ConfirmedSnapshot{snapshot = Snapshot{number}} -> number
 
+  Snapshot{version = confVersion} = getSnapshot confirmedSnapshot
+
+  confUTxOToDecommit = case confirmedSnapshot of
+    InitialSnapshot{} -> Nothing
+    ConfirmedSnapshot{snapshot = Snapshot{utxoToDecommit}} -> utxoToDecommit
+
   seenSn = seenSnapshotNumber seenSnapshot
 
   confirmedUTxO = case confirmedSnapshot of

hydra-node/src/Hydra/HeadLogic/Error.hs

@@ -35,6 +35,7 @@ data RequirementFailure tx
   = ReqSnNumberInvalid {requestedSn :: SnapshotNumber, lastSeenSn :: SnapshotNumber}
   | ReqSvNumberInvalid {requestedSv :: SnapshotVersion, lastSeenSv :: SnapshotVersion}
   | ReqSnNotLeader {requestedSn :: SnapshotNumber, leader :: Party}
+  | ReqSnDecommitNotSettled
   | InvalidMultisignature {multisig :: Text, vkeys :: [VerificationKey HydraKey]}
   | SnapshotAlreadySigned {knownSignatures :: [Party], receivedSignature :: Party}
   | AckSnNumberInvalid {requestedSn :: SnapshotNumber, lastSeenSn :: SnapshotNumber}

hydra-node/src/Hydra/HeadLogic/Outcome.hs

@@ -156,7 +156,8 @@ causes = Continue []
 
 data WaitReason tx
   = WaitOnNotApplicableTx {validationError :: ValidationError}
-  | WaitOnSnapshotNumber {waitingFor :: SnapshotNumber}
+  | WaitOnSnapshotNumber {waitingForNumber :: SnapshotNumber}
+  | WaitOnSnapshotVersion {waitingForVersion :: SnapshotVersion}
   | WaitOnSeenSnapshot
   | WaitOnTxs {waitingForTxIds :: [TxIdType tx]}
   | WaitOnContestationDeadline

hydra-node/test/Hydra/BehaviorSpec.hs

@@ -418,13 +418,13 @@ spec = parallel $ do
               withHydraNode bobSk [alice] chain $ \n2 -> do
                 openHead chain n1 n2
                 let decommitTx1 = SimpleTx 1 (utxoRef 1) (utxoRef 42)
-                send n2 (Decommit{decommitTx = decommitTx1})
+                send n1 (Decommit{decommitTx = decommitTx1})
                 waitUntil [n1, n2] $
                   DecommitRequested{headId = testHeadId, decommitTx = decommitTx1, utxoToDecommit = utxoRefs [42]}
 
                 let decommitTx2 = SimpleTx 2 (utxoRef 2) (utxoRef 22)
-                send n1 (Decommit{decommitTx = decommitTx2})
-                waitUntil [n1] $
+                send n2 (Decommit{decommitTx = decommitTx2})
+                waitUntil [n2] $
                   DecommitInvalid
                     { headId = testHeadId
                     , decommitTx = decommitTx2
@@ -433,10 +433,32 @@ spec = parallel $ do
 
                 waitUntil [n1, n2] $ DecommitFinalized{headId = testHeadId, decommitTxId = txId decommitTx1}
 
-                send n1 (Decommit{decommitTx = decommitTx2})
+                send n2 (Decommit{decommitTx = decommitTx2})
                 waitUntil [n1, n2] $ DecommitApproved{headId = testHeadId, decommitTxId = txId decommitTx2, utxoToDecommit = utxoRefs [22]}
                 waitUntil [n1, n2] $ DecommitFinalized{headId = testHeadId, decommitTxId = txId decommitTx2}
 
+      it "can process transactions while decommit pending" $
+        shouldRunInSim $ do
+          withSimulatedChainAndNetwork $ \chain ->
+            withHydraNode aliceSk [bob] chain $ \n1 ->
+              withHydraNode bobSk [alice] chain $ \n2 -> do
+                openHead chain n1 n2
+
+                let decommitTx = SimpleTx 1 (utxoRef 1) (utxoRef 42)
+                send n2 (Decommit{decommitTx})
+                waitUntil [n1, n2] $
+                  DecommitRequested{headId = testHeadId, decommitTx, utxoToDecommit = utxoRefs [42]}
+                waitUntil [n1, n2] $
+                  DecommitApproved{headId = testHeadId, decommitTxId = 1, utxoToDecommit = utxoRefs [42]}
+
+                let normalTx = SimpleTx 2 (utxoRef 2) (utxoRef 3)
+                send n2 (NewTx normalTx)
+                waitUntilMatch [n1, n2] $ \case
+                  SnapshotConfirmed{snapshot = Snapshot{confirmed}} -> 2 `elem` confirmed
+                  _ -> False
+
+                waitUntil [n1, n2] $ DecommitFinalized{headId = testHeadId, decommitTxId = 1}
+
     it "can close with decommit in flight" $
       shouldRunInSim $ do
         withSimulatedChainAndNetwork $ \chain ->
@@ -448,7 +470,12 @@ spec = parallel $ do
               send n1 Close
               waitUntil [n1, n2] $ ReadyToFanout{headId = testHeadId}
               send n1 Fanout
-              waitUntil [n1, n2] $ HeadIsFinalized{headId = testHeadId, utxo = utxoRefs [1, 2]}
+
+              waitMatch n2 $ \case
+                HeadIsContested{headId, snapshotNumber} -> guard $ headId == testHeadId && snapshotNumber == 1
+                _ -> Nothing
+
+              waitUntil [n1, n2] $ HeadIsFinalized{headId = testHeadId, utxo = utxoRefs [1]}
 
     it "fanout utxo is correct after a decommit" $
       shouldRunInSim $ do
@@ -535,12 +562,11 @@ spec = parallel $ do
               withHydraNode aliceSk [] chain $ \n1 -> do
                 send n1 Init
                 waitUntil [n1] $ HeadIsInitializing testHeadId (fromList [alice])
-                simulateCommit chain (alice, utxoRef 1)
 
           logs = selectTraceEventsDynamic @_ @(HydraNodeLog SimpleTx) result
 
-      logs `shouldContain` [BeginEffect alice 1 0 (ClientEffect $ HeadIsInitializing testHeadId $ fromList [alice])]
-      logs `shouldContain` [EndEffect alice 1 0]
+      logs `shouldContain` [BeginEffect alice 2 0 (ClientEffect $ HeadIsInitializing testHeadId $ fromList [alice])]
+      logs `shouldContain` [EndEffect alice 2 0]
 
   describe "rolling back & forward does not make the node crash" $ do
     it "does work for rollbacks past init" $
@@ -600,7 +626,7 @@ waitUntilMatch nodes predicate = do
       failure $
         toString $
           unlines
-            [ "waitUntilMatch did not match a message within " <> show oneMonth
+            [ "waitUntilMatch did not match a message within " <> show oneMonth <> ", seen messages:"
             , unlines (show <$> msgs)
             ]
  where
@@ -663,6 +689,7 @@ dummySimulatedChainNetwork =
 -- | With-pattern wrapper around 'simulatedChainAndNetwork' which does 'cancel'
 -- the 'tickThread'. Also, this will fix tx to 'SimpleTx' so that it can pick an
 -- initial chain state to play back to our test nodes.
+-- NOTE: The simulated network has a block time of 20 (simulated) seconds.
 withSimulatedChainAndNetwork ::
   (MonadTime m, MonadDelay m, MonadAsync m) =>
   (SimulatedChainNetwork SimpleTx m -> m ()) ->
@@ -710,7 +737,10 @@ simulatedChainAndNetwork initialChainState = do
                 Chain
                   { postTx = \tx -> do
                       now <- getCurrentTime
-                      createAndYieldEvent nodes history localChainState $ toOnChainTx now tx
+                      -- Only observe "after one block"
+                      void . async $ do
+                        threadDelay blockTime
+                        createAndYieldEvent nodes history localChainState $ toOnChainTx now tx
                   , draftCommitTx = \_ -> error "unexpected call to draftCommitTx"
                   , submitTx = \_ -> error "unexpected call to submitTx"
                   }
@@ -811,7 +841,7 @@ toOnChainTx now = \case
   DecrementTx{headId, decrementingSnapshot} ->
     OnDecrementTx
       { headId
-      , newVersion = version
+      , newVersion = version + 1
       , distributedOutputs = maybe mempty outputsOfUTxO utxoToDecommit
       }
    where
@@ -831,9 +861,8 @@ toOnChainTx now = \case
   FanoutTx{} ->
     OnFanoutTx{headId = testHeadId}
 
--- NOTE(SN): Deliberately long to emphasize that we run these tests in IOSim.
 testContestationPeriod :: ContestationPeriod
-testContestationPeriod = UnsafeContestationPeriod 3600
+testContestationPeriod = UnsafeContestationPeriod 10
 
 nothingHappensFor ::
   (MonadTimer m, MonadThrow m, IsChainState tx) =>