Mempool fixes

README.md

@@ -3,22 +3,58 @@
 The DMQ node allows for client peers to communicate efficiently by publishing
 and consuming messages which are diffused over a P2P network to other nodes.
 
-This repository provides the `dmq-node` executable to participate in the DMQ network.
+This repository provides the `dmq-node` executable to participate in the DMQ
+network.
+
+The `dmq-node` is developed with respect to the [CIP#0137].
+
+__NOTE__: This is still an early version of DMQ node, which comes with some
+quirks:
+
+* issue#6 - no support for ledger peers, which requires setting up static peers
+            using local roots.  We are aiming to add this feature for the
+            `cardano-node-10.7` release.
+* issue#13 - using `TxSubmission` mini-protocol for which roles are swapped,
+             e.g. server requests data, client servers the data.  This makes
+             configuration awekward, since your local roots specify who will get data from
+             you, rather than who you get data from.  We are working on a new
+             mini-protocol to address this.
 
 # Instructions
 
 ## Building the project
 
 We use cabal to build our project, potentially inside a Nix shell (nix develop or nix-shell). It should suffice with:
 ```bash
-> cabal build dmq-node
+> cabal build dmq-node:exe:dmq-node
 ```
 
 The executable can be run with:
 ```bash
 > cabal run dmq-node
 ```
 
+### Building with Nix
+
+To build the executable using Nix, one can use:
+```bash
+> nix build .#dmq-node
+```
+
+#### Static build
+
+To get a statically linked executable using [musl] library, use:
+```bash
+> nix build .#dmq-node-static
+```
+
+## Developing with Nix
+
+To enter a development shell with all dependencies available, use:
+```bash
+> nix develop
+```
+
 ## Testing the project
 
 To run the test suite, one can use:
@@ -27,6 +63,15 @@ To run the test suite, one can use:
 > cabal test all
 ```
 
+### CDDL
+
+This project comes with a CDDL specification for the DMQ protocols
+(`node-to-client` and `node-to-node`).  To check changes against the CDDL
+specification, use:
+```bash
+cabal run dmq-node:dmq-cddl
+```
+
 # Contributing
 
 The contributing guide is available [here][contributing-guide].
@@ -39,3 +84,5 @@ The code of conduct is available [here][code-of-conduct].
 [contributing-guide]: ./CONTRIBUTING.md
 [code-of-conduct]: ./CODE_OF_CONDUCT.md
 [style-guide]: ./docs/StyleGuide.md
+[musl]: https://musl.libc.org/
+[CIP#0137]: https://cips.cardano.org/cip/CIP-0137

cabal.project

@@ -18,7 +18,7 @@ index-state:
   , hackage.haskell.org 2025-11-10T01:36:00Z
 
   -- Bump this if you need newer packages from CHaP
-  , cardano-haskell-packages 2025-11-07T15:42:47Z
+  , cardano-haskell-packages 2026-01-09T12:04:47Z
 
 packages:
   ./dmq-node
@@ -46,8 +46,8 @@ source-repository-package
 source-repository-package
   type: git
   location: https://github.com/IntersectMBO/ouroboros-consensus
-  tag: a24a5618678bcdabe042fcad7e47569285e964a2
-  --sha256: sha256-1iW9i2UqaiopbluG9T22Hz7QwXY6+qCgqqmjDXlIp40=
+  tag: 454e2491a23d14161c159d815bd26c545455c9ef
+  --sha256: sha256-ODoM1dCyKM8MiMherH1xxcJtzoYdqbhDGkwzFB6SKeM=
   subdir:
     ouroboros-consensus-cardano
     ouroboros-consensus-diffusion
@@ -58,10 +58,14 @@ source-repository-package
 source-repository-package
   type: git
   location: https://github.com/IntersectMBO/ouroboros-network
-  tag: b87b35827c309d168af77571b6d44198136f1199
-  --sha256: sha256-SVsAcfyIpNGNI9x5MFUybhwjIZ+6nU4kdECCdVmfdeE=
+  tag: 3c4433d05ec012af6d1a26e6b5e86665627c08c4
+  --sha256: sha256-Jemp6PlzISA+l1wdXV6MrIxaBpAxdrLLAlbkB7ZqF2Y=
   subdir:
     acts-generic
     cardano-diffusion
     ouroboros-network
     network-mux
+
+if impl(ghc >= 9.12.0)
+  allow-newer: *:time,
+               *:nothunks,

dmq-node/app/Main.hs

@@ -12,12 +12,14 @@ module Main where
 
 import Control.Concurrent.Class.MonadSTM.Strict
 import Control.Monad (void, when)
+import Control.Monad.Class.MonadThrow
 import Control.Tracer (Tracer (..), nullTracer, traceWith)
 
 import Data.Act
 import Data.Aeson (ToJSON)
 import Data.ByteString.Lazy qualified as BSL
 import Data.Functor.Contravariant ((>$<))
+import Data.Foldable (traverse_)
 import Data.List.NonEmpty (NonEmpty)
 import Data.Maybe (maybeToList)
 import Data.Text qualified as Text
@@ -27,6 +29,7 @@ import Data.Void (Void)
 import Options.Applicative
 import System.Exit (exitSuccess)
 import System.Random (newStdGen, split)
+import System.IOManager (withIOManager)
 
 import Cardano.Git.Rev (gitRev)
 import Cardano.KESAgent.Protocols.StandardCrypto (StandardCrypto)
@@ -43,7 +46,6 @@ import DMQ.Handlers.TopLevel (toplevelExceptionHandler)
 import DMQ.NodeToClient qualified as NtC
 import DMQ.NodeToNode (NodeToNodeVersion, dmqCodecs, dmqLimitsAndTimeouts,
            ntnApps)
-import DMQ.Protocol.LocalMsgSubmission.Codec
 import DMQ.Protocol.SigSubmission.Type (Sig (..))
 import DMQ.Tracer
 
@@ -118,7 +120,7 @@ runDMQ commandLineConfig = do
     let (psRng, policyRng) = split stdGen
 
     -- TODO: this might not work, since `ouroboros-network` creates its own IO Completion Port.
-    Diffusion.withIOManager \iocp -> do
+    withIOManager \iocp -> do
       let localSnocket'      = localSnocket iocp
           mkStakePoolMonitor = connectToCardanoNode tracer localSnocket' snocketPath
 
@@ -137,11 +139,22 @@ runDMQ commandLineConfig = do
                                     then WithEventType "NtN Validation" >$< tracer
                                     else nullTracer
             dmqNtNApps =
-              let ntnMempoolWriter = Mempool.writerAdapter $
-                    Mempool.getWriter sigId
-                                      (poolValidationCtx $ stakePools nodeKernel)
-                                      (validateSig ntnValidationTracer (hashKey . VKey))
-                                      SigDuplicate
+              let ntnMempoolWriter =
+                    Mempool.getWriter SigDuplicate
+                                      sigId
+                                      (\now sigs ->
+                                        withPoolValidationCtx (stakePools nodeKernel) (validateSig (hashKey . VKey) now sigs)
+                                      )
+                                      (traverse_ $ \(sigid, reason) -> do
+                                        traceWith ntnValidationTracer $ InvalidSignature sigid reason
+                                        case reason of
+                                          SigDuplicate      -> return ()
+                                          SigExpired        -> return ()
+                                          NotInitialized    -> return ()
+                                          PoolNotEligible   -> return ()
+                                          UnrecognizedPool  -> return ()
+                                          err               -> throwIO (SigValidationException sigid err)
+                                      )
                                       (mempool nodeKernel)
                in ntnApps tracer
                           dmqConfig
@@ -161,14 +174,18 @@ runDMQ commandLineConfig = do
                                     else nullTracer
             dmqNtCApps =
               let ntcMempoolWriter =
-                    Mempool.getWriter sigId
-                                      (poolValidationCtx $ stakePools nodeKernel)
-                                      (validateSig ntcValidationTracer (hashKey . VKey))
-                                      SigDuplicate
+                    Mempool.getWriter SigDuplicate
+                                      sigId
+                                      (\now sigs ->
+                                        withPoolValidationCtx (stakePools nodeKernel) (validateSig (hashKey . VKey) now sigs)
+                                      )
+                                      (traverse_ $ \(sigid, reason) ->
+                                         traceWith ntcValidationTracer $ InvalidSignature sigid reason
+                                      )
                                       (mempool nodeKernel)
                in NtC.ntcApps tracer dmqConfig
                               mempoolReader ntcMempoolWriter
-                              (NtC.dmqCodecs encodeReject decodeReject)
+                              NtC.dmqCodecs
             dmqDiffusionArguments =
               diffusionArguments (if handshakeTracer
                                     then WithEventType "Handshake" >$< tracer

dmq-node/changelog.d/20260117_120143_coot_mempool_fixes.md

@@ -0,0 +1,15 @@
+### Breaking
+
+- Added `LocalMsgNotificationServerTracer` option (disabled by default)
+- `Sig`: improved instances `ToJSON` and `Show`
+- Refactored `validateSig`
+
+### Non-Breaking
+
+- By default enabled the folling traces:
+  - `MuxTracer`
+  - `TraceChurnCounters`
+  - `LocalCOnnectionManagerTracer`
+  - `SigSubmissionLogicTracer`
+- Removed pre-commit-hook in nix shell
+- Added negative test cases from `prop_validateSig` QuickCheck property

dmq-node/dmq-node.cabal

@@ -111,6 +111,7 @@ library
     io-classes:{io-classes, si-timers, strict-mvar, strict-stm} ^>=1.8.0.1,
     iproute ^>=1.7.15,
     kes-agent-crypto ^>=0.1,
+    mtl,
     network ^>=3.2.7,
     network-mux ^>=0.9.1,
     optparse-applicative >=0.18 && <0.20,
@@ -121,9 +122,8 @@ library
     random ^>=1.2,
     singletons,
     text >=1.2.4 && <2.2,
-    time ^>=1.12,
+    time >=1.12 && <1.15,
     transformers,
-    transformers-except,
     typed-protocols:{typed-protocols, cborg} ^>=1.1,
 
   hs-source-dirs: src
@@ -142,6 +142,7 @@ executable dmq-node
     -rtsopts
 
   build-depends:
+    Win32-network,
     acts,
     aeson,
     base,
@@ -150,7 +151,7 @@ executable dmq-node
     cardano-ledger-core,
     contra-tracer >=0.1 && <0.3,
     dmq-node,
-    io-classes:strict-stm,
+    io-classes:{io-classes, strict-stm},
     kes-agent-crypto,
     optparse-applicative,
     ouroboros-network:{ouroboros-network, api, framework},
@@ -180,7 +181,6 @@ test-suite dmq-tests
   build-depends:
     QuickCheck,
     base >=4.14 && <4.23,
-    binary,
     bytestring,
     cardano-crypto-class,
     cardano-crypto-tests,
@@ -189,16 +189,16 @@ test-suite dmq-tests
     containers,
     contra-tracer,
     dmq-node,
-    io-classes:{io-classes, strict-stm},
+    io-classes:{io-classes, si-timers},
     io-sim,
     kes-agent-crypto,
+    ouroboros-consensus-cardano,
     ouroboros-network:{api, framework, protocols, protocols-tests-lib, tests-lib},
     quickcheck-instances,
     serialise,
     tasty,
     tasty-quickcheck,
     time,
-    transformers,
     typed-protocols:{typed-protocols, codec-properties},
     with-utf8,
 

dmq-node/src/DMQ/Configuration.hs

@@ -152,6 +152,7 @@ data Configuration' f =
     dmqcSigSubmissionOutboundTracer                :: f Bool,
     dmqcSigSubmissionInboundTracer                 :: f Bool,
     dmqcLocalMsgSubmissionServerTracer             :: f Bool,
+    dmqcLocalMsgNotificationServerTracer           :: f Bool,
 
     dmqcVersion                                    :: f Bool
   }
@@ -230,7 +231,7 @@ defaultConfiguration = Configuration {
       dmqcChurnInterval                              = I defaultDeadlineChurnInterval,
       dmqcPeerSharing                                = I PeerSharingEnabled,
       dmqcPrettyLog                                  = I False,
-      dmqcMuxTracer                                  = I False,
+      dmqcMuxTracer                                  = I True,
       dmqcChannelTracer                              = I False,
       dmqcBearerTracer                               = I False,
       dmqcHandshakeTracer                            = I True,
@@ -243,7 +244,7 @@ defaultConfiguration = Configuration {
       dmqcTracePublicRootPeersTracer                 = I False,
       dmqcTraceLedgerPeersTracer                     = I False,
       dmqcTracePeerSelectionTracer                   = I True,
-      dmqcTraceChurnCounters                         = I False,
+      dmqcTraceChurnCounters                         = I True,
       dmqcDebugPeerSelectionInitiatorTracer          = I False,
       dmqcDebugPeerSelectionInitiatorResponderTracer = I False,
       dmqcTracePeerSelectionCounters                 = I True,
@@ -253,7 +254,7 @@ defaultConfiguration = Configuration {
       dmqcServerTracer                               = I True,
       dmqcInboundGovernorTracer                      = I True,
       dmqcInboundGovernorTransitionTracer            = I False,
-      dmqcLocalConnectionManagerTracer               = I False,
+      dmqcLocalConnectionManagerTracer               = I True,
       dmqcLocalServerTracer                          = I False,
       dmqcLocalInboundGovernorTracer                 = I False,
       dmqcDnsTracer                                  = I False,
@@ -270,8 +271,9 @@ defaultConfiguration = Configuration {
 
       dmqcSigSubmissionOutboundTracer                = I False,
       dmqcSigSubmissionInboundTracer                 = I True,
-      dmqcSigSubmissionLogicTracer                   = I False,
+      dmqcSigSubmissionLogicTracer                   = I True,
       dmqcLocalMsgSubmissionServerTracer             = I True,
+      dmqcLocalMsgNotificationServerTracer           = I False,
 
       -- CLI only options
       dmqcVersion                                    = I False
@@ -366,6 +368,7 @@ instance FromJSON PartialConfig where
       dmqcSigSubmissionInboundTracer                 <- Last <$> v .:? "SigSubmissionInboundTracer"
       dmqcSigSubmissionLogicTracer                   <- Last <$> v .:? "SigSubmissionLogicTracer"
       dmqcLocalMsgSubmissionServerTracer             <- Last <$> v .:? "LocalMsgSubmissionServerTracer"
+      dmqcLocalMsgNotificationServerTracer           <- Last <$> v .:? "LocalMsgNotificationServerTracer"
 
       pure $
         Configuration