Extend node kernel peer state with object diffusion state

Author: agustinmista

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Node/GSM.hs

@@ -56,8 +56,6 @@ import qualified Ouroboros.Consensus.HardFork.History.Qry as Qry
 import qualified Ouroboros.Consensus.Ledger.Basics as L
 import Ouroboros.Consensus.Node.GsmState
 import Ouroboros.Consensus.Storage.ChainDB.API (ChainDB)
-import Ouroboros.Consensus.Util.NormalForm.StrictTVar (StrictTVar)
-import qualified Ouroboros.Consensus.Util.NormalForm.StrictTVar as StrictSTM
 import System.FS.API
   ( HasFS
   , createDirectoryIfMissing
@@ -123,7 +121,7 @@ data GsmView m upstreamPeer selection peerState = GsmView
   , equivalent :: selection -> selection -> Bool
   -- ^ Whether the two selections are equivalent for the purpose of the
   -- Genesis State Machine
-  , getPeerStates :: STM m (Map.Map upstreamPeer (StrictTVar m peerState))
+  , getPeerStates :: STM m (Map.Map upstreamPeer peerState)
   -- ^ The current peer state with the latest candidates from the upstream peers
   , getCurrentSelection :: STM m selection
   -- ^ The node's current selection
@@ -371,11 +369,10 @@ realGsmEntryPoints tracerArgs gsmView =
     -- STAGE 1: all peers are idle, which means that
     --   * all ChainSync clients report no subsequent headers, and
     --   * all PerasCertDiffusion clients report no subsequent certificates
-    varsState <- getPeerStates
-    states <- traverse StrictSTM.readTVar varsState
+    peerStates <- getPeerStates
     check $
-      not (Map.null states)
-        && all peerIsIdle states
+      not (Map.null peerStates)
+        && all peerIsIdle peerStates
 
     -- STAGE 2: no candidate is better than the node's current
     -- selection
@@ -388,16 +385,15 @@ realGsmEntryPoints tracerArgs gsmView =
     -- block; general Praos reasoning ensures that won't take particularly
     -- long.
     selection <- getCurrentSelection
-    candidates <- traverse StrictSTM.readTVar varsState
     candidateOverSelection <- getCandidateOverSelection
     let ok candidate =
           WhetherCandidateIsBetter False
             == candidateOverSelection selection candidate
-    check $ all ok candidates
+    check $ all ok peerStates
 
     pure $
       GsmEventEnterCaughtUp
-        (Map.size states)
+        (Map.size peerStates)
         (cnvSelection selection)
 
   -- STAGE 3: the previous stages weren't so slow that the idler

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs

@@ -17,6 +17,7 @@ module Ouroboros.Consensus.NodeKernel
     MempoolCapacityBytesOverride (..)
   , NodeKernel (..)
   , NodeKernelArgs (..)
+  , NodeKernelPeerState (..)
   , TraceForgeEvent (..)
   , getImmTipSlot
   , getMempoolReader
@@ -49,6 +50,7 @@ import Data.Functor ((<&>))
 import Data.Hashable (Hashable)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as NE
+import qualified Data.Map.Strict as Map
 import Data.Maybe (isJust, mapMaybe)
 import Data.Proxy
 import qualified Data.Text as Text
@@ -80,6 +82,16 @@ import Ouroboros.Consensus.MiniProtocol.ChainSync.Client.HistoricityCheck
 import Ouroboros.Consensus.MiniProtocol.ChainSync.Client.InFutureCheck
   ( SomeHeaderInFutureCheck
   )
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.State
+  ( ObjectDiffusionInboundHandleCollection (..)
+  , ObjectDiffusionInboundState (..)
+  , newObjectDiffusionInboundHandleCollection
+  , odihState
+  )
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.PerasCert
+  ( PerasCertDiffusionInboundHandleCollection
+  , PerasCertDiffusionInboundState
+  )
 import Ouroboros.Consensus.Node.GSM (GsmNodeKernelArgs (..))
 import qualified Ouroboros.Consensus.Node.GSM as GSM
 import Ouroboros.Consensus.Node.Genesis
@@ -173,6 +185,9 @@ data NodeKernel m addrNTN addrNTC blk = NodeKernel
   -- from it with 'GSM.gsmStateToLedgerJudgement'.
   , getChainSyncHandles :: ChainSyncClientHandleCollection (ConnectionId addrNTN) m blk
   -- ^ The kill handle and exposed state for each ChainSync client.
+  , getObjectDiffusionInboundHandles ::
+      ObjectDiffusionInboundHandleCollection (ConnectionId addrNTN) m blk
+  -- ^ The exposed state for each ObjectDiffusion inbound client.
   , getPeerSharingRegistry :: PeerSharingRegistry addrNTN m
   -- ^ Read the current peer sharing registry, used for interacting with
   -- the PeerSharing protocol
@@ -217,6 +232,12 @@ data NodeKernelArgs m addrNTN addrNTC blk = NodeKernelArgs
   , getDiffusionPipeliningSupport :: DiffusionPipeliningSupport
   }
 
+-- | State about peers we are connected to
+data NodeKernelPeerState blk = NodeKernelPeerState
+  { chainSyncState :: ChainSyncState blk
+  , perasCertDiffusionInboundState :: PerasCertDiffusionInboundState blk
+  }
+
 initNodeKernel ::
   forall m addrNTN addrNTC blk.
   ( IOLike m
@@ -254,6 +275,7 @@ initNodeKernel
           , mempool
           , peerSharingRegistry
           , varChainSyncHandles
+          , varPerasCertDiffusionHandles
           , varGsmState
           } = st
 
