dmq: NtC applications

Author: crocodile-dentist

decentralized-message-queue/app/Main.hs

@@ -16,6 +16,7 @@ import DMQ.Configuration.Topology (readTopologyFileOrError)
 import DMQ.Diffusion.Applications (diffusionApplications)
 import DMQ.Diffusion.Arguments (diffusionArguments)
 import DMQ.Diffusion.NodeKernel (withNodeKernel)
+import DMQ.NodeToClient qualified as NtC
 import DMQ.NodeToNode (dmqCodecs, dmqLimitsAndTimeouts, mapNtNDMQtoOuroboros,
            ntnApps)
 
@@ -53,6 +54,9 @@ runDMQ cliopts@CLIOptions {
                              (decodeRemoteAddress (mapNtNDMQtoOuroboros maxBound)))
                   dmqLimitsAndTimeouts
                   defaultSigDecisionPolicy
+        dmqNtCApps =
+          NtC.ntcApps nodeKernel
+                      NtC.dmqCodecs
         dmqDiffusionArguments =
           diffusionArguments debugTracer
                              debugTracer
@@ -62,6 +66,7 @@ runDMQ cliopts@CLIOptions {
                                 dmqDiffusionConfiguration
                                 dmqLimitsAndTimeouts
                                 dmqNtNApps
+                                dmqNtCApps
                                 (policy policyRng)
 
     Diffusion.run dmqDiffusionArguments

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

@@ -54,6 +54,8 @@ library
     DMQ.Diffusion.PeerSelection
     DMQ.NodeToClient
     DMQ.NodeToNode
+    DMQ.NtC_Applications.LocalMsgSubmission
+    DMQ.NtC_Applications.LocalMsgNotification
     DMQ.Protocol.LocalMsgNotification.Client
     DMQ.Protocol.LocalMsgNotification.Codec
     DMQ.Protocol.LocalMsgNotification.Server
@@ -85,6 +87,7 @@ library
     ouroboros-network-framework ^>=0.19,
     ouroboros-network-protocols ^>=0.15,
     random ^>=1.2,
+    serialise,
     text >=1.2.4 && <2.2,
     time ^>= 1.12,
     typed-protocols:{typed-protocols, cborg} ^>=1.0,

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

@@ -10,11 +10,11 @@ import DMQ.NodeToClient (NodeToClientVersion, NodeToClientVersionData,
            stdVersionDataNTC)
 import DMQ.NodeToNode (NodeToNodeVersion, NodeToNodeVersionData,
            stdVersionDataNTN)
+import DMQ.NodeToClient qualified as NTC
 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 extraFlags 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 dmqcNetworkMagic)
