Merge pull request #1 from IntersectMBO/mw/with-cardano

Enhance validation logic

Author: crocodile-dentist

cabal.project

@@ -32,3 +32,36 @@ multi-repl: True
 if(os(windows))
   constraints:
     bitvec -simd
+
+-- kes-agent is not yet in CHaP, so we pull it from its GitHub repo
+source-repository-package
+  type: git
+  location: https://github.com/crocodile-dentist/kes-agent
+  tag: c0ef04dde5582a28415ff7c8c1bb197adeec6fc8
+  --sha256: sha256-slF7zuBy2DKWKlQfhBPW5FDRhueWrFcJkrHW4jEEELs=
+  subdir:
+    kes-agent
+    kes-agent-crypto
+
+source-repository-package
+  type: git
+  location: https://github.com/IntersectMBO/ouroboros-consensus
+  tag: a24a5618678bcdabe042fcad7e47569285e964a2
+  --sha256: sha256-1iW9i2UqaiopbluG9T22Hz7QwXY6+qCgqqmjDXlIp40=
+  subdir:
+    ouroboros-consensus-cardano
+    ouroboros-consensus-diffusion
+    sop-extras
+    ouroboros-consensus-protocol
+    ouroboros-consensus
+
+source-repository-package
+  type: git
+  location: https://github.com/IntersectMBO/ouroboros-network
+  tag: b87b35827c309d168af77571b6d44198136f1199
+  --sha256: sha256-SVsAcfyIpNGNI9x5MFUybhwjIZ+6nU4kdECCdVmfdeE=
+  subdir:
+    acts-generic
+    cardano-diffusion
+    ouroboros-network
+    network-mux

dmq-node/app/Main.hs

