Expose big ledger peers to dmq node

Author: crocodile-dentist

dmq-node/app/Main.hs

@@ -1,5 +1,6 @@
 {-# LANGUAGE DisambiguateRecordFields #-}
 {-# LANGUAGE MultiWayIf               #-}
+{-# LANGUAGE OverloadedRecordDot      #-}
 {-# LANGUAGE OverloadedStrings        #-}
 {-# LANGUAGE ScopedTypeVariables      #-}
 {-# LANGUAGE TemplateHaskell          #-}
@@ -8,6 +9,7 @@
 
 module Main where
 
+import Control.Concurrent.Class.MonadSTM.Strict
 import Control.Exception (throwIO)
 import Control.Monad (void, when)
 import Control.Tracer (Tracer (..), nullTracer, traceWith)
@@ -173,6 +175,7 @@ runDMQ commandLineConfig = do
                                  (if localHandshakeTracer
                                     then WithEventType "Handshake" >$< tracer
                                     else nullTracer)
+                                 $ readTVar $ nodeKernel.stakePools.ledgerPeersVar
             dmqDiffusionApplications =
               diffusionApplications nodeKernel
                                     dmqConfig

dmq-node/dmq-node.cabal

@@ -151,6 +151,7 @@ executable dmq-node
     cardano-ledger-core,
     contra-tracer >=0.1 && <0.3,
     dmq-node,
+    io-classes:{strict-stm},
     kes-agent-crypto,
     optparse-applicative,
     ouroboros-network:{ouroboros-network, api, framework},

dmq-node/src/DMQ/Diffusion/Arguments.hs

@@ -22,6 +22,7 @@ import Control.Monad.Class.MonadST (MonadST)
 import Control.Monad.Class.MonadThrow (MonadCatch)
 import Control.Monad.Class.MonadTimer.SI (MonadDelay, MonadTimer)
 import Control.Tracer (Tracer)
+import Data.List.NonEmpty (NonEmpty)
 import Network.DNS (Resolver)
 import Network.Socket (Socket)
 
@@ -35,7 +36,7 @@ import Ouroboros.Network.PeerSelection.Churn (peerChurnGovernor)
 import Ouroboros.Network.PeerSelection.Governor.Types
            (ExtraGuardedDecisions (..), PeerSelectionGovernorArgs (..))
 import Ouroboros.Network.PeerSelection.LedgerPeers.Type
-           (LedgerPeersConsensusInterface (..))
+           (LedgerPeersConsensusInterface (..), PoolStake, LedgerRelayAccessPoint)
 import Ouroboros.Network.PeerSelection.RelayAccessPoint (SRVPrefix)
 import Ouroboros.Network.PeerSelection.Types (nullPublicExtraPeersAPI)
 
@@ -49,6 +50,7 @@ diffusionArguments
      )
   => Tracer m (NtN.HandshakeTr ntnAddr)
   -> Tracer m (NtC.HandshakeTr ntcAddr)
+  -> STM m [(PoolStake, NonEmpty LedgerRelayAccessPoint)]
   -> Diffusion.Arguments
        NoExtraState NoExtraDebugState NoExtraFlags NoExtraPeers
        NoExtraAPI NoExtraChurnArgs NoExtraCounters NoExtraTracer
@@ -63,7 +65,8 @@ diffusionArguments
        NodeToClientVersion
        NodeToClientVersionData
 diffusionArguments handshakeNtNTracer
-                   handshakeNtCTracer =
+                   handshakeNtCTracer
+                   lpGetLedgerPeers =
   Diffusion.Arguments {
     Diffusion.daNtnDataFlow    = DMQ.ntnDataFlow
   , Diffusion.daNtnPeerSharing = peerSharing
@@ -74,7 +77,7 @@ diffusionArguments handshakeNtNTracer
   , Diffusion.daLedgerPeersCtx        =
       LedgerPeersConsensusInterface {
         lpGetLatestSlot  = return minBound
-      , lpGetLedgerPeers = return []
+      , lpGetLedgerPeers
       , lpExtraAPI       = NoExtraAPI
       }
   , Diffusion.daEmptyExtraState           = NoExtraState

dmq-node/src/DMQ/Diffusion/NodeKernel.hs

@@ -20,6 +20,7 @@ import Data.Aeson qualified as Aeson
 import Data.Function (on)
 import Data.Functor.Contravariant ((>$<))
 import Data.Hashable
+import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
 import Data.Map.Strict qualified as Map
 import Data.Sequence (Seq)
@@ -42,6 +43,8 @@ import Ouroboros.Network.BlockFetch (FetchClientRegistry,
 import Ouroboros.Network.ConnectionId (ConnectionId (..))
 import Ouroboros.Network.PeerSelection.Governor.Types
            (makePublicPeerSelectionStateVar)
+import Ouroboros.Network.PeerSelection.LedgerPeers.Type (PoolStake,
+           LedgerRelayAccessPoint)
 import Ouroboros.Network.PeerSharing (PeerSharingAPI, PeerSharingRegistry,
            newPeerSharingAPI, newPeerSharingRegistry,
            ps_POLICY_PEER_SHARE_MAX_PEERS, ps_POLICY_PEER_SHARE_STICKY_TIME)
@@ -83,6 +86,8 @@ data StakePools m = StakePools {
     stakePoolsVar     :: StrictTVar m (Map PoolId StakeSnapshot)
     -- | acquires validation context for signature validation
   , poolValidationCtx :: m PoolValidationCtx
+  , ledgerPeersVar
+      :: StrictTVar m [(PoolStake, NonEmpty LedgerRelayAccessPoint)]
   }
 
 data PoolValidationCtx =
@@ -109,15 +114,15 @@ newNodeKernel evolutionConfig rng = do
   sigMempoolSem <- newTxMempoolSem
   let (rng', rng'') = Random.split rng
   sigSharedTxStateVar <- newSharedTxStateVar rng'
-  nextEpochVar <- newTVarIO Nothing
-  stakePoolsVar <- newTVarIO Map.empty
+  (nextEpochVar, stakePoolsVar, ledgerPeersVar) <- atomically $
+    (,,) <$> newTVar Nothing <*> newTVar Map.empty <*> newTVar []
   let poolValidationCtx = do
         (nextEpochBoundary, stakePools') <-
           atomically $ (,) <$> readTVar nextEpochVar <*> readTVar stakePoolsVar
         now <- getCurrentTime
         return $ DMQPoolValidationCtx now nextEpochBoundary stakePools'
 
-      stakePools = StakePools { stakePoolsVar, poolValidationCtx }
+      stakePools = StakePools { stakePoolsVar, poolValidationCtx, ledgerPeersVar }
 
   peerSharingAPI <-
     newPeerSharingAPI

dmq-node/src/DMQ/NodeToClient/LocalStateQueryClient.hs

@@ -38,6 +38,7 @@ import Ouroboros.Consensus.Shelley.Ledger.SupportsProtocol ()
 import Ouroboros.Network.Block
 import Ouroboros.Network.Magic
 import Ouroboros.Network.Mux qualified as Mx
+import Ouroboros.Network.PeerSelection.LedgerPeers.Type (LedgerPeerSnapshot (..))
 import Ouroboros.Network.Protocol.LocalStateQuery.Client
 import Ouroboros.Network.Protocol.LocalStateQuery.Type
 
@@ -52,7 +53,8 @@ cardanoClient
   -> StakePools m
   -> StrictTVar m (Maybe UTCTime) -- ^ from node kernel
   -> LocalStateQueryClient (CardanoBlock crypto) (Point block) (Query block) m Void
-cardanoClient _tracer StakePools { stakePoolsVar } nextEpochVar = LocalStateQueryClient (idle Nothing)
+cardanoClient _tracer StakePools { stakePoolsVar, ledgerPeersVar } nextEpochVar =
+  LocalStateQueryClient (idle Nothing)
   where
     idle mSystemStart = pure $ SendMsgAcquire ImmutableTip acquire
       where
@@ -103,9 +105,18 @@ cardanoClient _tracer StakePools { stakePoolsVar } nextEpochVar = LocalStateQuer
       atomically do
         writeTVar stakePoolsVar ssStakeSnapshots
         writeTVar nextEpochVar $ Just nextEpoch
-      pure $ SendMsgRelease do
-        threadDelay $ min (realToFrac toNextEpoch) 86400 -- TODO fuzz this?
-        idle $ Just systemStart
+      pure $
+        SendMsgQuery (BlockQuery . QueryIfCurrentConway $ GetBigLedgerPeerSnapshot)
+        $ wrappingMismatch handleLedgerPeers
+      where
+        handleLedgerPeers (LedgerPeerSnapshot snapshot) = do
+          let bigRelays = fmap snd . snd $ snapshot
+          atomically do
+            writeTVar ledgerPeersVar bigRelays
+          pure $ SendMsgRelease do
+            threadDelay $ min (realToFrac toNextEpoch) 86400 -- TODO fuzz this?
+            idle $ Just systemStart
+
 
 connectToCardanoNode :: Tracer IO (WithEventType String)
                      -> LocalSnocket