From fcd15c1641de7cb62e661f8033e2c22c15de3772 Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Mon, 6 Jul 2026 10:14:53 +0200 Subject: [PATCH 1/5] local-msg-notification: explicit cbor decode error It's better to fail with an explicit cbor codec error than a pure exception from a partial function. --- .../Protocol/LocalMsgNotification/Codec.hs | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/dmq-node/src/DMQ/Protocol/LocalMsgNotification/Codec.hs b/dmq-node/src/DMQ/Protocol/LocalMsgNotification/Codec.hs index a0ffc1eb..00fa59d0 100644 --- a/dmq-node/src/DMQ/Protocol/LocalMsgNotification/Codec.hs +++ b/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) From 2e67cdc0d500b9a01ed0ae688e0a480b1d3b410d Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Mon, 6 Jul 2026 12:33:15 +0200 Subject: [PATCH 2/5] peer-selection: peer sharing timeouts Use the same timeouts as `cardano-node`. --- dmq-node/src/DMQ/Diffusion/PeerSelectionPolicy.hs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/dmq-node/src/DMQ/Diffusion/PeerSelectionPolicy.hs b/dmq-node/src/DMQ/Diffusion/PeerSelectionPolicy.hs index efa2ea54..3a724636 100644 --- a/dmq-node/src/DMQ/Diffusion/PeerSelectionPolicy.hs +++ b/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 } From bf78c1612b2ffe9767a3d8868d1ecc5b8d971ba7 Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Mon, 6 Jul 2026 12:34:04 +0200 Subject: [PATCH 3/5] peer-selection: pass RNG through node kernel This way we can create the policy in `diffusionApplications` rather than in the main module. --- dmq-node/app/Main.hs | 6 +----- dmq-node/src/DMQ/Diffusion/Applications.hs | 18 ++++++++++------ dmq-node/src/DMQ/Diffusion/NodeKernel.hs | 5 ++++- .../src/DMQ/Diffusion/NodeKernel/Types.hs | 21 ++++++++++--------- 4 files changed, 28 insertions(+), 22 deletions(-) diff --git a/dmq-node/app/Main.hs b/dmq-node/app/Main.hs index 621d92a2..4e7cfead 100644 --- a/dmq-node/app/Main.hs +++ b/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 diff --git a/dmq-node/src/DMQ/Diffusion/Applications.hs b/dmq-node/src/DMQ/Diffusion/Applications.hs index 9ab677d9..cc97a488 100644 --- a/dmq-node/src/DMQ/Diffusion/Applications.hs +++ b/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 } diff --git a/dmq-node/src/DMQ/Diffusion/NodeKernel.hs b/dmq-node/src/DMQ/Diffusion/NodeKernel.hs index cc167d58..6402872e 100644 --- a/dmq-node/src/DMQ/Diffusion/NodeKernel.hs +++ b/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 diff --git a/dmq-node/src/DMQ/Diffusion/NodeKernel/Types.hs b/dmq-node/src/DMQ/Diffusion/NodeKernel/Types.hs index 1069a1af..199da3ac 100644 --- a/dmq-node/src/DMQ/Diffusion/NodeKernel/Types.hs +++ b/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 { From 5cc184e87dcdc166fd0157e819f9715fb9d33108 Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Mon, 6 Jul 2026 16:24:44 +0200 Subject: [PATCH 4/5] Added tracers for NtC connection with cardano-node Partially addresses issue #59. --- dmq-node/src/DMQ/Diffusion/NodeKernel.hs | 17 +++++++++-- dmq-node/src/DMQ/Tracer.hs | 39 ++++++++++++++++++++++-- 2 files changed, 52 insertions(+), 4 deletions(-) diff --git a/dmq-node/src/DMQ/Diffusion/NodeKernel.hs b/dmq-node/src/DMQ/Diffusion/NodeKernel.hs index 6402872e..37182880 100644 --- a/dmq-node/src/DMQ/Diffusion/NodeKernel.hs +++ b/dmq-node/src/DMQ/Diffusion/NodeKernel.hs @@ -183,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 @@ -226,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 ()) diff --git a/dmq-node/src/DMQ/Tracer.hs b/dmq-node/src/DMQ/Tracer.hs index fb7410ad..93684ac1 100644 --- a/dmq-node/src/DMQ/Tracer.hs +++ b/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. From 98758a540a984e6abad7819f4841e3053aedc544 Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Tue, 7 Jul 2026 09:29:16 +0200 Subject: [PATCH 5/5] Added changelog fragment --- .../20260707_092212_coot_dmq_changes.md | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) create mode 100644 dmq-node/changelog.d/20260707_092212_coot_dmq_changes.md diff --git a/dmq-node/changelog.d/20260707_092212_coot_dmq_changes.md b/dmq-node/changelog.d/20260707_092212_coot_dmq_changes.md new file mode 100644 index 00000000..556129b8 --- /dev/null +++ b/dmq-node/changelog.d/20260707_092212_coot_dmq_changes.md @@ -0,0 +1,28 @@ + + +### 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 + +