@@ -1,17 +1,24 @@
-{-# LANGUAGE MultiWayIf          #-}
-{-# LANGUAGE OverloadedStrings   #-}
-{-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TemplateHaskell     #-}
-{-# LANGUAGE TypeApplications    #-}
+{-# LANGUAGE DataKinds                #-}
+{-# LANGUAGE DisambiguateRecordFields #-}
+{-# LANGUAGE MultiWayIf               #-}
+{-# LANGUAGE OverloadedRecordDot      #-}
+{-# LANGUAGE OverloadedStrings        #-}
+{-# LANGUAGE ScopedTypeVariables      #-}
+{-# LANGUAGE TemplateHaskell          #-}
+{-# LANGUAGE TypeApplications         #-}
+{-# LANGUAGE TypeOperators            #-}
 
 module Main where
 
+import Control.Concurrent.Class.MonadSTM.Strict
 import Control.Monad (void, when)
 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.List.NonEmpty (NonEmpty)
 import Data.Maybe (maybeToList)
 import Data.Text qualified as Text
 import Data.Text.IO qualified as Text
@@ -23,13 +30,15 @@ import System.Random (newStdGen, split)
 
 import Cardano.Git.Rev (gitRev)
 import Cardano.KESAgent.Protocols.StandardCrypto (StandardCrypto)
+import Cardano.Ledger.Keys (VKey (..))
+import Cardano.Ledger.Hashes (hashKey)
 
 import DMQ.Configuration
 import DMQ.Configuration.CLIOptions (parseCLIOptions)
 import DMQ.Configuration.Topology (readTopologyFileOrError)
 import DMQ.Diffusion.Applications (diffusionApplications)
 import DMQ.Diffusion.Arguments
-import DMQ.Diffusion.NodeKernel (mempool, withNodeKernel)
+import DMQ.Diffusion.NodeKernel
 import DMQ.Handlers.TopLevel (toplevelExceptionHandler)
 import DMQ.NodeToClient qualified as NtC
 import DMQ.NodeToNode (NodeToNodeVersion, dmqCodecs, dmqLimitsAndTimeouts,
@@ -39,9 +48,14 @@ import DMQ.Protocol.SigSubmission.Type (Sig (..))
 import DMQ.Tracer
 
 import DMQ.Diffusion.PeerSelection (policy)
+import DMQ.NodeToClient.LocalStateQueryClient
+import DMQ.Protocol.SigSubmission.Validate
 import Ouroboros.Network.Diffusion qualified as Diffusion
+import Ouroboros.Network.PeerSelection.LedgerPeers.Type
 import Ouroboros.Network.PeerSelection.PeerSharing.Codec (decodeRemoteAddress,
            encodeRemoteAddress)
+import Ouroboros.Network.SizeInBytes
+import Ouroboros.Network.Snocket
 import Ouroboros.Network.TxSubmission.Mempool.Simple qualified as Mempool
 
 import Paths_dmq_node qualified as Meta
@@ -70,6 +84,7 @@ runDMQ commandLineConfig = do
           dmqcTopologyFile         = I topologyFile,
           dmqcHandshakeTracer      = I handshakeTracer,
           dmqcLocalHandshakeTracer = I localHandshakeTracer,
+          dmqcCardanoNodeSocket    = I snocketPath,
           dmqcVersion              = I version
         } = config' <> commandLineConfig
             `act`
@@ -101,49 +116,75 @@ runDMQ commandLineConfig = do
     stdGen <- newStdGen
     let (psRng, policyRng) = split stdGen
 
-    withNodeKernel @StandardCrypto
-                   tracer
-                   dmqConfig
-                   psRng $ \nodeKernel -> do
-      dmqDiffusionConfiguration <- mkDiffusionConfiguration dmqConfig nt
-
-      let dmqNtNApps =
-            ntnApps tracer
-                    dmqConfig
-                    nodeKernel
-                    (dmqCodecs
-                       (encodeRemoteAddress (maxBound @NodeToNodeVersion))
-                       (decodeRemoteAddress (maxBound @NodeToNodeVersion)))
-                    dmqLimitsAndTimeouts
-                    defaultSigDecisionPolicy
-          dmqNtCApps =
-            let sigSize _ = 0 -- TODO
-                maxMsgs = 1000 -- TODO: make this dynamic?
-                mempoolReader = Mempool.getReader sigId sigSize (mempool nodeKernel)
-                mempoolWriter = Mempool.getWriter sigId (pure ())
-                                                        (\_ _ -> Right () :: Either Void ())
-                                                        (\_ -> True)
-                                                        (mempool nodeKernel)
-             in NtC.ntcApps tracer dmqConfig
-                            mempoolReader mempoolWriter maxMsgs
-                            (NtC.dmqCodecs encodeReject decodeReject)
-          dmqDiffusionArguments =
-            diffusionArguments (if handshakeTracer
-                                  then WithEventType "Handshake" >$< tracer
-                                  else nullTracer)
-                               (if localHandshakeTracer
-                                  then WithEventType "Handshake" >$< tracer
-                                  else nullTracer)
-          dmqDiffusionApplications =
-            diffusionApplications nodeKernel
-                                  dmqConfig
-                                  dmqDiffusionConfiguration
-                                  dmqLimitsAndTimeouts
-                                  dmqNtNApps
-                                  dmqNtCApps
-                                  (policy policyRng)
-
-      Diffusion.run dmqDiffusionArguments
-                    (dmqDiffusionTracers dmqConfig tracer)
-                    dmqDiffusionConfiguration
-                    dmqDiffusionApplications
+    -- TODO: this might not work, since `ouroboros-network` creates its own IO Completion Port.
+    Diffusion.withIOManager \iocp -> do
+      let localSnocket'      = localSnocket iocp
+          mkStakePoolMonitor = connectToCardanoNode tracer localSnocket' snocketPath
+
+      withNodeKernel @StandardCrypto
+                     tracer
+                     dmqConfig
+                     psRng
+                     mkStakePoolMonitor $ \nodeKernel -> do
+        dmqDiffusionConfiguration <-
+          mkDiffusionConfiguration dmqConfig nt nodeKernel.stakePools.ledgerBigPeersVar
+
+        let sigSize :: Sig StandardCrypto -> SizeInBytes
+            sigSize = fromIntegral . BSL.length . sigRawBytes
+            mempoolReader = Mempool.getReader sigId sigSize (mempool nodeKernel)
+            dmqNtNApps =
+              let ntnMempoolWriter = Mempool.writerAdapter $
+                    Mempool.getWriter sigId
+                                      (poolValidationCtx $ stakePools nodeKernel)
+                                      (validateSig (hashKey . VKey))
+                                      SigDuplicate
+                                      (mempool nodeKernel)
+               in ntnApps tracer
+                          dmqConfig
+                          mempoolReader
+                          ntnMempoolWriter
+                          sigSize
+                          nodeKernel
+                          (dmqCodecs
+                                   -- TODO: `maxBound :: Cardano.Network.NodeToNode.NodeToNodeVersion`
+                                   -- is unsafe here!
+                                   (encodeRemoteAddress (maxBound @NodeToNodeVersion))
+                                   (decodeRemoteAddress (maxBound @NodeToNodeVersion)))
+                          dmqLimitsAndTimeouts
+                          defaultSigDecisionPolicy
+            dmqNtCApps =
+              let ntcMempoolWriter =
+                    Mempool.getWriter sigId
+                                      (poolValidationCtx $ stakePools nodeKernel)
+                                      (validateSig (hashKey . VKey))
+                                      SigDuplicate
+                                      (mempool nodeKernel)
+               in NtC.ntcApps tracer dmqConfig
+                              mempoolReader ntcMempoolWriter
+                              (NtC.dmqCodecs encodeReject decodeReject)
+            dmqDiffusionArguments =
+              diffusionArguments (if handshakeTracer
+                                    then WithEventType "Handshake" >$< tracer
+                                    else nullTracer)
+                                 (if localHandshakeTracer
+                                    then WithEventType "Handshake" >$< tracer
+                                    else nullTracer)
+                                 $ maybe [] out <$> tryReadTMVar nodeKernel.stakePools.ledgerPeersVar
+              where
+                out :: LedgerPeerSnapshot AllLedgerPeers
+                    -> [(PoolStake, NonEmpty LedgerRelayAccessPoint)]
+                out (LedgerAllPeerSnapshotV23 _pt _magic relays) = relays
+
+            dmqDiffusionApplications =
+              diffusionApplications nodeKernel
+                                    dmqConfig
+                                    dmqDiffusionConfiguration
+                                    dmqLimitsAndTimeouts
+                                    dmqNtNApps
+                                    dmqNtCApps
+                                    (policy policyRng)
+
+        Diffusion.run dmqDiffusionArguments
+                      (dmqDiffusionTracers dmqConfig tracer)
+                      dmqDiffusionConfiguration
+                      dmqDiffusionApplications

dmq-node/dmq-node.cabal

@@ -68,6 +68,7 @@ library
     DMQ.NodeToClient
     DMQ.NodeToClient.LocalMsgNotification
     DMQ.NodeToClient.LocalMsgSubmission
+    DMQ.NodeToClient.LocalStateQueryClient
     DMQ.NodeToClient.Version
     DMQ.NodeToNode
     DMQ.NodeToNode.Version
@@ -81,6 +82,7 @@ library
     DMQ.Protocol.LocalMsgSubmission.Type
     DMQ.Protocol.SigSubmission.Codec
     DMQ.Protocol.SigSubmission.Type
+    DMQ.Protocol.SigSubmission.Validate
     DMQ.Tracer
 
   build-depends:
@@ -93,13 +95,17 @@ library
     bytestring >=0.10 && <0.13,
     cardano-binary,
     cardano-crypto-class,
+    cardano-crypto-wrapper,
+    cardano-diffusion,
+    cardano-ledger-byron,
+    cardano-ledger-core,
+    cardano-ledger-shelley,
+    cardano-slotting,
     cborg >=0.2.1 && <0.3,
-    containers >=0.5 && <0.9,
+    containers >=0.5 && <0.8,
     contra-tracer >=0.1 && <0.3,
     deepseq >=1.0 && <1.6,
-    directory,
     dns >=1.0 && <4.3,
-    filepath,
     generic-data,
     hashable >=1.0 && <1.6,
     io-classes:{io-classes, si-timers, strict-mvar, strict-stm} ^>=1.8.0.1,
@@ -108,11 +114,16 @@ library
     network ^>=3.2.7,
     network-mux ^>=0.9.1,
     optparse-applicative >=0.18 && <0.20,
+    ouroboros-consensus,
+    ouroboros-consensus-cardano,
+    ouroboros-consensus-diffusion,
     ouroboros-network:{ouroboros-network, api, framework, orphan-instances, protocols} ^>=0.23,
     random ^>=1.2,
     singletons,
     text >=1.2.4 && <2.2,
-    time >=1.12 && <1.16,
+    time ^>=1.12,
+    transformers,
+    transformers-except,
     typed-protocols:{typed-protocols, cborg} ^>=1.1,
 
   hs-source-dirs: src
@@ -134,12 +145,15 @@ executable dmq-node
     acts,
     aeson,
     base,
+    bytestring,
     cardano-git-rev,
+    cardano-ledger-core,
     contra-tracer >=0.1 && <0.3,
     dmq-node,
+    io-classes:strict-stm,
     kes-agent-crypto,
     optparse-applicative,
-    ouroboros-network:{ouroboros-network, api},
+    ouroboros-network:{ouroboros-network, api, framework},
     random,
     text,
 
@@ -166,20 +180,25 @@ test-suite dmq-tests
   build-depends:
     QuickCheck,
     base >=4.14 && <4.23,
+    binary,
     bytestring,
     cardano-crypto-class,
     cardano-crypto-tests,
+    cardano-ledger-core,
     cborg,
+    containers,
     contra-tracer,
     dmq-node,
-    io-classes,
+    io-classes:{io-classes, strict-stm},
     io-sim,
     kes-agent-crypto,
     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,