@@ -273,23 +295,38 @@ initNodeKernel
                 { GSM.antiThunderingHerd = Just gsmAntiThunderingHerd
                 , GSM.getCandidateOverSelection = do
                     weights <- ChainDB.getPerasWeightSnapshot chainDB
-                    pure $ \(headers, _lst) state ->
-                      case AF.intersectionPoint headers (csCandidate state) of
+                    pure $ \(headers, _lst) peerState -> do
+                      let candidate = csCandidate (chainSyncState peerState)
+                      case AF.intersectionPoint headers candidate of
                         Nothing -> GSM.CandidateDoesNotIntersect
                         Just{} ->
                           GSM.WhetherCandidateIsBetter $ -- precondition requires intersection
                             preferAnchoredCandidate
                               (configBlock cfg)
                               (forgetFingerprint weights)
                               headers
-                              (csCandidate state)
-                , GSM.peerIsIdle = csIdling
+                              candidate
+                , GSM.peerIsIdle = \peerState -> do
+                    csIdling (chainSyncState peerState)
+                      && odisIdling (perasCertDiffusionInboundState peerState)
                 , GSM.durationUntilTooOld =
                     gsmDurationUntilTooOld
                       <&> \wd (_headers, lst) ->
                         GSM.getDurationUntilTooOld wd (getTipSlot lst)
                 , GSM.equivalent = (==) `on` (AF.headPoint . fst)
-                , GSM.getPeerStates = fmap cschState <$> cschcMap varChainSyncHandles
+                , GSM.getPeerStates = do
+                    chainSyncPeers <- cschcMap varChainSyncHandles
+                    perasCertDiffusionPeers <- odihcMap varPerasCertDiffusionHandles
+                    let commonPeers = Map.intersectionWith (,) chainSyncPeers perasCertDiffusionPeers
+                    -- TODO understand whether map intersection provides the right semantics here
+                    forM commonPeers $ \(cscHandle, odiHandle) -> do
+                      chainSyncState <- readTVar (cschState cscHandle)
+                      perasCertDiffusionState <- readTVar (odihState odiHandle)
+                      pure
+                        NodeKernelPeerState
+                          { chainSyncState = chainSyncState
+                          , perasCertDiffusionInboundState = perasCertDiffusionState
+                          }
                 , GSM.getCurrentSelection = do
                     headers <- ChainDB.getCurrentChainWithTime chainDB
                     extLedgerState <- ChainDB.getCurrentLedger chainDB
@@ -366,6 +403,7 @@ initNodeKernel
         , getFetchMode = readFetchMode blockFetchInterface
         , getGsmState = readTVar varGsmState
         , getChainSyncHandles = varChainSyncHandles
+        , getObjectDiffusionInboundHandles = varPerasCertDiffusionHandles
         , getPeerSharingRegistry = peerSharingRegistry
         , getTracers = tracers
         , setBlockForging = \a -> atomically . LazySTM.putTMVar blockForgingVar $! a
@@ -416,6 +454,8 @@ data InternalState m addrNTN addrNTC blk = IS
       BlockFetchConsensusInterface (ConnectionId addrNTN) (HeaderWithTime blk) blk m
   , fetchClientRegistry :: FetchClientRegistry (ConnectionId addrNTN) (HeaderWithTime blk) blk m
   , varChainSyncHandles :: ChainSyncClientHandleCollection (ConnectionId addrNTN) m blk
+  , varPerasCertDiffusionHandles ::
+      PerasCertDiffusionInboundHandleCollection (ConnectionId addrNTN) m blk
   , varGsmState :: StrictTVar m GSM.GsmState
   , mempool :: Mempool m blk
   , peerSharingRegistry :: PeerSharingRegistry addrNTN m
@@ -454,6 +494,8 @@ initInternalState
       newTVarIO gsmState
 
     varChainSyncHandles <- atomically newChainSyncClientHandleCollection
+    varPerasCertDiffusionHandles <- atomically newObjectDiffusionInboundHandleCollection
+
     mempool <-
       openMempool
         registry

ouroboros-consensus-diffusion/test/consensus-test/Test/Consensus/GSM.hs

@@ -142,7 +142,7 @@ setupGsm isHaaSatisfied vars = do
       , GSM.peerIsIdle = isIdling
       , GSM.durationUntilTooOld = Just durationUntilTooOld
       , GSM.equivalent = (==) -- unsound, but harmless in this test
-      , GSM.getPeerStates = readTVar varStates
+      , GSM.getPeerStates = traverse readTVar =<< readTVar varStates
       , GSM.getCurrentSelection = readTVar varSelection
       , GSM.minCaughtUpDuration = thrashLimit
       , GSM.setCaughtUpPersistentMark = \b ->

ouroboros-consensus-diffusion/test/consensus-test/Test/Consensus/Genesis/Tests/LoE/CaughtUp.hs

@@ -283,7 +283,7 @@ mkGsmEntryPoints varChainSyncHandles chainDB writeGsmState =
       { GSM.getCandidateOverSelection = pure candidateOverSelection
       , GSM.peerIsIdle = csIdling
       , GSM.equivalent = (==) `on` AF.headPoint
-      , GSM.getPeerStates = fmap cschState <$> cschcMap varChainSyncHandles
+      , GSM.getPeerStates = traverse readTVar =<< fmap cschState <$> cschcMap varChainSyncHandles
       , GSM.getCurrentSelection = ChainDB.getCurrentChain chainDB
       , -- Make sure that we stay in CaughtUp for the duration of the test once we
         -- have entered it.