wip

Author: crocodile-dentist

dmq-node/src/DMQ/NodeToClient.hs

@@ -16,6 +16,7 @@ module DMQ.NodeToClient
 import Data.Aeson qualified as Aeson
 import Data.ByteString.Lazy (ByteString)
 import Data.Functor.Contravariant ((>$<))
+import Data.Typeable
 import Data.Void
 import Data.Word
 
@@ -132,20 +133,22 @@ data Apps ntcAddr m a =
 -- | Construct applications for the node-to-client protocols
 --
 ntcApps
-  :: forall crypto idx ntcAddr failure m.
+  :: forall crypto idx ntcAddr m.
      ( MonadThrow m
      , MonadThread m
      , MonadSTM m
      , Crypto crypto
      , Aeson.ToJSON ntcAddr
      , Aeson.ToJSON (MempoolAddFail (Sig crypto))
+     , Show (MempoolAddFail (Sig crypto))
      , ShowProxy (MempoolAddFail (Sig crypto))
      , ShowProxy (Sig crypto)
+     , Typeable crypto
      )
   => (forall ev. Aeson.ToJSON ev => Tracer m (WithEventType ev))
   -> Configuration
   -> TxSubmissionMempoolReader SigId (Sig crypto) idx m
-  -> MempoolWriter SigId (Sig crypto) failure idx m
+  -> MempoolWriter SigId (Sig crypto) idx m
   -> Word16
   -> Codecs crypto m
   -> Apps ntcAddr m ()

dmq-node/src/DMQ/NodeToClient/LocalMsgSubmission.hs

