dmq: NtC applications

Author: crocodile-dentist

decentralized-message-queue/app/Main.hs

@@ -17,14 +17,19 @@ import DMQ.Configuration.CLIOptions (parseCLIOptions)
 import DMQ.Configuration.Topology (readTopologyFileOrError)
 import DMQ.Diffusion.Applications (diffusionApplications)
 import DMQ.Diffusion.Arguments
-import DMQ.Diffusion.NodeKernel (withNodeKernel)
+import DMQ.Diffusion.NodeKernel (mempool, withNodeKernel)
+import DMQ.NodeToClient qualified as NtC
 import DMQ.NodeToNode (dmqCodecs, dmqLimitsAndTimeouts, ntnApps)
+import DMQ.Protocol.LocalMsgSubmission.Codec
+import DMQ.Protocol.SigSubmission.Codec
+import DMQ.Protocol.SigSubmission.Type (Sig (..))
 import DMQ.Tracer
 
 import DMQ.Diffusion.PeerSelection (policy)
 import Ouroboros.Network.Diffusion qualified as Diffusion
 import Ouroboros.Network.PeerSelection.PeerSharing.Codec (decodeRemoteAddress,
            encodeRemoteAddress)
+import Ouroboros.Network.TxSubmission.Mempool.Simple qualified as Mempool
 
 main :: IO ()
 main = void . runDMQ =<< execParser opts
@@ -78,6 +83,13 @@ runDMQ commandLineConfig = do
                                (decodeRemoteAddress maxBound))
                     dmqLimitsAndTimeouts
                     defaultSigDecisionPolicy
+          dmqNtCApps =
+            let sigSize _ = 0 -- TODO
+                maxMsgs = 1000 -- TODO: make this dynamic?
+                mempoolReader = Mempool.getReader sigId sigSize (mempool nodeKernel)
+                mempoolWriter = Mempool.getWriter sigId (const True) (mempool nodeKernel)
+             in NtC.ntcApps () mempoolReader mempoolWriter maxMsgs
+                            (NtC.dmqCodecs encodeSig decodeSig encodeReject decodeReject)
           dmqDiffusionArguments =
             diffusionArguments (if handshakeTracer
                                   then WithEventType "Handshake" >$< tracer
@@ -91,6 +103,7 @@ runDMQ commandLineConfig = do
                                   dmqDiffusionConfiguration
                                   dmqLimitsAndTimeouts
                                   dmqNtNApps
+                                  dmqNtCApps
                                   (policy policyRng)
 
       Diffusion.run dmqDiffusionArguments

decentralized-message-queue/decentralized-message-queue.cabal

@@ -56,8 +56,11 @@ library
     DMQ.NodeToClient.Version
     DMQ.NodeToNode
     DMQ.NodeToNode.Version
+    DMQ.NodeToClient.LocalMsgSubmission
+    DMQ.NodeToClient.LocalMsgNotification
     DMQ.Protocol.LocalMsgNotification.Client
     DMQ.Protocol.LocalMsgNotification.Codec
+    DMQ.Protocol.LocalMsgNotification.Examples
     DMQ.Protocol.LocalMsgNotification.Server
     DMQ.Protocol.LocalMsgNotification.Type
     DMQ.Protocol.LocalMsgSubmission.Client
@@ -143,8 +146,10 @@ test-suite dmq-test
     QuickCheck,
     base >=4.14 && <4.22,
     bytestring,
+    contra-tracer,
     decentralized-message-queue,
     ouroboros-network-api,
+    ouroboros-network-framework,
     ouroboros-network-protocols:testlib,
     ouroboros-network-testing,
     quickcheck-instances,
@@ -154,7 +159,8 @@ test-suite dmq-test
     time,
     typed-protocols:{typed-protocols, codec-properties},
     with-utf8,
-    io-classes
+    io-classes,
+    io-sim
 
   ghc-options:
     -fno-ignore-asserts

decentralized-message-queue/src/DMQ/Diffusion/Applications.hs

@@ -8,13 +8,13 @@ import DMQ.Configuration
 import DMQ.Diffusion.NodeKernel (NodeKernel (..))
 import DMQ.NodeToClient (NodeToClientVersion, NodeToClientVersionData,
            stdVersionDataNTC)
+import DMQ.NodeToClient qualified as NTC
 import DMQ.NodeToNode (NodeToNodeVersion, NodeToNodeVersionData,
            stdVersionDataNTN)
 import DMQ.NodeToNode qualified as NTN
 
 import Ouroboros.Network.Diffusion.Types qualified as Diffusion
 import Ouroboros.Network.ExitPolicy (RepromoteDelay (..))
-import Ouroboros.Network.Mux (OuroborosApplication (..))
 import Ouroboros.Network.PeerSelection.Governor.Types (PeerSelectionPolicy)
 import Ouroboros.Network.Protocol.Handshake.Version (combineVersions,
            simpleSingletonVersions)
@@ -27,6 +27,7 @@ diffusionApplications
   -> Diffusion.Configuration NoExtraFlags m ntnFd ntnAddr ntcFd ntcAddr
   -> NTN.LimitsAndTimeouts ntnAddr
   -> NTN.Apps ntnAddr m a ()
+  -> NTC.Apps ntcAddr m ()
   -> PeerSelectionPolicy ntnAddr m
   -> Diffusion.Applications ntnAddr NodeToNodeVersion   NodeToNodeVersionData
                             ntcAddr NodeToClientVersion NodeToClientVersionData
@@ -44,6 +45,7 @@ diffusionApplications
   }
   ntnLimitsAndTimeouts
   ntnApps
