TX Submission Logic

cabal.project

@@ -15,10 +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 2024-07-10T11:10:47Z
+  , hackage.haskell.org 2024-09-18T10:39:04Z
 
   -- Bump this if you need newer packages from CHaP
-  , cardano-haskell-packages 2024-07-09T10:22:18Z
+  , cardano-haskell-packages 2024-09-16T15:22:31Z
 
 packages: ./cardano-ping
           ./monoidal-synchronisation

ouroboros-network-api/src/Ouroboros/Network/SizeInBytes.hs

@@ -17,6 +17,7 @@ import Quiet (Quiet (..))
 newtype SizeInBytes = SizeInBytes { getSizeInBytes :: Word32 }
   deriving (Eq, Ord)
   deriving Show      via Quiet SizeInBytes
+  deriving Bounded   via Word32
   deriving Enum      via Word32
   deriving Num       via Word32
   deriving Real      via Word32

ouroboros-network-protocols/src/Ouroboros/Network/Protocol/TxSubmission2/Server.hs

@@ -83,9 +83,9 @@ data ServerStIdle (n :: N) txid tx m a where
   -- | Collect a pipelined result.
   --
   CollectPipelined
-    :: Maybe                 (ServerStIdle (S n) txid tx m a)
-    -> (Collect txid tx -> m (ServerStIdle    n  txid tx m a))
-    ->                        ServerStIdle (S n) txid tx m a
+    :: Maybe                 (m (ServerStIdle (S n) txid tx m a))
+    -> (Collect txid tx -> m (   ServerStIdle    n  txid tx m a))
+    ->                           ServerStIdle (S n) txid tx m a
 
 
 -- | Transform a 'TxSubmissionServerPipelined' into a 'PeerPipelined'.
@@ -145,6 +145,5 @@ txSubmissionServerPeerPipelined (TxSubmissionServerPipelined server) =
 
     go (CollectPipelined mNone collect) =
       SenderCollect
-        (fmap go mNone)
-        (SenderEffect . fmap go . collect)
-
+        (SenderEffect . fmap go <$> mNone)
+        (SenderEffect . fmap go  .  collect)

ouroboros-network-protocols/testlib/Ouroboros/Network/Protocol/TxSubmission2/Direct.hs

@@ -54,7 +54,8 @@ directPipelined (TxSubmissionServerPipelined mserver)
       SendMsgReplyTxs txs client' <- recvMsgRequestTxs txids
       directSender (enqueue (CollectTxs txids txs) q) server' client'
 
