Merge pull request #4887 from IntersectMBO/coot/tx-submission

TX Submission Logic

Author: coot

cabal.project

@@ -18,7 +18,7 @@ index-state:
   , hackage.haskell.org 2025-08-05T15:28:56Z
 
   -- Bump this if you need newer packages from CHaP
-  , cardano-haskell-packages 2025-02-15T18:39:38Z
+  , cardano-haskell-packages 2025-03-18T17:41:11Z
 
 packages: ./cardano-ping
           ./monoidal-synchronisation

flake.lock

@@ -221,7 +221,7 @@ socketAsBearer sduSize batchSize readBuffer_m sduTimeout egressInterval sd =
           let ts32 = Mx.timestampMicrosecondsLow32Bits ts
               buf  = map (Mx.encodeSDU .
                            (\sdu -> Mx.setTimestamp sdu (Mx.RemoteClockModel ts32))) sdus
-          r <- timeout ((fromIntegral $ length sdus) * sduTimeout) $
+          r <- timeout (fromIntegral (length sdus) * sduTimeout) $
               Socket.sendMany sd (concatMap BL.toChunks buf)
               `catch` Mx.handleIOException "sendAll errored"
           case r of

network-mux/src/Network/Mux/Bearer/Socket.hs

@@ -18,13 +18,22 @@ import Network.Mux.Types
 -- >  0                   1                   2                   3
 -- >  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 -- > +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
--- > |              transmission time                                |
+-- > |                        transmission time                      |
 -- > +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
--- > |M|    conversation id          |              length           |
+-- > |d|    mini-protocol number     |             length            |
 -- > +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 --
 -- All fields are in big endian byte order.
 --
+-- * transmission time: time when the SDU was sent
+-- * @d@: mini-protocol direction (`MiniProtocolDir`):
+--
+--     * 1 - initiator direction
+--     * 0 - responder direction
+--
+-- * mini-protocol number (`MiniProtocolNum`)
+-- * length: length of the payload
+--
 encodeSDU :: SDU -> BL.ByteString
 encodeSDU sdu =
   let hdr = Bin.runPut enc in

network-mux/src/Network/Mux/Codec.hs

@@ -114,7 +114,7 @@ let
         packages.ouroboros-network-protocols.components.tests.test.preCheck =
           if buildSystem == "x86_64-linux" then "export GHCRTS=-M800M" else "";
         packages.ouroboros-network.components.tests.sim-tests.preCheck =
-          if buildSystem == "x86_64-linux" then "export GHCRTS=-M2400M" else "";
+          if buildSystem == "x86_64-linux" then "export GHCRTS=-M7000M" else "";
       })
       ({ pkgs, ... }: lib.mkIf pkgs.stdenv.hostPlatform.isWindows {
         packages.basement.configureFlags = [ "--hsc2hs-options=--cflag=-Wno-int-conversion" ];

nix/ouroboros-network.nix

@@ -28,6 +28,7 @@
 
 * `IsLedgerPeer` added to `Ouroboros.Network.LedgerPeers.Types` module.
 * Added `ProtocolTimeLimitsWithRnd` to `Ouroboros.Network.Protocol.Limits`
+* Derived `Bounded` instance for `SizeInBytes`.
 
 ## 0.14.1.0 -- 2025-07-17
 

ouroboros-network-api/CHANGELOG.md

@@ -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-api/src/Ouroboros/Network/SizeInBytes.hs

@@ -219,7 +219,11 @@ instance Hashable LocalAddress where
 newtype TestAddress addr = TestAddress { getTestAddress :: addr }
   deriving (Eq, Ord, Generic, NFData)
   deriving NoThunks via InspectHeap (TestAddress addr)
-  deriving Show via Quiet (TestAddress addr)
+
+instance Show addr => Show (TestAddress addr) where
+   showsPrec d (TestAddress addr) = showParen (d > app_prec) $
+        showString "TestAddress " . showsPrec (app_prec+1) addr
+     where app_prec = 10
 
 instance Hashable addr => Hashable (TestAddress addr)
 

ouroboros-network-framework/src/Ouroboros/Network/Snocket.hs

@@ -16,6 +16,8 @@
 * `timeLimitsChainSync` changed type: it is a function which takes the idle timeout as an argument.
 * `Ouroboros.Network.Protocols.TxSubmission2.Codec.{encode,decode}TxSubmission2`
   are no longer exported.
+* `CollectPipelined` constructor for `TxSubmission2.Server` was modified: now
+  one can run a monadic action in the continuation when no message is available.
 
 ### Non-breaking changes
 

ouroboros-network-protocols/CHANGELOG.md

@@ -25,6 +25,8 @@ module Ouroboros.Network.Protocol.TxSubmission2.Server
   ) where
 
 import Data.List.NonEmpty (NonEmpty)
+import Data.Map.Strict (Map)
+import Data.Map.Strict qualified as Map
 
 import Network.TypedProtocol.Core
 import Network.TypedProtocol.Peer.Server
@@ -51,7 +53,7 @@ data Collect txid tx =
        -- contains the transactions sent, but this pairs them up with the
        -- transactions requested. This is because the peer can determine that
        -- some transactions are no longer needed.
-     | CollectTxs [txid] [tx]
+     | CollectTxs (Map txid SizeInBytes) [tx]
 
 
 data ServerStIdle (n :: N) txid tx m a where
@@ -77,16 +79,16 @@ data ServerStIdle (n :: N) txid tx m a where
   -- |
   --
   SendMsgRequestTxsPipelined
-    :: [txid]
+    :: Map txid SizeInBytes
     -> m (ServerStIdle (S n) txid tx m a)
     -> ServerStIdle       n  txid tx m a
 
   -- | 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'.
@@ -127,12 +129,12 @@ txSubmissionServerPeerPipelined (TxSubmissionServerPipelined server) =
 
     go (SendMsgRequestTxsPipelined txids k) =
       YieldPipelined
-        (MsgRequestTxs txids)
+        (MsgRequestTxs $ Map.keys txids)
         (ReceiverAwait \case
            MsgReplyTxs txs -> ReceiverDone (CollectTxs txids txs))
         (Effect (go <$> k))
 
     go (CollectPipelined mNone collect) =
-      Collect (fmap go mNone)
-              (Effect . fmap go . collect)
+      Collect (Effect . fmap go <$> mNone)
+              (Effect . fmap go  .  collect)