+  ntcApps
   peerSelectionPolicy =
   Diffusion.Applications {
     daApplicationInitiatorMode =
@@ -67,11 +69,7 @@ diffusionApplications
         [ simpleSingletonVersions
             version
             (stdVersionDataNTC networkMagic)
-            (\_versionData ->
-                OuroborosApplication
-                  [
-                  ]
-            )
+            (NTC.responders ntcApps version)
         | version <- [minBound..maxBound]
         ]
   , daRethrowPolicy                     =  muxErrorRethrowPolicy

decentralized-message-queue/src/DMQ/Diffusion/NodeKernel.hs

@@ -39,16 +39,16 @@ import DMQ.Protocol.SigSubmission.Type (Sig (..), SigId)
 data NodeKernel ntnAddr m =
   NodeKernel {
     -- | The fetch client registry, used for the keep alive clients.
-    fetchClientRegistry :: FetchClientRegistry (ConnectionId ntnAddr) () () m
+    fetchClientRegistry :: !(FetchClientRegistry (ConnectionId ntnAddr) () () m)
 
     -- | Read the current peer sharing registry, used for interacting with
     -- the PeerSharing protocol
-  , peerSharingRegistry :: PeerSharingRegistry ntnAddr m
-  , peerSharingAPI      :: PeerSharingAPI ntnAddr StdGen m
-  , mempool             :: Mempool m Sig
-  , sigChannelVar       :: TxChannelsVar m ntnAddr SigId Sig
-  , sigMempoolSem       :: TxMempoolSem m
-  , sigSharedTxStateVar :: SharedTxStateVar m ntnAddr SigId Sig
+  , peerSharingRegistry :: !(PeerSharingRegistry ntnAddr m)
+  , peerSharingAPI      :: !(PeerSharingAPI ntnAddr StdGen m)
+  , mempool             :: !(Mempool m Sig)
+  , sigChannelVar       :: !(TxChannelsVar m ntnAddr SigId Sig)
+  , sigMempoolSem       :: !(TxMempoolSem m)
+  , sigSharedTxStateVar :: !(SharedTxStateVar m ntnAddr SigId Sig)
   }
 
 newNodeKernel :: ( MonadLabelledSTM m

decentralized-message-queue/src/DMQ/NodeToClient.hs

@@ -1,35 +1,58 @@
-{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE DataKinds  #-}
+{-# LANGUAGE RankNTypes #-}
 
 module DMQ.NodeToClient
-  ( LocalAddress (..)
-  , module DMQ.NodeToClient.Version
+  ( module DMQ.NodeToClient.Version
   , Protocols (..)
   , HandshakeTr
+  , Apps
+  , dmqCodecs
+  , ntcApps
   , ntcHandshakeArguments
+  , responders
   ) where
 
+import Data.ByteString.Lazy (ByteString)
+import Data.Void
+import Data.Word
+
+import Control.Concurrent.Class.MonadSTM
+import Control.Monad.Class.MonadFork
 import Control.Monad.Class.MonadST (MonadST)
+import Control.Monad.Class.MonadThrow
 import Control.Tracer (Tracer, nullTracer)
 
+import Codec.CBOR.Decoding qualified as CBOR
+import Codec.CBOR.Encoding qualified as CBOR
 import Codec.CBOR.Term qualified as CBOR
 
 import Network.Mux qualified as Mx
+import Network.TypedProtocol.Codec hiding (decode, encode)
+import Network.TypedProtocol.Codec.CBOR qualified as CBOR
+
+import DMQ.NodeToClient.LocalMsgNotification
+import DMQ.NodeToClient.LocalMsgSubmission
+import DMQ.NodeToClient.Version
+import DMQ.Protocol.LocalMsgNotification.Codec
+import DMQ.Protocol.LocalMsgNotification.Server
+import DMQ.Protocol.LocalMsgNotification.Type
+import DMQ.Protocol.LocalMsgSubmission.Codec
+import DMQ.Protocol.LocalMsgSubmission.Server
+import DMQ.Protocol.LocalMsgSubmission.Type
 
-import Ouroboros.Network.ConnectionId (ConnectionId)
-import Ouroboros.Network.Driver.Simple (TraceSendRecv)
+import Ouroboros.Network.Context
+import Ouroboros.Network.Driver.Simple
 import Ouroboros.Network.Handshake.Acceptable (Acceptable (..))
 import Ouroboros.Network.Handshake.Queryable (Queryable (..))
+import Ouroboros.Network.Mux
 import Ouroboros.Network.Protocol.Handshake (Handshake, HandshakeArguments (..))
 import Ouroboros.Network.Protocol.Handshake.Codec (cborTermVersionDataCodec,
            codecHandshake, noTimeLimitsHandshake)
-import Ouroboros.Network.Snocket (LocalAddress (..))
-
-import DMQ.NodeToClient.Version
-
+import Ouroboros.Network.TxSubmission.Inbound.V2.Types
+           (TxSubmissionMempoolWriter)
+import Ouroboros.Network.TxSubmission.Mempool.Reader
+import Ouroboros.Network.Util.ShowProxy
 
-data Protocols =
-  Protocols {
-  }
 
 type HandshakeTr ntcAddr = Mx.WithBearer (ConnectionId ntcAddr) (TraceSendRecv (Handshake NodeToClientVersion CBOR.Term))
 
@@ -53,3 +76,154 @@ ntcHandshakeArguments tracer =
   , haQueryVersion  = queryVersion
   , haTimeLimits    = noTimeLimitsHandshake
   }
+
+
+data Codecs m sig reject =
+  Codecs {
+    msgSubmissionCodec
+      :: !(Codec (LocalMsgSubmission sig reject)
+                 CBOR.DeserialiseFailure m ByteString)
+  , msgNotificationCodec
+      :: !(Codec (LocalMsgNotification sig)
+               CBOR.DeserialiseFailure m ByteString)
+  }
+
+dmqCodecs :: MonadST m
+          => (sig -> CBOR.Encoding)
+          -> (forall s. CBOR.Decoder s sig)
+          -> (reject -> CBOR.Encoding)
+          -> (forall s. CBOR.Decoder s reject)
+          -> Codecs m sig reject
+dmqCodecs encodeSig decodeSig encodeReject' decodeReject' =
+  Codecs {
+    msgSubmissionCodec = codecLocalMsgSubmission encodeSig decodeSig encodeReject' decodeReject'
+  , msgNotificationCodec = codecLocalMsgNotification encodeSig decodeSig
+  }
+
+
+-- | A node-to-client application
+--
+type App ntcAddr m a =
+     NodeToClientVersion
+  -> ResponderContext ntcAddr
+  -> Mx.Channel m ByteString
+  -> m (a, Maybe ByteString)
+
+
+data Apps ntcAddr m a =
+  Apps {
+    -- | Start a sig-submission client
+    aLocalMsgSubmission   :: !(App ntcAddr m a)
+
+    -- | Start a sig-submission server
+  , aLocalMsgNotification :: !(App ntcAddr m a)
+  }
+
+
+-- | Construct applications for the node-to-client protocols
+--
+ntcApps
+  :: (MonadThrow m, MonadThread m, MonadSTM m, ShowProxy reject, ShowProxy sig)
+  => reject
+  -> TxSubmissionMempoolReader msgid sig idx m
+  -> TxSubmissionMempoolWriter msgid sig idx m
+  -> Word16
+  -> Codecs m sig reject
+  -> Apps ntcAddr m ()
+ntcApps reject mempoolReader mempoolWriter maxMsgs
+        Codecs { msgSubmissionCodec, msgNotificationCodec } =
+  Apps {
+    aLocalMsgSubmission
+  , aLocalMsgNotification
+  }
+  where
+    aLocalMsgSubmission _version _ctx channel = do
+      labelThisThread "LocalMsgSubmissionServer"
+      runPeer
+        nullTracer
+        msgSubmissionCodec
+        channel
+        (localMsgSubmissionServerPeer $
+          localMsgSubmissionServer nullTracer reject mempoolWriter)
+
+    aLocalMsgNotification _version _ctx channel = do
+      labelThisThread "LocalMsgNotificationServer"
+      runPeer
+        nullTracer
+        msgNotificationCodec
+        channel
+        (localMsgNotificationServerPeer $
+          localMsgNotificationServer nullTracer (pure ()) maxMsgs mempoolReader)
+
+
+data Protocols appType ntcAddr bytes m a b =
+  Protocols {
+    msgSubmissionProtocol   :: !(RunMiniProtocolWithMinimalCtx appType ntcAddr bytes m a b)
+  , msgNotificationProtocol :: !(RunMiniProtocolWithMinimalCtx appType ntcAddr bytes m a b)
+  }
+
+responders
+  :: Apps ntcAddr m a
+  -> NodeToClientVersion
+  -> NodeToClientVersionData
+  -> OuroborosApplicationWithMinimalCtx Mx.ResponderMode ntcAddr ByteString m Void a
+responders Apps {
+             aLocalMsgSubmission
+           , aLocalMsgNotification
+           }
+           version =
+  nodeToClientProtocols
+    Protocols {
+      msgSubmissionProtocol =
+        ResponderProtocolOnly $
+           MiniProtocolCb $ aLocalMsgSubmission version
+    , msgNotificationProtocol =
+        ResponderProtocolOnly $
+           MiniProtocolCb $ aLocalMsgNotification version
+    }
+    version
+
+
+-- | Make an 'OuroborosApplication' for the bundle of mini-protocols that
+-- make up the overall node-to-client protocol.
+--
+-- This function specifies the wire format protocol numbers as well as the
+-- protocols that run for each 'NodeToClientVersion'.
+--
+-- They are chosen to not overlap with the node to node protocol numbers.
+-- This is not essential for correctness, but is helpful to allow a single
+-- shared implementation of tools that can analyse both protocols, e.g.
+-- wireshark plugins.
+--
+nodeToClientProtocols
+  :: Protocols appType ntcAddr bytes m a b
+  -> NodeToClientVersion
+  -> NodeToClientVersionData
+  -> OuroborosApplicationWithMinimalCtx appType ntcAddr bytes m a b
+nodeToClientProtocols protocols _version _versionData =
+  OuroborosApplication $
+    case protocols of
+      Protocols {
+        msgSubmissionProtocol
+      , msgNotificationProtocol
+      } ->
+        [ localMsgSubmission msgSubmissionProtocol
+        , localMsgNotification msgNotificationProtocol
+        ]
+  where
+    localMsgSubmission protocol = MiniProtocol {
+        miniProtocolNum    = MiniProtocolNum 14,
+        miniProtocolStart  = StartOnDemand,
+        miniProtocolLimits = maximumMiniProtocolLimits,
+        miniProtocolRun    = protocol
+      }
+    localMsgNotification protocol = MiniProtocol {
+        miniProtocolNum    = MiniProtocolNum 15,
+        miniProtocolStart  = StartOnDemand,
+        miniProtocolLimits = maximumMiniProtocolLimits,
+        miniProtocolRun    = protocol
+    }
+    maximumMiniProtocolLimits =
+      MiniProtocolLimits {
+        maximumIngressQueue = 0xffffffff
+      }