diff --git a/cabal.project b/cabal.project index bf7f9fd..8d02365 100644 --- a/cabal.project +++ b/cabal.project @@ -15,14 +15,10 @@ repository cardano-haskell-packages -- repeat the index-state for hackage to work around haskell.nix parsing limitation index-state: -- Bump this if you need newer packages from Hackage - , hackage.haskell.org 2026-06-20T00:51:00Z + , hackage.haskell.org 2026-06-29T22:49:53Z -- Bump this if you need newer packages from CHaP - , cardano-haskell-packages 2026-05-11T20:15:43Z - -active-repositories: - , :rest - , cardano-haskell-packages:override + , cardano-haskell-packages 2026-07-02T10:10:00Z packages: ./dmq-node @@ -44,25 +40,63 @@ if impl(ghc >= 9.12.0) -- rejecting: cardano-crypto-class-2.3.1.0 (conflict: cardano-crypto-tests => cardano-crypto-class>=2.2.2 && <2.2.4) -- allow-newer: cardano-crypto-tests:cardano-crypto-class --- ouroboros-network with PR#5376, but without contra-tracer-0.2.1.0 (PR#5368) source-repository-package type: git - location: https://github.com/IntersectMBO/ouroboros-network - tag: 5a80699bca834a4de47aca4fe4d322832309e374 - --sha256: sha256-FXp0gIfsUDVEE4ZuEM5y55qx0I7d5+/wQ7AjUz6UDcE= - subdir: network-mux - cardano-diffusion - ouroboros-network - acts-generic + location: https://github.com/IntersectMBO/ouroboros-network.git + tag: 9e45c79049bd5f20e3de308ee8203c201aa61f9e + --sha256: sha256-dK8taHnenl4D18bOxepXstO0hq7RWj2jnaxECPOT+ew= + subdir: + cardano-diffusion + network-mux + ouroboros-network + acts-generic + +source-repository-package + type: git + location: https://github.com/IntersectMBO/ouroboros-consensus.git + tag: e37ba209ffdbdcb3cc8e7a9cb0fb6d6fa7cce8a3 + --sha256: sha256-srdli4YUIo3HtdWwpk2Tq5okk4GQ5Fpn/ipFO8eykVU= + subdir: + . source-repository-package type: git - location: https://github.com/IntersectMBO/ouroboros-consensus - tag: ec9df6ee1d7ab918e75219107fc17e1b2cf1a0e9 - --sha256: sha256-c+yGPftvlas2CcnFLUPPqjnitz5yTu/Qud5LaSMyY34= + location: https://github.com/IntersectMBO/cardano-ledger.git + tag: 245e63dcd335e3ca29823609e4aaa8f5f096a425 + --sha256: sha256-chsapfNnY/V7y1snBzjP0LSVnYl/Axt6GjahTPNOKj0= + subdir: + eras/allegra/impl + eras/alonzo/impl + eras/babbage/impl + eras/byron/chain/executable-spec + eras/byron/crypto + eras/byron/ledger/executable-spec + eras/byron/ledger/impl + eras/conway/impl + eras/dijkstra/impl + eras/mary/impl + eras/shelley-ma/test-suite + eras/shelley/impl + eras/shelley/test-suite + libs/cardano-data + libs/cardano-ledger-api + libs/cardano-ledger-binary + libs/cardano-ledger-core + libs/cardano-protocol-tpraos + libs/non-integral + libs/small-steps + libs/vector-map + +source-repository-package + type: git + location: https://github.com/f-f/kes-agent.git + tag: 32c1ed675d22a30735d9f22f7afa436a3ef3e64a + --sha256: sha256-o7hFX1JnraS6Xq0WoXQwd9Z8GsPPv0Ls2DWvZ08o0ZU= + subdir: + kes-agent + kes-agent-crypto constraints: semialign >= 1.4, - validation < 1.2 + validation >= 1.2 && < 2.0 allow-newer: hedgehog-quickcheck:QuickCheck, monoidal-containers:semialign - diff --git a/dmq-node/changelog.d/20260710_092801_coot_prepare_11_1.md b/dmq-node/changelog.d/20260710_092801_coot_prepare_11_1.md new file mode 100644 index 0000000..6163a35 --- /dev/null +++ b/dmq-node/changelog.d/20260710_092801_coot_prepare_11_1.md @@ -0,0 +1,25 @@ + + +### Breaking + +- Using `tx-submission` undecision logic: + - `NodeKernel` data type changed to adapt to the new tx-submission logic + - `SigMempoolResult` is provided by `DMQ.Diffusion.PeerSelection.PeerMetric` module. + - `sigSubmissionLogicPeerTracer` was removed from `DMQTracers` + - `sigDecisionPolicy :: TxDecisionPolicy` was adapted to the new tx-submission logic + +### Non-Breaking + +- Integrated with `trace-dispatcher ^>= 2.13.0` and `contra-tracer >= 0.2.1.0` + + diff --git a/dmq-node/dmq-node.cabal b/dmq-node/dmq-node.cabal index 4a1701a..4c632eb 100644 --- a/dmq-node/dmq-node.cabal +++ b/dmq-node/dmq-node.cabal @@ -127,7 +127,7 @@ library hashable >=1.0 && <1.6, io-classes:{io-classes, si-timers, strict-mvar, strict-stm} ^>=1.8.0.1, iproute ^>=1.7.15, - kes-agent-crypto ^>=1.1.0.0, + kes-agent-crypto ^>=1.2, mmap, mtl, network ^>=3.2.7, @@ -142,7 +142,7 @@ library singletons, text >=1.2.4 && <2.2, time >=1.12 && <1.15, - trace-dispatcher ^>=2.12.1, + trace-dispatcher ^>=2.13.0, transformers, transformers-except, typed-protocols:{typed-protocols, cborg} ^>=1.2, @@ -219,14 +219,12 @@ test-suite dmq-tests contra-tracer, deepseq, dmq-node, - hashable, io-classes:{io-classes, si-timers, strict-mvar, strict-stm}, io-sim, kes-agent-crypto, ouroboros-network:{ouroboros-network, api, framework, ouroboros-network-tests-lib, protocols, protocols-tests-lib, tests-lib}, psqueues, quickcheck-instances, - random, serialise, tasty, tasty-quickcheck, diff --git a/dmq-node/src/DMQ/Configuration/Topology.hs b/dmq-node/src/DMQ/Configuration/Topology.hs index 8deb6de..0456531 100644 --- a/dmq-node/src/DMQ/Configuration/Topology.hs +++ b/dmq-node/src/DMQ/Configuration/Topology.hs @@ -18,20 +18,11 @@ import Data.ByteString.Lazy qualified as LBS import Data.Text (Text) import Data.Text qualified as Text import Ouroboros.Network.Diffusion.Topology (NetworkTopology (..)) -import Ouroboros.Network.Diffusion.Types (NoExtraFlags (..)) +import Ouroboros.Network.Diffusion.Types (NoExtraConfig (..), NoExtraFlags (..)) import Ouroboros.Network.OrphanInstances (localRootPeersGroupsFromJSON, networkTopologyFromJSON, networkTopologyToJSON) import System.Exit (die) --- TODO: move `NoExtraConfig` and the `ToJSON NoExtraFlags` to --- `ouroboros-network`. - -data NoExtraConfig = NoExtraConfig - deriving Show - -instance ToJSON NoExtraFlags where - toJSON _ = Null - omitField _ = True instance FromJSON (NetworkTopology NoExtraConfig NoExtraFlags) where parseJSON = networkTopologyFromJSON diff --git a/dmq-node/src/DMQ/Diffusion/NodeKernel.hs b/dmq-node/src/DMQ/Diffusion/NodeKernel.hs index 3718288..c2d86ba 100644 --- a/dmq-node/src/DMQ/Diffusion/NodeKernel.hs +++ b/dmq-node/src/DMQ/Diffusion/NodeKernel.hs @@ -20,7 +20,6 @@ import Control.Monad.Class.MonadTimer.SI import "contra-tracer" Control.Tracer (nullTracer) import Data.Function (on) -import Data.Hashable import Data.Map.Strict qualified as Map import Data.OrdPSQ qualified as OrdPSQ import Data.Proxy @@ -53,8 +52,7 @@ import Ouroboros.Network.PeerSelection.Governor.Types import Ouroboros.Network.PeerSharing (newPeerSharingAPI, newPeerSharingRegistry, ps_POLICY_PEER_SHARE_MAX_PEERS, ps_POLICY_PEER_SHARE_STICKY_TIME) import Ouroboros.Network.Protocol.Handshake (Acceptable (..)) -import Ouroboros.Network.Protocol.Handshake.Codec (cborTermVersionDataCodec, - noTimeLimitsHandshake) +import Ouroboros.Network.Protocol.Handshake.Codec (noTimeLimitsHandshake) import Ouroboros.Network.Protocol.LocalStateQuery.Client import Ouroboros.Network.Protocol.LocalStateQuery.Type import Ouroboros.Network.Snocket (Snocket, localAddressFromPath) @@ -90,12 +88,11 @@ newNodeKernel rng ShelleyGenesis {sgMaxKESEvolutions} = do peerSharingRegistry <- newPeerSharingRegistry mempool <- Mempool.empty - sigChannelVar <- newTxChannelsVar - sigMempoolSem <- newTxMempoolSem - let (rng', rng'') = Random.splitGen rng - (peerSharingRng, peerSelectionPolicyRng) = Random.splitGen rng'' + let (peerSharingRng, peerSelectionPolicyRng) = Random.splitGen rng peerSelectionPolicyRngVar <- newTVarIO peerSelectionPolicyRng - sigSharedTxStateVar <- newSharedTxStateVar rng' + sigSharedTxStateVar <- newSharedTxStateVar emptySharedTxState + sigPeerRegistry <- newPeerTxRegistry + sigCountersVar <- newTxSubmissionCountersVar mempty (readinessVar, ocertCountersVar, stakePoolsVar, @@ -145,9 +142,9 @@ newNodeKernel rng ShelleyGenesis {sgMaxKESEvolutions} = do , peerSharingAPI , peerSelectionPolicyRngVar , mempool - , sigChannelVar - , sigMempoolSem , sigSharedTxStateVar + , sigPeerRegistry + , sigCountersVar , readinessVar , stakePools , peerMetric @@ -170,7 +167,6 @@ withNodeKernel :: forall crypto ntnAddr ntcAddr m a. , MonadTime m , MonadTimer m , Ord ntnAddr - , Hashable ntnAddr ) => DMQTracers crypto ntnAddr ntcAddr m -> Snocket m Cardano.NtoC.LocalSocket LocalAddress @@ -183,6 +179,7 @@ withNodeKernel :: forall crypto ntnAddr ntcAddr m a. -- decision logic threads will be killed -> m a withNodeKernel DMQTracers { sigSubmissionLogicTracer, + sigCountersTracer, localStateQueryClientTracer, cardanoNodeHandshakeProtocolTracer, cardanoNodeMuxTracer, @@ -200,18 +197,20 @@ withNodeKernel DMQTracers { sigSubmissionLogicTracer, rng k = do nodeKernel@NodeKernel { mempool, - sigChannelVar, - sigSharedTxStateVar + sigSharedTxStateVar, + sigPeerRegistry, + sigCountersVar } <- newNodeKernel rng shelleyGenesis withAsync (mempoolWorker mempool) $ \mempoolThread -> - withAsync (decisionLogicThreads - sigSubmissionLogicTracer - nullTracer + withAsync (txCountersThreadV2 Policy.sigDecisionPolicy - sigChannelVar - sigSharedTxStateVar) + sigCountersTracer + sigSubmissionLogicTracer + sigCountersVar + sigSharedTxStateVar + sigPeerRegistry) $ \sigLogicThread -> withAsync (connectToCardanoNode nodeKernel) \spmAid -> do link mempoolThread @@ -229,7 +228,7 @@ withNodeKernel DMQTracers { sigSubmissionLogicTracer, ConnectToArgs { ctaHandshakeCodec = Cardano.NtoC.nodeToClientHandshakeCodec, ctaHandshakeTimeLimits = noTimeLimitsHandshake, - ctaVersionDataCodec = cborTermVersionDataCodec Cardano.NtoC.nodeToClientCodecCBORTerm, + ctaVersionDataCodec = Cardano.NtoC.nodeToClientVersionDataCodec, ctaConnectTracers = Cardano.NtoC.NetworkConnectTracers { Cardano.NtoC.nctMuxTracers = Mx.Tracers { diff --git a/dmq-node/src/DMQ/Diffusion/NodeKernel/Types.hs b/dmq-node/src/DMQ/Diffusion/NodeKernel/Types.hs index 199da3a..25a890f 100644 --- a/dmq-node/src/DMQ/Diffusion/NodeKernel/Types.hs +++ b/dmq-node/src/DMQ/Diffusion/NodeKernel/Types.hs @@ -44,7 +44,7 @@ data Readiness = NotReady | Ready data NodeKernel crypto ntnAddr m = NodeKernel { -- | The keep alive registry, used for the keep alive clients. - keepAliveRegistry :: !(KeepAliveRegistry (ConnectionId ntnAddr) m) + keepAliveRegistry :: !(KeepAliveRegistry (ConnectionId ntnAddr) m) -- | Read the current peer sharing registry, used for interacting with -- the PeerSharing protocol @@ -52,9 +52,9 @@ data NodeKernel crypto 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)) + , sigSharedTxStateVar :: !(SharedTxStateVar m ntnAddr SigId) + , sigPeerRegistry :: !(PeerTxRegistry m ntnAddr) + , sigCountersVar :: !(TxSubmissionCountersVar m) , stakePools :: !(StakePools m) , readinessVar :: !(StrictTVar m Readiness) , peerMetric :: !(PeerMetric m SigId ntnAddr) diff --git a/dmq-node/src/DMQ/Diffusion/PeerSelection/PeerMetric.hs b/dmq-node/src/DMQ/Diffusion/PeerSelection/PeerMetric.hs index 4c139f1..b6d4ad9 100644 --- a/dmq-node/src/DMQ/Diffusion/PeerSelection/PeerMetric.hs +++ b/dmq-node/src/DMQ/Diffusion/PeerSelection/PeerMetric.hs @@ -18,7 +18,7 @@ module DMQ.Diffusion.PeerSelection.PeerMetric , announciness -- * Re-exports , TraceLabelPeer (..) - , TxMempoolResult (..) + , SigMempoolResult (..) -- * For testing purposes , peerMetricVar -- * Pure API exported for testing purposes @@ -42,8 +42,9 @@ import Data.OrdPSQ qualified as OrdPSQ import Network.Mux.Trace (TraceLabelPeer (..)) -import Ouroboros.Network.TxSubmission.Inbound.V2 (TxMempoolResult (..)) +data SigMempoolResult = SigAccepted | SigRejected + deriving (Eq, Show) newtype PeerMetricConfiguration = PeerMetricConfiguration { timeWindowToKeep :: DiffTime @@ -143,7 +144,7 @@ data ReportPeerMetric' m sigid f = ReportPeerMetric { -- | Report a received `sig` reportSig :: LocalPeerMetricState sigid - -> f (sigid, TxMempoolResult) + -> f (sigid, SigMempoolResult) -> STM m (LocalPeerMetricState sigid) } @@ -220,13 +221,13 @@ reportSigImpl Ord sigid => PeerMetricConfiguration -> LocalPeerMetricState sigid - -> TraceLabelPeer peeraddr (sigid, TxMempoolResult) + -> TraceLabelPeer peeraddr (sigid, SigMempoolResult) -> PeerMetricState sigid peeraddr -> (LocalPeerMetricState sigid, PeerMetricState sigid peeraddr) reportSigImpl config localState - (TraceLabelPeer _peeraddr (sigid, TxRejected)) + (TraceLabelPeer _peeraddr (sigid, SigRejected)) metricState = ( localState' @@ -238,7 +239,7 @@ reportSigImpl reportSigImpl config localState - (TraceLabelPeer peeraddr (sigid, TxAccepted)) + (TraceLabelPeer peeraddr (sigid, SigAccepted)) st = ( localState' diff --git a/dmq-node/src/DMQ/NodeToClient.hs b/dmq-node/src/DMQ/NodeToClient.hs index 0d9d240..3048ca9 100644 --- a/dmq-node/src/DMQ/NodeToClient.hs +++ b/dmq-node/src/DMQ/NodeToClient.hs @@ -51,8 +51,8 @@ import Ouroboros.Network.Handshake.Queryable (Queryable (..)) import Ouroboros.Network.Mux import Ouroboros.Network.OrphanInstances () import Ouroboros.Network.Protocol.Handshake (HandshakeArguments (..)) -import Ouroboros.Network.Protocol.Handshake.Codec (cborTermVersionDataCodec, - codecHandshake, noTimeLimitsHandshake) +import Ouroboros.Network.Protocol.Handshake.Codec (codecHandshake, + mkVersionedCodecCBORTerm, noTimeLimitsHandshake) import Ouroboros.Network.TxSubmission.Mempool.Reader import Ouroboros.Network.TxSubmission.Mempool.Simple import Ouroboros.Network.Util.ShowProxy @@ -71,7 +71,7 @@ ntcHandshakeArguments tracer = , haBearerTracer = nullTracer -- TODO , haHandshakeCodec = codecHandshake nodeToClientVersionCodec , haVersionDataCodec = - cborTermVersionDataCodec + mkVersionedCodecCBORTerm nodeToClientCodecCBORTerm , haAcceptVersion = acceptableVersion , haQueryVersion = queryVersion diff --git a/dmq-node/src/DMQ/NodeToNode.hs b/dmq-node/src/DMQ/NodeToNode.hs index 09be700..f441185 100644 --- a/dmq-node/src/DMQ/NodeToNode.hs +++ b/dmq-node/src/DMQ/NodeToNode.hs @@ -99,8 +99,8 @@ import Ouroboros.Network.TxSubmission.Mempool.Reader import Ouroboros.Network.OrphanInstances () import Ouroboros.Network.Protocol.Handshake (HandshakeArguments (..)) -import Ouroboros.Network.Protocol.Handshake.Codec (cborTermVersionDataCodec, - codecHandshake, timeLimitsHandshake) +import Ouroboros.Network.Protocol.Handshake.Codec (codecHandshake, + mkVersionedCodecCBORTerm, timeLimitsHandshake) import Ouroboros.Network.Protocol.KeepAlive.Client (keepAliveClientPeer) import Ouroboros.Network.Protocol.KeepAlive.Codec (byteLimitsKeepAlive, codecKeepAlive_v2, timeLimitsKeepAlive) @@ -189,8 +189,7 @@ ntnApps -> Apps ntnAddr m () () ntnApps DMQTracers { - sigSubmissionLogicPeerTracer - , sigSubmissionV2ProtocolTracer + sigSubmissionV2ProtocolTracer , sigSubmissionInboundTracer , sigSubmissionV1ProtocolTracer , sigSubmissionOutboundV1Tracer @@ -206,9 +205,9 @@ ntnApps keepAliveRegistry , peerSharingRegistry , peerSharingAPI - , sigChannelVar - , sigMempoolSem , sigSharedTxStateVar + , sigPeerRegistry + , sigCountersVar , peerMetric } Codecs { @@ -257,14 +256,11 @@ ntnApps $ peerMetric in withPeer - (Mx.WithBearer connId >$< sigSubmissionLogicPeerTracer) - sigChannelVar - sigMempoolSem sigDecisionPolicy - sigSharedTxStateVar mempoolReader - mempoolWriter - sigSize + sigSharedTxStateVar + sigPeerRegistry + sigCountersVar (remoteAddress connId) ( \(peerSigAPI :: PeerTxAPI m SigId (Sig crypto)) -> runPipelinedAnnotatedPeerWithLimits @@ -276,7 +272,9 @@ ntnApps $ sigSubmissionV2InboundPeerPipelined $ sigSubmissionInbound (Mx.WithBearer connId >$< sigSubmissionInboundTracer) + sigDecisionPolicy mempoolWriter + sigSize peerSigAPI reportPeerMetrics controlMessage @@ -336,14 +334,11 @@ ntnApps -> m ((), Maybe BL.ByteString) aSigSubmissionV1Server _version ResponderContext { rcConnectionId = connId } channel = withPeer - (Mx.WithBearer connId >$< sigSubmissionLogicPeerTracer) - sigChannelVar - sigMempoolSem sigDecisionPolicy - sigSharedTxStateVar mempoolReader - mempoolWriter - sigSize + sigSharedTxStateVar + sigPeerRegistry + sigCountersVar (remoteAddress connId) $ \(peerSigAPI :: PeerTxAPI m SigId (Sig crypto)) -> runPipelinedAnnotatedPeerWithLimits @@ -356,7 +351,9 @@ ntnApps $ txSubmissionInboundV2 (Mx.WithBearer connId >$< sigSubmissionInboundTracer) _SIG_SUBMISSION_INIT_DELAY + sigDecisionPolicy mempoolWriter + sigSize peerSigAPI @@ -716,7 +713,7 @@ ntnHandshakeArguments tracer = haHandshakeTracer = tracer , haBearerTracer = nullTracer -- TODO , haHandshakeCodec = codecHandshake nodeToNodeVersionCodec - , haVersionDataCodec = cborTermVersionDataCodec nodeToNodeCodecCBORTerm + , haVersionDataCodec = mkVersionedCodecCBORTerm nodeToNodeCodecCBORTerm , haAcceptVersion = acceptableVersion , haQueryVersion = queryVersion , haTimeLimits = timeLimitsHandshake diff --git a/dmq-node/src/DMQ/Policy.hs b/dmq-node/src/DMQ/Policy.hs index dd2281d..710e0f7 100644 --- a/dmq-node/src/DMQ/Policy.hs +++ b/dmq-node/src/DMQ/Policy.hs @@ -95,13 +95,19 @@ maxSigIdsInflight = 33 -- sigDecisionPolicy :: TxDecisionPolicy sigDecisionPolicy = TxDecisionPolicy { - maxNumTxIdsToRequest = maxSigIdsInflight, - maxUnacknowledgedTxIds = 4 * maxSigIdsInflight, - txsSizeInflightPerPeer = maxSigSize * fromIntegral maxSigIdsInflight, - txInflightMultiplicity = 1, - bufferedTxsMinLifetime = 0, - scoreRate = 0.1, - scoreMax = 15 * 60 + maxNumTxIdsToRequest = maxSigIdsInflight, + maxUnacknowledgedTxIds = 4 * maxSigIdsInflight, + txsSizeInflightPerPeer = maxSigSize * fromIntegral maxSigIdsInflight, + maxOutstandingTxBatchesPerPeer = 4, + txInflightMultiplicity = 1, + bufferedTxsMinLifetime = 0, + scoreRate = 0.1, + scoreMax = 15 * 60, + scoreAcceptDecrement = 3, + interTxSpace = 0.250, + inflightTimeout = 0.600, + maxPeerClaimDelay = 0.250, + disablePipelinedTxIdRequests = False } diff --git a/dmq-node/src/DMQ/Protocol/SigSubmission/Type.hs b/dmq-node/src/DMQ/Protocol/SigSubmission/Type.hs index f449573..77b1b32 100644 --- a/dmq-node/src/DMQ/Protocol/SigSubmission/Type.hs +++ b/dmq-node/src/DMQ/Protocol/SigSubmission/Type.hs @@ -1,14 +1,15 @@ -{-# LANGUAGE DerivingStrategies #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE NamedFieldPuns #-} -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE PatternSynonyms #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE TypeData #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE UndecidableInstances #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TypeData #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE UndecidableInstances #-} module DMQ.Protocol.SigSubmission.Type ( -- * Data types @@ -35,6 +36,7 @@ module DMQ.Protocol.SigSubmission.Type , KESPeriod (..) ) where +import Control.DeepSeq (NFData) import Control.Exception (Exception (..)) import Control.Monad.Class.MonadTime.SI import Data.Aeson @@ -47,6 +49,8 @@ import Data.Time.Clock.POSIX (POSIXTime) import Data.Typeable import Data.Word (Word64) +import NoThunks.Class (NoThunks) + import Cardano.Crypto.DSIGN.Class (DSIGNAlgorithm, VerKeyDSIGN) import Cardano.Crypto.Hash.Blake2b (Blake2b_256) import Cardano.Crypto.Hash.Class (Hash) @@ -61,6 +65,7 @@ import Cardano.Logging qualified as Logging import Ouroboros.Network.Protocol.TxSubmission2.Type as SigSubmission hiding (TxSubmission2) import Ouroboros.Network.Protocol.TxSubmission2.Type as TxSubmission2 +import Ouroboros.Network.Tx (HasRawTxId (..)) import Ouroboros.Network.Util.ShowProxy @@ -73,6 +78,11 @@ type data SigPayload newtype SigId = SigId { getSigId :: Hash Blake2b_256 SigPayload } deriving stock (Show, Eq, Ord) + deriving newtype (NFData, NoThunks) + +instance HasRawTxId SigId where + type RawTxId SigId = SigId + getRawTxId = id instance ToJSON SigId where toJSON (SigId sigId) = toJSON sigId diff --git a/dmq-node/src/DMQ/SigSubmissionV2/Inbound.hs b/dmq-node/src/DMQ/SigSubmissionV2/Inbound.hs index 0743569..2a74477 100644 --- a/dmq-node/src/DMQ/SigSubmissionV2/Inbound.hs +++ b/dmq-node/src/DMQ/SigSubmissionV2/Inbound.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE BangPatterns #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE KindSignatures #-} @@ -14,62 +13,80 @@ module DMQ.SigSubmissionV2.Inbound import Data.Functor.Identity (Identity (..)) import Data.Map.Strict qualified as Map +import Data.Maybe (mapMaybe) import Data.Sequence.Strict qualified as StrictSeq import Data.Set qualified as Set +import Data.Typeable (Typeable) import Control.Concurrent.Class.MonadSTM.Strict -import Control.Exception (assert) import Control.Monad (foldM, unless) import Control.Monad.Class.MonadAsync (MonadAsync (..)) import Control.Monad.Class.MonadThrow import Control.Monad.Class.MonadTime.SI +import Control.Monad.Class.MonadTimer.SI import "contra-tracer" Control.Tracer (Tracer, traceWith) import Network.TypedProtocol -import Ouroboros.Network.ControlMessage (ControlMessage (..), ControlMessageSTM) +import Ouroboros.Network.ControlMessage (ControlMessage (..), ControlMessageSTM, + timeoutWithControlMessage) import Ouroboros.Network.Protocol.TxSubmission2.Type (NumTxIdsToAck (..), NumTxIdsToReq (..)) +import Ouroboros.Network.RegisteredDelay qualified as RegisteredDelay import Ouroboros.Network.TxSubmission.Inbound.V2 (PeerTxAPI (..), - TraceTxSubmissionInbound (..), TxDecision (..), TxsToMempool (..)) -import Ouroboros.Network.TxSubmission.Inbound.V2.Types - (TxSubmissionMempoolWriter (..)) + TxDecisionPolicy) +import Ouroboros.Network.TxSubmission.Inbound.V2.State qualified as State +import Ouroboros.Network.TxSubmission.Inbound.V2.Types (PeerAction (..), + PeerPhase (..), PeerTxLocalState (..), ProcessedTxCount (..), + TraceTxSubmissionInbound (..), TxKey, TxSubmissionCounters (..), + TxSubmissionMempoolWriter (..), TxSubmissionProtocolError (..), + const_MAX_TX_SIZE_DISCREPANCY, diffTimeToMilliseconds, + emptyPeerTxLocalState) import DMQ.Diffusion.PeerSelection.PeerMetric (LocalPeerMetricState, - ReportPeerMetric' (..), ReportPeerMetricI, TxMempoolResult (..)) + ReportPeerMetric' (..), ReportPeerMetricI, SigMempoolResult (..)) import DMQ.Diffusion.PeerSelection.PeerMetric qualified as PeerMetric import DMQ.Protocol.SigSubmissionV2.Inbound -import DMQ.Protocol.SigSubmissionV2.Type (NumIdsAck (NumIdsAck), NumIdsReq (..)) +import DMQ.Protocol.SigSubmissionV2.Type (NumIdsAck (..), NumIdsReq (..), + SizeInBytes) import DMQ.SigSubmissionV2.Types -- | A sig-submission inbound side (client, sic!). -- --- The client blocks on receiving `SigDecision` from the decision logic. If --- there are sig's to download it pipelines two requests: first for sig's second --- for sigid's. If there are no sig's to download, it either sends a blocking or --- non-blocking request for sigid's. -- sigSubmissionInbound :: forall sigid sig idx m failure. - ( MonadThrow m - , MonadAsync m - , Ord sigid, MonadMonotonicTime m + ( MonadAsync m + , MonadThrow m + , MonadTimer m + , Ord sigid + , Show sigid + , Typeable sigid ) => Tracer m (TraceTxSubmissionInbound sigid sig) + -> TxDecisionPolicy -> TxSubmissionMempoolWriter sigid sig idx m failure + -> (sig -> SizeInBytes) -> PeerTxAPI m sigid sig -> ReportPeerMetricI m sigid -> ControlMessageSTM m -> SigSubmissionInboundPipelined sigid sig m () sigSubmissionInbound tracer - TxSubmissionMempoolWriter { txId } + policy + TxSubmissionMempoolWriter { txId, mempoolAddTxs } + sigSize PeerTxAPI { - readTxDecision, - handleReceivedTxIds, - handleReceivedTxs, - submitTxToMempool + awaitSharedChange, + runNextPeerAction, + runNextPeerActionPipelined, + applyReceivedTxIds, + applyReceivedTxs, + applySubmittedTxs, + resolveTxRequest, + resolveBufferedTxs, + addCounters } ReportPeerMetric { reportSigIds, @@ -77,194 +94,294 @@ sigSubmissionInbound } controlMessageSTM = - SigSubmissionInboundPipelined (inboundIdle PeerMetric.emptyLocalPeerMetricState) + SigSubmissionInboundPipelined + (serverIdle emptyPeerTxLocalState PeerMetric.emptyLocalPeerMetricState) where - inboundIdle - :: LocalPeerMetricState sigid - -> m (InboundStIdle Z sigid sig m ()) - inboundIdle localState = do + -- Entry point / reset state of the non-pipelined loop. Called on start, + -- when the pipeline drains, and after an idle peer wakes up. + serverIdle :: PeerTxLocalState sig + -> LocalPeerMetricState sigid + -> m (InboundStIdle Z sigid sig m ()) + serverIdle peerState metricState = do + -- non-blocking check for termination + atomically controlMessageSTM >>= \case + Terminate -> pure (SendMsgDone (return ())) + _ -> do + now <- getMonotonicTime + -- when the pipeline fully drained, emit the body-download episode + -- duration + peerState' <- case peerDownloadStartTime peerState of + Nothing -> pure peerState + Just startTime -> do + addCounters mempty { + txPipelineWaitMs = + diffTimeToMilliseconds (now `diffTime` startTime) + } + pure peerState { peerDownloadStartTime = Nothing } + -- traceCanRequest Zero peerState' + (peerAction, peerState'') <- + runNextPeerAction now (State.drainPeerScore policy now peerState') + case peerAction of + PeerDoNothing generation mDelay -> do + -- An Active->Idle transition means this peer has just become + -- eligible for actions it could not take before (e.g. claiming + -- expired leases as an idle claimant). Re-run the scheduler + -- immediately rather than parking on a wake condition that may + -- not fire. + let cameToIdle = peerPhase peerState' /= PeerIdle + && peerPhase peerState'' == PeerIdle + if cameToIdle + then serverIdle peerState'' metricState + else do + mRegisteredDelay <- traverse RegisteredDelay.new mDelay + timeoutWithControlMessage controlMessageSTM (awaitSharedChange generation mRegisteredDelay) >>= \case + Nothing -> pure (SendMsgDone (return ())) + Just {} -> serverIdle peerState'' metricState + PeerSubmitTxs txKeys -> + submitBufferedSigs txKeys peerState'' metricState serverIdle + PeerRequestTxs txKeys -> + requestSigBodies Zero txKeys peerState'' metricState + PeerRequestTxIds txIdsToAck txIdsToReq -> + serverReqSigIds Zero txIdsToAck txIdsToReq peerState'' metricState - -- NOTE: The `tx-logic` is using `MVar` API, while `controlMessage` an - -- `STM`, we need to compose both. We do that in two steps to avoid - -- Situation where the `tx-logic` is much quicker to reply, and wins the - -- race each time. - k <- - -- 1. non-blocking check of `ControlMessage` - atomically controlMessageSTM >>= \case - Terminate -> return (Left ()) - _ -> - -- 2. race `readTXDecision` and `controlMessageSTM` - atomically (do - cntrlMsg <- controlMessageSTM - case cntrlMsg of - Terminate -> return () - Continue -> retry - Quiesce -> retry - ) - `race` - readTxDecision + -- Submit buffered sig bodies to the mempool, feed the announciness metric, + -- and continue with @k@. + submitBufferedSigs :: forall (n :: N). + [TxKey] + -> PeerTxLocalState sig + -> LocalPeerMetricState sigid + -> (PeerTxLocalState sig -> LocalPeerMetricState sigid + -> m (InboundStIdle n sigid sig m ())) + -> m (InboundStIdle n sigid sig m ()) + submitBufferedSigs txKeys peerState metricState k = do + bufferedSigs <- resolveBufferedTxs peerState txKeys - case k of - Left{} -> pure (SendMsgDone $ return ()) - Right sigd@TxDecision - { txdTxsToRequest = sigsToRequest - , txdTxsToMempool = TxsToMempool { listOfTxsToMempool } - } -> do - traceWith tracer (TraceTxInboundDecision sigd) + start <- getMonotonicTime + let submitted = [ (txKey, sigid') | (txKey, sigid', _) <- bufferedSigs ] + toSubmit = [ sig | (_, _, sig) <- bufferedSigs ] + (acceptedSigIds, rejectedSigs) <- if null toSubmit + then pure ([], []) + else mempoolAddTxs toSubmit + end <- getMonotonicTime - let !collected = length listOfTxsToMempool + -- 'mempoolAddTxs' partitions the batch into accepted/rejected; map each + -- verdict's sigids back to our 'TxKey's via the submitted batch. + let submittedKeyOf = Map.fromList [ (sigid', txKey) | (txKey, sigid') <- submitted ] + resolvedKeys = mapMaybe (`Map.lookup` submittedKeyOf) acceptedSigIds + rejectedKeys = mapMaybe ((`Map.lookup` submittedKeyOf) . fst) rejectedSigs + rejectedForTrace = fmap fst rejectedSigs + acceptedCount = length acceptedSigIds + rejectedCount = length rejectedSigs + delta = end `diffTime` start - -- Only attempt to add sigs if we have some work to do - localState' <- - if collected > 0 - then do - -- submitTxToMempool traces: - -- * `TraceTxSubmissionProcessed`, - -- * `TraceTxInboundAddedToMempool`, and - -- * `TraceTxInboundRejectedFromMempool` - -- events. - foldM (\st (sigid, sig) -> do - r <- submitTxToMempool tracer sigid sig - atomically $ reportSig st (Identity (sigid, r)) - ) - localState - listOfTxsToMempool - else return localState + addCounters mempty { txSubmissionWaitMs = diffTimeToMilliseconds delta } + peerState' <- applySubmittedTxs end resolvedKeys rejectedKeys peerState + let (score, peerState'') = + State.applyPeerEvents policy end acceptedCount rejectedCount peerState' + traceWith tracer $ + TraceTxSubmissionProcessed ProcessedTxCount { + ptxcAccepted = acceptedCount, + ptxcRejected = rejectedCount, + ptxcScore = score + } + unless (null acceptedSigIds) $ + traceWith tracer (TraceTxInboundAddedToMempool acceptedSigIds delta) + unless (null rejectedForTrace) $ + traceWith tracer (TraceTxInboundRejectedFromMempool rejectedForTrace delta) - -- TODO: - -- We can update the state so that other `sig-submission` servers will - -- not try to add these sigs to the mempool. - if Map.null sigsToRequest - then clientReqSigIds Zero localState' sigd - else clientReqSigs localState' sigd + -- Bridge the batch verdict into DMQ's per-sig announciness metric. + metricState'' <- atomically $ do + metricState' + <- foldM (\st sigid' -> reportSig st (Identity (sigid', SigAccepted))) + metricState acceptedSigIds + let rejectedSigIds = fst <$> rejectedSigs + foldM (\st sigid' -> reportSig st (Identity (sigid', SigRejected))) + metricState' rejectedSigIds + k peerState'' metricState'' + -- Send a pipelined request for sig bodies. + requestSigBodies :: forall (n :: N). + Nat n + -> [TxKey] + -> PeerTxLocalState sig + -> LocalPeerMetricState sigid + -> m (InboundStIdle n sigid sig m ()) + requestSigBodies n txKeys peerState metricState = do + sigsToRequest <- resolveTxRequest peerState txKeys + traceWith tracer (TraceTxInboundRequestTxs (Map.keys sigsToRequest)) - -- Pipelined request of sigs - clientReqSigs :: LocalPeerMetricState sigid - -> TxDecision sigid sig - -> m (InboundStIdle Z sigid sig m ()) - clientReqSigs localState sigd@TxDecision { txdTxsToRequest = sigdSigsToRequest } = - pure $ SendMsgRequestSigsPipelined sigdSigsToRequest - (clientReqSigIds (Succ Zero) localState sigd) + -- Record the start of the download episode on the first outstanding + -- body request. Subsequent pipelined requests leave the start time + -- unchanged so we measure from first-send to last-receive. + sendTime <- getMonotonicTime + let peerState' = case peerDownloadStartTime peerState of + Nothing -> peerState { peerDownloadStartTime = Just sendTime } + Just _ -> peerState + pure $ SendMsgRequestSigsPipelined sigsToRequest + (continueAfterBodyRequests (Succ n) peerState' metricState) - clientReqSigIds :: forall (n :: N). - Nat n - -> LocalPeerMetricState sigid - -> TxDecision sigid sig - -> m (InboundStIdle n sigid sig m ()) - clientReqSigIds - n localState TxDecision { txdTxIdsToRequest = 0 } - = - case n of - Zero -> inboundIdle localState - Succ _ -> handleReplies n localState + -- Post-reply scheduling; dispatch the next action or drain the pipeline. + continueAfterReplies :: forall (n :: N). + Nat n + -> PeerTxLocalState sig + -> LocalPeerMetricState sigid + -> m (InboundStIdle n sigid sig m ()) + continueAfterReplies Zero peerState metricState = serverIdle peerState metricState + continueAfterReplies n@Succ{} peerState metricState = + pipelinedDispatch n peerState metricState - clientReqSigIds - -- if there are no unacknowledged sigids, the protocol requires sending - -- a blocking `MsgRequestSigIds` request. This is important, as otherwise - -- the client side wouldn't have a chance to terminate the - -- mini-protocol. - Zero - localState - TxDecision { txdTxIdsToAcknowledge = sigIdsToAck, - txdPipelineTxIds = False, - txdTxIdsToRequest = sigIdsToReq - } - = - pure $ SendMsgRequestSigIdsBlocking - (NumIdsAck . getNumTxIdsToAck $ sigIdsToAck) - (NumIdsReq . getNumTxIdsToReq $ sigIdsToReq) - (\sigids -> do - time <- getMonotonicTime - let sigidsLst = fst <$> sigids - sigidsSeq = StrictSeq.fromList sigidsLst - sigidsMap = Map.fromList sigids - unless (StrictSeq.length sigidsSeq <= fromIntegral sigIdsToReq) $ - throwIO ProtocolErrorSigIdsNotRequested - let localState' = reportSigIds sigidsLst time localState - handleReceivedTxIds sigIdsToReq sigidsSeq sigidsMap - inboundIdle localState' - ) + -- Post-body-request scheduling; same dispatch, guaranteed pipelined depth. + continueAfterBodyRequests :: forall (n :: N). + Nat (S n) + -> PeerTxLocalState sig + -> LocalPeerMetricState sigid + -> m (InboundStIdle (S n) sigid sig m ()) + continueAfterBodyRequests = pipelinedDispatch - clientReqSigIds - n@Zero - localState - TxDecision { txdTxIdsToAcknowledge = sigIdsToAck, - txdPipelineTxIds = True, - txdTxIdsToRequest = sigIdsToReq - } - = - pure $ SendMsgRequestSigIdsPipelined - (NumIdsAck . getNumTxIdsToAck $ sigIdsToAck) - (NumIdsReq . getNumTxIdsToReq $ sigIdsToReq) - (handleReplies (Succ n) localState) + pipelinedDispatch :: forall (n :: N). + Nat (S n) + -> PeerTxLocalState sig + -> LocalPeerMetricState sigid + -> m (InboundStIdle (S n) sigid sig m ()) + pipelinedDispatch n peerState metricState = do + now <- getMonotonicTime + (peerAction, peerState') <- + runNextPeerActionPipelined now (State.drainPeerScore policy now peerState) + case peerAction of + PeerSubmitTxs txKeys -> + submitBufferedSigs txKeys peerState' metricState (continueAfterReplies n) + PeerRequestTxs txKeys -> + requestSigBodies n txKeys peerState' metricState + PeerRequestTxIds txIdsToAck txIdsToReq -> + serverReqSigIds n txIdsToAck txIdsToReq peerState' metricState + PeerDoNothing{} -> + pure $ handleReplies n peerState' metricState - clientReqSigIds - n@Succ{} - localState - TxDecision { txdTxIdsToAcknowledge = sigIdsToAck, - txdPipelineTxIds, - txdTxIdsToRequest = sigIdsToReq - } - = - -- it is impossible that we have had `sig`'s to request (Succ{} - is an - -- evidence for that), but no unacknowledged `sigid`s. - assert txdPipelineTxIds $ - pure $ SendMsgRequestSigIdsPipelined - (NumIdsAck . getNumTxIdsToAck $ sigIdsToAck) - (NumIdsReq . getNumTxIdsToReq $ sigIdsToReq) - (handleReplies (Succ n) localState) + -- Construct and send a sigid request (blocking or pipelined). + serverReqSigIds :: forall (n :: N). + Nat n + -> NumTxIdsToAck + -> NumTxIdsToReq + -> PeerTxLocalState sig + -> LocalPeerMetricState sigid + -> m (InboundStIdle n sigid sig m ()) + -- nothing to ack or request: back to idle + serverReqSigIds Zero 0 0 peerState metricState = + serverIdle peerState metricState + -- nothing to do but pipeline not empty: drain remaining replies + serverReqSigIds n@Succ{} 0 0 peerState metricState = + pure $ handleReplies n peerState metricState + -- non-pipelined; may send a blocking request + serverReqSigIds Zero txIdsToAck txIdsToReq peerState metricState = + if StrictSeq.null (peerUnacknowledgedTxIds peerState) + then do + sendTime <- getMonotonicTime + addCounters mempty { txIdBlockingReqsSent = 1 } + pure $ SendMsgRequestSigIdsBlocking + (NumIdsAck (getNumTxIdsToAck txIdsToAck)) + (NumIdsReq (getNumTxIdsToReq txIdsToReq)) + (\sigids -> do + now <- getMonotonicTime + addCounters mempty { + txIdBlockingWaitMs = + diffTimeToMilliseconds (now `diffTime` sendTime) + } + unless (length sigids <= fromIntegral txIdsToReq) $ + throwIO ProtocolErrorSigIdsNotRequested + let metricState' = reportSigIds (fst <$> sigids) now metricState + peerState' <- applyReceivedTxIds now txIdsToReq sigids peerState + serverIdle peerState' metricState') + else do + addCounters mempty { txIdPipelinedReqsSent = 1 } + pure $ SendMsgRequestSigIdsPipelined + (NumIdsAck (getNumTxIdsToAck txIdsToAck)) + (NumIdsReq (getNumTxIdsToReq txIdsToReq)) + (pure $ handleReplies (Succ Zero) peerState metricState) + + -- pipelined request at depth > 0 + serverReqSigIds n@Succ{} txIdsToAck txIdsToReq peerState metricState = do + addCounters mempty { txIdPipelinedReqsSent = 1 } + pure $ SendMsgRequestSigIdsPipelined + (NumIdsAck (getNumTxIdsToAck txIdsToAck)) + (NumIdsReq (getNumTxIdsToReq txIdsToReq)) + (pure $ handleReplies (Succ n) peerState metricState) + + -- Prepare to collect a single pipelined reply. handleReplies :: forall (n :: N). Nat (S n) + -> PeerTxLocalState sig -> LocalPeerMetricState sigid - -> m (InboundStIdle (S n) sigid sig m ()) - handleReplies (Succ n'@Succ{}) localState = - pure $ CollectPipelined - Nothing - (handleReply localState (handleReplies n')) - - handleReplies (Succ Zero) localState = - pure $ CollectPipelined - Nothing - (handleReply localState inboundIdle) + -> InboundStIdle (S n) sigid sig m () + handleReplies (Succ n) peerState metricState = + CollectPipelined Nothing (handleReply n peerState metricState) + -- Process a single pipelined reply. handleReply :: forall (n :: N). - LocalPeerMetricState sigid - -> (LocalPeerMetricState sigid -> m (InboundStIdle n sigid sig m ())) - -- continuation + Nat n + -> PeerTxLocalState sig + -> LocalPeerMetricState sigid -> Collect sigid sig -> m (InboundStIdle n sigid sig m ()) - handleReply localState k = \case + handleReply n peerState metricState = \case CollectSigIds sigIdsToReq sigids -> do - time <- getMonotonicTime - let sigidsLst = fst <$> sigids - sigidsSeq = StrictSeq.fromList sigidsLst - sigidsMap = Map.fromList sigids - unless (StrictSeq.length sigidsSeq <= fromIntegral sigIdsToReq) $ + unless (length sigids <= fromIntegral (getNumIdsReq sigIdsToReq)) $ throwIO ProtocolErrorSigIdsNotRequested - let localState' = reportSigIds sigidsLst time localState - handleReceivedTxIds (NumTxIdsToReq . getNumIdsReq $ sigIdsToReq) sigidsSeq sigidsMap - k localState' - - CollectSigs sigids sigs -> do - let requested = Map.keysSet sigids - received = Map.fromList [ (txId sig, sig) | sig <- sigs ] - notReceived = requested Set.\\ Map.keysSet received + now <- getMonotonicTime + let metricState' = reportSigIds (fst <$> sigids) now metricState + peerState' <- + applyReceivedTxIds now + (NumTxIdsToReq (getNumIdsReq sigIdsToReq)) sigids peerState + continueAfterReplies n peerState' metricState' - unless (Map.keysSet received `Set.isSubsetOf` requested) $ + CollectSigs requested sigs -> do + let received = Map.fromList [ (txId sig, sig) | sig <- sigs ] + wrongSizedSigs = collectWrongSizedSigs requested received + unless (Map.keysSet received `Set.isSubsetOf` Map.keysSet requested) $ throwIO ProtocolErrorSigNotRequested + traceWith tracer $ TraceTxSubmissionCollected (txId <$> sigs) + unless (null wrongSizedSigs) $ do + let protocolError = ProtocolErrorTxSizeError wrongSizedSigs + traceWith tracer (TraceTxInboundError protocolError) + throwIO protocolError + now <- getMonotonicTime + (penaltyCount, peerState') <- + applyReceivedTxs now [ (txId sig, sig) | sig <- sigs ] peerState + peerState'' <- + if penaltyCount == 0 + then pure peerState' + else do + let (score, ps) = State.applyPeerEvents policy now 0 penaltyCount peerState' + traceWith tracer $ + TraceTxSubmissionProcessed ProcessedTxCount { + ptxcAccepted = 0, + ptxcRejected = penaltyCount, + ptxcScore = score + } + pure ps + -- sigs that were requested but not received count as rejected for the + -- announciness metric (mirrors the old @CollectSigs@ handling). + let notReceived = Map.keysSet requested Set.\\ Map.keysSet received + metricState' <- atomically $ + foldM (\st sigid -> reportSig st (Identity (sigid, SigRejected))) + metricState (Set.toList notReceived) + continueAfterReplies n peerState'' metricState' - localState' <- atomically do - foldM (\st sigid -> reportSig st (Identity (sigid, TxRejected))) - localState - (Set.toList notReceived) + -- Sigs whose received size does not match the advertised size (within a + -- fuzzy tolerance). + collectWrongSizedSigs :: Map.Map sigid SizeInBytes + -> Map.Map sigid sig + -> [(sigid, SizeInBytes, SizeInBytes)] + collectWrongSizedSigs requested received = + [ (sigid', receivedSize, advertisedSize) + | (sigid', sig) <- Map.toList received + , let receivedSize = sigSize sig + , Just advertisedSize <- [Map.lookup sigid' requested] + , not (checkSigSize receivedSize advertisedSize) + ] - mbe <- handleReceivedTxs sigids received - traceWith tracer $ TraceTxSubmissionCollected (txId `map` sigs) - case mbe of - -- one of `sig`s had a wrong size - Just e -> traceWith tracer (TraceTxInboundError e) - >> throwIO e - Nothing -> k localState' + checkSigSize :: SizeInBytes -> SizeInBytes -> Bool + checkSigSize received advertised + | received > advertised = received - advertised <= const_MAX_TX_SIZE_DISCREPANCY + | otherwise = advertised - received <= const_MAX_TX_SIZE_DISCREPANCY diff --git a/dmq-node/src/DMQ/Tracer.hs b/dmq-node/src/DMQ/Tracer.hs index 93684ac..7ba27a8 100644 --- a/dmq-node/src/DMQ/Tracer.hs +++ b/dmq-node/src/DMQ/Tracer.hs @@ -17,9 +17,9 @@ module DMQ.Tracer , Diffusion.NoExtraState (..) , Diffusion.NoExtraDebugState (..) , Diffusion.NoExtraFlags (..) - , NoExtraConfig (..) - , NoExtraAPI (..) - , NoExtraChurnArgs (..) + , Diffusion.NoExtraConfig (..) + , Diffusion.NoExtraAPI (..) + , Diffusion.NoExtraChurnArgs (..) ) where import Codec.CBOR.Term (Term) @@ -60,7 +60,7 @@ import Ouroboros.Network.Protocol.TxSubmission2.Type qualified as STX import Ouroboros.Network.Snocket (RemoteAddress) import Ouroboros.Network.TxSubmission.Inbound.V2 (TraceTxLogic) import Ouroboros.Network.TxSubmission.Inbound.V2.Types - (TraceTxSubmissionInbound) + (TraceTxSubmissionInbound, TxSubmissionCounters) import Ouroboros.Network.TxSubmission.Outbound (TraceTxSubmissionOutbound) import Cardano.KESAgent.KES.Crypto (Crypto) @@ -93,6 +93,8 @@ import DMQ.SigSubmissionV2.Outbound qualified as SigSubV2 data DMQTracers crypto ntnAddr ntcAddr m = DMQTracers { sigSubmissionLogicTracer :: Tracer m (TraceTxLogic ntnAddr SigId (Sig crypto)), + sigCountersTracer + :: Tracer m TxSubmissionCounters, sigSubmissionLogicPeerTracer :: Tracer m (Mx.WithBearer (ConnectionId ntnAddr) (TraceTxLogic ntnAddr SigId (Sig crypto))), localMsgSubmissionProtocolTracer @@ -255,7 +257,7 @@ mkDMQTracers ekgStore dmqConfigFilePath = do Logging.Info Logging.DNormal (Logging.Stdout Logging.MachineFormat) - dmqConfigFilePath + (Logging.FromFile dmqConfigFilePath) Logging.emptyTraceConfig { Logging.tcMetricsPrefix = Just metricsPrefix } else @@ -273,291 +275,297 @@ mkDMQTracers ekgStore dmqConfigFilePath = do let trForward = mempty mbTrEkg = Just ekgTrace - !dtMuxTracer <- mkTracer traceConfig configReflection + !dtMuxTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Mux", "Remote"] - !dtChannelTracer <- mkTracer + !dtChannelTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Mux", "Remote", "Channel"] - !dtBearerTracer <- mkTracer + !dtBearerTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Mux", "Remote", "Bearer"] - !dtHandshakeTracer <- mkTracer + !dtHandshakeTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Handshake", "Remote"] - !dtLocalMuxTracer <- mkTracer + !dtLocalMuxTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Mux", "Local"] - !dtLocalChannelTracer <- mkTracer + !dtLocalChannelTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Mux", "Local", "Channel"] - !dtLocalBearerTracer <- mkTracer + !dtLocalBearerTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Mux", "Local", "Bearer"] - !dtLocalHandshakeTracer <- mkTracer + !dtLocalHandshakeTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Handshake", "Local"] - !dtDiffusionTracer <- mkTracer + !dtDiffusionTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Startup", "DiffusionInit"] - !dtTraceLocalRootPeersTracer <- mkTracer + !dtTraceLocalRootPeersTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Peers", "LocalRoot"] - !dtTracePublicRootPeersTracer <- mkTracer + !dtTracePublicRootPeersTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Peers", "PublicRoot"] - !dtTraceLedgerPeersTracer <- mkTracer + !dtTraceLedgerPeersTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Peers", "Ledger"] - !dtTracePeerSelectionTracer <- mkTracer + !dtTracePeerSelectionTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "PeerSelection", "Selection"] - !dtDebugPeerSelectionTracer <- mkTracer + !dtDebugPeerSelectionTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "PeerSelection"] - !dtTracePeerSelectionCounters <- mkTracer + !dtTracePeerSelectionCounters <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "PeerSelection"] - !dtPeerSelectionActionsTracer <- mkTracer + !dtPeerSelectionActionsTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "PeerSelection", "Actions"] - !dtConnectionManagerTracer <- mkTracer + !dtConnectionManagerTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "ConnectionManager", "Remote"] - !dtConnectionManagerTransitionTracer <- mkTracer + !dtConnectionManagerTransitionTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "ConnectionManager", "Transition"] - !dtServerTracer <- mkTracer + !dtServerTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Server", "Local"] - !dtInboundGovernorTracer <- mkTracer + !dtInboundGovernorTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "InboundGovernor", "Remote"] - !dtInboundGovernorTransitionTracer <- mkTracer + !dtInboundGovernorTransitionTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "InboundGovernor", "Transition"] - !dtLocalConnectionManagerTracer <- mkTracer + !dtLocalConnectionManagerTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward Nothing -- never conflate metrics of the same name with those originating from `connectionManagerTr` ["Net", "ConnectionManager", "Local"] - !dtLocalServerTracer <- mkTracer + !dtLocalServerTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Server", "Local"] - !dtLocalInboundGovernorTracer <- mkTracer + !dtLocalInboundGovernorTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "InboundGovernor", "Local"] - !dtDnsTracer <- mkTracer + !dtDnsTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "DNS"] let dmqDifussionTracers = Diffusion.Tracers { - Diffusion.dtMuxTracer = Tracer $ Logging.traceWith dtMuxTracer, - Diffusion.dtChannelTracer = Tracer $ Logging.traceWith dtChannelTracer, - Diffusion.dtBearerTracer = Tracer $ Logging.traceWith dtBearerTracer, - Diffusion.dtHandshakeTracer = Tracer $ Logging.traceWith dtHandshakeTracer, - Diffusion.dtLocalMuxTracer = Tracer $ Logging.traceWith dtLocalMuxTracer, - Diffusion.dtLocalChannelTracer = Tracer $ Logging.traceWith dtLocalChannelTracer, - Diffusion.dtLocalBearerTracer = Tracer $ Logging.traceWith dtLocalBearerTracer, - Diffusion.dtLocalHandshakeTracer = Tracer $ Logging.traceWith dtLocalHandshakeTracer, - Diffusion.dtDiffusionTracer = Tracer $ Logging.traceWith dtDiffusionTracer, - Diffusion.dtTraceLocalRootPeersTracer = Tracer $ Logging.traceWith dtTraceLocalRootPeersTracer, - Diffusion.dtTracePublicRootPeersTracer = Tracer $ Logging.traceWith dtTracePublicRootPeersTracer, - Diffusion.dtTraceLedgerPeersTracer = Tracer $ Logging.traceWith dtTraceLedgerPeersTracer, - Diffusion.dtTracePeerSelectionTracer = Tracer $ Logging.traceWith dtTracePeerSelectionTracer, - Diffusion.dtDebugPeerSelectionTracer = Tracer $ Logging.traceWith dtDebugPeerSelectionTracer, - Diffusion.dtTracePeerSelectionCounters = Tracer $ Logging.traceWith dtTracePeerSelectionCounters, - Diffusion.dtPeerSelectionActionsTracer = Tracer $ Logging.traceWith dtPeerSelectionActionsTracer, - Diffusion.dtConnectionManagerTracer = Tracer $ Logging.traceWith dtConnectionManagerTracer, - Diffusion.dtConnectionManagerTransitionTracer = Tracer $ Logging.traceWith dtConnectionManagerTransitionTracer, - Diffusion.dtServerTracer = Tracer $ Logging.traceWith dtServerTracer, - Diffusion.dtInboundGovernorTracer = Tracer $ Logging.traceWith dtInboundGovernorTracer, - Diffusion.dtInboundGovernorTransitionTracer = Tracer $ Logging.traceWith dtInboundGovernorTransitionTracer, - Diffusion.dtLocalConnectionManagerTracer = Tracer $ Logging.traceWith dtLocalConnectionManagerTracer, - Diffusion.dtLocalServerTracer = Tracer $ Logging.traceWith dtLocalServerTracer, - Diffusion.dtLocalInboundGovernorTracer = Tracer $ Logging.traceWith dtLocalInboundGovernorTracer, - Diffusion.dtDnsTracer = Tracer $ Logging.traceWith dtDnsTracer + Diffusion.dtMuxTracer = mkTracer $ Logging.traceWith dtMuxTracer, + Diffusion.dtChannelTracer = mkTracer $ Logging.traceWith dtChannelTracer, + Diffusion.dtBearerTracer = mkTracer $ Logging.traceWith dtBearerTracer, + Diffusion.dtHandshakeTracer = mkTracer $ Logging.traceWith dtHandshakeTracer, + Diffusion.dtLocalMuxTracer = mkTracer $ Logging.traceWith dtLocalMuxTracer, + Diffusion.dtLocalChannelTracer = mkTracer $ Logging.traceWith dtLocalChannelTracer, + Diffusion.dtLocalBearerTracer = mkTracer $ Logging.traceWith dtLocalBearerTracer, + Diffusion.dtLocalHandshakeTracer = mkTracer $ Logging.traceWith dtLocalHandshakeTracer, + Diffusion.dtDiffusionTracer = mkTracer $ Logging.traceWith dtDiffusionTracer, + Diffusion.dtTraceLocalRootPeersTracer = mkTracer $ Logging.traceWith dtTraceLocalRootPeersTracer, + Diffusion.dtTracePublicRootPeersTracer = mkTracer $ Logging.traceWith dtTracePublicRootPeersTracer, + Diffusion.dtTraceLedgerPeersTracer = mkTracer $ Logging.traceWith dtTraceLedgerPeersTracer, + Diffusion.dtTracePeerSelectionTracer = mkTracer $ Logging.traceWith dtTracePeerSelectionTracer, + Diffusion.dtDebugPeerSelectionTracer = mkTracer $ Logging.traceWith dtDebugPeerSelectionTracer, + Diffusion.dtTracePeerSelectionCounters = mkTracer $ Logging.traceWith dtTracePeerSelectionCounters, + Diffusion.dtPeerSelectionActionsTracer = mkTracer $ Logging.traceWith dtPeerSelectionActionsTracer, + Diffusion.dtConnectionManagerTracer = mkTracer $ Logging.traceWith dtConnectionManagerTracer, + Diffusion.dtConnectionManagerTransitionTracer = mkTracer $ Logging.traceWith dtConnectionManagerTransitionTracer, + Diffusion.dtServerTracer = mkTracer $ Logging.traceWith dtServerTracer, + Diffusion.dtInboundGovernorTracer = mkTracer $ Logging.traceWith dtInboundGovernorTracer, + Diffusion.dtInboundGovernorTransitionTracer = mkTracer $ Logging.traceWith dtInboundGovernorTransitionTracer, + Diffusion.dtLocalConnectionManagerTracer = mkTracer $ Logging.traceWith dtLocalConnectionManagerTracer, + Diffusion.dtLocalServerTracer = mkTracer $ Logging.traceWith dtLocalServerTracer, + Diffusion.dtLocalInboundGovernorTracer = mkTracer $ Logging.traceWith dtLocalInboundGovernorTracer, + Diffusion.dtDnsTracer = mkTracer $ Logging.traceWith dtDnsTracer } - !sigSubmissionLogicTracer <- mkTracer + !sigSubmissionLogicTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "SigSubmission", "Logic"] - !sigSubmissionLogicPeerTracer <- mkTracer + !sigCountersTracer <- mkLoggingTracer + traceConfig configReflection + stdoutTrace trForward mbTrEkg + ["Net", "SigSubmission", "Counters"] + + !sigSubmissionLogicPeerTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "SigSubmission", "Logic"] - !localMsgSubmissionProtocolTracer <- mkTracer + !localMsgSubmissionProtocolTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "SigSubmission", "Protocol"] - !localMsgSubmissionServerTracer <- mkTracer + !localMsgSubmissionServerTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "SigSubmission", "Server"] - !localMsgNotificationProtocolTracer <- mkTracer + !localMsgNotificationProtocolTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "MsgNotification", "Protocol"] - !localMsgNotificationServerTracer <- mkTracer + !localMsgNotificationServerTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "MsgNotification", "Server"] - !sigSubmissionV2ProtocolTracer <- mkTracer + !sigSubmissionV2ProtocolTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "SigSubmission", "V2", "Protocol"] - !sigSubmissionInboundTracer <- mkTracer + !sigSubmissionInboundTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "SigSubmission", "Inbound"] - !sigSubmissionV1ProtocolTracer <- mkTracer + !sigSubmissionV1ProtocolTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "SigSubmission", "V1", "Protocol"] - !sigSubmissionOutboundV1Tracer <- mkTracer + !sigSubmissionOutboundV1Tracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "SigSubmission", "V1", "Outbound"] - !sigSubmissionOutboundV2Tracer <- mkTracer + !sigSubmissionOutboundV2Tracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "SigSubmission", "V2", "Outbound"] - !keepAliveProtocolTracer <- mkTracer + !keepAliveProtocolTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "KeepAlive", "Protocol"] - !peerSharingProtocolTracer <- mkTracer + !peerSharingProtocolTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "PeerShare", "Protocol"] - !dmqStartupTracer' <- mkTracer + !dmqStartupTracer' <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Startup"] - !localStateQueryClientTracer <- mkTracer + !localStateQueryClientTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "LocalStateQuery"] - !sigValidationTracer <- mkTracer + !sigValidationTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Sig", "Validation"] - !localSigValidationTracer <- mkTracer + !localSigValidationTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "Sig", "Validation"] - !cardanoNodeHandshakeTracer <- mkTracer + !cardanoNodeHandshakeTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "Cardano", "Handshake"] - !cardanoNodeHandshakeProtocolTracer <- mkTracer + !cardanoNodeHandshakeProtocolTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "Cardano", "Handshake", "Protocol"] - !cardanoNodeMuxTracer <- mkTracer + !cardanoNodeMuxTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "Cardano", "Mux", "Local"] - !cardanoNodeChannelTracer <- mkTracer + !cardanoNodeChannelTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "Cardano", "Mux", "Local", "Channel"] - !cardanoNodeBearerTracer <- mkTracer + !cardanoNodeBearerTracer <- mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg ["Net", "Local", "Cardano", "Mux", "Local", "Bearer"] let dmqTracers = DMQTracers { - sigSubmissionLogicTracer = Tracer $ Logging.traceWith sigSubmissionLogicTracer, - sigSubmissionLogicPeerTracer = Tracer $ Logging.traceWith sigSubmissionLogicPeerTracer, - localMsgSubmissionProtocolTracer = Tracer $ Logging.traceWith localMsgSubmissionProtocolTracer, - localMsgSubmissionServerTracer = Tracer $ Logging.traceWith localMsgSubmissionServerTracer, - localMsgNotificationProtocolTracer = Tracer $ Logging.traceWith localMsgNotificationProtocolTracer, - localMsgNotificationServerTracer = Tracer $ Logging.traceWith localMsgNotificationServerTracer, - sigSubmissionV2ProtocolTracer = Tracer $ Logging.traceWith sigSubmissionV2ProtocolTracer, - sigSubmissionInboundTracer = Tracer $ Logging.traceWith sigSubmissionInboundTracer, - sigSubmissionV1ProtocolTracer = Tracer $ Logging.traceWith sigSubmissionV1ProtocolTracer, - sigSubmissionOutboundV1Tracer = Tracer $ Logging.traceWith sigSubmissionOutboundV1Tracer, - sigSubmissionOutboundV2Tracer = Tracer $ Logging.traceWith sigSubmissionOutboundV2Tracer, - keepAliveProtocolTracer = Tracer $ Logging.traceWith keepAliveProtocolTracer, - peerSharingProtocolTracer = Tracer $ Logging.traceWith peerSharingProtocolTracer, - dmqStartupTracer = Tracer $ Logging.traceWith dmqStartupTracer', - localStateQueryClientTracer = Tracer $ Logging.traceWith localStateQueryClientTracer, - sigValidationTracer = Tracer $ Logging.traceWith sigValidationTracer, - localSigValidationTracer = Tracer $ Logging.traceWith localSigValidationTracer, - 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 + sigSubmissionLogicTracer = mkTracer $ Logging.traceWith sigSubmissionLogicTracer, + sigCountersTracer = mkTracer $ Logging.traceWith sigCountersTracer, + sigSubmissionLogicPeerTracer = mkTracer $ Logging.traceWith sigSubmissionLogicPeerTracer, + localMsgSubmissionProtocolTracer = mkTracer $ Logging.traceWith localMsgSubmissionProtocolTracer, + localMsgSubmissionServerTracer = mkTracer $ Logging.traceWith localMsgSubmissionServerTracer, + localMsgNotificationProtocolTracer = mkTracer $ Logging.traceWith localMsgNotificationProtocolTracer, + localMsgNotificationServerTracer = mkTracer $ Logging.traceWith localMsgNotificationServerTracer, + sigSubmissionV2ProtocolTracer = mkTracer $ Logging.traceWith sigSubmissionV2ProtocolTracer, + sigSubmissionInboundTracer = mkTracer $ Logging.traceWith sigSubmissionInboundTracer, + sigSubmissionV1ProtocolTracer = mkTracer $ Logging.traceWith sigSubmissionV1ProtocolTracer, + sigSubmissionOutboundV1Tracer = mkTracer $ Logging.traceWith sigSubmissionOutboundV1Tracer, + sigSubmissionOutboundV2Tracer = mkTracer $ Logging.traceWith sigSubmissionOutboundV2Tracer, + keepAliveProtocolTracer = mkTracer $ Logging.traceWith keepAliveProtocolTracer, + peerSharingProtocolTracer = mkTracer $ Logging.traceWith peerSharingProtocolTracer, + dmqStartupTracer = mkTracer $ Logging.traceWith dmqStartupTracer', + localStateQueryClientTracer = mkTracer $ Logging.traceWith localStateQueryClientTracer, + sigValidationTracer = mkTracer $ Logging.traceWith sigValidationTracer, + localSigValidationTracer = mkTracer $ Logging.traceWith localSigValidationTracer, + cardanoNodeHandshakeTracer = mkTracer $ Logging.traceWith cardanoNodeHandshakeTracer, + cardanoNodeHandshakeProtocolTracer = mkTracer $ Logging.traceWith cardanoNodeHandshakeProtocolTracer, + cardanoNodeMuxTracer = mkTracer $ Logging.traceWith cardanoNodeMuxTracer, + cardanoNodeChannelTracer = mkTracer $ Logging.traceWith cardanoNodeChannelTracer, + cardanoNodeBearerTracer = mkTracer $ Logging.traceWith cardanoNodeBearerTracer } -- This backend can only be used globally, i.e. will always apply to the namespace root. @@ -575,7 +583,7 @@ mkDMQTracers ekgStore dmqConfigFilePath = do -- | Create and configure a tracer. -- -mkTracer :: Logging.LogFormatting a +mkLoggingTracer :: Logging.LogFormatting a => Logging.MetaTrace a => Logging.TraceConfig -> Logging.ConfigReflection @@ -584,7 +592,7 @@ mkTracer :: Logging.LogFormatting a -> Maybe (Logging.Trace IO Logging.FormattedMessage) -> [Text] -> IO (Logging.Trace IO a) -mkTracer traceConfig configReflection stdoutTrace trForward mbTrEkg as = do +mkLoggingTracer traceConfig configReflection stdoutTrace trForward mbTrEkg as = do tracer <- Logging.mkCardanoTracer stdoutTrace trForward mbTrEkg as Logging.configureTracers configReflection traceConfig [tracer] return tracer @@ -1102,9 +1110,6 @@ instance Logging.MetaTrace (AnyMessage (PS.PeerSharing addr)) where , Namespace [] ["PeerShareDone"] ] --- An orphan instance needed for `Handshake versionNumber Term` -instance ToJSON Term where - toJSON term = String (Text.pack . show $ term) instance ToJSON (PublicRootPeers (Diffusion.NoExtraPeers RemoteAddress) RemoteAddress) where toJSON prp = @@ -1112,12 +1117,3 @@ instance ToJSON (PublicRootPeers (Diffusion.NoExtraPeers RemoteAddress) RemoteAd , "ledgerPeers" .= PublicRootPeers.getLedgerPeers prp , "bigLedgerPeers" .= PublicRootPeers.getBigLedgerPeers prp ] - --- TODO: move to this instance, `NoExtraChurnArgs` and `NoExtraAPI` to --- `Ouroboros.Network.Diffusion.Types`. - -instance ToJSON Diffusion.NoExtraDebugState where - toJSON _ = Null - omitField _ = True -data NoExtraChurnArgs = NoExtraChurnArgs -data NoExtraAPI = NoExtraAPI diff --git a/dmq-node/test/DMQ/Protocol/LocalMsgNotification/Test.hs b/dmq-node/test/DMQ/Protocol/LocalMsgNotification/Test.hs index df894ae..c79be4a 100644 --- a/dmq-node/test/DMQ/Protocol/LocalMsgNotification/Test.hs +++ b/dmq-node/test/DMQ/Protocol/LocalMsgNotification/Test.hs @@ -45,7 +45,7 @@ import Ouroboros.Network.Protocol.Codec.Utils qualified as Utils import Test.Ouroboros.Network.Protocol.Utils import Test.Ouroboros.Network.Utils -testTracer :: ({-Show a,-}Applicative m) => Tracer m a +testTracer :: ({-Show a,-}Monad m) => Tracer m a testTracer = nullTracer -- debugTracer tests :: TestTree diff --git a/dmq-node/test/DMQ/Protocol/SigSubmission/Test.hs b/dmq-node/test/DMQ/Protocol/SigSubmission/Test.hs index 922cae8..3cd142a 100644 --- a/dmq-node/test/DMQ/Protocol/SigSubmission/Test.hs +++ b/dmq-node/test/DMQ/Protocol/SigSubmission/Test.hs @@ -97,11 +97,11 @@ tests = , testProperty "codec" prop_codec_mockcrypto , testProperty "codec id" prop_codec_id_mockcrypto , testProperty "codec 2-splits" $ withMaxSize 20 - $ withMaxSuccess 20 + $ withNumTests 20 prop_codec_splits2_mockcrypto -- MockCrypto produces too large messages for this test to run: -- , testProperty "codec 3-splits" $ withMaxSize 10 - -- $ withMaxSuccess 10 + -- $ withNumTests 10 -- prop_codec_splits3_mockcrypto , testProperty "codec cbor" prop_codec_cbor_mockcrypto , testProperty "codec valid cbor" prop_codec_valid_cbor_mockcrypto @@ -115,12 +115,12 @@ tests = , testProperty "codec" prop_codec_standardcrypto , testProperty "codec id" prop_codec_id_standardcrypto , testProperty "codec 2-splits" $ withMaxSize 20 - $ withMaxSuccess 20 + $ withNumTests 20 prop_codec_splits2_standardcrypto -- StandardCrypto produces too large messages for this test to run: {- , testProperty "codec 3-splits" $ withMaxSize 10 - $ withMaxSuccess 10 + $ withNumTests 10 prop_codec_splits3_standardcrypto -} , testProperty "codec cbor" prop_codec_cbor_standardcrypto diff --git a/dmq-node/test/Test/DMQ/PeerSelection/PeerMetric.hs b/dmq-node/test/Test/DMQ/PeerSelection/PeerMetric.hs index c73e94f..40fa6b2 100644 --- a/dmq-node/test/Test/DMQ/PeerSelection/PeerMetric.hs +++ b/dmq-node/test/Test/DMQ/PeerSelection/PeerMetric.hs @@ -130,12 +130,12 @@ reportSigType config (localStMap, st) = \case SigValid { sigId, peerAddr, sigTime } -> let localSt = reportSigIdsImpl config [sigId] sigTime $ Map.findWithDefault emptyLocalPeerMetricState peerAddr localStMap - (localSt', st') = reportSigImpl config localSt (TraceLabelPeer peerAddr (sigId, TxAccepted)) st + (localSt', st') = reportSigImpl config localSt (TraceLabelPeer peerAddr (sigId, SigAccepted)) st in (Map.insert peerAddr localSt' localStMap, st') SigInvalid { sigId, peerAddr, sigTime } -> let localSt = reportSigIdsImpl config [sigId] sigTime $ Map.findWithDefault emptyLocalPeerMetricState peerAddr localStMap - (localSt', st') = reportSigImpl config localSt (TraceLabelPeer peerAddr (sigId, TxRejected)) st + (localSt', st') = reportSigImpl config localSt (TraceLabelPeer peerAddr (sigId, SigRejected)) st in (Map.insert peerAddr localSt' localStMap, st') @@ -376,7 +376,7 @@ instance Arbitrary ConfigWithTimes where ] --- | When two peers both get 'TxAccepted' for the same sigid (e.g. after a +-- | When two peers both get 'SigAccepted' for the same sigid (e.g. after a -- mempool eviction and re-submission), the peer with the *earlier* announcement -- time keeps the entry. 'peer1' is accepted first, 'peer2' second; the times -- are independent so the test covers both displacement and no-displacement. @@ -390,9 +390,9 @@ prop_competingPeers ConfigWithTimes { cwtConfig = config let lst1 = reportSigIdsImpl config [sigId] time1 emptyLocalPeerMetricState lst2 = reportSigIdsImpl config [sigId] time2 emptyLocalPeerMetricState -- peer1 is accepted first, claiming the entry - (_, st1) = reportSigImpl config lst1 (TraceLabelPeer peer1 (sigId, TxAccepted)) emptyPeerMetricState + (_, st1) = reportSigImpl config lst1 (TraceLabelPeer peer1 (sigId, SigAccepted)) emptyPeerMetricState -- peer2 is accepted second - (_, st2) = reportSigImpl config lst2 (TraceLabelPeer peer2 (sigId, TxAccepted)) st1 + (_, st2) = reportSigImpl config lst2 (TraceLabelPeer peer2 (sigId, SigAccepted)) st1 a = announcinessImpl st2 PeerMetricConfiguration { timeWindowToKeep } = config cutoff = (-timeWindowToKeep) `addTime` time2 diff --git a/dmq-node/test/Test/DMQ/SigSubmission/App.hs b/dmq-node/test/Test/DMQ/SigSubmission/App.hs index 840a9d6..9bc7354 100644 --- a/dmq-node/test/Test/DMQ/SigSubmission/App.hs +++ b/dmq-node/test/Test/DMQ/SigSubmission/App.hs @@ -25,15 +25,13 @@ import Control.Monad.Class.MonadThrow import Control.Monad.Class.MonadTime.SI import Control.Monad.Class.MonadTimer.SI import Control.Monad.IOSim -import Control.Tracer (Tracer (..), contramap) -import System.Random (mkStdGen) +import Control.Tracer (Tracer (..), contramap, mkTracer) import Data.ByteString.Lazy qualified as BSL import Data.Foldable (toList, traverse_) import Data.Foldable qualified as Foldable import Data.Function (on) import Data.Functor.Identity (runIdentity) -import Data.Hashable import Data.List (nubBy) import Data.List qualified as List import Data.Map.Strict (Map) @@ -259,7 +257,6 @@ runSigSubmissionV2 , Typeable sigid , Show peeraddr , Ord peeraddr - , Hashable peeraddr , Typeable peeraddr , sigid ~ Int @@ -281,21 +278,19 @@ runSigSubmissionV2 tracer tracerSigLogic st0 sigDecisionPolicy = do return (mempool, c, d, e, outChannel, inChannel) ) st0 inboundMempool <- emptyMempool - let sigRng = mkStdGen 42 -- TODO - - sigChannelsVar <- newMVar (TxChannels Map.empty) - sigMempoolSem <- newTxMempoolSem - sharedSigStateVar <- newSharedTxStateVar sigRng + sharedSigStateVar <- newSharedTxStateVar emptySharedTxState traceTVarIO sharedSigStateVar \_ -> return . TraceDynamic . SigStateTrace labelTVarIO sharedSigStateVar "shared-sig-state" + sigPeerRegistry <- newPeerTxRegistry + sigCountersVar <- newTxSubmissionCountersVar mempty duplicateSigsVar <- LazySTM.newTVarIO [] - withAsync (decisionLogicThreads tracerSigLogic sayTracer - sigDecisionPolicy sigChannelsVar sharedSigStateVar) $ \a -> do + withAsync (txCountersThreadV2 sigDecisionPolicy sayTracer tracerSigLogic + sigCountersVar sharedSigStateVar sigPeerRegistry) $ \a -> do let outbounds = (\(addr, (mempool, _, outDelay, _, outChannel, _)) -> do labelThisThread ("outbound-" ++ show addr) let outbound = sigSubmissionOutbound - (Tracer $ say . show) + (mkTracer $ say . show) (NumIdsAck $ getNumTxIdsToReq $ maxUnacknowledgedTxIds sigDecisionPolicy) (getMempoolReader mempool) (maxBound :: TestVersion) @@ -311,18 +306,17 @@ runSigSubmissionV2 tracer tracerSigLogic st0 sigDecisionPolicy = do let inbounds = (\(addr, (_, ctrlMsgSTM, _, inDelay, _, inChannel)) -> do labelThisThread ("inbound-" ++ show addr) - withPeer tracerSigLogic - sigChannelsVar - sigMempoolSem - sigDecisionPolicy - sharedSigStateVar + withPeer sigDecisionPolicy (getMempoolReader inboundMempool) - (getMempoolWriter duplicateSigsVar inboundMempool) - getTxSize + sharedSigStateVar + sigPeerRegistry + sigCountersVar addr $ \(api :: PeerTxAPI m TxId (Tx TxId))-> do let inbound = sigSubmissionInbound verboseTracer + sigDecisionPolicy (getMempoolWriter duplicateSigsVar inboundMempool) + getTxSize api PeerMetric.nullMetrics ctrlMsgSTM @@ -468,13 +462,11 @@ runSigSubmissionV2WithMetric tracer tracerSigLogic config st0 sigDecisionPolicy return (mempool, c, d, e, outChannel, inChannel) ) st0 inboundMempool <- emptyMempool - let sigRng = mkStdGen 42 - - sigChannelsVar <- newMVar (TxChannels Map.empty) - sigMempoolSem <- newTxMempoolSem - sharedSigStateVar <- newSharedTxStateVar sigRng + sharedSigStateVar <- newSharedTxStateVar emptySharedTxState traceTVarIO sharedSigStateVar \_ -> return . TraceDynamic . SigStateTrace labelTVarIO sharedSigStateVar "shared-sig-state" + sigPeerRegistry <- newPeerTxRegistry + sigCountersVar <- newTxSubmissionCountersVar mempty duplicateSigsVar <- LazySTM.newTVarIO [] peerMetric <- PeerMetric.mkPeerMetric @@ -484,12 +476,12 @@ runSigSubmissionV2WithMetric tracer tracerSigLogic config st0 sigDecisionPolicy -> DontTrace _ -> TraceValue (Just (SimMetricSnapshot new)) (Just $ show new) - withAsync (decisionLogicThreads tracerSigLogic sayTracer - sigDecisionPolicy sigChannelsVar sharedSigStateVar) $ \a -> do + withAsync (txCountersThreadV2 sigDecisionPolicy sayTracer tracerSigLogic + sigCountersVar sharedSigStateVar sigPeerRegistry) $ \a -> do let outbounds = (\(addr, (mempool, _, outDelay, _, outChannel, _)) -> do labelThisThread ("outbound-" ++ show addr) let outbound = sigSubmissionOutbound - (Tracer $ say . show) + (mkTracer $ say . show) (NumIdsAck $ getNumTxIdsToReq $ maxUnacknowledgedTxIds sigDecisionPolicy) (getMempoolReader mempool) (maxBound :: TestVersion) @@ -505,19 +497,18 @@ runSigSubmissionV2WithMetric tracer tracerSigLogic config st0 sigDecisionPolicy let inbounds = (\(addr, (_, ctrlMsgSTM, _, inDelay, _, inChannel)) -> do labelThisThread ("inbound-" ++ show addr) - withPeer tracerSigLogic - sigChannelsVar - sigMempoolSem - sigDecisionPolicy - sharedSigStateVar + withPeer sigDecisionPolicy (getMempoolReader inboundMempool) - (getMempoolWriter duplicateSigsVar inboundMempool) - getTxSize + sharedSigStateVar + sigPeerRegistry + sigCountersVar addr $ \(api :: PeerTxAPI (IOSim s) TxId (Tx TxId)) -> do let inbound = sigSubmissionInbound (contramap (SimAppEvent . TraceLabelPeer addr) (dynamicTracer <> sayTracer :: Tracer (IOSim s) SimTraceEvent)) + sigDecisionPolicy (getMempoolWriter duplicateSigsVar inboundMempool) + getTxSize api (PeerMetric.hoist (TraceLabelPeer addr . runIdentity) diff --git a/dmq-node/test/Test/DMQ/SigSubmission/Types.hs b/dmq-node/test/Test/DMQ/SigSubmission/Types.hs index a7e4579..85cdaed 100644 --- a/dmq-node/test/Test/DMQ/SigSubmission/Types.hs +++ b/dmq-node/test/Test/DMQ/SigSubmission/Types.hs @@ -138,4 +138,4 @@ instance Arbitrary SigSubmissionState where ] newtype SigStateTrace peeraddr sigid = - SigStateTrace (SharedTxState peeraddr sigid (Tx sigid)) + SigStateTrace (SharedTxState peeraddr sigid) diff --git a/flake.lock b/flake.lock index 9134400..1fb641f 100644 --- a/flake.lock +++ b/flake.lock @@ -3,11 +3,11 @@ "CHaP": { "flake": false, "locked": { - "lastModified": 1779260980, - "narHash": "sha256-5SvJkobPmCtSYKwQWu1M/iFXj3g1PDy2bMJ9g1vm/No=", + "lastModified": 1783608062, + "narHash": "sha256-b7vCPkrTeKFDfJ3x9TMB9pwUm0eqrR28mxvunZMi/0c=", "owner": "IntersectMBO", "repo": "cardano-haskell-packages", - "rev": "4b6a661aa6463529c0ab0207567d62f85a6b899c", + "rev": "28088c447cffe5dce9816c3a1ea5df693b350dba", "type": "github" }, "original": { @@ -155,11 +155,11 @@ "hackage": { "flake": false, "locked": { - "lastModified": 1782113726, - "narHash": "sha256-Uwh+sHSOKIzicEyHFqnwexReNBvX9jbFiuExI0r51iM=", + "lastModified": 1782826502, + "narHash": "sha256-G6bt7DeWkDXJWI/fHME467V/SOaUdfG6hk7bTRuWyKg=", "owner": "input-output-hk", "repo": "hackage.nix", - "rev": "7dfb22c38f71a464f1c352d88d16e67b23ada26c", + "rev": "6a87e2657c145eb81d5f0f53702d4f26c08f9cbf", "type": "github" }, "original": {