-            (\_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,3 +1,4 @@
+{-# LANGUAGE DataKinds      #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DeriveGeneric  #-}
 {-# LANGUAGE LambdaCase     #-}
@@ -6,28 +7,51 @@
 module DMQ.NodeToClient where
 
 import Codec.CBOR.Term qualified as CBOR
+import Codec.Serialise (Serialise (decode, encode))
 import Control.DeepSeq (NFData)
 import Control.Monad ((>=>))
 import Data.Bits (Bits (..))
+import Data.ByteString.Lazy (ByteString)
 import Data.Text (Text)
 import Data.Text qualified as T
+import Data.Void
 import GHC.Generics (Generic)
 
+import Control.Monad.Class.MonadFork
 import Control.Monad.Class.MonadST (MonadST)
+import Control.Concurrent.Class.MonadSTM
+import Control.Monad.Class.MonadThrow
 import Control.Tracer (Tracer, nullTracer)
 
 import Network.Mux qualified as Mx
+import Network.TypedProtocol.Codec hiding (encode, decode)
+import Network.TypedProtocol.Codec.CBOR qualified as CBOR
 
+import DMQ.Diffusion.NodeKernel
+import DMQ.NtC_Applications.LocalMsgSubmission
+import DMQ.NtC_Applications.LocalMsgNotification
+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 DMQ.Protocol.SigSubmission.Codec
+import DMQ.Protocol.SigSubmission.Type
+
+import Ouroboros.Network.Context
 import Ouroboros.Network.CodecCBORTerm (CodecCBORTerm (..))
-import Ouroboros.Network.ConnectionId (ConnectionId)
-import Ouroboros.Network.Driver.Simple (TraceSendRecv)
+import Ouroboros.Network.Driver.Simple
 import Ouroboros.Network.Handshake.Acceptable (Acceptable (..))
 import Ouroboros.Network.Handshake.Queryable (Queryable (..))
 import Ouroboros.Network.Magic (NetworkMagic (..))
+import Ouroboros.Network.Mux
 import Ouroboros.Network.Protocol.Handshake (Accept (..), Handshake,
            HandshakeArguments (..))
 import Ouroboros.Network.Protocol.Handshake.Codec (cborTermVersionDataCodec,
            codecHandshake, noTimeLimitsHandshake)
+import Ouroboros.Network.TxSubmission.Mempool.Simple qualified as Mempool
+
 
 data NodeToClientVersion =
   NodeToClientV_1
@@ -106,9 +130,6 @@ nodeToClientCodecCBORTerm _v = CodecCBORTerm {encodeTerm, decodeTerm}
       decoder x query | x >= 0 && x <= 0xffffffff = Right (NodeToClientVersionData (NetworkMagic $ fromIntegral x) query)
                       | otherwise                 = Left $ T.pack $ "networkMagic out of bound: " <> show x
 
-data Protocols =
-  Protocols {
-  }
 
 type HandshakeTr ntcAddr = Mx.WithBearer (ConnectionId ntcAddr) (TraceSendRecv (Handshake NodeToClientVersion CBOR.Term))
 
@@ -139,3 +160,159 @@ stdVersionDataNTC networkMagic =
     { networkMagic
     , query        = False
     }
+
+
+-- TODO: delete these aliases
+type LocalMsgSubmission' = LocalMsgSubmission Sig Int
+type LocalMsgNotification' = LocalMsgNotification Sig
+
+data Codecs m =
+  Codecs {
+    msgSubmissionCodec
+      :: Codec LocalMsgSubmission'
+               CBOR.DeserialiseFailure m ByteString
+  , msgNotificationCodec
+      :: Codec LocalMsgNotification'
+               CBOR.DeserialiseFailure m ByteString
+  }
+
+dmqCodecs :: MonadST m
+          => Codecs m
+dmqCodecs =
+  Codecs {
+    msgSubmissionCodec = codecLocalMsgSubmission encodeSig decodeSig encode decode
+  , 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)
+  => NodeKernel ntnAddr m
+  -> Codecs m
+  -> Apps ntcAddr m ()
+ntcApps NodeKernel { mempool }
+        Codecs { msgSubmissionCodec, msgNotificationCodec }=
+  Apps {
+    aLocalMsgSubmission
+  , aLocalMsgNotification
+  }
+  where
+    sigSize :: Sig -> SizeInBytes
+    sigSize _ = 0 -- TODO
+
+    mempoolReader = Mempool.getReader sigId sigSize mempool
+    mempoolWriter = Mempool.getWriter sigId (const True) mempool
+
+    aLocalMsgSubmission _version _ctx channel = do
+      labelThisThread "LocalMsgSubmissionServer"
+      runPeer
+        nullTracer
+        msgSubmissionCodec
+        channel
+        (localMsgSubmissionServerPeer $
+          localMsgSubmissionServer undefined mempoolWriter)
+
+    aLocalMsgNotification _version _ctx channel = do
+      labelThisThread "LocalMsgNotificationServer"
+      runPeer
+        nullTracer
+        msgNotificationCodec
+        channel
+        (localMsgNotificationServerPeer $
+          localMsgNotificationServer undefined nullTracer undefined 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
+    -- TODO: verify protocol numbers
+    localMsgSubmission protocol = MiniProtocol {
+        miniProtocolNum    = MiniProtocolNum 10,
+        miniProtocolStart  = StartOnDemand,
+        miniProtocolLimits = maximumMiniProtocolLimits,
+        miniProtocolRun    = protocol
+      }
+    localMsgNotification protocol = MiniProtocol {
+        miniProtocolNum    = MiniProtocolNum 11,
+        miniProtocolStart  = StartOnDemand,
+        miniProtocolLimits = maximumMiniProtocolLimits,
+        miniProtocolRun    = protocol
+    }
+    maximumMiniProtocolLimits =
+      MiniProtocolLimits {
+        maximumIngressQueue = 0xffffffff
+      }