local msg notification codec & peer sharing configuration

dmq-node/app/Main.hs

@@ -48,7 +48,6 @@ import DMQ.Configuration.Topology (readTopologyFile)
 import DMQ.Diffusion.Applications (diffusionApplications)
 import DMQ.Diffusion.Arguments
 import DMQ.Diffusion.NodeKernel as NodeKernel
-import DMQ.Diffusion.PeerSelectionPolicy (policy)
 import DMQ.Genesis
 import DMQ.Handlers.TopLevel (toplevelExceptionHandler)
 import DMQ.Mempool qualified as Mempool
@@ -180,9 +179,7 @@ runDMQ commandLineConfig = do
           ekgStore ps
           >>= link
 
-    stdGen <- Random.newStdGen
-    let (psRng, policyRng) = Random.splitGen stdGen
-    policyRngVar <- newTVarIO policyRng
+    psRng <- Random.newStdGen
 
     (shelleyGenesis, _) <-
       runExceptT (readGenesis genesisFile genesisHash) >>= either throwIO pure
@@ -269,7 +266,6 @@ runDMQ commandLineConfig = do
                                     dmqLimitsAndTimeouts
                                     dmqNtNApps
                                     dmqNtCApps
-                                    (policy policyRngVar nodeKernel.peerMetric)
 
         Diffusion.run dmqDiffusionArguments
                       dmqDiffusionTracers

dmq-node/changelog.d/20260707_092212_coot_dmq_changes.md

@@ -0,0 +1,28 @@
+<!--
+A new scriv changelog fragment.
+
+Uncomment the section that is right (remove the HTML comment wrapper).
+For top level release notes, leave all the headers commented out.
+-->
+
+### Breaking
+
+- Added PeerSelectionPolicy's rng to `NodeKernel`.  This allowed to create the
+  policy in `diffusionApplications` rather than in the `main` module. 
+
+### Non-Breaking
+
+- Peer sharing delays and timeouts in `PeerSelectionPolicy`
+- Explicit cborg error in local-msg-notification protocol
+- Added tracers for NtC connection with `cardano-node`:
+  - hanshake mini-protocol tracer
+  - mux tracer
+  - mux channel tracer
+  - mux bearer tracer
+
+<!--
+### Patch
+
+- A bullet item for the Patch category.
+
+-->

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

@@ -3,8 +3,11 @@
 
 module DMQ.Diffusion.Applications where
 
+import Control.Concurrent.Class.MonadSTM.Strict
+
 import DMQ.Configuration
 import DMQ.Diffusion.NodeKernel (NodeKernel (..))
+import DMQ.Diffusion.PeerSelectionPolicy qualified as PeerSelectionPolicy
 import DMQ.NodeToClient (NodeToClientVersion, NodeToClientVersionData,
            stdVersionDataNTC)
 import DMQ.NodeToClient qualified as NTC
@@ -14,26 +17,29 @@ import DMQ.NodeToNode qualified as NTN
 
 import Ouroboros.Network.Diffusion.Types qualified as Diffusion
 import Ouroboros.Network.ExitPolicy (RepromoteDelay (..))
-import Ouroboros.Network.PeerSelection.Governor.Types (PeerSelectionPolicy)
 import Ouroboros.Network.Protocol.Handshake.Version (combineVersions,
            simpleSingletonVersions)
 import Ouroboros.Network.RethrowPolicy (ioErrorRethrowPolicy,
            muxErrorRethrowPolicy)
 
 diffusionApplications
-  :: NodeKernel crypto ntnAddr m
+  :: ( MonadSTM m
+     , Ord ntnAddr
+     )
+  => NodeKernel crypto ntnAddr m
   -> Configuration
   -> Diffusion.Configuration NoExtraFlags m ntnFd ntnAddr ntcFd ntcAddr
   -> NTN.LimitsAndTimeouts crypto ntnAddr
   -> NTN.Apps ntnAddr m a ()
   -> NTC.Apps ntcAddr m ()