-    directSender q (CollectPipelined (Just server') _) client =
+    directSender q (CollectPipelined (Just server) _) client = do
+      server' <- server
       directSender q server' client
 
     directSender (ConsQ c q) (CollectPipelined _ collect) client = do

ouroboros-network-protocols/testlib/Ouroboros/Network/Protocol/TxSubmission2/Examples.hs

@@ -272,7 +272,7 @@ txSubmissionServer tracer txId maxUnacked maxTxIdsToRequest maxTxToRequest =
         --
       | canRequestMoreTxs st
       = CollectPipelined
-          (Just (serverReqTxs accum (Succ n) st))
+          (Just (pure $ serverReqTxs accum (Succ n) st))
           (handleReply accum n st)
 
         -- In this case there is nothing else to do so we block until we

ouroboros-network/ouroboros-network.cabal

@@ -69,6 +69,12 @@ library
                        Ouroboros.Network.PeerSelection.RootPeersDNS.PublicRootPeers
                        Ouroboros.Network.PeerSharing
                        Ouroboros.Network.TxSubmission.Inbound
+                       Ouroboros.Network.TxSubmission.Inbound.Decision
+                       Ouroboros.Network.TxSubmission.Inbound.Policy
+                       Ouroboros.Network.TxSubmission.Inbound.Registry
+                       Ouroboros.Network.TxSubmission.Inbound.Server
+                       Ouroboros.Network.TxSubmission.Inbound.State
+                       Ouroboros.Network.TxSubmission.Inbound.Types
                        Ouroboros.Network.TxSubmission.Mempool.Reader
                        Ouroboros.Network.TxSubmission.Outbound
   other-modules:       Ouroboros.Network.Diffusion.Common
@@ -139,6 +145,7 @@ library
                        io-classes-mtl   ^>=0.1,
                        network-mux,
                        si-timers,
+                       strict-mvar,
                        ouroboros-network-api       ^>=0.9.0,
                        ouroboros-network-framework ^>=0.13.2.2,
                        ouroboros-network-protocols ^>=0.10,
@@ -198,6 +205,7 @@ library sim-tests-lib
 
                        cardano-prelude,
                        cardano-slotting,
+                       cardano-strict-containers,
                        contra-tracer,
                        nothunks,
 
@@ -215,6 +223,7 @@ library sim-tests-lib
                        ouroboros-network-testing ^>= 0.7.0,
                        si-timers,
                        strict-stm,
+                       strict-mvar,
                        typed-protocols,
                        typed-protocols-examples,
   exposed-modules:     Ouroboros.Network.BlockFetch.Examples
@@ -246,6 +255,10 @@ library sim-tests-lib
                        Test.Ouroboros.Network.Testnet
                        Test.Ouroboros.Network.Testnet.Simulation.Node
                        Test.Ouroboros.Network.TxSubmission
+                       Test.Ouroboros.Network.TxSubmission.Types
+                       Test.Ouroboros.Network.TxSubmission.TxLogic
+                       Test.Ouroboros.Network.TxSubmission.AppV1
+                       Test.Ouroboros.Network.TxSubmission.AppV2
                        Test.Ouroboros.Network.Version
 
   ghc-options:         -Wall

ouroboros-network/sim-tests-lib/Test/Ouroboros/Network/Diffusion/Node.hs

@@ -92,6 +92,7 @@ import Ouroboros.Network.Testing.Data.Script (Script (..), stepScriptSTM')
 
 import Simulation.Network.Snocket (AddressType (..), FD)
 
+import Ouroboros.Network.BlockFetch.ClientRegistry (readPeerGSVs)
 import Ouroboros.Network.PeerSelection.Bootstrap (UseBootstrapPeers)
 import Ouroboros.Network.PeerSelection.LedgerPeers.Type
            (LedgerPeersConsensusInterface,
@@ -105,6 +106,10 @@ import Ouroboros.Network.PeerSelection.RelayAccessPoint (DomainAccessPoint,
 import Ouroboros.Network.PeerSelection.RootPeersDNS.DNSActions (DNSLookupType)
 import Ouroboros.Network.PeerSelection.State.LocalRootPeers (HotValency,
            WarmValency)
+import Ouroboros.Network.TxSubmission.Inbound.Policy (TxDecisionPolicy)
+import Ouroboros.Network.TxSubmission.Inbound.Registry (decisionLogicThread)
+import Ouroboros.Network.TxSubmission.Inbound.Types (TraceTxLogic,
+           TraceTxSubmissionInbound)
 import Test.Ouroboros.Network.Diffusion.Node.ChainDB (addBlock,
            getBlockPointSet)
 import Test.Ouroboros.Network.Diffusion.Node.MiniProtocols qualified as Node
@@ -114,6 +119,7 @@ import Test.Ouroboros.Network.Diffusion.Node.NodeKernel (NodeKernel (..),
 import Test.Ouroboros.Network.Diffusion.Node.NodeKernel qualified as Node
 import Test.Ouroboros.Network.PeerSelection.RootPeersDNS (DNSLookupDelay,
            DNSTimeout, mockDNSActions)
+import Test.Ouroboros.Network.TxSubmission.Types (Tx)
 
 
 data Interfaces m = Interfaces
@@ -158,6 +164,8 @@ data Arguments m = Arguments
     , aDNSTimeoutScript     :: Script DNSTimeout
     , aDNSLookupDelayScript :: Script DNSLookupDelay
     , aDebugTracer          :: Tracer m String
+    , aTxDecisionPolicy     :: TxDecisionPolicy
+    , aTxs                  :: [Tx Int]
     }
 
 -- The 'mockDNSActions' is not using \/ specifying 'resolverException', thus we
@@ -195,9 +203,11 @@ run :: forall resolver m.
                              NtCAddr NtCVersion NtCVersionData
                              ResolverException m
     -> Tracer m (TraceLabelPeer NtNAddr (TraceFetchClientState BlockHeader))
+    -> Tracer m (TraceTxSubmissionInbound Int (Tx Int))
+    -> Tracer m (TraceTxLogic NtNAddr Int (Tx Int))
     -> m Void
-run blockGeneratorArgs limits ni na tracersExtra tracerBlockFetch =
-    Node.withNodeKernelThread blockGeneratorArgs
+run blockGeneratorArgs limits ni na tracersExtra tracerBlockFetch tracerTxSubmissionInbound tracerTxLogic =
+    Node.withNodeKernelThread blockGeneratorArgs (aTxs na)
       $ \ nodeKernel nodeKernelThread -> do
         dnsTimeoutScriptVar <- newTVarIO (aDNSTimeoutScript na)
         dnsLookupDelayScriptVar <- newTVarIO (aDNSLookupDelayScript na)
@@ -270,7 +280,7 @@ run blockGeneratorArgs limits ni na tracersExtra tracerBlockFetch =
               , Diff.P2P.daPeerSharingRegistry    = nkPeerSharingRegistry nodeKernel
               }
 
-        let apps = Node.applications (aDebugTracer na) nodeKernel Node.cborCodecs limits appArgs blockHeader
+        let apps = Node.applications (aDebugTracer na) tracerTxSubmissionInbound tracerTxLogic nodeKernel Node.cborCodecs limits appArgs blockHeader
 
         withAsync
            (Diff.P2P.runM interfaces
@@ -280,11 +290,19 @@ run blockGeneratorArgs limits ni na tracersExtra tracerBlockFetch =
                           (mkArgsExtra useBootstrapPeersScriptVar) apps appsExtra)
            $ \ diffusionThread ->
                withAsync (blockFetch nodeKernel) $ \blockFetchLogicThread ->
-                 wait diffusionThread
-              <> wait blockFetchLogicThread
-              <> wait nodeKernelThread
+
+                 withAsync (decisionLogicThread
+                              tracerTxLogic
+                              (aTxDecisionPolicy na)
+                              (readPeerGSVs (nkFetchClientRegistry nodeKernel))
+                              (nkTxChannelsVar nodeKernel)
+                              (nkSharedTxStateVar nodeKernel)) $ \decLogicThread ->
+                      wait diffusionThread
+                   <> wait blockFetchLogicThread
+                   <> wait nodeKernelThread
+                   <> wait decLogicThread
   where
-    blockFetch :: NodeKernel BlockHeader Block s m
+    blockFetch :: NodeKernel BlockHeader Block s txid m
                -> m Void
     blockFetch nodeKernel = do
       blockFetchLogic
@@ -300,7 +318,7 @@ run blockGeneratorArgs limits ni na tracersExtra tracerBlockFetch =
           bfcSalt                   = 0
         })
 
-    blockFetchPolicy :: NodeKernel BlockHeader Block s m
+    blockFetchPolicy :: NodeKernel BlockHeader Block s txid m
                      -> BlockFetchConsensusInterface NtNAddr BlockHeader Block m
     blockFetchPolicy nodeKernel =
         BlockFetchConsensusInterface {
@@ -422,6 +440,7 @@ run blockGeneratorArgs limits ni na tracersExtra tracerBlockFetch =
       , Node.aaOwnPeerSharing           = aOwnPeerSharing na
       , Node.aaUpdateOutboundConnectionsState =
           iUpdateOutboundConnectionsState ni
+      , Node.aaTxDecisionPolicy         = aTxDecisionPolicy na
       }
 
 --- Utils