Hotfix Checkpoint handling of API server projections (#2152)

The projections in the API server were not correctly handing
`Checkpoint` events (introduced with event log rotation).

I tried to create a regression test, but we don't have a `Gen
[StateChanged tx]` generators that honors the constraints required for
`aggregate` so I could not get hold of an expected list of
`StateChanged` events with a consistent `Checkpoint` to do property
testing.

The whole `Projection` business is a bit contrived since we have the
full `StateChanged` event stream available (the API server is now an
`EventSink`). As we are also not needing to track changes onto the
projected values (resource-specific websockets / subscriptions would
require that), we should consider dropping the whole mechanism and just
stick with a `getHeadState :: STM (HeadState tx)` + getter functions (or
lenses) to acquire the relevant things in the HTTP and WS API.

This PR though is a hotfix and I kept the diff to a minimum (especially
as its not covered by tests!)

---

* [x] CHANGELOG updated
* [x] Documentation update not needed
* [x] Haddocks updated
* [ ] No new TODOs introduced
  - An XXX note on the same as above

Author: noonio

CHANGELOG.md

@@ -10,8 +10,32 @@ changes.
 
 ## [0.23.0] - UNRELEASED
 
+## [0.22.4] - UNRELEASED
+
 - Accept additional field `amount` when depositing to specify the amount of Lovelace that should be depositted to a Head returning any leftover to the user.
 
+- Fix API not correctly handling event log rotation. This was evident in not
+  being able to use `/commit` although the head is initializing or outdated
+  information in the `Greetings` message.
+
+- Ignore snapshot signatures of already confirmed snapshots. This was previously
+  resulting in the node waiting for the accompanying snapshot request and
+  occurred when running heads with mirror nodes.
+
+- Fix an internal persistent queue blocking after restart when it reached
+  capacity.
+
+- Handle failing lease keep alive in network component and avoid bursts in
+  heartbeating.
+
+## [0.22.3] - 2025-07-21
+
+* Change behavior of `Hydra.Network.Etcd` to fallback to earliest possible
+  revision if `last-known-revision` is missing or too old. This can happen if a
+  node is down for a long time and the `etcd` cluster compacted the last known
+  revision in the meantime
+  [#2136](https://github.com/cardano-scaling/hydra/issues/2136).
+
 - Don't keep around invalid transactions as they could lead to stuck Head.
 
 - Hydra API server responds with the correct `Content-Type` header `application-json`.

hydra-node/src/Hydra/API/Server.hs

@@ -12,6 +12,7 @@ import Control.Concurrent.STM.TChan (newBroadcastTChanIO, writeTChan)
 import Control.Exception (IOException)
 import Data.Conduit.Combinators (map)
 import Data.Conduit.List (catMaybes)
+import Data.Map qualified as Map
 import Hydra.API.APIServerLog (APIServerLog (..))
 import Hydra.API.ClientInput (ClientInput)
 import Hydra.API.HTTPServer (httpApp)
@@ -31,19 +32,22 @@ import Hydra.Chain (Chain (..))
 import Hydra.Chain.ChainState (IsChainState)
 import Hydra.Chain.Direct.State ()
 import Hydra.Events (EventSink (..), EventSource (..))
-import Hydra.HeadLogic (aggregate)
-import Hydra.HeadLogic.Outcome qualified as StateChanged
-import Hydra.HeadLogic.State (
+import Hydra.HeadLogic (
+  CoordinatedHeadState (..),
   Deposit (..),
-  HeadState (Idle),
+  HeadState (..),
   IdleState (..),
+  InitialState (..),
+  OpenState (..),
+  aggregate,
  )
+import Hydra.HeadLogic.Outcome qualified as StateChanged
 import Hydra.HeadLogic.StateEvent (StateEvent (..))
 import Hydra.Logging (Tracer, traceWith)
 import Hydra.Network (IP, PortNumber)
 import Hydra.Node.ApiTransactionTimeout (ApiTransactionTimeout)
 import Hydra.Node.Environment (Environment)
-import Hydra.Tx (HeadId, IsTx (..), Party, txId)
+import Hydra.Tx (IsTx (..), Party, txId)
 import Network.HTTP.Types (status500)
 import Network.Wai (responseLBS)
 import Network.Wai.Handler.Warp (
@@ -96,8 +100,10 @@ withAPIServer config env party eventSource tracer chain pparams serverOutputFilt
     -- Initialize our read models from stored events
     -- NOTE: we do not keep the stored events around in memory
     headStateP <- mkProjection (Idle $ IdleState mkChainState) aggregate
+    -- XXX: We never subscribe to changes of commitInfoP et al directly so a
+    -- single read model and normal functions mapping from HeadState ->
+    -- CommitInfo etc. would suffice and are less fragile
     commitInfoP <- mkProjection CannotCommit projectCommitInfo
-    headIdP <- mkProjection Nothing projectInitializingHeadId
     pendingDepositsP <- mkProjection [] projectPendingDeposits
     let historyTimedOutputs = sourceEvents .| map mkTimedServerOutputFromStateEvent .| catMaybes
     _ <-
@@ -108,7 +114,6 @@ withAPIServer config env party eventSource tracer chain pparams serverOutputFilt
                 lift $ atomically $ do
                   update headStateP stateChanged
                   update commitInfoP stateChanged
-                  update headIdP stateChanged
                   update pendingDepositsP stateChanged
             )
     (notifyServerRunning, waitForServerRunning) <- setupServerNotification
@@ -127,7 +132,7 @@ withAPIServer config env party eventSource tracer chain pparams serverOutputFilt
             . simpleCors
             $ websocketsOr
               defaultConnectionOptions
-              (wsApp env party tracer historyTimedOutputs callback headStateP headIdP responseChannel serverOutputFilter)
+              (wsApp env party tracer historyTimedOutputs callback headStateP responseChannel serverOutputFilter)
               ( httpApp
                   tracer
                   chain
@@ -150,7 +155,6 @@ withAPIServer config env party eventSource tracer chain pparams serverOutputFilt
                     atomically $ do
                       update headStateP stateChanged
                       update commitInfoP stateChanged
-                      update headIdP stateChanged
                       update pendingDepositsP stateChanged
                     -- Send to the client if it maps to a server output
                     case mkTimedServerOutputFromStateEvent event of
@@ -260,6 +264,9 @@ mkTimedServerOutputFromStateEvent event =
 -- | Projection to obtain the list of pending deposits.
 projectPendingDeposits :: IsTx tx => [TxIdType tx] -> StateChanged.StateChanged tx -> [TxIdType tx]
 projectPendingDeposits txIds = \case
+  StateChanged.Checkpoint{state} -> case state of
+    Open OpenState{coordinatedHeadState = CoordinatedHeadState{pendingDeposits}} -> Map.keys pendingDeposits
+    _ -> txIds
   StateChanged.DepositRecorded{depositTxId} -> depositTxId : txIds
   StateChanged.DepositRecovered{depositTxId} -> filter (/= depositTxId) txIds
   StateChanged.CommitFinalized{depositTxId} -> filter (/= depositTxId) txIds
@@ -270,18 +277,12 @@ projectPendingDeposits txIds = \case
 -- state since this is when Head parties need to commit some funds.
 projectCommitInfo :: CommitInfo -> StateChanged.StateChanged tx -> CommitInfo
 projectCommitInfo commitInfo = \case
+  StateChanged.Checkpoint{state} -> case state of
+    Initial InitialState{headId} -> NormalCommit headId
+    Open OpenState{headId} -> IncrementalCommit headId
+    _ -> CannotCommit
   StateChanged.HeadInitialized{headId} -> NormalCommit headId
   StateChanged.HeadOpened{headId} -> IncrementalCommit headId
   StateChanged.HeadAborted{} -> CannotCommit
   StateChanged.HeadClosed{} -> CannotCommit
   _other -> commitInfo
-
--- | Projection to obtain the 'HeadId' needed to draft a commit transaction.
--- NOTE: We only want to project 'HeadId' when the Head is in the 'Initializing'
--- state since this is when Head parties need to commit some funds.
-projectInitializingHeadId :: Maybe HeadId -> StateChanged.StateChanged tx -> Maybe HeadId
-projectInitializingHeadId mHeadId = \case
-  StateChanged.HeadInitialized{headId} -> Just headId
-  StateChanged.HeadOpened{} -> Nothing
-  StateChanged.HeadAborted{} -> Nothing
-  _other -> mHeadId

hydra-node/src/Hydra/API/WSServer.hs

@@ -40,13 +40,12 @@ import Hydra.Chain.ChainState (
   IsChainState,
  )
 import Hydra.Chain.Direct.State ()
-import Hydra.HeadLogic (ClosedState (ClosedState, readyToFanoutSent), HeadState, StateChanged)
+import Hydra.HeadLogic (ClosedState (ClosedState, readyToFanoutSent), HeadState, InitialState (..), OpenState (..), StateChanged)
 import Hydra.HeadLogic.State qualified as HeadState
 import Hydra.Logging (Tracer, traceWith)
 import Hydra.NetworkVersions qualified as NetworkVersions
 import Hydra.Node.Environment (Environment (..))
 import Hydra.Tx (Party)
-import Hydra.Tx.HeadId (HeadId (..))
 import Network.WebSockets (
   PendingConnection (pendingRequest),
   RequestHead (..),
@@ -68,13 +67,11 @@ wsApp ::
   (ClientInput tx -> IO ()) ->
   -- | Read model to enhance 'Greetings' messages with 'HeadStatus'.
   Projection STM.STM (StateChanged tx) (HeadState tx) ->
-  -- | Read model to enhance 'Greetings' messages with 'HeadId'.
-  Projection STM.STM (StateChanged tx) (Maybe HeadId) ->
   TChan (Either (TimedServerOutput tx) (ClientMessage tx)) ->
   ServerOutputFilter tx ->
   PendingConnection ->
   IO ()
-wsApp env party tracer history callback headStateP headIdP responseChannel ServerOutputFilter{txContainsAddr} pending = do
+wsApp env party tracer history callback headStateP responseChannel ServerOutputFilter{txContainsAddr} pending = do
   traceWith tracer NewAPIConnection
   let path = requestPath $ pendingRequest pending
   queryParams <- uriQuery <$> mkURIBs path
@@ -97,21 +94,19 @@ wsApp env party tracer history callback headStateP headIdP responseChannel Serve
   -- client.
   forwardGreetingOnly config con = do
     headState <- atomically getLatest
-    hydraHeadId <- atomically getLatestHeadId
     sendTextData con $
       handleUtxoInclusion config (atKey "snapshotUtxo" .~ Nothing) $
         Aeson.encode
           Greetings
             { me = party
             , headStatus = getHeadStatus headState
-            , hydraHeadId
+            , hydraHeadId = getHeadId headState
             , snapshotUtxo = getSnapshotUtxo headState
             , hydraNodeVersion = showVersion NetworkVersions.hydraNodeVersion
             , env
             }
 
   Projection{getLatest} = headStateP
-  Projection{getLatest = getLatestHeadId} = headIdP
 
   mkServerOutputConfig :: [QueryParam] -> ServerOutputConfig
   mkServerOutputConfig qp =
@@ -185,12 +180,16 @@ wsApp env party tracer history callback headStateP headIdP responseChannel Serve
         WithAddressedTx addr -> txContainsAddr tx addr
         WithoutAddressedTx -> True
 
--- | Get the content of 'headStatus' field in 'Greetings' message from the full 'HeadState'.
-getHeadStatus :: HeadState tx -> HeadStatus
-getHeadStatus = \case
-  HeadState.Idle{} -> Idle
-  HeadState.Initial{} -> Initializing
-  HeadState.Open{} -> Open
-  HeadState.Closed ClosedState{readyToFanoutSent}
-    | readyToFanoutSent -> FanoutPossible
-    | otherwise -> Closed
+  getHeadStatus = \case
+    HeadState.Idle{} -> Idle
+    HeadState.Initial{} -> Initializing
+    HeadState.Open{} -> Open
+    HeadState.Closed ClosedState{readyToFanoutSent}
+      | readyToFanoutSent -> FanoutPossible
+      | otherwise -> Closed
+
+  getHeadId = \case
+    HeadState.Idle{} -> Nothing
+    HeadState.Initial InitialState{headId} -> Just headId
+    HeadState.Open OpenState{headId} -> Just headId
+    HeadState.Closed ClosedState{headId} -> Just headId