@@ -1,13 +1,17 @@
-{-# LANGUAGE FlexibleContexts   #-}
-{-# LANGUAGE OverloadedStrings  #-}
-{-# LANGUAGE StandaloneDeriving #-}
+{-# LANGUAGE FlexibleContexts     #-}
+{-# LANGUAGE OverloadedStrings    #-}
+{-# LANGUAGE StandaloneDeriving   #-}
+{-# LANGUAGE TypeApplications     #-}
+{-# LANGUAGE UndecidableInstances #-}
 
 module DMQ.NodeToClient.LocalMsgSubmission where
 
 import Control.Concurrent.Class.MonadSTM
+import Control.Monad.Class.MonadThrow
 import Control.Tracer
 import Data.Aeson (ToJSON (..), object, (.=))
 import Data.Aeson qualified as Aeson
+import Data.Typeable
 
 import DMQ.Protocol.LocalMsgSubmission.Server
 import DMQ.Protocol.LocalMsgSubmission.Type
@@ -16,55 +20,78 @@ import Ouroboros.Network.TxSubmission.Mempool.Simple
 -- | Local transaction submission server, for adding txs to the 'Mempool'
 --
 localMsgSubmissionServer ::
-     MonadSTM m
-  => (sig -> sigid)
+     forall msgid msg idx m.
+     ( MonadSTM m
+     , MonadThrow m
+     , Typeable msgid
+     , Typeable msg
+     , Show msgid
+     , Show (MempoolAddFail msg))
+  => (msg -> msgid)
   -- ^ get message id
-  -> Tracer m (TraceLocalMsgSubmission sig sigid)
-  -> MempoolWriter sigid sig failure idx m
+  -> Tracer m (TraceLocalMsgSubmission msg msgid)
+  -> MempoolWriter msgid msg idx m
   -- ^ duplicate error tag in case the mempool returns the empty list on failure
-  -> m (LocalMsgSubmissionServer sig m ())
+  -> m (LocalMsgSubmissionServer msg m ())
 localMsgSubmissionServer getMsgId tracer MempoolWriter { mempoolAddTxs } =
     pure server
   where
-    process (sigid, e@(SubmitFail reason)) =
-      (e, server) <$ traceWith tracer (TraceSubmitFailure sigid reason)
-    process (sigid, success) =
-      (success, server) <$ traceWith tracer (TraceSubmitAccept sigid)
+    process (Left (msgid, reason)) = do
+      traceWith tracer (TraceSubmitFailure msgid reason)
+      throwIO $ MsgValidationException msgid reason
+    process (Right [(msgid, e@(SubmitFail reason))]) =
+      (e, server) <$ traceWith tracer (TraceSubmitFailure msgid reason)
+    process (Right [(msgid, SubmitSuccess)]) =
+      (SubmitSuccess, server) <$ traceWith tracer (TraceSubmitAccept msgid)
+    process _ = throwIO (TooManyMessages @msgid @msg)
 
     server = LocalTxSubmissionServer {
-      recvMsgSubmitTx = \sig -> do
-        traceWith tracer $ TraceReceivedMsg (getMsgId sig)
-        process . head =<< mempoolAddTxs [sig]
+      recvMsgSubmitTx = \msg -> do
+        traceWith tracer $ TraceReceivedMsg (getMsgId msg)
+        process =<< mempoolAddTxs [msg]
 
     , recvMsgDone = ()
     }
 
 
-data TraceLocalMsgSubmission sig sigid =
-    TraceReceivedMsg sigid
+data TraceLocalMsgSubmission msg msgid =
+    TraceReceivedMsg msgid
   -- ^ A signature was received.
-  | TraceSubmitFailure sigid (MempoolAddFail sig)
-  | TraceSubmitAccept sigid
+  | TraceSubmitFailure msgid (MempoolAddFail msg)
+  | TraceSubmitAccept msgid
 
 deriving instance
-     (Show sig, Show sigid, Show (MempoolAddFail sig))
-  => Show (TraceLocalMsgSubmission sig sigid)
+     (Show msg, Show msgid, Show (MempoolAddFail msg))
+  => Show (TraceLocalMsgSubmission msg msgid)
 
-instance (ToJSON sigid, ToJSON (MempoolAddFail sig))
-      => ToJSON (TraceLocalMsgSubmission sig sigid) where
-  toJSON (TraceReceivedMsg sigid) =
+
+
+data MsgSubmissionServerException msgid msg =
+    MsgValidationException msgid (MempoolAddFail msg)
+  | TooManyMessages
+
+deriving instance (Show (MempoolAddFail msg), Show msgid)
+  => Show (MsgSubmissionServerException msgid msg)
+
+instance (Typeable msgid, Typeable msg, Show (MempoolAddFail msg), Show msgid)
+  => Exception (MsgSubmissionServerException msgid msg) where
+
+
+instance (ToJSON msgid, ToJSON (MempoolAddFail msg))
+      => ToJSON (TraceLocalMsgSubmission msg msgid) where
+  toJSON (TraceReceivedMsg msgid) =
     -- TODO: once we have verbosity levels, we could include the full tx, for
     -- now one can use `TraceSendRecv` tracer for the mini-protocol to see full
     -- msgs.
     object [ "kind" .= Aeson.String "TraceReceivedMsg"
-           , "sigId" .= sigid
+           , "sigId" .= msgid
            ]
-  toJSON (TraceSubmitFailure sigid reject) =
+  toJSON (TraceSubmitFailure msgid reject) =
     object [ "kind" .= Aeson.String "TraceSubmitFailure"
-           , "sigId" .= sigid
+           , "sigId" .= msgid
            , "reason" .= reject
            ]
-  toJSON (TraceSubmitAccept sigid) =
+  toJSON (TraceSubmitAccept msgid) =
     object [ "kind" .= Aeson.String "TraceSubmitAccept"
-           , "sigId" .= sigid
+           , "sigId" .= msgid
            ]

dmq-node/src/DMQ/Protocol/LocalMsgSubmission/Codec.hs

@@ -10,8 +10,11 @@ import Codec.CBOR.Encoding qualified as CBOR
 import Codec.CBOR.Read qualified as CBOR
 import Control.Monad.Class.MonadST
 import Data.ByteString.Lazy (ByteString)
+import Data.Text qualified as T
+import Data.Tuple (swap)
 import Text.Printf
 
+import Cardano.Binary
 import Cardano.KESAgent.KES.Crypto (Crypto (..))
 
 import DMQ.Protocol.LocalMsgSubmission.Type
@@ -35,7 +38,25 @@ codecLocalMsgSubmission =
 
 encodeReject :: MempoolAddFail (Sig crypto) -> CBOR.Encoding
 encodeReject = \case
-  SigInvalid reason -> CBOR.encodeListLen 2 <> CBOR.encodeWord 0 <> CBOR.encodeString reason
+  SigInvalid reason -> CBOR.encodeListLen 2 <> CBOR.encodeWord 0 <> e
+    where
+      e = case reason of
+        InvalidKESSignature ocertKESPeriod sigKESPeriod err ->  mconcat [
+          CBOR.encodeListLen 4, CBOR.encodeWord 0, toCBOR ocertKESPeriod, toCBOR sigKESPeriod, CBOR.encodeString (T.pack err)
+          ]
+        InvalidSignatureOCERT ocertN sigKESPeriod err -> mconcat [
+          CBOR.encodeListLen 4, CBOR.encodeWord 1, CBOR.encodeWord64 ocertN, toCBOR sigKESPeriod, CBOR.encodeString (T.pack err)
+          ]
+        KESBeforeStartOCERT startKESPeriod sigKESPeriod -> mconcat [
+          CBOR.encodeListLen 3, CBOR.encodeWord 2, toCBOR startKESPeriod, toCBOR sigKESPeriod
+          ]
+        KESAfterEndOCERT endKESPeriod sigKESPeriod ->  mconcat [
+          CBOR.encodeListLen 3, CBOR.encodeWord 3, toCBOR endKESPeriod, toCBOR sigKESPeriod
+          ]
+        UnrecognizedPool -> CBOR.encodeListLen 1 <> CBOR.encodeWord 4
+        ExpiredPool      -> CBOR.encodeListLen 1 <> CBOR.encodeWord 5
+        NotInitialized   -> CBOR.encodeListLen 1 <> CBOR.encodeWord 6
+        ClockSkew        -> CBOR.encodeListLen 1 <> CBOR.encodeWord 7
   SigDuplicate      -> CBOR.encodeListLen 1 <> CBOR.encodeWord 1
   SigExpired        -> CBOR.encodeListLen 1 <> CBOR.encodeWord 2
   SigResultOther reason
@@ -46,7 +67,22 @@ decodeReject = do
   len <- CBOR.decodeListLen
   tag <- CBOR.decodeWord
   case (tag, len) of
-    (0, 2)     -> SigInvalid <$> CBOR.decodeString
+    (0, 2) -> SigInvalid <$> decSigValidError
+      where
+        decSigValidError :: CBOR.Decoder s SigValidationError
+        decSigValidError = do
+          lenTag <- (,) <$> CBOR.decodeListLen <*> CBOR.decodeWord
+          case swap lenTag of
+            (0, 4) -> InvalidKESSignature <$> fromCBOR <*> fromCBOR <*> (T.unpack <$> CBOR.decodeString)
+            (1, 4) -> InvalidSignatureOCERT <$> CBOR.decodeWord64 <*> fromCBOR <*> (T.unpack <$> CBOR.decodeString)
+            (2, 3) -> KESBeforeStartOCERT <$> fromCBOR <*> fromCBOR
+            (3, 4) -> KESAfterEndOCERT <$> fromCBOR <*> fromCBOR
+            (4, 1) -> pure UnrecognizedPool
+            (5, 1) -> pure ExpiredPool
+            (6, 1) -> pure NotInitialized
+            (7, 1) -> pure ClockSkew
+            _otherwise -> fail $ printf "unrecognized (tag,len) = (%d, %d) when decoding SigInvalid tag" tag len
+
     (1, 1)     -> pure SigDuplicate
     (2, 1)     -> pure SigExpired
     (3, 2)     -> SigResultOther <$> CBOR.decodeString