Integrate New Tx Submission Protocol

cabal.project

@@ -14,9 +14,9 @@ repository cardano-haskell-packages
 -- update either of these.
 index-state:
   -- Bump this if you need newer packages from Hackage
-  , hackage.haskell.org 2024-08-26T10:41:44Z
+  , hackage.haskell.org 2024-09-16T12:20:25Z
   -- Bump this if you need newer packages from CHaP
-  , cardano-haskell-packages 2024-09-03T00:18:11Z
+  , cardano-haskell-packages 2024-09-11T14:05:05Z
 
 packages:
   ouroboros-consensus
@@ -56,3 +56,21 @@ if impl(ghc >= 9.10)
     , cardano-ledger-binary:plutus-ledger-api
     , cardano-ledger-conway:plutus-ledger-api
 
+source-repository-package
+  type: git
+  location: https://github.com/IntersectMBO/ouroboros-network
+  tag: 388cc6906b83f41ac2da192b1fd89ab986b4af74
+  --sha256: sha256-LUwryrP5jK+/c4lDitJf/oKg/DqLgbIc68bn83FsHI0=
+  subdir:
+    cardano-client
+    cardano-ping
+    monoidal-synchronisation
+    network-mux
+    ntp-client
+    ouroboros-network-api
+    ouroboros-network-framework
+    ouroboros-network-mock
+    ouroboros-network-protocols
+    ouroboros-network-testing
+    ouroboros-network
+    quickcheck-monoids

ouroboros-consensus-cardano/changelog.d/20240808_125131_marcin.wojtowicz_tx_wire_size.md

@@ -0,0 +1,21 @@
+<!--
+A new scriv changelog fragment.
+
+Uncomment the section that is right (remove the HTML comment wrapper).
+-->
+
+<!--
+### Patch
+
+- A bullet item for the Patch category.
+
+-->
+<!--
+### Non-Breaking
+
+- A bullet item for the Non-Breaking category.
+
+-->
+### Breaking
+
+- Implement txWireSize of TxLimits instantiations for Byron and Shelley

ouroboros-consensus-cardano/src/byron/Ouroboros/Consensus/Byron/Ledger/Mempool.hs

@@ -73,6 +73,7 @@ import           Ouroboros.Consensus.Ledger.Abstract
 import           Ouroboros.Consensus.Ledger.SupportsMempool
 import           Ouroboros.Consensus.Util (ShowProxy (..))
 import           Ouroboros.Consensus.Util.Condense
