TX Submission Logic

cabal.project

@@ -15,7 +15,7 @@ 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-09-26T15:16:07Z

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

@@ -84,9 +84,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'.
@@ -136,6 +136,6 @@ txSubmissionServerPeerPipelined (TxSubmissionServerPipelined server) =
         (Effect (go <$> k))
 
     go (CollectPipelined mNone collect) =
-      Collect (fmap go mNone)
-              (Effect . fmap go . collect)
+      Collect (Effect . fmap go <$> mNone)
+              (Effect . 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

@@ -90,6 +90,12 @@ library
     Ouroboros.Network.PeerSharing
     Ouroboros.Network.Tracers
     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
 
@@ -168,6 +174,7 @@ library
     random,
     si-timers,
     strict-checked-vars ^>=0.2,
+    strict-mvar,
     strict-stm,
     typed-protocols ^>=0.3,
     typed-protocols-stateful,
@@ -199,6 +206,7 @@ library sim-tests-lib
     cardano-binary,
     cardano-prelude,
     cardano-slotting,
+    cardano-strict-containers,
     cborg,
     containers,
     contra-tracer,
@@ -228,6 +236,7 @@ library sim-tests-lib
     random,
     serialise,
     si-timers,
+    strict-mvar,
     strict-stm,
     tasty,
     tasty-hunit,
@@ -267,6 +276,10 @@ library sim-tests-lib
     Test.Ouroboros.Network.Testnet
     Test.Ouroboros.Network.Testnet.Simulation.Node
     Test.Ouroboros.Network.TxSubmission
+    Test.Ouroboros.Network.TxSubmission.AppV1
+    Test.Ouroboros.Network.TxSubmission.AppV2
+    Test.Ouroboros.Network.TxSubmission.TxLogic
+    Test.Ouroboros.Network.TxSubmission.Types
     Test.Ouroboros.Network.Version
 
 -- Simulation tests, and IO tests which don't require native system calls.

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
@@ -193,9 +201,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)
@@ -268,7 +278,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
@@ -278,11 +288,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
@@ -298,7 +316,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 {
@@ -420,6 +438,7 @@ run blockGeneratorArgs limits ni na tracersExtra tracerBlockFetch =
       , Node.aaOwnPeerSharing           = aOwnPeerSharing na
       , Node.aaUpdateOutboundConnectionsState =
           iUpdateOutboundConnectionsState ni
+      , Node.aaTxDecisionPolicy         = aTxDecisionPolicy na
       }
 
 --- Utils