-  -> PeerSelectionPolicy ntnAddr m
   -> Diffusion.Applications ntnAddr NodeToNodeVersion   NodeToNodeVersionData
                             ntcAddr NodeToClientVersion NodeToClientVersionData
                             NoExtraFlags m a
 diffusionApplications
   NodeKernel {
-    peerSharingRegistry
+    peerSharingRegistry,
+    peerSelectionPolicyRngVar,
+    peerMetric
   }
   Configuration {
     dmqcNetworkMagic = I networkMagic
@@ -45,7 +51,7 @@ diffusionApplications
   ntnLimitsAndTimeouts
   ntnApps
   ntcApps
-  peerSelectionPolicy =
+  =
   Diffusion.Applications {
     Diffusion.daApplicationInitiatorMode =
       combineVersions
@@ -76,7 +82,7 @@ diffusionApplications
   , Diffusion.daReturnPolicy        = const dmqRepromoteDelay
   , Diffusion.daRepromoteErrorDelay = dmqRepromoteDelay
   , Diffusion.daLocalRethrowPolicy  = mempty
-  , Diffusion.daPeerSelectionPolicy = peerSelectionPolicy
+  , Diffusion.daPeerSelectionPolicy = PeerSelectionPolicy.policy peerSelectionPolicyRngVar peerMetric
   , Diffusion.daPeerSharingRegistry = peerSharingRegistry
   }
 

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

@@ -93,6 +93,8 @@ newNodeKernel rng ShelleyGenesis {sgMaxKESEvolutions} = do
   sigChannelVar <- newTxChannelsVar
   sigMempoolSem <- newTxMempoolSem
   let (rng', rng'') = Random.splitGen rng
+      (peerSharingRng, peerSelectionPolicyRng) = Random.splitGen rng''
+  peerSelectionPolicyRngVar <- newTVarIO peerSelectionPolicyRng
   sigSharedTxStateVar <- newSharedTxStateVar rng'
   (readinessVar,
     ocertCountersVar,
@@ -132,7 +134,7 @@ newNodeKernel rng ShelleyGenesis {sgMaxKESEvolutions} = do
   peerSharingAPI <-
     newPeerSharingAPI
       publicPeerSelectionStateVar
-      rng''
+      peerSharingRng
       ps_POLICY_PEER_SHARE_STICKY_TIME
       ps_POLICY_PEER_SHARE_MAX_PEERS
 
@@ -141,6 +143,7 @@ newNodeKernel rng ShelleyGenesis {sgMaxKESEvolutions} = do
   pure NodeKernel { keepAliveRegistry
                   , peerSharingRegistry
                   , peerSharingAPI
+                  , peerSelectionPolicyRngVar
                   , mempool
                   , sigChannelVar
                   , sigMempoolSem
@@ -180,7 +183,11 @@ withNodeKernel :: forall crypto ntnAddr ntcAddr m a.
                -- decision logic threads will be killed
                -> m a
 withNodeKernel DMQTracers { sigSubmissionLogicTracer,
-                            localStateQueryClientTracer
+                            localStateQueryClientTracer,
+                            cardanoNodeHandshakeProtocolTracer,
+                            cardanoNodeMuxTracer,
+                            cardanoNodeChannelTracer,
+                            cardanoNodeBearerTracer
                           }
                localSnocket
                mkLocalBearer
@@ -223,7 +230,16 @@ withNodeKernel DMQTracers { sigSubmissionLogicTracer,
           ctaHandshakeCodec      = Cardano.NtoC.nodeToClientHandshakeCodec,
           ctaHandshakeTimeLimits = noTimeLimitsHandshake,
           ctaVersionDataCodec    = cborTermVersionDataCodec Cardano.NtoC.nodeToClientCodecCBORTerm,
-          ctaConnectTracers      = Cardano.NtoC.nullNetworkConnectTracers, --debuggingNetworkConnectTracers,
+          ctaConnectTracers      = Cardano.NtoC.NetworkConnectTracers {
+              Cardano.NtoC.nctMuxTracers =
+                Mx.Tracers {
+                  Mx.tracer        = cardanoNodeMuxTracer,
+                  Mx.channelTracer = cardanoNodeChannelTracer,
+                  Mx.bearerTracer  = cardanoNodeBearerTracer
+                },
+              Cardano.NtoC.nctHandshakeTracer =
+                cardanoNodeHandshakeProtocolTracer
+            },
           ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion
         }
         (\_ -> return ())

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

@@ -48,16 +48,17 @@ data NodeKernel crypto ntnAddr m =
 
     -- | Read the current peer sharing registry, used for interacting with
     -- the PeerSharing protocol
-  , peerSharingRegistry :: !(PeerSharingRegistry ntnAddr m)
-  , peerSharingAPI      :: !(PeerSharingAPI ntnAddr StdGen m)
-  , mempool             :: !(Mempool m SigId (Sig crypto))
-  , sigChannelVar       :: !(TxChannelsVar m ntnAddr SigId (Sig crypto))
-  , sigMempoolSem       :: !(TxMempoolSem m)
-  , sigSharedTxStateVar :: !(SharedTxStateVar m ntnAddr SigId (Sig crypto))
-  , stakePools          :: !(StakePools m)
-  , readinessVar        :: !(StrictTVar m Readiness)
-  , peerMetric          :: !(PeerMetric m SigId ntnAddr)
-  , lastSigByPoolIdVar  :: !(StrictTVar m (OrdPSQ PoolId Time ()))
+  , peerSharingRegistry       :: !(PeerSharingRegistry ntnAddr m)
+  , peerSharingAPI            :: !(PeerSharingAPI ntnAddr StdGen m)
+  , peerSelectionPolicyRngVar :: !(StrictTVar m StdGen)
+  , mempool                   :: !(Mempool m SigId (Sig crypto))
+  , sigChannelVar             :: !(TxChannelsVar m ntnAddr SigId (Sig crypto))
+  , sigMempoolSem             :: !(TxMempoolSem m)
+  , sigSharedTxStateVar       :: !(SharedTxStateVar m ntnAddr SigId (Sig crypto))
+  , stakePools                :: !(StakePools m)
+  , readinessVar              :: !(StrictTVar m Readiness)
+  , peerMetric                :: !(PeerMetric m SigId ntnAddr)
+  , lastSigByPoolIdVar        :: !(StrictTVar m (OrdPSQ PoolId Time ()))
   }
 
 data StakePools m = StakePools {

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

@@ -12,7 +12,7 @@ import DMQ.Diffusion.PeerSelection.PeerMetric qualified as PeerMetric
 
 import Ouroboros.Network.PeerSelection hiding (PeerMetrics)
 
--- | Trivial peer selection policy used as dummy value
+-- | DMQ PeerSelectionPolicy
 --
 policy :: forall sigId peerAddr m.
           ( MonadSTM m
@@ -34,10 +34,10 @@ policy rngVar peerMetrics =
 
     policyFindPublicRootTimeout      = 5,
     policyMaxInProgressPeerShareReqs = 0,
-    policyPeerShareRetryTime         = 0, -- seconds
-    policyPeerShareBatchWaitTime     = 0, -- seconds
-    policyPeerShareOverallTimeout    = 0, -- seconds
-    policyPeerShareActivationDelay   = 2, -- seconds
+    policyPeerShareRetryTime         = 900,  -- seconds
+    policyPeerShareBatchWaitTime     = 3,    -- seconds
+    policyPeerShareOverallTimeout    = 10,   -- seconds
+    policyPeerShareActivationDelay   = 300,  -- seconds
     policyMaxConnectionRetries       = 5,
     policyClearFailCountDelay        = 120   -- seconds
   }

dmq-node/src/DMQ/Protocol/LocalMsgNotification/Codec.hs

@@ -139,12 +139,21 @@ codecLocalMsgNotification' mkWithBytes encodeMsg decodeMsgWithBytes =
           msgs <- CBOR.decodeSequenceLenIndef
                     (flip (:)) [] reverse
                     (Utils.decodeWithByteSpan decodeMsgWithBytes)
-          -- Issue #15
-          -- more <- bool DoesNotHaveMore HasMore <$> CBOR.decodeBool
-          -- return (Annotator \bytes ->
-          --          SomeMessage $ MsgReply (BlockingReply (mkWithBytes bytes <$> NonEmpty.fromList msgs)) more)
-          return (Annotator \bytes ->
-                   SomeMessage $ MsgReply (BlockingReply (mkWithBytes bytes <$> NonEmpty.fromList msgs)) DoesNotHaveMore)
+          case NonEmpty.nonEmpty msgs of
+            Nothing -> fail (printf "codecLocalMsgNotification (%s, %s) empty list"
+                                    (show (activeAgency :: ActiveAgency st))
+                                    (show stok)
+                            )
+            Just msgs' ->
+              -- Issue #15
+              -- more <- bool DoesNotHaveMore HasMore <$> CBOR.decodeBool
+              -- return (Annotator \bytes ->
+              --          SomeMessage $ MsgReply (BlockingReply (mkWithBytes bytes <$> msgs')) more)
+              return (Annotator \bytes ->
+                       SomeMessage $
+                         MsgReply
+                           (BlockingReply (mkWithBytes bytes <$> msgs'))
+                           DoesNotHaveMore)
 
         (SingIdle, 1, 3) -> return (Annotator \_ -> SomeMessage MsgClientDone)
 

dmq-node/src/DMQ/Tracer.hs

@@ -50,6 +50,7 @@ import Ouroboros.Network.Magic (NetworkMagic (..))
 import Ouroboros.Network.OrphanInstances ()
 import Ouroboros.Network.PeerSelection.PublicRootPeers (PublicRootPeers)
 import Ouroboros.Network.PeerSelection.PublicRootPeers qualified as PublicRootPeers
+import Ouroboros.Network.Protocol.Handshake.Type (Handshake)
 import Ouroboros.Network.Protocol.KeepAlive.Type (KeepAlive)
 import Ouroboros.Network.Protocol.KeepAlive.Type qualified as KA
 import Ouroboros.Network.Protocol.PeerSharing.Type (PeerSharing)
@@ -66,6 +67,7 @@ import Cardano.KESAgent.KES.Crypto (Crypto)
 import Cardano.Logging (Namespace (..))
 import Cardano.Logging qualified as Logging
 import Cardano.Logging.Prometheus.TCPServer qualified as Logging
+import Cardano.Network.NodeToClient qualified as Cardano.NtC
 
 import DMQ.Configuration
 import DMQ.Diffusion.NodeKernel.Types (ValidationCfg)
@@ -125,7 +127,16 @@ data DMQTracers crypto ntnAddr ntcAddr m = DMQTracers {
     localSigValidationTracer
       :: Tracer m SigValidationTrace,
     cardanoNodeHandshakeTracer
-      :: Tracer m (NtC.HandshakeTr LocalAddress)
+      :: Tracer m (NtC.HandshakeTr LocalAddress),
+    cardanoNodeHandshakeProtocolTracer
+      :: Tracer m (Mx.WithBearer (ConnectionId LocalAddress)
+                                 (TraceSendRecv (Handshake Cardano.NtC.NodeToClientVersion Term))),
+    cardanoNodeMuxTracer
+      :: Tracer m (Mx.WithBearer (ConnectionId LocalAddress) Mx.Trace),
+    cardanoNodeChannelTracer
+      :: Tracer m (Mx.WithBearer (ConnectionId LocalAddress) Mx.ChannelTrace),
+    cardanoNodeBearerTracer
+      :: Tracer m (Mx.WithBearer (ConnectionId LocalAddress) Mx.BearerTrace)
     }
 
 data DMQStartupTrace
@@ -504,6 +515,26 @@ mkDMQTracers ekgStore dmqConfigFilePath = do
     stdoutTrace trForward mbTrEkg
     ["Net", "Local", "Cardano", "Handshake"]
 
+  !cardanoNodeHandshakeProtocolTracer <- mkTracer
+    traceConfig configReflection
+    stdoutTrace trForward mbTrEkg
+    ["Net", "Local", "Cardano", "Handshake", "Protocol"]
+
+  !cardanoNodeMuxTracer <- mkTracer
+    traceConfig configReflection
+    stdoutTrace trForward mbTrEkg
+    ["Net", "Local", "Cardano", "Mux", "Local"]
+
+  !cardanoNodeChannelTracer <- mkTracer
+    traceConfig configReflection
+    stdoutTrace trForward mbTrEkg
+    ["Net", "Local", "Cardano", "Mux", "Local", "Channel"]
+
+  !cardanoNodeBearerTracer <- mkTracer
+    traceConfig configReflection
+    stdoutTrace trForward mbTrEkg
+    ["Net", "Local", "Cardano", "Mux", "Local", "Bearer"]
+
   let dmqTracers = DMQTracers {
         sigSubmissionLogicTracer           = Tracer $ Logging.traceWith sigSubmissionLogicTracer,
         sigSubmissionLogicPeerTracer       = Tracer $ Logging.traceWith sigSubmissionLogicPeerTracer,
@@ -522,7 +553,11 @@ mkDMQTracers ekgStore dmqConfigFilePath = do
         localStateQueryClientTracer        = Tracer $ Logging.traceWith localStateQueryClientTracer,
         sigValidationTracer                = Tracer $ Logging.traceWith sigValidationTracer,
         localSigValidationTracer           = Tracer $ Logging.traceWith localSigValidationTracer,
-        cardanoNodeHandshakeTracer         = Tracer $ Logging.traceWith cardanoNodeHandshakeTracer
+        cardanoNodeHandshakeTracer         = Tracer $ Logging.traceWith cardanoNodeHandshakeTracer,
+        cardanoNodeHandshakeProtocolTracer = Tracer $ Logging.traceWith cardanoNodeHandshakeProtocolTracer,
+        cardanoNodeMuxTracer               = Tracer $ Logging.traceWith cardanoNodeMuxTracer,
+        cardanoNodeChannelTracer           = Tracer $ Logging.traceWith cardanoNodeChannelTracer,
+        cardanoNodeBearerTracer            = Tracer $ Logging.traceWith cardanoNodeBearerTracer
       }
 
   -- This backend can only be used globally, i.e. will always apply to the namespace root.