+import           Ouroboros.Network.SizeInBytes as Network
 
 {-------------------------------------------------------------------------------
   Transactions
@@ -127,6 +128,11 @@ instance LedgerSupportsMempool ByronBlock where
 instance TxLimits ByronBlock where
   type TxMeasure ByronBlock = IgnoringOverflow ByteSize32
 
+  txWireSize =   fromIntegral
+               . Strict.length
+               . CC.mempoolPayloadRecoverBytes
+               . toMempoolPayload
+
   blockCapacityTxMeasure _cfg st =
       IgnoringOverflow
     $ ByteSize32

ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Mempool.hs

@@ -39,7 +39,7 @@ module Ouroboros.Consensus.Shelley.Ledger.Mempool (
 import qualified Cardano.Crypto.Hash as Hash
 import qualified Cardano.Ledger.Allegra.Rules as AllegraEra
 import           Cardano.Ledger.Alonzo.Core (Tx, TxSeq, bodyTxL, eraProtVerLow,
-                     fromTxSeq, ppMaxBBSizeL, ppMaxBlockExUnitsL, sizeTxF)
+                     fromTxSeq, ppMaxBBSizeL, ppMaxBlockExUnitsL, sizeTxF, wireSizeTxF)
 import qualified Cardano.Ledger.Alonzo.Rules as AlonzoEra
 import           Cardano.Ledger.Alonzo.Scripts (ExUnits, ExUnits',
                      pointWiseExUnits, unWrapExUnits)
@@ -385,16 +385,19 @@ instance MaxTxSizeUTxO (ConwayEra c) where
 
 instance ShelleyCompatible p (ShelleyEra c) => TxLimits (ShelleyBlock p (ShelleyEra c)) where
   type TxMeasure (ShelleyBlock p (ShelleyEra c)) = IgnoringOverflow ByteSize32
+  txWireSize (ShelleyTx _ tx) = fromIntegral (tx ^. wireSizeTxF)
   txMeasure              _cfg st tx = runValidation $ txInBlockSize st tx
   blockCapacityTxMeasure _cfg       = txsMaxBytes
 
 instance ShelleyCompatible p (AllegraEra c) => TxLimits (ShelleyBlock p (AllegraEra c)) where
   type TxMeasure (ShelleyBlock p (AllegraEra c)) = IgnoringOverflow ByteSize32
+  txWireSize (ShelleyTx _ tx) = fromIntegral (tx ^. wireSizeTxF)
   txMeasure              _cfg st tx = runValidation $ txInBlockSize st tx
   blockCapacityTxMeasure _cfg       = txsMaxBytes
 
 instance ShelleyCompatible p (MaryEra c) => TxLimits (ShelleyBlock p (MaryEra c)) where
   type TxMeasure (ShelleyBlock p (MaryEra c)) = IgnoringOverflow ByteSize32
+  txWireSize (ShelleyTx _ tx) = fromIntegral (tx ^. wireSizeTxF)
   txMeasure              _cfg st tx = runValidation $ txInBlockSize st tx
   blockCapacityTxMeasure _cfg       = txsMaxBytes
 
@@ -485,6 +488,7 @@ instance ( ShelleyCompatible p (AlonzoEra c)
          ) => TxLimits (ShelleyBlock p (AlonzoEra c)) where
 
   type TxMeasure (ShelleyBlock p (AlonzoEra c)) = AlonzoMeasure
+  txWireSize (ShelleyTx _ tx) = fromIntegral (tx ^. wireSizeTxF)
   txMeasure              _cfg st tx = runValidation $ txMeasureAlonzo st tx
   blockCapacityTxMeasure _cfg       = blockCapacityAlonzoMeasure
 
@@ -582,12 +586,14 @@ instance ( ShelleyCompatible p (BabbageEra c)
          ) => TxLimits (ShelleyBlock p (BabbageEra c)) where
 
   type TxMeasure (ShelleyBlock p (BabbageEra c)) = ConwayMeasure
+  txWireSize (ShelleyTx _ tx) = fromIntegral (tx ^. wireSizeTxF)
   txMeasure              _cfg st tx = runValidation $ txMeasureBabbage st tx
   blockCapacityTxMeasure _cfg       = blockCapacityConwayMeasure
 
 instance ( ShelleyCompatible p (ConwayEra c)
          ) => TxLimits (ShelleyBlock p (ConwayEra c)) where
 
   type TxMeasure (ShelleyBlock p (ConwayEra c)) = ConwayMeasure
+  txWireSize (ShelleyTx _ tx) = fromIntegral (tx ^. wireSizeTxF)
   txMeasure              _cfg st tx = runValidation $ txMeasureConway st tx
   blockCapacityTxMeasure _cfg       = blockCapacityConwayMeasure

ouroboros-consensus-cardano/src/unstable-byronspec/Ouroboros/Consensus/ByronSpec/Ledger/Mempool.hs

@@ -57,6 +57,7 @@ instance TxLimits ByronSpecBlock where
   type TxMeasure ByronSpecBlock = IgnoringOverflow ByteSize32
 
   -- Dummy values, as these are not used in practice.
+  txWireSize = const . fromIntegral $ (0 :: Int)
   blockCapacityTxMeasure _cfg _st = IgnoringOverflow $ ByteSize32 1
 
   txMeasure _cfg _st _tx = pure $ IgnoringOverflow $ ByteSize32 0

ouroboros-consensus-diffusion/changelog.d/20240808_125632_marcin.wojtowicz_tx_wire_size.md

@@ -0,0 +1,23 @@
+<!--
+A new scriv changelog fragment.
+
+Uncomment the section that is right (remove the HTML comment wrapper).
+-->
+
+<!--
+### Patch
+
+- A bullet item for the Patch category.
+
+-->
+
+### Non-Breaking
+
+- Provide txWireSize to tx-submission protocol
+
+<!--
+### Breaking
+
+- A bullet item for the Breaking category.
+
+-->

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs

@@ -118,6 +118,13 @@ import           Ouroboros.Network.Protocol.TxSubmission2.Codec
 import           Ouroboros.Network.Protocol.TxSubmission2.Server
 import           Ouroboros.Network.Protocol.TxSubmission2.Type
 import           Ouroboros.Network.TxSubmission.Inbound
+import           Ouroboros.Network.TxSubmission.Inbound.Policy
+                     (TxDecisionPolicy (..))
+import           Ouroboros.Network.TxSubmission.Inbound.Registry (PeerTxAPI,
+                     withPeer)
+import           Ouroboros.Network.TxSubmission.Inbound.Server
+                     (EnableNewTxSubmissionProtocol (..), txSubmissionInboundV2)
+import           Ouroboros.Network.TxSubmission.Inbound.Types (TraceTxLogic)
 import           Ouroboros.Network.TxSubmission.Mempool.Reader
                      (mapTxSubmissionMempoolReader)
 import           Ouroboros.Network.TxSubmission.Outbound
@@ -168,7 +175,13 @@ data Handlers m addr blk = Handlers {
     , hTxSubmissionServer
         :: NodeToNodeVersion
         -> ConnectionId addr
-        -> TxSubmissionServerPipelined (GenTxId blk) (GenTx blk) m ()
+        -> Either
+            (TxSubmissionServerPipelined (GenTxId blk) (GenTx blk) m ())
+            (PeerTxAPI m (GenTxId blk) (GenTx blk)
+               -> TxSubmissionServerPipelined (GenTxId blk) (GenTx blk) m ())
+        -- ^ Either we use the legacy tx submission protocol or the newest one
+        -- which require PeerTxAPI. This is decided by
+        -- 'EnableNewTxSubmissionProtocol' flag.
 
     , hKeepAliveClient
         :: NodeToNodeVersion
@@ -209,10 +222,12 @@ mkHandlers ::
      )
   => NodeKernelArgs m addrNTN addrNTC blk
   -> NodeKernel     m addrNTN addrNTC blk
+  -> EnableNewTxSubmissionProtocol
   -> Handlers       m addrNTN           blk
 mkHandlers
       NodeKernelArgs {chainSyncFutureCheck, chainSyncHistoricityCheck, keepAliveRng, miniProtocolParameters}
-      NodeKernel {getChainDB, getMempool, getTopLevelConfig, getTracers = tracers, getPeerSharingAPI, getGsmState} =
+      NodeKernel {getChainDB, getMempool, getTopLevelConfig, getTracers = tracers, getPeerSharingAPI, getGsmState}
+      enableNewTxSubmissionProtocol =
     Handlers {
         hChainSyncClient = \peer _isBigLedgerpeer dynEnv ->
           CsClient.chainSyncClient
@@ -243,17 +258,32 @@ mkHandlers
       , hTxSubmissionClient = \version controlMessageSTM peer ->
           txSubmissionOutbound
             (contramap (TraceLabelPeer peer) (Node.txOutboundTracer tracers))
-            (NumTxIdsToAck $ txSubmissionMaxUnacked miniProtocolParameters)
+            (NumTxIdsToAck $ getNumTxIdsToReq
+                           $ maxUnacknowledgedTxIds
+                           $ txDecisionPolicy
+                           $ miniProtocolParameters)
             (mapTxSubmissionMempoolReader txForgetValidated $ getMempoolReader getMempool)
             version
             controlMessageSTM
       , hTxSubmissionServer = \version peer ->
-          txSubmissionInbound
-            (contramap (TraceLabelPeer peer) (Node.txInboundTracer tracers))
-            (NumTxIdsToAck $ txSubmissionMaxUnacked miniProtocolParameters)
-            (mapTxSubmissionMempoolReader txForgetValidated $ getMempoolReader getMempool)
-            (getMempoolWriter getMempool)
-            version
+          case enableNewTxSubmissionProtocol of
+            EnableNewTxSubmissionProtocol ->
+              Right $ \api ->
+                txSubmissionInboundV2
+                  (contramap (TraceLabelPeer peer) (Node.txInboundTracer tracers))
+                  (getMempoolWriter getMempool)
+                  api
+            DisableNewTxSubmissionProtocol ->
+                Left
+              $ txSubmissionInbound
+                  (contramap (TraceLabelPeer peer) (Node.txInboundTracer tracers))
+                  (NumTxIdsToAck $ getNumTxIdsToReq
+                                 $ maxUnacknowledgedTxIds
+                                 $ txDecisionPolicy
+                                 $ miniProtocolParameters)
+                  (mapTxSubmissionMempoolReader txForgetValidated $ getMempoolReader getMempool)
+                  (getMempoolWriter getMempool)
+                  version
       , hKeepAliveClient = \_version -> keepAliveClient (Node.keepAliveClientTracer tracers) keepAliveRng
       , hKeepAliveServer = \_version _peer -> keepAliveServer
       , hPeerSharingClient = \_version controlMessageSTM _peer -> peerSharingClient controlMessageSTM
@@ -375,6 +405,7 @@ data Tracers' peer blk e f = Tracers {
     , tBlockFetchTracer           :: f (TraceLabelPeer peer (TraceSendRecv (BlockFetch blk (Point blk))))
     , tBlockFetchSerialisedTracer :: f (TraceLabelPeer peer (TraceSendRecv (BlockFetch (Serialised blk) (Point blk))))
     , tTxSubmission2Tracer        :: f (TraceLabelPeer peer (TraceSendRecv (TxSubmission2 (GenTxId blk) (GenTx blk))))
+    , tTxLogicTracer              :: f (TraceLabelPeer peer (TraceTxLogic peer (GenTxId blk) (GenTx blk)))
     }
 
 instance (forall a. Semigroup (f a)) => Semigroup (Tracers' peer blk e f) where
@@ -384,6 +415,7 @@ instance (forall a. Semigroup (f a)) => Semigroup (Tracers' peer blk e f) where
       , tBlockFetchTracer           = f tBlockFetchTracer
       , tBlockFetchSerialisedTracer = f tBlockFetchSerialisedTracer
       , tTxSubmission2Tracer        = f tTxSubmission2Tracer
+      , tTxLogicTracer              = f tTxLogicTracer
       }
     where
       f :: forall a. Semigroup a
@@ -399,6 +431,7 @@ nullTracers = Tracers {
     , tBlockFetchTracer           = nullTracer
     , tBlockFetchSerialisedTracer = nullTracer
     , tTxSubmission2Tracer        = nullTracer
+    , tTxLogicTracer              = nullTracer
     }
 
 showTracers :: ( Show blk
@@ -416,6 +449,7 @@ showTracers tr = Tracers {
     , tBlockFetchTracer           = showTracing tr
     , tBlockFetchSerialisedTracer = showTracing tr
     , tTxSubmission2Tracer        = showTracing tr
+    , tTxLogicTracer              = showTracing tr
     }
 
 {-------------------------------------------------------------------------------
@@ -533,7 +567,7 @@ mkApps ::
      , ShowProxy blk
      , ShowProxy (Header blk)
      , ShowProxy (TxId (GenTx blk))
-     , ShowProxy (GenTx blk)
+     , ShowProxy (GenTx blk), HasTxId (GenTx blk), LedgerSupportsMempool blk, Show addrNTN
      )
   => NodeKernel m addrNTN addrNTC blk -- ^ Needed for bracketing only
   -> Tracers m (ConnectionId addrNTN) blk e
@@ -695,13 +729,28 @@ mkApps kernel Tracers {..} mkCodecs ByteLimits {..} genChainSyncTimeout lopBucke
       -> m ((), Maybe bTX)
     aTxSubmission2Server version ResponderContext { rcConnectionId = them } channel = do
       labelThisThread "TxSubmissionServer"
-      runPipelinedPeerWithLimits
-        (contramap (TraceLabelPeer them) tTxSubmission2Tracer)
-        (cTxSubmission2Codec (mkCodecs version))
-        blTxSubmission2
-        timeLimitsTxSubmission2
-        channel
-        (txSubmissionServerPeerPipelined (hTxSubmissionServer version them))
+
+      let runServer serverApi =
+              runPipelinedPeerWithLimits
+                  (contramap (TraceLabelPeer them) tTxSubmission2Tracer)
+                  (cTxSubmission2Codec (mkCodecs version))
+                  blTxSubmission2
+                  timeLimitsTxSubmission2
+                  channel
+                  (txSubmissionServerPeerPipelined serverApi)
+
+      case hTxSubmissionServer version them of
+        Left legacyTxSubmissionServer ->
+            runServer legacyTxSubmissionServer
+        Right newTxSubmissionServer ->
+          withPeer (contramap (TraceLabelPeer them) tTxLogicTracer)
+                   (getTxChannelsVar kernel)
+                   (getSharedTxStateVar kernel)
+                   (mapTxSubmissionMempoolReader txForgetValidated
+                   $ getMempoolReader (getMempool kernel))
+                   txWireSize
+                   them $ \api ->
+                     runServer (newTxSubmissionServer api)
 
     aKeepAliveClient
       :: NodeToNodeVersion

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Node.hs

@@ -136,6 +136,8 @@ import           Ouroboros.Network.PeerSelection.PeerSharing (PeerSharing)
 import           Ouroboros.Network.PeerSelection.PeerSharing.Codec
                      (decodeRemoteAddress, encodeRemoteAddress)
 import           Ouroboros.Network.RethrowPolicy
+import           Ouroboros.Network.TxSubmission.Inbound.Server
+                     (EnableNewTxSubmissionProtocol)
 import qualified SafeWildCards
 import           System.Exit (ExitCode (..))
 import           System.FilePath ((</>))
@@ -200,6 +202,9 @@ data RunNodeArgs m addrNTN addrNTC blk (p2p :: Diffusion.P2P) = RunNodeArgs {
     , rnGetUseBootstrapPeers :: STM m UseBootstrapPeers
 
     , rnGenesisConfig :: GenesisConfig
+
+    -- | Enable or disable the new tx submission protocol
+    , rnEnableNewTxSubmissionProtocol :: EnableNewTxSubmissionProtocol
     }
 
 
@@ -400,6 +405,7 @@ runWith :: forall m addrNTN addrNTC versionDataNTN versionDataNTC blk p2p.
      , Hashable addrNTN -- the constraint comes from `initNodeKernel`
      , NetworkIO m
      , NetworkAddr addrNTN
+     , Show addrNTN
      )
   => RunNodeArgs m addrNTN addrNTC blk p2p
   -> (NodeToNodeVersion -> addrNTN -> CBOR.Encoding)
@@ -567,7 +573,7 @@ runWith RunNodeArgs{..} encAddrNtN decAddrNtN LowLevelRunNodeArgs{..} =
           (gcChainSyncLoPBucketConfig llrnGenesisConfig)
           (gcCSJConfig llrnGenesisConfig)
           (reportMetric Diffusion.peerMetricsConfiguration peerMetrics)
-          (NTN.mkHandlers nodeKernelArgs nodeKernel)
+          (NTN.mkHandlers nodeKernelArgs nodeKernel rnEnableNewTxSubmissionProtocol)
 
     mkNodeToClientApps
       :: NodeKernelArgs m addrNTN (ConnectionId addrNTC) blk