4.4.1

* server_key_hash fields

* test

* refactor

* fix

* order

* use sync command in test

* refactor

---------

Co-authored-by: Evgeny Poberezkin <[email protected]>

CHANGELOG.md

@@ -1,3 +1,10 @@
+# 4.4.1
+
+SMP agent:
+
+- fix handling of server addresses.
+- prevent message delivery getting stuck on IO error in SMP client.
+
 # 4.4.0
 
 SMP agent:

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 4.4.0
+version: 4.4.1
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -112,7 +112,7 @@ executables:
       - -threaded
 
 tests:
-  smp-server-test:
+  simplexmq-test:
     source-dirs: tests
     main: Test.hs
     dependencies:

simplexmq.cabal

@@ -1,11 +1,11 @@
 cabal-version: 1.12
 
--- This file has been generated from package.yaml by hpack version 0.35.0.
+-- This file has been generated from package.yaml by hpack version 0.35.1.
 --
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        4.4.0
+version:        4.4.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -58,11 +58,13 @@ library
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230110_users
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230117_fkey_indexes
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230120_delete_errors
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230217_server_key_hash
       Simplex.Messaging.Agent.TAsyncs
       Simplex.Messaging.Agent.TRcvQueues
       Simplex.Messaging.Client
       Simplex.Messaging.Client.Agent
       Simplex.Messaging.Crypto
+      Simplex.Messaging.Crypto.Lazy
       Simplex.Messaging.Crypto.Ratchet
       Simplex.Messaging.Encoding
       Simplex.Messaging.Encoding.String
@@ -92,6 +94,7 @@ library
       Simplex.Messaging.Server.StoreLog
       Simplex.Messaging.TMap
       Simplex.Messaging.Transport
+      Simplex.Messaging.Transport.Buffer
       Simplex.Messaging.Transport.Client
       Simplex.Messaging.Transport.HTTP2
       Simplex.Messaging.Transport.HTTP2.Client
@@ -346,7 +349,7 @@ executable smp-server
   if flag(swift)
     cpp-options: -DswiftJSON
 
-test-suite smp-server-test
+test-suite simplexmq-test
   type: exitcode-stdio-1.0
   main-is: Test.hs
   other-modules:

src/Simplex/Messaging/Agent.hs

@@ -757,7 +757,7 @@ sendMessage' c connId msgFlags msg = withConnLock c connId "sendMessage" $ do
 enqueueCommand :: AgentMonad m => AgentClient -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> m ()
 enqueueCommand c corrId connId server aCommand = do
   resumeSrvCmds c server
-  commandId <- withStore' c $ \db -> createCommand db corrId connId server aCommand
+  commandId <- withStore c $ \db -> createCommand db corrId connId server aCommand
   queuePendingCommands c server [commandId]
 
 resumeSrvCmds :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m ()

src/Simplex/Messaging/Agent/Client.hs

@@ -5,6 +5,7 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE FunctionalDependencies #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -162,7 +163,7 @@ import UnliftIO (mapConcurrently)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
-type ClientVar msg = TMVar (Either AgentErrorType (ProtocolClient msg))
+type ClientVar err msg = TMVar (Either AgentErrorType (ProtocolClient err msg))
 
 type SMPClientVar = TMVar (Either AgentErrorType SMPClient)
 
@@ -319,15 +320,15 @@ newAgentClient InitialAgentServers {smp, ntf, netCfg} agentEnv = do
 agentClientStore :: AgentClient -> SQLiteStore
 agentClientStore AgentClient {agentEnv = Env {store}} = store
 
-class ProtocolServerClient msg where
-  getProtocolServerClient :: AgentMonad m => AgentClient -> TransportSession msg -> m (ProtocolClient msg)
-  clientProtocolError :: ErrorType -> AgentErrorType
+class ProtocolServerClient err msg | msg -> err where
+  getProtocolServerClient :: AgentMonad m => AgentClient -> TransportSession msg -> m (ProtocolClient err msg)
+  clientProtocolError :: err -> AgentErrorType
 
-instance ProtocolServerClient BrokerMsg where
+instance ProtocolServerClient ErrorType BrokerMsg where
   getProtocolServerClient = getSMPServerClient
   clientProtocolError = SMP
 
-instance ProtocolServerClient NtfResponse where
+instance ProtocolServerClient ErrorType NtfResponse where
   getProtocolServerClient = getNtfServerClient
   clientProtocolError = NTF
 
@@ -428,7 +429,7 @@ getClientVar tSess clients = maybe (Left <$> newClientVar) (pure . Right) =<< TM
       TM.insert tSess var clients
       pure var
 
-waitForProtocolClient :: (AgentMonad m, ProtocolTypeI (ProtoType msg)) => AgentClient -> TransportSession msg -> ClientVar msg -> m (ProtocolClient msg)
+waitForProtocolClient :: (AgentMonad m, ProtocolTypeI (ProtoType msg)) => AgentClient -> TransportSession msg -> ClientVar err msg -> m (ProtocolClient err msg)
 waitForProtocolClient c (_, srv, _) clientVar = do
   NetworkConfig {tcpConnectTimeout} <- readTVarIO $ useNetworkConfig c
   client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar clientVar)
@@ -438,18 +439,18 @@ waitForProtocolClient c (_, srv, _) clientVar = do
     Nothing -> Left $ BROKER (B.unpack $ strEncode srv) TIMEOUT
 
 newProtocolClient ::
-  forall msg m.
+  forall err msg m.
   (AgentMonad m, ProtocolTypeI (ProtoType msg)) =>
   AgentClient ->
   TransportSession msg ->
-  TMap (TransportSession msg) (ClientVar msg) ->
-  m (ProtocolClient msg) ->
+  TMap (TransportSession msg) (ClientVar err msg) ->
+  m (ProtocolClient err msg) ->
   (AgentClient -> TransportSession msg -> m ()) ->
-  ClientVar msg ->
-  m (ProtocolClient msg)
+  ClientVar err msg ->
+  m (ProtocolClient err msg)
 newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient reconnectClient clientVar = tryConnectClient pure tryConnectAsync
   where
-    tryConnectClient :: (ProtocolClient msg -> m a) -> m () -> m a
+    tryConnectClient :: (ProtocolClient err msg -> m a) -> m () -> m a
     tryConnectClient successAction retryAction =
       tryError connectClient >>= \r -> case r of
         Right client -> do
@@ -474,7 +475,7 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient reconne
       withRetryInterval ri $ \loop -> void $ tryConnectClient (const $ reconnectClient c tSess) loop
       atomically . removeAsyncAction aId $ asyncClients c
 
-hostEvent :: forall msg. ProtocolTypeI (ProtoType msg) => (AProtocolType -> TransportHost -> ACommand 'Agent) -> ProtocolClient msg -> ACommand 'Agent
+hostEvent :: forall err msg. ProtocolTypeI (ProtoType msg) => (AProtocolType -> TransportHost -> ACommand 'Agent) -> ProtocolClient err msg -> ACommand 'Agent
 hostEvent event client = event (AProtocolType $ protocolTypeI @(ProtoType msg)) $ transportHost' client
 
 getClientConfig :: AgentMonad m => AgentClient -> (AgentConfig -> ProtocolClientConfig) -> m ProtocolClientConfig
@@ -518,7 +519,7 @@ throwWhenNoDelivery c SndQueue {server, sndId} =
   where
     k = (server, sndId)
 
-closeProtocolServerClients :: AgentClient -> (AgentClient -> TMap (TransportSession msg) (ClientVar msg)) -> IO ()
+closeProtocolServerClients :: AgentClient -> (AgentClient -> TMap (TransportSession msg) (ClientVar err msg)) -> IO ()
 closeProtocolServerClients c clientsSel =
   atomically (swapTVar cs M.empty) >>= mapM_ (forkIO . closeClient)
   where
@@ -541,32 +542,32 @@ withLockMap_ locks key = withGetLock $ TM.lookup key locks >>= maybe newLock pur
   where
     newLock = createLock >>= \l -> TM.insert key l locks $> l
 
-withClient_ :: forall a m msg. (AgentMonad m, ProtocolServerClient msg) => AgentClient -> TransportSession msg -> ByteString -> (ProtocolClient msg -> m a) -> m a
+withClient_ :: forall a m err msg. (AgentMonad m, ProtocolServerClient err msg) => AgentClient -> TransportSession msg -> ByteString -> (ProtocolClient err msg -> m a) -> m a
 withClient_ c tSess@(userId, srv, _) statCmd action = do
   cl <- getProtocolServerClient c tSess
   (action cl <* stat cl "OK") `catchError` logServerError cl
   where
     stat cl = liftIO . incClientStat c userId cl statCmd
-    logServerError :: ProtocolClient msg -> AgentErrorType -> m a
+    logServerError :: ProtocolClient err msg -> AgentErrorType -> m a
     logServerError cl e = do
       logServer "<--" c srv "" $ strEncode e
       stat cl $ strEncode e
       throwError e
 
-withLogClient_ :: (AgentMonad m, ProtocolServerClient msg) => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (ProtocolClient msg -> m a) -> m a
+withLogClient_ :: (AgentMonad m, ProtocolServerClient err msg) => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (ProtocolClient err msg -> m a) -> m a
 withLogClient_ c tSess@(_, srv, _) entId cmdStr action = do
   logServer "-->" c srv entId cmdStr
   res <- withClient_ c tSess cmdStr action
   logServer "<--" c srv entId "OK"
   return res
 
-withClient :: forall m msg a. (AgentMonad m, ProtocolServerClient msg, ProtocolTypeI (ProtoType msg)) => AgentClient -> TransportSession msg -> ByteString -> (ProtocolClient msg -> ExceptT ProtocolClientError IO a) -> m a
-withClient c tSess statKey action = withClient_ c tSess statKey $ \client -> liftClient (clientProtocolError @msg) (clientServer client) $ action client
+withClient :: forall m err msg a. (AgentMonad m, ProtocolServerClient err msg, ProtocolTypeI (ProtoType msg), Encoding err, Show err) => AgentClient -> TransportSession msg -> ByteString -> (ProtocolClient err msg -> ExceptT (ProtocolClientError err) IO a) -> m a
+withClient c tSess statKey action = withClient_ c tSess statKey $ \client -> liftClient (clientProtocolError @err @msg) (clientServer client) $ action client
 
-withLogClient :: forall m msg a. (AgentMonad m, ProtocolServerClient msg, ProtocolTypeI (ProtoType msg)) => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (ProtocolClient msg -> ExceptT ProtocolClientError IO a) -> m a
-withLogClient c tSess entId cmdStr action = withLogClient_ c tSess entId cmdStr $ \client -> liftClient (clientProtocolError @msg) (clientServer client) $ action client
+withLogClient :: forall m err msg a. (AgentMonad m, ProtocolServerClient err msg, ProtocolTypeI (ProtoType msg), Encoding err, Show err) => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (ProtocolClient err msg -> ExceptT (ProtocolClientError err) IO a) -> m a
+withLogClient c tSess entId cmdStr action = withLogClient_ c tSess entId cmdStr $ \client -> liftClient (clientProtocolError @err @msg) (clientServer client) $ action client
 
-withSMPClient :: (AgentMonad m, SMPQueueRec q) => AgentClient -> q -> ByteString -> (SMPClient -> ExceptT ProtocolClientError IO a) -> m a
+withSMPClient :: (AgentMonad m, SMPQueueRec q) => AgentClient -> q -> ByteString -> (SMPClient -> ExceptT SMPClientError IO a) -> m a
 withSMPClient c q cmdStr action = do
   tSess <- mkSMPTransportSession c q
   withLogClient c tSess (queueId q) cmdStr action
@@ -576,23 +577,23 @@ withSMPClient_ c q cmdStr action = do
   tSess <- mkSMPTransportSession c q
   withLogClient_ c tSess (queueId q) cmdStr action
 
-withNtfClient :: forall m a. AgentMonad m => AgentClient -> NtfServer -> EntityId -> ByteString -> (NtfClient -> ExceptT ProtocolClientError IO a) -> m a
+withNtfClient :: forall m a. AgentMonad m => AgentClient -> NtfServer -> EntityId -> ByteString -> (NtfClient -> ExceptT NtfClientError IO a) -> m a
 withNtfClient c srv = withLogClient c (0, srv, Nothing)
 
-liftClient :: AgentMonad m => (ErrorType -> AgentErrorType) -> HostName -> ExceptT ProtocolClientError IO a -> m a
+liftClient :: (AgentMonad m, Show err, Encoding err) => (err -> AgentErrorType) -> HostName -> ExceptT (ProtocolClientError err) IO a -> m a
 liftClient protocolError_ = liftError . protocolClientError protocolError_
 
-protocolClientError :: (ErrorType -> AgentErrorType) -> HostName -> ProtocolClientError -> AgentErrorType
+protocolClientError :: (Show err, Encoding err) => (err -> AgentErrorType) -> HostName -> ProtocolClientError err -> AgentErrorType
 protocolClientError protocolError_ host = \case
   PCEProtocolError e -> protocolError_ e
-  PCEResponseError e -> BROKER host $ RESPONSE e
+  PCEResponseError e -> BROKER host $ RESPONSE $ B.unpack $ smpEncode e
   PCEUnexpectedResponse _ -> BROKER host UNEXPECTED
   PCEResponseTimeout -> BROKER host TIMEOUT
   PCENetworkError -> BROKER host NETWORK
   PCEIncompatibleHost -> BROKER host HOST
   PCETransportError e -> BROKER host $ TRANSPORT e
   e@PCESignatureError {} -> INTERNAL $ show e
-  e@PCEIOError {} -> INTERNAL $ show e
+  PCEIOError {} -> BROKER host NETWORK
 
 data SMPTestStep = TSConnect | TSCreateQueue | TSSecureQueue | TSDeleteQueue | TSDisconnect
   deriving (Eq, Show, Generic)
@@ -632,7 +633,7 @@ runSMPServerTest c userId (ProtoServerWithAuth srv auth) = do
       Left e -> pure (Just $ testErr TSConnect e)
   where
     addr = B.unpack $ strEncode srv
-    testErr :: SMPTestStep -> ProtocolClientError -> SMPTestFailure
+    testErr :: SMPTestStep -> SMPClientError -> SMPTestFailure
     testErr step = SMPTestFailure step . protocolClientError SMP addr
 
 mkTransportSession :: AgentMonad m => AgentClient -> UserId -> ProtoServer msg -> EntityId -> m (TransportSession msg)
@@ -682,7 +683,7 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange = do
           }
   pure (rq, SMPQueueUri vRange $ SMPQueueAddress srv sndId e2eDhKey)
 
-processSubResult :: AgentClient -> RcvQueue -> Either ProtocolClientError () -> IO (Either ProtocolClientError ())
+processSubResult :: AgentClient -> RcvQueue -> Either SMPClientError () -> IO (Either SMPClientError ())
 processSubResult c rq r = do
   case r of
     Left e ->
@@ -691,10 +692,11 @@ processSubResult c rq r = do
     _ -> addSubscription c rq
   pure r
 
-temporaryClientError :: ProtocolClientError -> Bool
+temporaryClientError :: ProtocolClientError err -> Bool
 temporaryClientError = \case
   PCENetworkError -> True
   PCEResponseTimeout -> True
+  PCEIOError _ -> True
   _ -> False
 
 temporaryAgentError :: AgentErrorType -> Bool
@@ -731,7 +733,7 @@ subscribeQueues c qs = do
 type BatchResponses e = (NonEmpty (RcvQueue, Either e ()))
 
 -- statBatchSize is not used to batch the commands, only for traffic statistics
-sendTSessionBatches :: forall m. AgentMonad m => ByteString -> Int -> (SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses ProtocolClientError)) -> AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
+sendTSessionBatches :: forall m. AgentMonad m => ByteString -> Int -> (SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError)) -> AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
 sendTSessionBatches statCmd statBatchSize action c qs =
   concatMap L.toList <$> (mapConcurrently sendClientBatch =<< batchQueues)
   where
@@ -758,7 +760,7 @@ sendTSessionBatches statCmd statBatchSize action c qs =
               let n = (length qs - 1) `div` statBatchSize + 1
                in incClientStatN c userId smp n (statCmd <> "S") "OK"
 
-sendBatch :: (SMPClient -> NonEmpty (SMP.RcvPrivateSignKey, SMP.RecipientId) -> IO (NonEmpty (Either ProtocolClientError ()))) -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses ProtocolClientError)
+sendBatch :: (SMPClient -> NonEmpty (SMP.RcvPrivateSignKey, SMP.RecipientId) -> IO (NonEmpty (Either SMPClientError ()))) -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError)
 sendBatch smpCmdFunc smp qs = L.zip qs <$> smpCmdFunc smp (L.map queueCreds qs)
   where
     queueCreds RcvQueue {rcvPrivateKey, rcvId} = (rcvPrivateKey, rcvId)
@@ -1046,7 +1048,7 @@ incStat AgentClient {agentStats} n k = do
     Just v -> modifyTVar' v (+ n)
     _ -> newTVar n >>= \v -> TM.insert k v agentStats
 
-incClientStat :: AgentClient -> UserId -> ProtocolClient msg -> ByteString -> ByteString -> IO ()
+incClientStat :: AgentClient -> UserId -> ProtocolClient err msg -> ByteString -> ByteString -> IO ()
 incClientStat c userId pc = incClientStatN c userId pc 1
 
 incServerStat :: AgentClient -> UserId -> ProtocolServer p -> ByteString -> ByteString -> IO ()
@@ -1056,7 +1058,7 @@ incServerStat c userId ProtocolServer {host} cmd res = do
   where
     statsKey = AgentStatsKey {userId, host = strEncode $ L.head host, clientTs = "", cmd, res}
 
-incClientStatN :: AgentClient -> UserId -> ProtocolClient msg -> Int -> ByteString -> ByteString -> IO ()
+incClientStatN :: AgentClient -> UserId -> ProtocolClient err msg -> Int -> ByteString -> ByteString -> IO ()
 incClientStatN c userId pc n cmd res = do
   atomically $ incStat c n statsKey
   where

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -123,13 +123,13 @@ defaultMessageRetryInterval =
             maxInterval = 60_000000
           },
       riSlow =
-        -- TODO: these timeouts can be increased in v4.6 once most clients are updated
+        -- TODO: these timeouts can be increased in v5.0 once most clients are updated
         -- to resume sending on QCONT messages.
         -- After that local message expiration period should be also increased.
         RetryInterval
-          { initialInterval = 30_000000,
-            increaseAfter = 30_000000,
-            maxInterval = 600_000000
+          { initialInterval = 60_000000,
+            increaseAfter = 60_000000,
+            maxInterval = 1200_000000 -- 20min
           }
     }
 

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -82,11 +82,11 @@ processNtfSub c (connId, cmd) = do
             case clientNtfCreds of
               Just ClientNtfCreds {notifierId} -> do
                 let newSub = newNtfSubscription connId smpServer (Just notifierId) ntfServer NASKey
-                withStore' c $ \db -> createNtfSubscription db newSub $ NtfSubNTFAction NSACreate
+                withStore c $ \db -> createNtfSubscription db newSub $ NtfSubNTFAction NSACreate
                 addNtfNTFWorker ntfServer
               Nothing -> do
                 let newSub = newNtfSubscription connId smpServer Nothing ntfServer NASNew
-                withStore' c $ \db -> createNtfSubscription db newSub $ NtfSubSMPAction NSASmpKey
+                withStore c $ \db -> createNtfSubscription db newSub $ NtfSubSMPAction NSASmpKey
                 addNtfSMPWorker smpServer
         (Just (sub@NtfSubscription {ntfSubStatus, ntfServer = subNtfServer, smpServer = smpServer', ntfQueueId}, action_)) -> do
           case (clientNtfCreds, ntfQueueId) of

src/Simplex/Messaging/Agent/Protocol.hs

@@ -1121,7 +1121,7 @@ instance ToJSON ConnectionErrorType where
 -- | SMP server errors.
 data BrokerErrorType
   = -- | invalid server response (failed to parse)
-    RESPONSE {smpErr :: ErrorType}
+    RESPONSE {smpErr :: String}
   | -- | unexpected response
     UNEXPECTED
   | -- | network error
@@ -1164,27 +1164,27 @@ instance StrEncoding AgentErrorType where
       <|> "CONN " *> (CONN <$> parseRead1)
       <|> "SMP " *> (SMP <$> strP)
       <|> "NTF " *> (NTF <$> strP)
-      <|> "BROKER " *> (BROKER <$> srvP <* " RESPONSE " <*> (RESPONSE <$> strP))
-      <|> "BROKER " *> (BROKER <$> srvP <* " TRANSPORT " <*> (TRANSPORT <$> transportErrorP))
-      <|> "BROKER " *> (BROKER <$> srvP <* A.space <*> parseRead1)
+      <|> "BROKER " *> (BROKER <$> textP <* " RESPONSE " <*> (RESPONSE <$> textP))
+      <|> "BROKER " *> (BROKER <$> textP <* " TRANSPORT " <*> (TRANSPORT <$> transportErrorP))
+      <|> "BROKER " *> (BROKER <$> textP <* A.space <*> parseRead1)
       <|> "AGENT QUEUE " *> (AGENT . A_QUEUE <$> parseRead A.takeByteString)
       <|> "AGENT " *> (AGENT <$> parseRead1)
       <|> "INTERNAL " *> (INTERNAL <$> parseRead A.takeByteString)
     where
-      srvP = T.unpack . safeDecodeUtf8 <$> A.takeTill (== ' ')
+      textP = T.unpack . safeDecodeUtf8 <$> A.takeTill (== ' ')
   strEncode = \case
     CMD e -> "CMD " <> bshow e
     CONN e -> "CONN " <> bshow e
     SMP e -> "SMP " <> strEncode e
     NTF e -> "NTF " <> strEncode e
-    BROKER srv (RESPONSE e) -> "BROKER " <> addr srv <> " RESPONSE " <> strEncode e
-    BROKER srv (TRANSPORT e) -> "BROKER " <> addr srv <> " TRANSPORT " <> serializeTransportError e
-    BROKER srv e -> "BROKER " <> addr srv <> " " <> bshow e
+    BROKER srv (RESPONSE e) -> "BROKER " <> text srv <> " RESPONSE " <> text e
+    BROKER srv (TRANSPORT e) -> "BROKER " <> text srv <> " TRANSPORT " <> serializeTransportError e
+    BROKER srv e -> "BROKER " <> text srv <> " " <> bshow e
     AGENT (A_QUEUE e) -> "AGENT QUEUE " <> bshow e
     AGENT e -> "AGENT " <> bshow e
     INTERNAL e -> "INTERNAL " <> bshow e
     where
-      addr = encodeUtf8 . T.pack
+      text = encodeUtf8 . T.pack
 
 instance Arbitrary AgentErrorType where arbitrary = genericArbitraryU
 

src/Simplex/Messaging/Agent/Store.hs

@@ -510,6 +510,8 @@ data StoreError
     SEUserNotFound
   | -- | Connection not found (or both queues absent).
     SEConnNotFound
+  | -- | Server not found.
+    SEServerNotFound
   | -- | Connection already used.
     SEConnDuplicate
   | -- | Wrong connection type, e.g. "send" connection when "receive" or "duplex" is expected, or vice versa.

src/Simplex/Messaging/Agent/Store/SQLite.hs

@@ -400,16 +400,16 @@ updateNewConnSnd db connId sq =
 createRcvConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> IO (Either StoreError ConnId)
 createRcvConn db gVar cData@ConnData {userId, connAgentVersion, enableNtfs, duplexHandshake} q@RcvQueue {server} cMode =
   createConn_ gVar cData $ \connId -> do
-    upsertServer_ db server
+    serverKeyHash_ <- createServer_ db server
     DB.execute db "INSERT INTO connections (user_id, conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?,?,?,?,?,?)" (userId, connId, cMode, connAgentVersion, enableNtfs, duplexHandshake)
-    void $ insertRcvQueue_ db connId q
+    void $ insertRcvQueue_ db connId q serverKeyHash_
 
 createSndConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> SndQueue -> IO (Either StoreError ConnId)
 createSndConn db gVar cData@ConnData {userId, connAgentVersion, enableNtfs, duplexHandshake} q@SndQueue {server} =
   createConn_ gVar cData $ \connId -> do
-    upsertServer_ db server
+    serverKeyHash_ <- createServer_ db server
     DB.execute db "INSERT INTO connections (user_id, conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?,?,?,?,?,?)" (userId, connId, SCMInvitation, connAgentVersion, enableNtfs, duplexHandshake)
-    void $ insertSndQueue_ db connId q
+    void $ insertSndQueue_ db connId q serverKeyHash_
 
 getRcvConn :: DB.Connection -> SMPServer -> SMP.RecipientId -> IO (Either StoreError (RcvQueue, SomeConn))
 getRcvConn db ProtocolServer {host, port} rcvId = runExceptT $ do
@@ -447,8 +447,8 @@ addConnRcvQueue db connId rq =
 
 addConnRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO Int64
 addConnRcvQueue_ db connId rq@RcvQueue {server} = do
-  upsertServer_ db server
-  insertRcvQueue_ db connId rq
+  serverKeyHash_ <- createServer_ db server
+  insertRcvQueue_ db connId rq serverKeyHash_
 
 addConnSndQueue :: DB.Connection -> ConnId -> SndQueue -> IO (Either StoreError Int64)
 addConnSndQueue db connId sq =
@@ -459,8 +459,8 @@ addConnSndQueue db connId sq =
 
 addConnSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO Int64
 addConnSndQueue_ db connId sq@SndQueue {server} = do
-  upsertServer_ db server
-  insertSndQueue_ db connId sq
+  serverKeyHash_ <- createServer_ db server
+  insertSndQueue_ db connId sq serverKeyHash_
 
 setRcvQueueStatus :: DB.Connection -> RcvQueue -> QueueStatus -> IO ()
 setRcvQueueStatus db RcvQueue {rcvId, server = ProtocolServer {host, port}} status =
@@ -858,18 +858,21 @@ updateRatchet db connId rc skipped = do
         forM_ (M.assocs mks) $ \(msgN, mk) ->
           DB.execute db "INSERT INTO skipped_messages (conn_id, header_key, msg_n, msg_key) VALUES (?, ?, ?, ?)" (connId, hk, msgN, mk)
 
-createCommand :: DB.Connection -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> IO AsyncCmdId
-createCommand db corrId connId srv cmd = do
-  DB.execute
-    db
-    "INSERT INTO commands (host, port, corr_id, conn_id, command_tag, command) VALUES (?,?,?,?,?,?)"
-    (host_, port_, corrId, connId, agentCommandTag cmd, cmd)
-  insertedRowId db
+createCommand :: DB.Connection -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> IO (Either StoreError AsyncCmdId)
+createCommand db corrId connId srv_ cmd = runExceptT $ do
+  (host_, port_, serverKeyHash_) <- serverFields
+  liftIO $ do
+    DB.execute
+      db
+      "INSERT INTO commands (host, port, corr_id, conn_id, command_tag, command, server_key_hash) VALUES (?,?,?,?,?,?,?)"
+      (host_, port_, corrId, connId, agentCommandTag cmd, cmd, serverKeyHash_)
+    insertedRowId db
   where
-    (host_, port_) =
-      case srv of
-        Just (SMPServer host port _) -> (Just host, Just port)
-        _ -> (Nothing, Nothing)
+    serverFields :: ExceptT StoreError IO (Maybe (NonEmpty TransportHost), Maybe ServiceName, Maybe C.KeyHash)
+    serverFields = case srv_ of
+      Just srv@(SMPServer host port _) ->
+        (Just host,Just port,) <$> ExceptT (getServerKeyHash_ db srv)
+      Nothing -> pure (Nothing, Nothing, Nothing)
 
 insertedRowId :: DB.Connection -> IO Int64
 insertedRowId db = fromOnly . head <$> DB.query_ db "SELECT last_insert_rowid()"
@@ -880,7 +883,7 @@ getPendingCommands db connId = do
     <$> DB.query
       db
       [sql|
-        SELECT c.host, c.port, s.key_hash, c.command_id
+        SELECT c.host, c.port, COALESCE(c.server_key_hash, s.key_hash), c.command_id
         FROM commands c
         LEFT JOIN servers s ON s.host = c.host AND s.port = c.port
         WHERE conn_id = ?
@@ -1003,7 +1006,7 @@ getNtfSubscription db connId =
     DB.query
       db
       [sql|
-        SELECT s.host, s.port, s.key_hash, ns.ntf_host, ns.ntf_port, ns.ntf_key_hash,
+        SELECT s.host, s.port, COALESCE(nsb.smp_server_key_hash, s.key_hash), ns.ntf_host, ns.ntf_port, ns.ntf_key_hash,
           nsb.smp_ntf_id, nsb.ntf_sub_id, nsb.ntf_sub_status, nsb.ntf_sub_action, nsb.ntf_sub_smp_action, nsb.ntf_sub_action_ts
         FROM ntf_subscriptions nsb
         JOIN servers s ON s.host = nsb.smp_host AND s.port = nsb.smp_port
@@ -1021,21 +1024,23 @@ getNtfSubscription db connId =
             _ -> Nothing
        in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus}, action)
 
-createNtfSubscription :: DB.Connection -> NtfSubscription -> NtfSubAction -> IO ()
-createNtfSubscription db ntfSubscription action = do
-  let NtfSubscription {connId, smpServer = (SMPServer host port _), ntfQueueId, ntfServer = (NtfServer ntfHost ntfPort _), ntfSubId, ntfSubStatus} = ntfSubscription
+createNtfSubscription :: DB.Connection -> NtfSubscription -> NtfSubAction -> IO (Either StoreError ())
+createNtfSubscription db ntfSubscription action = runExceptT $ do
+  let NtfSubscription {connId, smpServer = smpServer@(SMPServer host port _), ntfQueueId, ntfServer = (NtfServer ntfHost ntfPort _), ntfSubId, ntfSubStatus} = ntfSubscription
+  smpServerKeyHash_ <- ExceptT $ getServerKeyHash_ db smpServer
   actionTs <- liftIO getCurrentTime
-  DB.execute
-    db
-    [sql|
-      INSERT INTO ntf_subscriptions
-        (conn_id, smp_host, smp_port, smp_ntf_id, ntf_host, ntf_port, ntf_sub_id,
-          ntf_sub_status, ntf_sub_action, ntf_sub_smp_action, ntf_sub_action_ts)
-      VALUES (?,?,?,?,?,?,?,?,?,?,?)
-    |]
-    ( (connId, host, port, ntfQueueId, ntfHost, ntfPort, ntfSubId)
-        :. (ntfSubStatus, ntfSubAction, ntfSubSMPAction, actionTs)
-    )
+  liftIO $
+    DB.execute
+      db
+      [sql|
+        INSERT INTO ntf_subscriptions
+          (conn_id, smp_host, smp_port, smp_ntf_id, ntf_host, ntf_port, ntf_sub_id,
+            ntf_sub_status, ntf_sub_action, ntf_sub_smp_action, ntf_sub_action_ts, smp_server_key_hash)
+        VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
+      |]
+      ( (connId, host, port, ntfQueueId, ntfHost, ntfPort, ntfSubId)
+          :. (ntfSubStatus, ntfSubAction, ntfSubSMPAction, actionTs, smpServerKeyHash_)
+      )
   where
     (ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction action
 
@@ -1136,7 +1141,7 @@ getNextNtfSubNTFAction db ntfServer@(NtfServer ntfHost ntfPort _) = do
       DB.query
         db
         [sql|
-          SELECT ns.conn_id, s.host, s.port, s.key_hash,
+          SELECT ns.conn_id, s.host, s.port, COALESCE(ns.smp_server_key_hash, s.key_hash),
             ns.smp_ntf_id, ns.ntf_sub_id, ns.ntf_sub_status, ns.ntf_sub_action_ts, ns.ntf_sub_action
           FROM ntf_subscriptions ns
           JOIN servers s ON s.host = ns.smp_host AND s.port = ns.smp_port
@@ -1321,20 +1326,25 @@ instance (ToField a, ToField b, ToField c, ToField d, ToField e, ToField f,
 
 {- ORMOLU_ENABLE -}
 
--- * Server upsert helper
+-- * Server helper
 
-upsertServer_ :: DB.Connection -> SMPServer -> IO ()
-upsertServer_ dbConn ProtocolServer {host, port, keyHash} = do
-  DB.executeNamed
-    dbConn
-    [sql|
-      INSERT INTO servers (host, port, key_hash) VALUES (:host,:port,:key_hash)
-      ON CONFLICT (host, port) DO UPDATE SET
-        host=excluded.host,
-        port=excluded.port,
-        key_hash=excluded.key_hash;
-    |]
-    [":host" := host, ":port" := port, ":key_hash" := keyHash]
+-- | Creates a new server, if it doesn't exist, and returns the passed key hash if it is different from stored.
+createServer_ :: DB.Connection -> SMPServer -> IO (Maybe C.KeyHash)
+createServer_ db newSrv@ProtocolServer {host, port, keyHash} =
+  getServerKeyHash_ db newSrv >>= \case
+    Right keyHash_ -> pure keyHash_
+    Left _ -> insertNewServer_ $> Nothing
+  where
+    insertNewServer_ =
+      DB.execute db "INSERT INTO servers (host, port, key_hash) VALUES (?,?,?)" (host, port, keyHash)
+
+-- | Returns the stored server key hash if it is different from the passed one, or the error if the server does not exist.
+getServerKeyHash_ :: DB.Connection -> SMPServer -> IO (Either StoreError (Maybe C.KeyHash))
+getServerKeyHash_ db ProtocolServer {host, port, keyHash} = do
+  firstRow useKeyHash SEServerNotFound $
+    DB.query db "SELECT key_hash FROM servers WHERE host = ? AND port = ?" (host, port)
+  where
+    useKeyHash (Only keyHash') = if keyHash /= keyHash' then Just keyHash else Nothing
 
 upsertNtfServer_ :: DB.Connection -> NtfServer -> IO ()
 upsertNtfServer_ db ProtocolServer {host, port, keyHash} = do
@@ -1351,30 +1361,30 @@ upsertNtfServer_ db ProtocolServer {host, port, keyHash} = do
 
 -- * createRcvConn helpers
 
-insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO Int64
-insertRcvQueue_ db connId' RcvQueue {..} = do
+insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> Maybe C.KeyHash -> IO Int64
+insertRcvQueue_ db connId' RcvQueue {..} serverKeyHash_ = do
   qId <- newQueueId_ <$> DB.query db "SELECT rcv_queue_id FROM rcv_queues WHERE conn_id = ? ORDER BY rcv_queue_id DESC LIMIT 1" (Only connId')
   DB.execute
     db
     [sql|
       INSERT INTO rcv_queues
-        (host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status, rcv_queue_id, rcv_primary, replace_rcv_queue_id, smp_client_version) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?);
+        (host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status, rcv_queue_id, rcv_primary, replace_rcv_queue_id, smp_client_version, server_key_hash) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?);
     |]
-    ((host server, port server, rcvId, connId', rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret) :. (sndId, status, qId, primary, dbReplaceQueueId, smpClientVersion))
+    ((host server, port server, rcvId, connId', rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret) :. (sndId, status, qId, primary, dbReplaceQueueId, smpClientVersion, serverKeyHash_))
   pure qId
 
 -- * createSndConn helpers
 
-insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO Int64
-insertSndQueue_ db connId' SndQueue {..} = do
+insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> Maybe C.KeyHash -> IO Int64
+insertSndQueue_ db connId' SndQueue {..} serverKeyHash_ = do
   qId <- newQueueId_ <$> DB.query db "SELECT snd_queue_id FROM snd_queues WHERE conn_id = ? ORDER BY snd_queue_id DESC LIMIT 1" (Only connId')
   DB.execute
     db
     [sql|
       INSERT INTO snd_queues
-        (host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status, snd_queue_id, snd_primary, replace_snd_queue_id, smp_client_version) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?);
+        (host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status, snd_queue_id, snd_primary, replace_snd_queue_id, smp_client_version, server_key_hash) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?);
     |]
-    ((host server, port server, sndId, connId', sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret) :. (status, qId, primary, dbReplaceQueueId, smpClientVersion))
+    ((host server, port server, sndId, connId', sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret) :. (status, qId, primary, dbReplaceQueueId, smpClientVersion, serverKeyHash_))
   pure qId
 
 newQueueId_ :: [Only Int64] -> Int64
@@ -1443,7 +1453,7 @@ getRcvQueuesByConnId_ db connId =
 rcvQueueQuery :: Query
 rcvQueueQuery =
   [sql|
-    SELECT c.user_id, s.key_hash, q.conn_id, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
+    SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
       q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.status,
       q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id, q.smp_client_version, q.delete_errors,
       q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret
@@ -1477,7 +1487,7 @@ getSndQueuesByConnId_ dbConn connId =
     <$> DB.query
       dbConn
       [sql|
-        SELECT c.user_id, s.key_hash, q.host, q.port, q.snd_id, q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status, q.snd_queue_id, q.snd_primary, q.replace_snd_queue_id, q.smp_client_version
+        SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.host, q.port, q.snd_id, q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status, q.snd_queue_id, q.snd_primary, q.replace_snd_queue_id, q.smp_client_version
         FROM snd_queues q
         JOIN servers s ON q.host = s.host AND q.port = s.port
         JOIN connections c ON q.conn_id = c.conn_id

src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs

@@ -40,6 +40,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220915_connection_queu
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230110_users
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230117_fkey_indexes
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230120_delete_errors
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230217_server_key_hash
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Transport.Client (TransportHost)
 
@@ -59,7 +60,8 @@ schemaMigrations =
     ("m20220915_connection_queues", m20220915_connection_queues),
     ("m20230110_users", m20230110_users),
     ("m20230117_fkey_indexes", m20230117_fkey_indexes),
-    ("m20230120_delete_errors", m20230120_delete_errors)
+    ("m20230120_delete_errors", m20230120_delete_errors),
+    ("m20230217_server_key_hash", m20230217_server_key_hash)
   ]
 
 -- | The list of migrations in ascending order by date

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20230217_server_key_hash.hs

@@ -0,0 +1,20 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230217_server_key_hash where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+-- server_key_hash is not null for records whose entities refer to a server
+-- that was previously saved with the same host and port but different key hash
+m20230217_server_key_hash :: Query
+m20230217_server_key_hash =
+  [sql|
+ALTER TABLE rcv_queues ADD COLUMN server_key_hash BLOB;
+
+ALTER TABLE snd_queues ADD COLUMN server_key_hash BLOB;
+
+ALTER TABLE ntf_subscriptions ADD COLUMN smp_server_key_hash BLOB;
+
+ALTER TABLE commands ADD COLUMN server_key_hash BLOB;
+|]

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -48,6 +48,7 @@ CREATE TABLE rcv_queues(
   rcv_primary INTEGER CHECK(rcv_primary NOT NULL),
   replace_rcv_queue_id INTEGER NULL,
   delete_errors INTEGER DEFAULT 0 CHECK(delete_errors NOT NULL),
+  server_key_hash BLOB,
   PRIMARY KEY(host, port, rcv_id),
   FOREIGN KEY(host, port) REFERENCES servers
   ON DELETE RESTRICT ON UPDATE CASCADE,
@@ -68,6 +69,7 @@ CREATE TABLE snd_queues(
   snd_queue_id INTEGER CHECK(snd_queue_id NOT NULL),
   snd_primary INTEGER CHECK(snd_primary NOT NULL),
   replace_snd_queue_id INTEGER NULL,
+  server_key_hash BLOB,
   PRIMARY KEY(host, port, snd_id),
   FOREIGN KEY(host, port) REFERENCES servers
   ON DELETE RESTRICT ON UPDATE CASCADE
@@ -199,6 +201,7 @@ CREATE TABLE ntf_subscriptions(
   updated_by_supervisor INTEGER NOT NULL DEFAULT 0, -- to be checked on updates by workers to not overwrite supervisor command(state still should be updated)
   created_at TEXT NOT NULL DEFAULT(datetime('now')),
   updated_at TEXT NOT NULL DEFAULT(datetime('now')),
+  smp_server_key_hash BLOB,
   PRIMARY KEY(conn_id),
   FOREIGN KEY(smp_host, smp_port) REFERENCES servers(host, port)
   ON DELETE SET NULL ON UPDATE CASCADE,
@@ -214,6 +217,7 @@ CREATE TABLE commands(
   command_tag BLOB NOT NULL,
   command BLOB NOT NULL,
   agent_version INTEGER NOT NULL DEFAULT 1,
+  server_key_hash BLOB,
   FOREIGN KEY(host, port) REFERENCES servers
   ON DELETE RESTRICT ON UPDATE CASCADE
 );

src/Simplex/Messaging/Client.hs

@@ -55,6 +55,7 @@ module Simplex.Messaging.Client
 
     -- * Supporting types and client configuration
     ProtocolClientError (..),
+    SMPClientError,
     ProtocolClientConfig (..),
     NetworkConfig (..),
     TransportSessionMode (..),
@@ -62,7 +63,9 @@ module Simplex.Messaging.Client
     defaultNetworkConfig,
     transportClientConfig,
     chooseTransportHost,
+    proxyUsername,
     ServerTransmission,
+    ClientCommand,
   )
 where
 
@@ -105,28 +108,28 @@ import System.Timeout (timeout)
 -- | 'SMPClient' is a handle used to send commands to a specific SMP server.
 --
 -- Use 'getSMPClient' to connect to an SMP server and create a client handle.
-data ProtocolClient msg = ProtocolClient
+data ProtocolClient err msg = ProtocolClient
   { action :: Maybe (Async ()),
     sessionId :: SessionId,
     sessionTs :: UTCTime,
     thVersion :: Version,
-    client_ :: PClient msg
+    client_ :: PClient err msg
   }
 
-data PClient msg = PClient
+data PClient err msg = PClient
   { connected :: TVar Bool,
     transportSession :: TransportSession msg,
     transportHost :: TransportHost,
     tcpTimeout :: Int,
     pingErrorCount :: TVar Int,
     clientCorrId :: TVar Natural,
-    sentCommands :: TMap CorrId (Request msg),
+    sentCommands :: TMap CorrId (Request err msg),
     sndQ :: TBQueue (NonEmpty SentRawTransmission),
-    rcvQ :: TBQueue (NonEmpty (SignedTransmission msg)),
+    rcvQ :: TBQueue (NonEmpty (SignedTransmission err msg)),
     msgQ :: Maybe (TBQueue (ServerTransmission msg))
   }
 
-type SMPClient = ProtocolClient SMP.BrokerMsg
+type SMPClient = ProtocolClient ErrorType SMP.BrokerMsg
 
 -- | Type for client command data
 type ClientCommand msg = (Maybe C.APrivateSignKey, QueueId, ProtoCommand msg)
@@ -229,14 +232,14 @@ defaultClientConfig =
       smpServerVRange = supportedSMPServerVRange
     }
 
-data Request msg = Request
+data Request err msg = Request
   { queueId :: QueueId,
-    responseVar :: TMVar (Response msg)
+    responseVar :: TMVar (Response err msg)
   }
 
-type Response msg = Either ProtocolClientError msg
+type Response err msg = Either (ProtocolClientError err) msg
 
-chooseTransportHost :: NetworkConfig -> NonEmpty TransportHost -> Either ProtocolClientError TransportHost
+chooseTransportHost :: NetworkConfig -> NonEmpty TransportHost -> Either (ProtocolClientError err) TransportHost
 chooseTransportHost NetworkConfig {socksProxy, hostMode, requiredHostMode} hosts =
   firstOrError $ case hostMode of
     HMOnionViaSocks -> maybe publicHost (const onionHost) socksProxy
@@ -250,15 +253,15 @@ chooseTransportHost NetworkConfig {socksProxy, hostMode, requiredHostMode} hosts
     onionHost = find isOnionHost hosts
     publicHost = find (not . isOnionHost) hosts
 
-clientServer :: ProtocolTypeI (ProtoType msg) => ProtocolClient msg -> String
+clientServer :: ProtocolTypeI (ProtoType msg) => ProtocolClient err msg -> String
 clientServer = B.unpack . strEncode . snd3 . transportSession . client_
   where
     snd3 (_, s, _) = s
 
-transportHost' :: ProtocolClient msg -> TransportHost
+transportHost' :: ProtocolClient err msg -> TransportHost
 transportHost' = transportHost . client_
 
-transportSession' :: ProtocolClient msg -> TransportSession msg
+transportSession' :: ProtocolClient err msg -> TransportSession msg
 transportSession' = transportSession . client_
 
 type UserId = Int64
@@ -271,7 +274,7 @@ type TransportSession msg = (UserId, ProtoServer msg, Maybe EntityId)
 --
 -- A single queue can be used for multiple 'SMPClient' instances,
 -- as 'SMPServerTransmission' includes server information.
-getProtocolClient :: forall msg. Protocol msg => TransportSession msg -> ProtocolClientConfig -> Maybe (TBQueue (ServerTransmission msg)) -> (ProtocolClient msg -> IO ()) -> IO (Either ProtocolClientError (ProtocolClient msg))
+getProtocolClient :: forall err msg. Protocol err msg => TransportSession msg -> ProtocolClientConfig -> Maybe (TBQueue (ServerTransmission msg)) -> (ProtocolClient err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient err msg))
 getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, smpServerVRange} msgQ disconnected = do
   case chooseTransportHost networkConfig (host srv) of
     Right useHost ->
@@ -280,7 +283,7 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
     Left e -> pure $ Left e
   where
     NetworkConfig {tcpConnectTimeout, tcpTimeout, smpPingInterval} = networkConfig
-    mkProtocolClient :: TransportHost -> STM (PClient msg)
+    mkProtocolClient :: TransportHost -> STM (PClient err msg)
     mkProtocolClient transportHost = do
       connected <- newTVar False
       pingErrorCount <- newTVar 0
@@ -302,7 +305,7 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
             msgQ
           }
 
-    runClient :: (ServiceName, ATransport) -> TransportHost -> PClient msg -> IO (Either ProtocolClientError (ProtocolClient msg))
+    runClient :: (ServiceName, ATransport) -> TransportHost -> PClient err msg -> IO (Either (ProtocolClientError err) (ProtocolClient err msg))
     runClient (port', ATransport t) useHost c = do
       cVar <- newEmptyTMVarIO
       let tcConfig = transportClientConfig networkConfig
@@ -317,18 +320,15 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
         Just (Left e) -> Left e
         Nothing -> Left PCENetworkError
 
-    proxyUsername :: TransportSession msg -> ByteString
-    proxyUsername (userId, _, entityId_) = C.sha256Hash $ bshow userId <> maybe "" (":" <>) entityId_
-
     useTransport :: (ServiceName, ATransport)
     useTransport = case port srv of
       "" -> defaultTransport cfg
       "80" -> ("80", transport @WS)
       p -> (p, transport @TLS)
 
-    client :: forall c. Transport c => TProxy c -> PClient msg -> TMVar (Either ProtocolClientError (ProtocolClient msg)) -> c -> IO ()
+    client :: forall c. Transport c => TProxy c -> PClient err msg -> TMVar (Either (ProtocolClientError err) (ProtocolClient err msg)) -> c -> IO ()
     client _ c cVar h =
-      runExceptT (protocolClientHandshake @msg h (keyHash srv) smpServerVRange) >>= \case
+      runExceptT (protocolClientHandshake @err @msg h (keyHash srv) smpServerVRange) >>= \case
         Left e -> atomically . putTMVar cVar . Left $ PCETransportError e
         Right th@THandle {sessionId, thVersion} -> do
           sessionTs <- getCurrentTime
@@ -339,16 +339,16 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
           raceAny_ ([send c' th, process c', receive c' th] <> [ping c' | smpPingInterval > 0])
             `finally` disconnected c'
 
-    send :: Transport c => ProtocolClient msg -> THandle c -> IO ()
+    send :: Transport c => ProtocolClient err msg -> THandle c -> IO ()
     send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= tPut h
 
-    receive :: Transport c => ProtocolClient msg -> THandle c -> IO ()
+    receive :: Transport c => ProtocolClient err msg -> THandle c -> IO ()
     receive ProtocolClient {client_ = PClient {rcvQ}} h = forever $ tGet h >>= atomically . writeTBQueue rcvQ
 
-    ping :: ProtocolClient msg -> IO ()
+    ping :: ProtocolClient err msg -> IO ()
     ping c@ProtocolClient {client_ = PClient {pingErrorCount}} = do
       threadDelay smpPingInterval
-      runExceptT (sendProtocolCommand c Nothing "" protocolPing) >>= \case
+      runExceptT (sendProtocolCommand c Nothing "" $ protocolPing @err @msg) >>= \case
         Left PCEResponseTimeout -> do
           cnt <- atomically $ stateTVar pingErrorCount $ \cnt -> (cnt + 1, cnt + 1)
           when (maxCnt == 0 || cnt < maxCnt) $ ping c
@@ -356,10 +356,10 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
       where
         maxCnt = smpPingCount networkConfig
 
-    process :: ProtocolClient msg -> IO ()
+    process :: ProtocolClient err msg -> IO ()
     process c = forever $ atomically (readTBQueue $ rcvQ $ client_ c) >>= mapM_ (processMsg c)
 
-    processMsg :: ProtocolClient msg -> SignedTransmission msg -> IO ()
+    processMsg :: ProtocolClient err msg -> SignedTransmission err msg -> IO ()
     processMsg c@ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, qId, respOrErr)) =
       if B.null $ bs corrId
         then sendMsg respOrErr
@@ -377,23 +377,26 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
                       _ -> Right r
                   else Left . PCEUnexpectedResponse $ bshow respOrErr
       where
-        sendMsg :: Either ErrorType msg -> IO ()
+        sendMsg :: Either err msg -> IO ()
         sendMsg = \case
           Right msg -> atomically $ mapM_ (`writeTBQueue` serverTransmission c qId msg) msgQ
           Left e -> putStrLn $ "SMP client error: " <> show e
 
+proxyUsername :: TransportSession msg -> ByteString
+proxyUsername (userId, _, entityId_) = C.sha256Hash $ bshow userId <> maybe "" (":" <>) entityId_
+
 -- | Disconnects client from the server and terminates client threads.
-closeProtocolClient :: ProtocolClient msg -> IO ()
+closeProtocolClient :: ProtocolClient err msg -> IO ()
 closeProtocolClient = mapM_ uninterruptibleCancel . action
 
 -- | SMP client error type.
-data ProtocolClientError
+data ProtocolClientError err
   = -- | Correctly parsed SMP server ERR response.
     -- This error is forwarded to the agent client as `ERR SMP err`.
-    PCEProtocolError ErrorType
+    PCEProtocolError err
   | -- | Invalid server response that failed to parse.
     -- Forwarded to the agent client as `ERR BROKER RESPONSE`.
-    PCEResponseError ErrorType
+    PCEResponseError err
   | -- | Different response from what is expected to a certain SMP command,
     -- e.g. server should respond `IDS` or `ERR` to `NEW` command,
     -- other responses would result in this error.
@@ -416,6 +419,8 @@ data ProtocolClientError
     PCEIOError IOException
   deriving (Eq, Show, Exception)
 
+type SMPClientError = ProtocolClientError ErrorType
+
 -- | Create a new SMP queue.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#create-queue-command
@@ -425,7 +430,7 @@ createSMPQueue ::
   RcvPublicVerifyKey ->
   RcvPublicDhKey ->
   Maybe BasicAuth ->
-  ExceptT ProtocolClientError IO QueueIdsKeys
+  ExceptT SMPClientError IO QueueIdsKeys
 createSMPQueue c rpKey rKey dhKey auth =
   sendSMPCommand c (Just rpKey) "" (NEW rKey dhKey auth) >>= \case
     IDS qik -> pure qik
@@ -434,7 +439,7 @@ createSMPQueue c rpKey rKey dhKey auth =
 -- | Subscribe to the SMP queue.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#subscribe-to-queue
-subscribeSMPQueue :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT ProtocolClientError IO ()
+subscribeSMPQueue :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT SMPClientError IO ()
 subscribeSMPQueue c rpKey rId =
   sendSMPCommand c (Just rpKey) rId SUB >>= \case
     OK -> return ()
@@ -442,7 +447,7 @@ subscribeSMPQueue c rpKey rId =
     r -> throwE . PCEUnexpectedResponse $ bshow r
 
 -- | Subscribe to multiple SMP queues batching commands if supported.
-subscribeSMPQueues :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId) -> IO (NonEmpty (Either ProtocolClientError ()))
+subscribeSMPQueues :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId) -> IO (NonEmpty (Either SMPClientError ()))
 subscribeSMPQueues c qs = sendProtocolCommands c cs >>= mapM response . L.zip qs
   where
     cs = L.map (\(rpKey, rId) -> (Just rpKey, rId, Cmd SRecipient SUB)) qs
@@ -455,14 +460,14 @@ subscribeSMPQueues c qs = sendProtocolCommands c cs >>= mapM response . L.zip qs
 writeSMPMessage :: SMPClient -> RecipientId -> BrokerMsg -> IO ()
 writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c rId msg) (msgQ $ client_ c)
 
-serverTransmission :: ProtocolClient msg -> RecipientId -> msg -> ServerTransmission msg
+serverTransmission :: ProtocolClient err msg -> RecipientId -> msg -> ServerTransmission msg
 serverTransmission ProtocolClient {thVersion, sessionId, client_ = PClient {transportSession}} entityId message =
   (transportSession, thVersion, sessionId, entityId, message)
 
 -- | Get message from SMP queue. The server returns ERR PROHIBITED if a client uses SUB and GET via the same transport connection for the same queue
 --
 -- https://github.covm/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#receive-a-message-from-the-queue
-getSMPMessage :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT ProtocolClientError IO (Maybe RcvMessage)
+getSMPMessage :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT SMPClientError IO (Maybe RcvMessage)
 getSMPMessage c rpKey rId =
   sendSMPCommand c (Just rpKey) rId GET >>= \case
     OK -> pure Nothing
@@ -472,26 +477,26 @@ getSMPMessage c rpKey rId =
 -- | Subscribe to the SMP queue notifications.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#subscribe-to-queue-notifications
-subscribeSMPQueueNotifications :: SMPClient -> NtfPrivateSignKey -> NotifierId -> ExceptT ProtocolClientError IO ()
+subscribeSMPQueueNotifications :: SMPClient -> NtfPrivateSignKey -> NotifierId -> ExceptT SMPClientError IO ()
 subscribeSMPQueueNotifications = okSMPCommand NSUB
 
 -- | Secure the SMP queue by adding a sender public key.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#secure-queue-command
-secureSMPQueue :: SMPClient -> RcvPrivateSignKey -> RecipientId -> SndPublicVerifyKey -> ExceptT ProtocolClientError IO ()
+secureSMPQueue :: SMPClient -> RcvPrivateSignKey -> RecipientId -> SndPublicVerifyKey -> ExceptT SMPClientError IO ()
 secureSMPQueue c rpKey rId senderKey = okSMPCommand (KEY senderKey) c rpKey rId
 
 -- | Enable notifications for the queue for push notifications server.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#enable-notifications-command
-enableSMPQueueNotifications :: SMPClient -> RcvPrivateSignKey -> RecipientId -> NtfPublicVerifyKey -> RcvNtfPublicDhKey -> ExceptT ProtocolClientError IO (NotifierId, RcvNtfPublicDhKey)
+enableSMPQueueNotifications :: SMPClient -> RcvPrivateSignKey -> RecipientId -> NtfPublicVerifyKey -> RcvNtfPublicDhKey -> ExceptT SMPClientError IO (NotifierId, RcvNtfPublicDhKey)
 enableSMPQueueNotifications c rpKey rId notifierKey rcvNtfPublicDhKey =
   sendSMPCommand c (Just rpKey) rId (NKEY notifierKey rcvNtfPublicDhKey) >>= \case
     NID nId rcvNtfSrvPublicDhKey -> pure (nId, rcvNtfSrvPublicDhKey)
     r -> throwE . PCEUnexpectedResponse $ bshow r
 
 -- | Enable notifications for the multiple queues for push notifications server.
-enableSMPQueuesNtfs :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId, NtfPublicVerifyKey, RcvNtfPublicDhKey) -> IO (NonEmpty (Either ProtocolClientError (NotifierId, RcvNtfPublicDhKey)))
+enableSMPQueuesNtfs :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId, NtfPublicVerifyKey, RcvNtfPublicDhKey) -> IO (NonEmpty (Either SMPClientError (NotifierId, RcvNtfPublicDhKey)))
 enableSMPQueuesNtfs c qs = L.map response <$> sendProtocolCommands c cs
   where
     cs = L.map (\(rpKey, rId, notifierKey, rcvNtfPublicDhKey) -> (Just rpKey, rId, Cmd SRecipient $ NKEY notifierKey rcvNtfPublicDhKey)) qs
@@ -503,17 +508,17 @@ enableSMPQueuesNtfs c qs = L.map response <$> sendProtocolCommands c cs
 -- | Disable notifications for the queue for push notifications server.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#disable-notifications-command
-disableSMPQueueNotifications :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT ProtocolClientError IO ()
+disableSMPQueueNotifications :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT SMPClientError IO ()
 disableSMPQueueNotifications = okSMPCommand NDEL
 
 -- | Disable notifications for multiple queues for push notifications server.
-disableSMPQueuesNtfs :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId) -> IO (NonEmpty (Either ProtocolClientError ()))
+disableSMPQueuesNtfs :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId) -> IO (NonEmpty (Either SMPClientError ()))
 disableSMPQueuesNtfs = okSMPCommands NDEL
 
 -- | Send SMP message.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#send-message
-sendSMPMessage :: SMPClient -> Maybe SndPrivateSignKey -> SenderId -> MsgFlags -> MsgBody -> ExceptT ProtocolClientError IO ()
+sendSMPMessage :: SMPClient -> Maybe SndPrivateSignKey -> SenderId -> MsgFlags -> MsgBody -> ExceptT SMPClientError IO ()
 sendSMPMessage c spKey sId flags msg =
   sendSMPCommand c spKey sId (SEND flags msg) >>= \case
     OK -> pure ()
@@ -522,7 +527,7 @@ sendSMPMessage c spKey sId flags msg =
 -- | Acknowledge message delivery (server deletes the message).
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#acknowledge-message-delivery
-ackSMPMessage :: SMPClient -> RcvPrivateSignKey -> QueueId -> MsgId -> ExceptT ProtocolClientError IO ()
+ackSMPMessage :: SMPClient -> RcvPrivateSignKey -> QueueId -> MsgId -> ExceptT SMPClientError IO ()
 ackSMPMessage c rpKey rId msgId =
   sendSMPCommand c (Just rpKey) rId (ACK msgId) >>= \case
     OK -> return ()
@@ -533,26 +538,26 @@ ackSMPMessage c rpKey rId msgId =
 -- The existing messages from the queue will still be delivered.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#suspend-queue
-suspendSMPQueue :: SMPClient -> RcvPrivateSignKey -> QueueId -> ExceptT ProtocolClientError IO ()
+suspendSMPQueue :: SMPClient -> RcvPrivateSignKey -> QueueId -> ExceptT SMPClientError IO ()
 suspendSMPQueue = okSMPCommand OFF
 
 -- | Irreversibly delete SMP queue and all messages in it.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#delete-queue
-deleteSMPQueue :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT ProtocolClientError IO ()
+deleteSMPQueue :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT SMPClientError IO ()
 deleteSMPQueue = okSMPCommand DEL
 
 -- | Delete multiple SMP queues batching commands if supported.
-deleteSMPQueues :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId) -> IO (NonEmpty (Either ProtocolClientError ()))
+deleteSMPQueues :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId) -> IO (NonEmpty (Either SMPClientError ()))
 deleteSMPQueues = okSMPCommands DEL
 
-okSMPCommand :: PartyI p => Command p -> SMPClient -> C.APrivateSignKey -> QueueId -> ExceptT ProtocolClientError IO ()
+okSMPCommand :: PartyI p => Command p -> SMPClient -> C.APrivateSignKey -> QueueId -> ExceptT SMPClientError IO ()
 okSMPCommand cmd c pKey qId =
   sendSMPCommand c (Just pKey) qId cmd >>= \case
     OK -> return ()
     r -> throwE . PCEUnexpectedResponse $ bshow r
 
-okSMPCommands :: PartyI p => Command p -> SMPClient -> NonEmpty (C.APrivateSignKey, QueueId) -> IO (NonEmpty (Either ProtocolClientError ()))
+okSMPCommands :: PartyI p => Command p -> SMPClient -> NonEmpty (C.APrivateSignKey, QueueId) -> IO (NonEmpty (Either SMPClientError ()))
 okSMPCommands cmd c qs = L.map response <$> sendProtocolCommands c cs
   where
     aCmd = Cmd sParty cmd
@@ -563,11 +568,11 @@ okSMPCommands cmd c qs = L.map response <$> sendProtocolCommands c cs
       Left e -> Left e
 
 -- | Send SMP command
-sendSMPCommand :: PartyI p => SMPClient -> Maybe C.APrivateSignKey -> QueueId -> Command p -> ExceptT ProtocolClientError IO BrokerMsg
+sendSMPCommand :: PartyI p => SMPClient -> Maybe C.APrivateSignKey -> QueueId -> Command p -> ExceptT SMPClientError IO BrokerMsg
 sendSMPCommand c pKey qId cmd = sendProtocolCommand c pKey qId (Cmd sParty cmd)
 
 -- | Send multiple commands with batching and collect responses
-sendProtocolCommands :: forall msg. ProtocolEncoding (ProtoCommand msg) => ProtocolClient msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Either ProtocolClientError msg))
+sendProtocolCommands :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Either (ProtocolClientError err) msg))
 sendProtocolCommands c@ProtocolClient {client_ = PClient {sndQ}} cs = do
   ts <- mapM (runExceptT . mkTransmission c) cs
   mapM_ (atomically . writeTBQueue sndQ . L.map fst) . L.nonEmpty . rights $ L.toList ts
@@ -576,22 +581,22 @@ sendProtocolCommands c@ProtocolClient {client_ = PClient {sndQ}} cs = do
     Left e -> pure $ Left e
 
 -- | Send Protocol command
-sendProtocolCommand :: forall msg. ProtocolEncoding (ProtoCommand msg) => ProtocolClient msg -> Maybe C.APrivateSignKey -> QueueId -> ProtoCommand msg -> ExceptT ProtocolClientError IO msg
+sendProtocolCommand :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> Maybe C.APrivateSignKey -> QueueId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
 sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}} pKey qId cmd = do
   (t, r) <- mkTransmission c (pKey, qId, cmd)
   ExceptT $ sendRecv t r
   where
     -- two separate "atomically" needed to avoid blocking
-    sendRecv :: SentRawTransmission -> TMVar (Response msg) -> IO (Response msg)
+    sendRecv :: SentRawTransmission -> TMVar (Response err msg) -> IO (Response err msg)
     sendRecv t r = atomically (writeTBQueue sndQ [t]) >> withTimeout c (atomically $ takeTMVar r)
 
-withTimeout :: ProtocolClient msg -> IO (Either ProtocolClientError msg) -> IO (Either ProtocolClientError msg)
+withTimeout :: ProtocolClient err msg -> IO (Either (ProtocolClientError err) msg) -> IO (Either (ProtocolClientError err) msg)
 withTimeout ProtocolClient {client_ = PClient {tcpTimeout, pingErrorCount}} a =
   timeout tcpTimeout a >>= \case
     Just r -> atomically (writeTVar pingErrorCount 0) >> pure r
     _ -> pure $ Left PCEResponseTimeout
 
-mkTransmission :: forall msg. ProtocolEncoding (ProtoCommand msg) => ProtocolClient msg -> ClientCommand msg -> ExceptT ProtocolClientError IO (SentRawTransmission, TMVar (Response msg))
+mkTransmission :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> ClientCommand msg -> ExceptT (ProtocolClientError err) IO (SentRawTransmission, TMVar (Response err msg))
 mkTransmission ProtocolClient {sessionId, thVersion, client_ = PClient {clientCorrId, sentCommands}} (pKey, qId, cmd) = do
   corrId <- liftIO $ atomically getNextCorrId
   let t = signTransmission $ encodeTransmission thVersion sessionId (corrId, qId, cmd)
@@ -604,7 +609,7 @@ mkTransmission ProtocolClient {sessionId, thVersion, client_ = PClient {clientCo
       pure . CorrId $ bshow i
     signTransmission :: ByteString -> SentRawTransmission
     signTransmission t = ((`C.sign` t) <$> pKey, t)
-    mkRequest :: CorrId -> STM (TMVar (Response msg))
+    mkRequest :: CorrId -> STM (TMVar (Response err msg))
     mkRequest corrId = do
       r <- newEmptyTMVar
       TM.insert corrId (Request qId r) sentCommands

src/Simplex/Messaging/Client/Agent.hs

@@ -38,14 +38,14 @@ import UnliftIO.Exception (Exception)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
-type SMPClientVar = TMVar (Either ProtocolClientError SMPClient)
+type SMPClientVar = TMVar (Either SMPClientError SMPClient)
 
 data SMPClientAgentEvent
   = CAConnected SMPServer
   | CADisconnected SMPServer (Set SMPSub)
   | CAReconnected SMPServer
   | CAResubscribed SMPServer SMPSub
-  | CASubError SMPServer SMPSub ProtocolClientError
+  | CASubError SMPServer SMPSub SMPClientError
 
 data SMPSubParty = SPRecipient | SPNotifier
   deriving (Eq, Ord, Show)
@@ -111,7 +111,7 @@ newSMPClientAgent agentCfg@SMPClientAgentConfig {msgQSize, agentQSize} = do
   asyncClients <- newTVar []
   pure SMPClientAgent {agentCfg, msgQ, agentQ, smpClients, srvSubs, pendingSrvSubs, reconnections, asyncClients}
 
-getSMPServerClient' :: SMPClientAgent -> SMPServer -> ExceptT ProtocolClientError IO SMPClient
+getSMPServerClient' :: SMPClientAgent -> SMPServer -> ExceptT SMPClientError IO SMPClient
 getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
   atomically getClientVar >>= either newSMPClient waitForSMPClient
   where
@@ -124,7 +124,7 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
       TM.insert srv smpVar smpClients
       pure smpVar
 
-    waitForSMPClient :: SMPClientVar -> ExceptT ProtocolClientError IO SMPClient
+    waitForSMPClient :: SMPClientVar -> ExceptT SMPClientError IO SMPClient
     waitForSMPClient smpVar = do
       let ProtocolClientConfig {networkConfig = NetworkConfig {tcpConnectTimeout}} = smpCfg agentCfg
       smpClient_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar smpVar)
@@ -133,10 +133,10 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
         Just (Left e) -> Left e
         Nothing -> Left PCEResponseTimeout
 
-    newSMPClient :: SMPClientVar -> ExceptT ProtocolClientError IO SMPClient
+    newSMPClient :: SMPClientVar -> ExceptT SMPClientError IO SMPClient
     newSMPClient smpVar = tryConnectClient pure tryConnectAsync
       where
-        tryConnectClient :: (SMPClient -> ExceptT ProtocolClientError IO a) -> ExceptT ProtocolClientError IO () -> ExceptT ProtocolClientError IO a
+        tryConnectClient :: (SMPClient -> ExceptT SMPClientError IO a) -> ExceptT SMPClientError IO () -> ExceptT SMPClientError IO a
         tryConnectClient successAction retryAction =
           tryE connectClient >>= \r -> case r of
             Right smp -> do
@@ -150,16 +150,16 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
                   putTMVar smpVar (Left e)
                   TM.delete srv smpClients
               throwE e
-        tryConnectAsync :: ExceptT ProtocolClientError IO ()
+        tryConnectAsync :: ExceptT SMPClientError IO ()
         tryConnectAsync = do
           a <- async connectAsync
           atomically $ modifyTVar' (asyncClients ca) (a :)
-        connectAsync :: ExceptT ProtocolClientError IO ()
+        connectAsync :: ExceptT SMPClientError IO ()
         connectAsync =
           withRetryInterval (reconnectInterval agentCfg) $ \loop ->
             void $ tryConnectClient (const reconnectClient) loop
 
-    connectClient :: ExceptT ProtocolClientError IO SMPClient
+    connectClient :: ExceptT SMPClientError IO SMPClient
     connectClient = ExceptT $ getProtocolClient (1, srv, Nothing) (smpCfg agentCfg) (Just msgQ) clientDisconnected
 
     clientDisconnected :: SMPClient -> IO ()
@@ -188,17 +188,17 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
       notify . CADisconnected srv $ M.keysSet ss
       reconnectServer
 
-    reconnectServer :: ExceptT ProtocolClientError IO ()
+    reconnectServer :: ExceptT SMPClientError IO ()
     reconnectServer = do
       a <- async tryReconnectClient
       atomically $ modifyTVar' (reconnections ca) (a :)
 
-    tryReconnectClient :: ExceptT ProtocolClientError IO ()
+    tryReconnectClient :: ExceptT SMPClientError IO ()
     tryReconnectClient = do
       withRetryInterval (reconnectInterval agentCfg) $ \loop ->
         reconnectClient `catchE` const loop
 
-    reconnectClient :: ExceptT ProtocolClientError IO ()
+    reconnectClient :: ExceptT SMPClientError IO ()
     reconnectClient = do
       withSMP ca srv $ \smp -> do
         notify $ CAReconnected srv
@@ -207,13 +207,13 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
           unlessM (atomically $ hasSub (srvSubs ca) srv s) $
             subscribe_ smp sub `catchE` handleError s
       where
-        subscribe_ :: SMPClient -> (SMPSub, C.APrivateSignKey) -> ExceptT ProtocolClientError IO ()
+        subscribe_ :: SMPClient -> (SMPSub, C.APrivateSignKey) -> ExceptT SMPClientError IO ()
         subscribe_ smp sub@(s, _) = do
           smpSubscribe smp sub
           atomically $ addSubscription ca srv sub
           notify $ CAResubscribed srv s
 
-        handleError :: SMPSub -> ProtocolClientError -> ExceptT ProtocolClientError IO ()
+        handleError :: SMPSub -> SMPClientError -> ExceptT SMPClientError IO ()
         handleError s = \case
           e@PCEResponseTimeout -> throwE e
           e@PCENetworkError -> throwE e
@@ -221,7 +221,7 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
             notify $ CASubError srv s e
             atomically $ removePendingSubscription ca srv s
 
-    notify :: SMPClientAgentEvent -> ExceptT ProtocolClientError IO ()
+    notify :: SMPClientAgentEvent -> ExceptT SMPClientError IO ()
     notify evt = atomically $ writeTBQueue (agentQ ca) evt
 
 closeSMPClientAgent :: MonadUnliftIO m => SMPClientAgent -> m ()
@@ -241,15 +241,15 @@ closeSMPServerClients c = readTVarIO (smpClients c) >>= mapM_ (forkIO . closeCli
 cancelActions :: Foldable f => TVar (f (Async ())) -> IO ()
 cancelActions as = readTVarIO as >>= mapM_ uninterruptibleCancel
 
-withSMP :: SMPClientAgent -> SMPServer -> (SMPClient -> ExceptT ProtocolClientError IO a) -> ExceptT ProtocolClientError IO a
+withSMP :: SMPClientAgent -> SMPServer -> (SMPClient -> ExceptT SMPClientError IO a) -> ExceptT SMPClientError IO a
 withSMP ca srv action = (getSMPServerClient' ca srv >>= action) `catchE` logSMPError
   where
-    logSMPError :: ProtocolClientError -> ExceptT ProtocolClientError IO a
+    logSMPError :: SMPClientError -> ExceptT SMPClientError IO a
     logSMPError e = do
       liftIO $ putStrLn $ "SMP error (" <> show srv <> "): " <> show e
       throwE e
 
-subscribeQueue :: SMPClientAgent -> SMPServer -> (SMPSub, C.APrivateSignKey) -> ExceptT ProtocolClientError IO ()
+subscribeQueue :: SMPClientAgent -> SMPServer -> (SMPSub, C.APrivateSignKey) -> ExceptT SMPClientError IO ()
 subscribeQueue ca srv sub = do
   atomically $ addPendingSubscription ca srv sub
   withSMP ca srv $ \smp -> subscribe_ smp `catchE` handleError
@@ -267,7 +267,7 @@ showServer :: SMPServer -> ByteString
 showServer ProtocolServer {host, port} =
   strEncode host <> B.pack (if null port then "" else ':' : port)
 
-smpSubscribe :: SMPClient -> (SMPSub, C.APrivateSignKey) -> ExceptT ProtocolClientError IO ()
+smpSubscribe :: SMPClient -> (SMPSub, C.APrivateSignKey) -> ExceptT SMPClientError IO ()
 smpSubscribe smp ((party, queueId), privKey) = subscribe_ smp privKey queueId
   where
     subscribe_ = case party of

src/Simplex/Messaging/Crypto.hs

@@ -103,6 +103,7 @@ module Simplex.Messaging.Crypto
 
     -- * NaCl crypto_box
     CbNonce (unCbNonce),
+    pattern CbNonce,
     cbEncrypt,
     cbEncryptMaxLenBS,
     cbDecrypt,
@@ -110,6 +111,14 @@ module Simplex.Messaging.Crypto
     randomCbNonce,
     pseudoRandomCbNonce,
 
+    -- * NaCl crypto_secretbox
+    SbKey (unSbKey),
+    pattern SbKey,
+    sbEncrypt,
+    sbDecrypt,
+    sbKey,
+    randomSbKey,
+
     -- * pseudo-random bytes
     pseudoRandomBytes,
 
@@ -153,6 +162,7 @@ import Data.ASN1.Types
 import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (bimap, first)
+import Data.ByteArray (ByteArrayAccess)
 import qualified Data.ByteArray as BA
 import Data.ByteString.Base64 (decode, encode)
 import qualified Data.ByteString.Base64.URL as U
@@ -690,6 +700,8 @@ data CryptoError
     AESDecryptError
   | -- CryptoBox decryption error
     CBDecryptError
+  | -- Poly1305 initialization error
+    CryptoPoly1305Error CE.CryptoError
   | -- | message is larger that allowed padded length minus 2 (to prepend message length)
     -- (or required un-padded length is larger than the message length)
     CryptoLargeMsgError
@@ -915,13 +927,20 @@ dh' (PublicKeyX448 k) (PrivateKeyX448 pk _) = DhSecretX448 $ X448.dh k pk
 
 -- | NaCl @crypto_box@ encrypt with a shared DH secret and 192-bit nonce.
 cbEncrypt :: DhSecret X25519 -> CbNonce -> ByteString -> Int -> Either CryptoError ByteString
-cbEncrypt secret (CbNonce nonce) msg paddedLen = cryptoBox secret nonce <$> pad msg paddedLen
+cbEncrypt (DhSecretX25519 secret) = sbEncrypt_ secret
+
+-- | NaCl @secret_box@ encrypt with a symmetric 256-bit key and 192-bit nonce.
+sbEncrypt :: SbKey -> CbNonce -> ByteString -> Int -> Either CryptoError ByteString
+sbEncrypt (SbKey key) = sbEncrypt_ key
+
+sbEncrypt_ :: ByteArrayAccess key => key -> CbNonce -> ByteString -> Int -> Either CryptoError ByteString
+sbEncrypt_ secret (CbNonce nonce) msg paddedLen = cryptoBox secret nonce <$> pad msg paddedLen
 
 -- | NaCl @crypto_box@ encrypt with a shared DH secret and 192-bit nonce.
 cbEncryptMaxLenBS :: KnownNat i => DhSecret X25519 -> CbNonce -> MaxLenBS i -> ByteString
-cbEncryptMaxLenBS secret (CbNonce nonce) = cryptoBox secret nonce . unMaxLenBS . padMaxLenBS
+cbEncryptMaxLenBS (DhSecretX25519 secret) (CbNonce nonce) = cryptoBox secret nonce . unMaxLenBS . padMaxLenBS
 
-cryptoBox :: DhSecret 'X25519 -> ByteString -> ByteString -> ByteString
+cryptoBox :: ByteArrayAccess key => key -> ByteString -> ByteString -> ByteString
 cryptoBox secret nonce s = BA.convert tag <> c
   where
     (rs, c) = xSalsa20 secret nonce s
@@ -929,7 +948,15 @@ cryptoBox secret nonce s = BA.convert tag <> c
 
 -- | NaCl @crypto_box@ decrypt with a shared DH secret and 192-bit nonce.
 cbDecrypt :: DhSecret X25519 -> CbNonce -> ByteString -> Either CryptoError ByteString
-cbDecrypt secret (CbNonce nonce) packet
+cbDecrypt (DhSecretX25519 secret) = sbDecrypt_ secret
+
+-- | NaCl @secret_box@ decrypt with a symmetric 256-bit key and 192-bit nonce.
+sbDecrypt :: SbKey -> CbNonce -> ByteString -> Either CryptoError ByteString
+sbDecrypt (SbKey key) = sbDecrypt_ key
+
+-- | NaCl @crypto_box@ decrypt with a shared DH secret and 192-bit nonce.
+sbDecrypt_ :: ByteArrayAccess key => key -> CbNonce -> ByteString -> Either CryptoError ByteString
+sbDecrypt_ secret (CbNonce nonce) packet
   | B.length packet < 16 = Left CBDecryptError
   | BA.constEq tag' tag = unPad msg
   | otherwise = Left CBDecryptError
@@ -938,9 +965,14 @@ cbDecrypt secret (CbNonce nonce) packet
     (rs, msg) = xSalsa20 secret nonce c
     tag = Poly1305.auth rs c
 
-newtype CbNonce = CbNonce {unCbNonce :: ByteString}
+newtype CbNonce = CryptoBoxNonce {unCbNonce :: ByteString}
   deriving (Eq, Show)
 
+pattern CbNonce :: ByteString -> CbNonce
+pattern CbNonce s <- CryptoBoxNonce s
+
+{-# COMPLETE CbNonce #-}
+
 instance StrEncoding CbNonce where
   strEncode (CbNonce s) = strEncode s
   strP = cbNonce <$> strP
@@ -951,31 +983,58 @@ instance ToJSON CbNonce where
 
 cbNonce :: ByteString -> CbNonce
 cbNonce s
-  | len == 24 = CbNonce s
-  | len > 24 = CbNonce . fst $ B.splitAt 24 s
-  | otherwise = CbNonce $ s <> B.replicate (24 - len) (toEnum 0)
+  | len == 24 = CryptoBoxNonce s
+  | len > 24 = CryptoBoxNonce . fst $ B.splitAt 24 s
+  | otherwise = CryptoBoxNonce $ s <> B.replicate (24 - len) (toEnum 0)
   where
     len = B.length s
 
 randomCbNonce :: IO CbNonce
-randomCbNonce = CbNonce <$> getRandomBytes 24
+randomCbNonce = CryptoBoxNonce <$> getRandomBytes 24
 
 pseudoRandomCbNonce :: TVar ChaChaDRG -> STM CbNonce
-pseudoRandomCbNonce gVar = CbNonce <$> pseudoRandomBytes 24 gVar
+pseudoRandomCbNonce gVar = CryptoBoxNonce <$> pseudoRandomBytes 24 gVar
 
 pseudoRandomBytes :: Int -> TVar ChaChaDRG -> STM ByteString
 pseudoRandomBytes n gVar = stateTVar gVar $ randomBytesGenerate n
 
 instance Encoding CbNonce where
   smpEncode = unCbNonce
-  smpP = CbNonce <$> A.take 24
+  smpP = CryptoBoxNonce <$> A.take 24
 
-xSalsa20 :: DhSecret X25519 -> ByteString -> ByteString -> (ByteString, ByteString)
-xSalsa20 (DhSecretX25519 shared) nonce msg = (rs, msg')
+newtype SbKey = SecretBoxKey {unSbKey :: ByteString}
+  deriving (Eq, Show)
+
+pattern SbKey :: ByteString -> SbKey
+pattern SbKey s <- SecretBoxKey s
+
+{-# COMPLETE SbKey #-}
+
+instance StrEncoding SbKey where
+  strEncode (SbKey s) = strEncode s
+  strP = sbKey <$> strP
+
+instance ToJSON SbKey where
+  toJSON = strToJSON
+  toEncoding = strToJEncoding
+
+sbKey :: ByteString -> SbKey
+sbKey s
+  | len == 32 = SecretBoxKey s
+  | len > 32 = SecretBoxKey . fst $ B.splitAt 32 s
+  | otherwise = SecretBoxKey $ s <> B.replicate (32 - len) (toEnum 0)
+  where
+    len = B.length s
+
+randomSbKey :: IO SbKey
+randomSbKey = SecretBoxKey <$> getRandomBytes 32
+
+xSalsa20 :: ByteArrayAccess key => key -> ByteString -> ByteString -> (ByteString, ByteString)
+xSalsa20 secret nonce msg = (rs, msg')
   where
     zero = B.replicate 16 $ toEnum 0
     (iv0, iv1) = B.splitAt 8 nonce
-    state0 = XSalsa.initialize 20 shared (zero `B.append` iv0)
+    state0 = XSalsa.initialize 20 secret (zero `B.append` iv0)
     state1 = XSalsa.derive state0 iv1
     (rs, state2) = XSalsa.generate state1 32
     (msg', _) = XSalsa.combine state2 msg

src/Simplex/Messaging/Crypto/Lazy.hs

@@ -0,0 +1,118 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE PatternSynonyms #-}
+
+module Simplex.Messaging.Crypto.Lazy
+  ( sha512Hash,
+    pad,
+    unPad,
+    sbEncrypt,
+    sbDecrypt,
+    fastReplicate,
+  )
+where
+
+import qualified Crypto.Cipher.XSalsa as XSalsa
+import qualified Crypto.Error as CE
+import Crypto.Hash (Digest, hashlazy)
+import Crypto.Hash.Algorithms (SHA512)
+import qualified Crypto.MAC.Poly1305 as Poly1305
+import Data.ByteArray (ByteArrayAccess)
+import qualified Data.ByteArray as BA
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString.Lazy.Char8 as LB
+import Data.Int (Int64)
+import Foreign (sizeOf)
+import Simplex.Messaging.Crypto (CbNonce, CryptoError (..), SbKey, pattern CbNonce, pattern SbKey)
+import Simplex.Messaging.Encoding
+
+type LazyByteString = LB.ByteString
+
+-- | SHA512 digest of a lazy bytestring.
+sha512Hash :: LazyByteString -> ByteString
+sha512Hash = BA.convert . (hashlazy :: LazyByteString -> Digest SHA512)
+
+-- this function does not validate the length of the message to avoid consuming all chunks,
+-- but if the passed string is longer it will truncate it to specified length
+pad :: LazyByteString -> Int64 -> Int64 -> Either CryptoError LazyByteString
+pad msg len paddedLen
+  | padLen >= 0 = Right $ LB.fromStrict encodedLen <> LB.take len msg <> fastReplicate padLen '#'
+  | otherwise = Left CryptoLargeMsgError
+  where
+    encodedLen = smpEncode len -- 8 bytes Int64 encoded length
+    padLen = paddedLen - len - 8
+
+fastReplicate :: Int64 -> Char -> LazyByteString
+fastReplicate n c
+  | n <= 0 = LB.empty
+  | n < chSize' = LB.fromStrict $ B.replicate (fromIntegral n) c
+  | otherwise = LB.fromChunks $ B.replicate (fromIntegral r) c : replicate (fromIntegral q) chPad
+  where
+    chSize = 65536 - 2 * sizeOf (undefined :: Int)
+    chPad = B.replicate chSize c
+    chSize' = fromIntegral chSize
+    (q, r) = quotRem n chSize'
+
+-- this function does not validate the length of the message to avoid consuming all chunks,
+-- so it can return a shorter string than expected
+unPad :: LazyByteString -> Either CryptoError LazyByteString
+unPad padded
+  | LB.length lenStr == 8 = case smpDecode $ LB.toStrict lenStr of
+    Right len
+      | len < 0 -> Left CryptoInvalidMsgError
+      | otherwise -> Right $ LB.take len rest
+    Left _ -> Left CryptoInvalidMsgError
+  | otherwise = Left CryptoInvalidMsgError
+  where
+    (lenStr, rest) = LB.splitAt 8 padded
+
+-- | NaCl @secret_box@ lazy encrypt with a symmetric 256-bit key and 192-bit nonce.
+sbEncrypt :: SbKey -> CbNonce -> LazyByteString -> Int64 -> Int64 -> Either CryptoError LazyByteString
+sbEncrypt (SbKey key) = sbEncrypt_ key
+
+sbEncrypt_ :: ByteArrayAccess key => key -> CbNonce -> LazyByteString -> Int64 -> Int64 -> Either CryptoError LazyByteString
+sbEncrypt_ secret (CbNonce nonce) msg len paddedLen = cryptoBox secret nonce =<< pad msg len paddedLen
+
+-- | NaCl @secret_box@ decrypt with a symmetric 256-bit key and 192-bit nonce.
+sbDecrypt :: SbKey -> CbNonce -> LazyByteString -> Either CryptoError LazyByteString
+sbDecrypt (SbKey key) = sbDecrypt_ key
+
+-- | NaCl @crypto_box@ decrypt with a shared DH secret and 192-bit nonce.
+sbDecrypt_ :: ByteArrayAccess key => key -> CbNonce -> LazyByteString -> Either CryptoError LazyByteString
+sbDecrypt_ secret (CbNonce nonce) packet
+  | LB.length tag' < 16 = Left CBDecryptError
+  | otherwise = case poly1305auth rs c of
+    Right tag
+      | BA.constEq (LB.toStrict tag') tag -> unPad msg
+      | otherwise -> Left CBDecryptError
+    Left e -> Left e
+  where
+    (tag', c) = LB.splitAt 16 packet
+    (rs, msg) = xSalsa20 secret nonce c
+
+cryptoBox :: ByteArrayAccess key => key -> ByteString -> LazyByteString -> Either CryptoError LazyByteString
+cryptoBox secret nonce s = (<> c) . LB.fromStrict . BA.convert <$> tag
+  where
+    (rs, c) = xSalsa20 secret nonce s
+    tag = poly1305auth rs c
+
+poly1305auth :: ByteString -> LazyByteString -> Either CryptoError Poly1305.Auth
+poly1305auth rs c = authTag <$> cryptoPassed (Poly1305.initialize rs)
+  where
+    authTag state = Poly1305.finalize $ Poly1305.updates state $ LB.toChunks c
+    cryptoPassed = \case
+      CE.CryptoPassed a -> Right a
+      CE.CryptoFailed e -> Left $ CryptoPoly1305Error e
+
+xSalsa20 :: ByteArrayAccess key => key -> ByteString -> LazyByteString -> (ByteString, LazyByteString)
+xSalsa20 secret nonce msg = (rs, msg')
+  where
+    zero = B.replicate 16 $ toEnum 0
+    (iv0, iv1) = B.splitAt 8 nonce
+    state0 = XSalsa.initialize 20 secret (zero `B.append` iv0)
+    state1 = XSalsa.derive state0 iv1
+    (rs, state2) = XSalsa.generate state1 32
+    (msg', _) = foldl update (LB.empty, state2) $ LB.toChunks msg
+    update (acc, state) chunk =
+      let (c, state') = XSalsa.combine state chunk
+       in (acc `LB.append` LB.fromStrict c, state')

src/Simplex/Messaging/Notifications/Client.hs

@@ -10,54 +10,57 @@ import Data.Word (Word16)
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol
+import Simplex.Messaging.Protocol (ErrorType)
 import Simplex.Messaging.Util (bshow)
 
-type NtfClient = ProtocolClient NtfResponse
+type NtfClient = ProtocolClient ErrorType NtfResponse
 
-ntfRegisterToken :: NtfClient -> C.APrivateSignKey -> NewNtfEntity 'Token -> ExceptT ProtocolClientError IO (NtfTokenId, C.PublicKeyX25519)
+type NtfClientError = ProtocolClientError ErrorType
+
+ntfRegisterToken :: NtfClient -> C.APrivateSignKey -> NewNtfEntity 'Token -> ExceptT NtfClientError IO (NtfTokenId, C.PublicKeyX25519)
 ntfRegisterToken c pKey newTkn =
   sendNtfCommand c (Just pKey) "" (TNEW newTkn) >>= \case
     NRTknId tknId dhKey -> pure (tknId, dhKey)
     r -> throwE . PCEUnexpectedResponse $ bshow r
 
-ntfVerifyToken :: NtfClient -> C.APrivateSignKey -> NtfTokenId -> NtfRegCode -> ExceptT ProtocolClientError IO ()
+ntfVerifyToken :: NtfClient -> C.APrivateSignKey -> NtfTokenId -> NtfRegCode -> ExceptT NtfClientError IO ()
 ntfVerifyToken c pKey tknId code = okNtfCommand (TVFY code) c pKey tknId
 
-ntfCheckToken :: NtfClient -> C.APrivateSignKey -> NtfTokenId -> ExceptT ProtocolClientError IO NtfTknStatus
+ntfCheckToken :: NtfClient -> C.APrivateSignKey -> NtfTokenId -> ExceptT NtfClientError IO NtfTknStatus
 ntfCheckToken c pKey tknId =
   sendNtfCommand c (Just pKey) tknId TCHK >>= \case
     NRTkn stat -> pure stat
     r -> throwE . PCEUnexpectedResponse $ bshow r
 
-ntfReplaceToken :: NtfClient -> C.APrivateSignKey -> NtfTokenId -> DeviceToken -> ExceptT ProtocolClientError IO ()
+ntfReplaceToken :: NtfClient -> C.APrivateSignKey -> NtfTokenId -> DeviceToken -> ExceptT NtfClientError IO ()
 ntfReplaceToken c pKey tknId token = okNtfCommand (TRPL token) c pKey tknId
 
-ntfDeleteToken :: NtfClient -> C.APrivateSignKey -> NtfTokenId -> ExceptT ProtocolClientError IO ()
+ntfDeleteToken :: NtfClient -> C.APrivateSignKey -> NtfTokenId -> ExceptT NtfClientError IO ()
 ntfDeleteToken = okNtfCommand TDEL
 
-ntfEnableCron :: NtfClient -> C.APrivateSignKey -> NtfTokenId -> Word16 -> ExceptT ProtocolClientError IO ()
+ntfEnableCron :: NtfClient -> C.APrivateSignKey -> NtfTokenId -> Word16 -> ExceptT NtfClientError IO ()
 ntfEnableCron c pKey tknId int = okNtfCommand (TCRN int) c pKey tknId
 
-ntfCreateSubscription :: NtfClient -> C.APrivateSignKey -> NewNtfEntity 'Subscription -> ExceptT ProtocolClientError IO NtfSubscriptionId
+ntfCreateSubscription :: NtfClient -> C.APrivateSignKey -> NewNtfEntity 'Subscription -> ExceptT NtfClientError IO NtfSubscriptionId
 ntfCreateSubscription c pKey newSub =
   sendNtfCommand c (Just pKey) "" (SNEW newSub) >>= \case
     NRSubId subId -> pure subId
     r -> throwE . PCEUnexpectedResponse $ bshow r
 
-ntfCheckSubscription :: NtfClient -> C.APrivateSignKey -> NtfSubscriptionId -> ExceptT ProtocolClientError IO NtfSubStatus
+ntfCheckSubscription :: NtfClient -> C.APrivateSignKey -> NtfSubscriptionId -> ExceptT NtfClientError IO NtfSubStatus
 ntfCheckSubscription c pKey subId =
   sendNtfCommand c (Just pKey) subId SCHK >>= \case
     NRSub stat -> pure stat
     r -> throwE . PCEUnexpectedResponse $ bshow r
 
-ntfDeleteSubscription :: NtfClient -> C.APrivateSignKey -> NtfSubscriptionId -> ExceptT ProtocolClientError IO ()
+ntfDeleteSubscription :: NtfClient -> C.APrivateSignKey -> NtfSubscriptionId -> ExceptT NtfClientError IO ()
 ntfDeleteSubscription = okNtfCommand SDEL
 
 -- | Send notification server command
-sendNtfCommand :: NtfEntityI e => NtfClient -> Maybe C.APrivateSignKey -> NtfEntityId -> NtfCommand e -> ExceptT ProtocolClientError IO NtfResponse
+sendNtfCommand :: NtfEntityI e => NtfClient -> Maybe C.APrivateSignKey -> NtfEntityId -> NtfCommand e -> ExceptT NtfClientError IO NtfResponse
 sendNtfCommand c pKey entId cmd = sendProtocolCommand c pKey entId (NtfCmd sNtfEntity cmd)
 
-okNtfCommand :: NtfEntityI e => NtfCommand e -> NtfClient -> C.APrivateSignKey -> NtfEntityId -> ExceptT ProtocolClientError IO ()
+okNtfCommand :: NtfEntityI e => NtfCommand e -> NtfClient -> C.APrivateSignKey -> NtfEntityId -> ExceptT NtfClientError IO ()
 okNtfCommand cmd c pKey entId =
   sendNtfCommand c (Just pKey) entId cmd >>= \case
     NROk -> return ()

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -1,6 +1,7 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
@@ -146,7 +147,7 @@ instance Encoding ANewNtfEntity where
       'S' -> ANE SSubscription <$> (NewNtfSub <$> smpP <*> smpP <*> smpP)
       _ -> fail "bad ANewNtfEntity"
 
-instance Protocol NtfResponse where
+instance Protocol ErrorType NtfResponse where
   type ProtoCommand NtfResponse = NtfCmd
   type ProtoType NtfResponse = 'PNTF
   protocolClientHandshake = ntfClientHandshake
@@ -183,7 +184,7 @@ data NtfCmd = forall e. NtfEntityI e => NtfCmd (SNtfEntity e) (NtfCommand e)
 
 deriving instance Show NtfCmd
 
-instance NtfEntityI e => ProtocolEncoding (NtfCommand e) where
+instance NtfEntityI e => ProtocolEncoding ErrorType (NtfCommand e) where
   type Tag (NtfCommand e) = NtfCommandTag e
   encodeProtocol _v = \case
     TNEW newTkn -> e (TNEW_, ' ', newTkn)
@@ -202,6 +203,9 @@ instance NtfEntityI e => ProtocolEncoding (NtfCommand e) where
 
   protocolP _v tag = (\(NtfCmd _ c) -> checkEntity c) <$?> protocolP _v (NCT (sNtfEntity @e) tag)
 
+  fromProtocolError = fromProtocolError @ErrorType @NtfResponse
+  {-# INLINE fromProtocolError #-}
+
   checkCredentials (sig, _, entityId, _) cmd = case cmd of
     -- TNEW and SNEW must have signature but NOT token/subscription IDs
     TNEW {} -> sigNoEntity
@@ -219,7 +223,7 @@ instance NtfEntityI e => ProtocolEncoding (NtfCommand e) where
         | not (B.null entityId) = Left $ CMD HAS_AUTH
         | otherwise = Right cmd
 
-instance ProtocolEncoding NtfCmd where
+instance ProtocolEncoding ErrorType NtfCmd where
   type Tag NtfCmd = NtfCmdTag
   encodeProtocol _v (NtfCmd _ c) = encodeProtocol _v c
 
@@ -239,6 +243,9 @@ instance ProtocolEncoding NtfCmd where
         SDEL_ -> pure SDEL
         PING_ -> pure PING
 
+  fromProtocolError = fromProtocolError @ErrorType @NtfResponse
+  {-# INLINE fromProtocolError #-}
+
   checkCredentials t (NtfCmd e c) = NtfCmd e <$> checkCredentials t c
 
 data NtfResponseTag
@@ -283,7 +290,7 @@ data NtfResponse
   | NRPong
   deriving (Show)
 
-instance ProtocolEncoding NtfResponse where
+instance ProtocolEncoding ErrorType NtfResponse where
   type Tag NtfResponse = NtfResponseTag
   encodeProtocol _v = \case
     NRTknId entId dhKey -> e (NRTknId_, ' ', entId, dhKey)
@@ -306,6 +313,13 @@ instance ProtocolEncoding NtfResponse where
     NRSub_ -> NRSub <$> _smpP
     NRPong_ -> pure NRPong
 
+  fromProtocolError = \case
+    PECmdSyntax -> CMD SYNTAX
+    PECmdUnknown -> CMD UNKNOWN
+    PESession -> SESSION
+    PEBlock -> BLOCK
+  {-# INLINE fromProtocolError #-}
+
   checkCredentials (_, _, entId, _) cmd = case cmd of
     -- IDTKN response must not have queue ID
     NRTknId {} -> noEntity

src/Simplex/Messaging/Notifications/Server.hs

@@ -27,7 +27,7 @@ import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import Network.Socket (ServiceName)
-import Simplex.Messaging.Client (ProtocolClientError (..))
+import Simplex.Messaging.Client (ProtocolClientError (..), SMPClientError)
 import Simplex.Messaging.Client.Agent
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
@@ -227,7 +227,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
       where
         showServer' = decodeLatin1 . strEncode . host
 
-    handleSubError :: SMPQueueNtf -> ProtocolClientError -> M ()
+    handleSubError :: SMPQueueNtf -> SMPClientError -> M ()
     handleSubError smpQueue = \case
       PCEProtocolError AUTH -> updateSubStatus smpQueue NSAuth
       PCEProtocolError e -> updateErr "SMP error " e
@@ -319,7 +319,7 @@ clientDisconnected NtfServerClient {connected} = atomically $ writeTVar connecte
 
 receive :: Transport c => THandle c -> NtfServerClient -> M ()
 receive th NtfServerClient {rcvQ, sndQ, activeAt} = forever $ do
-  ts <- tGet th
+  ts <- liftIO $ tGet th
   forM_ ts $ \t@(_, _, (corrId, entId, cmdOrError)) -> do
     atomically . writeTVar activeAt =<< liftIO getSystemTime
     logDebug "received transmission"
@@ -343,7 +343,7 @@ send h@THandle {thVersion = v} NtfServerClient {sndQ, sessionId, activeAt} = for
 
 data VerificationResult = VRVerified NtfRequest | VRFailed
 
-verifyNtfTransmission :: SignedTransmission NtfCmd -> NtfCmd -> M VerificationResult
+verifyNtfTransmission :: SignedTransmission ErrorType NtfCmd -> NtfCmd -> M VerificationResult
 verifyNtfTransmission (sig_, signed, (corrId, entId, _)) cmd = do
   st <- asks store
   case cmd of

src/Simplex/Messaging/Notifications/Server/Push/APNS.hs

@@ -32,10 +32,12 @@ import qualified Data.ByteString.Lazy.Char8 as LB
 import qualified Data.CaseInsensitive as CI
 import Data.Int (Int64)
 import Data.Map.Strict (Map)
+import Data.Maybe (isNothing)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock.System
 import qualified Data.X509 as X
+import qualified Data.X509.CertificateStore as XS
 import GHC.Generics (Generic)
 import Network.HTTP.Types (HeaderName, Status)
 import qualified Network.HTTP.Types as N
@@ -47,6 +49,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Store (NtfTknData (..))
 import Simplex.Messaging.Protocol (EncNMsgMeta)
+import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..))
 import Simplex.Messaging.Transport.HTTP2.Client
 import Simplex.Messaging.Util (safeDecodeUtf8)
 import System.Environment (getEnv)
@@ -97,8 +100,8 @@ readECPrivateKey f = do
 data PushNotification
   = PNVerification NtfRegCode
   | PNMessage PNMessageData
-  -- | PNAlert Text
-  | PNCheckMessages
+  | -- | PNAlert Text
+    PNCheckMessages
   deriving (Show)
 
 data PNMessageData = PNMessageData
@@ -194,9 +197,9 @@ data APNSPushClientConfig = APNSPushClientConfig
     appName :: ByteString,
     appTeamId :: Text,
     apnsPort :: ServiceName,
-    http2cfg :: HTTP2ClientConfig
+    http2cfg :: HTTP2ClientConfig,
+    caStoreFile :: FilePath
   }
-  deriving (Show)
 
 apnsProviderHost :: PushProvider -> HostName
 apnsProviderHost = \case
@@ -215,7 +218,8 @@ defaultAPNSPushClientConfig =
       appName = "chat.simplex.app",
       appTeamId = "5NN7GUYB6T",
       apnsPort = "443",
-      http2cfg = defaultHTTP2ClientConfig
+      http2cfg = defaultHTTP2ClientConfig {bufferSize = 16384},
+      caStoreFile = "/etc/ssl/cert.pem"
     }
 
 data APNSPushClient = APNSPushClient
@@ -259,8 +263,10 @@ mkApnsJWTToken appTeamId jwtHeader privateKey = do
   pure (jwt, signedJWT)
 
 connectHTTPS2 :: HostName -> APNSPushClientConfig -> TVar (Maybe HTTP2Client) -> IO (Either HTTP2ClientError HTTP2Client)
-connectHTTPS2 apnsHost APNSPushClientConfig {apnsPort, http2cfg} https2Client = do
-  r <- getHTTP2Client apnsHost apnsPort http2cfg disconnected
+connectHTTPS2 apnsHost APNSPushClientConfig {apnsPort, http2cfg, caStoreFile} https2Client = do
+  caStore_ <- XS.readCertificateStore caStoreFile
+  when (isNothing caStore_) $ putStrLn $ "Error loading CertificateStore from " <> caStoreFile
+  r <- getHTTP2Client apnsHost apnsPort caStore_ http2cfg disconnected
   case r of
     Right client -> atomically . writeTVar https2Client $ Just client
     Left e -> putStrLn $ "Error connecting to APNS: " <> show e
@@ -294,7 +300,8 @@ apnsNotification NtfTknData {tknDhSecret} nonce paddedLen = \case
     encrypt ntfData f = f . safeDecodeUtf8 . U.encode <$> C.cbEncrypt tknDhSecret nonce ntfData paddedLen
     apn aps notificationData = APNSNotification {aps, notificationData}
     apnMutableContent = APNSMutableContent {mutableContent = 1, alert = APNSAlertText "Encrypted message or another app event", category = Just ntfCategoryCheckMessage}
-    -- apnAlert alert = APNSAlert {alert, badge = Nothing, sound = Nothing, category = Nothing}
+
+-- apnAlert alert = APNSAlert {alert, badge = Nothing, sound = Nothing, category = Nothing}
 
 apnsRequest :: APNSPushClient -> ByteString -> APNSNotification -> IO Request
 apnsRequest c tkn ntf@APNSNotification {aps} = do
@@ -336,9 +343,9 @@ apnsPushProviderClient c@APNSPushClient {nonceDrg, apnsCfg} tkn@NtfTknData {toke
   nonce <- atomically $ C.pseudoRandomCbNonce nonceDrg
   apnsNtf <- liftEither $ first PPCryptoError $ apnsNotification tkn nonce (paddedNtfLength apnsCfg) pn
   req <- liftIO $ apnsRequest c tknStr apnsNtf
-  HTTP2Response {response, respBody} <- liftHTTPS2 $ sendRequest http2 req
+  HTTP2Response {response, respBody = HTTP2Body {bodyHead}} <- liftHTTPS2 $ sendRequest http2 req
   let status = H.responseStatus response
-      reason' = maybe "" reason $ J.decodeStrict' respBody
+      reason' = maybe "" reason $ J.decodeStrict' bodyHead
   logDebug $ "APNS response: " <> T.pack (show status) <> " " <> reason'
   result status reason'
   where

src/Simplex/Messaging/Protocol.hs

@@ -4,6 +4,7 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE FunctionalDependencies #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -19,11 +20,10 @@
 {-# LANGUAGE TypeFamilyDependencies #-}
 {-# LANGUAGE TypeOperators #-}
 {-# LANGUAGE UndecidableInstances #-}
+{-# HLINT ignore "Use newtype instead of data" #-}
 {-# OPTIONS_GHC -Wno-unrecognised-pragmas #-}
 {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
 
-{-# HLINT ignore "Use newtype instead of data" #-}
-
 -- |
 -- Module      : Simplex.Messaging.ProtocolEncoding
 -- Copyright   : (c) simplex.chat
@@ -52,6 +52,7 @@ module Simplex.Messaging.Protocol
     SParty (..),
     PartyI (..),
     QueueIdsKeys (..),
+    ProtocolErrorType (..),
     ErrorType (..),
     CommandError (..),
     Transmission,
@@ -126,6 +127,10 @@ module Simplex.Messaging.Protocol
     -- * TCP transport functions
     tPut,
     tGet,
+    tParse,
+    tDecodeParseValidate,
+    tEncode,
+    tEncodeBatch,
 
     -- * exports for tests
     CommandTag (..),
@@ -220,7 +225,7 @@ deriving instance Show Cmd
 type Transmission c = (CorrId, EntityId, c)
 
 -- | signed parsed transmission, with original raw bytes and parsing error.
-type SignedTransmission c = (Maybe C.ASignature, Signed, Transmission (Either ErrorType c))
+type SignedTransmission e c = (Maybe C.ASignature, Signed, Transmission (Either e c))
 
 type Signed = ByteString
 
@@ -870,6 +875,8 @@ type MsgId = ByteString
 -- | SMP message body.
 type MsgBody = ByteString
 
+data ProtocolErrorType = PECmdSyntax | PECmdUnknown | PESession | PEBlock
+
 -- | Type for protocol errors.
 data ErrorType
   = -- | incorrect block format, encoding or signature size
@@ -940,16 +947,16 @@ transmissionP = do
       command <- A.takeByteString
       pure RawTransmission {signature, signed, sessId, corrId, entityId, command}
 
-class (ProtocolEncoding msg, ProtocolEncoding (ProtoCommand msg), Show msg) => Protocol msg where
+class (ProtocolEncoding err msg, ProtocolEncoding err (ProtoCommand msg), Show err, Show msg) => Protocol err msg | msg -> err where
   type ProtoCommand msg = cmd | cmd -> msg
   type ProtoType msg = (sch :: ProtocolType) | sch -> msg
   protocolClientHandshake :: forall c. Transport c => c -> C.KeyHash -> VersionRange -> ExceptT TransportError IO (THandle c)
   protocolPing :: ProtoCommand msg
-  protocolError :: msg -> Maybe ErrorType
+  protocolError :: msg -> Maybe err
 
 type ProtoServer msg = ProtocolServer (ProtoType msg)
 
-instance Protocol BrokerMsg where
+instance Protocol ErrorType BrokerMsg where
   type ProtoCommand BrokerMsg = Cmd
   type ProtoType BrokerMsg = 'PSMP
   protocolClientHandshake = smpClientHandshake
@@ -958,13 +965,14 @@ instance Protocol BrokerMsg where
     ERR e -> Just e
     _ -> Nothing
 
-class ProtocolMsgTag (Tag msg) => ProtocolEncoding msg where
+class ProtocolMsgTag (Tag msg) => ProtocolEncoding err msg | msg -> err where
   type Tag msg
   encodeProtocol :: Version -> msg -> ByteString
   protocolP :: Version -> Tag msg -> Parser msg
-  checkCredentials :: SignedRawTransmission -> msg -> Either ErrorType msg
+  fromProtocolError :: ProtocolErrorType -> err
+  checkCredentials :: SignedRawTransmission -> msg -> Either err msg
 
-instance PartyI p => ProtocolEncoding (Command p) where
+instance PartyI p => ProtocolEncoding ErrorType (Command p) where
   type Tag (Command p) = CommandTag p
   encodeProtocol v = \case
     NEW rKey dhKey auth_ -> case auth_ of
@@ -995,6 +1003,9 @@ instance PartyI p => ProtocolEncoding (Command p) where
 
   protocolP v tag = (\(Cmd _ c) -> checkParty c) <$?> protocolP v (CT (sParty @p) tag)
 
+  fromProtocolError = fromProtocolError @ErrorType @BrokerMsg
+  {-# INLINE fromProtocolError #-}
+
   checkCredentials (sig, _, queueId, _) cmd = case cmd of
     -- NEW must have signature but NOT queue ID
     NEW {}
@@ -1014,7 +1025,7 @@ instance PartyI p => ProtocolEncoding (Command p) where
       | isNothing sig || B.null queueId -> Left $ CMD NO_AUTH
       | otherwise -> Right cmd
 
-instance ProtocolEncoding Cmd where
+instance ProtocolEncoding ErrorType Cmd where
   type Tag Cmd = CmdTag
   encodeProtocol v (Cmd _ c) = encodeProtocol v c
 
@@ -1044,9 +1055,12 @@ instance ProtocolEncoding Cmd where
         PING_ -> pure PING
     CT SNotifier NSUB_ -> pure $ Cmd SNotifier NSUB
 
+  fromProtocolError = fromProtocolError @ErrorType @BrokerMsg
+  {-# INLINE fromProtocolError #-}
+
   checkCredentials t (Cmd p c) = Cmd p <$> checkCredentials t c
 
-instance ProtocolEncoding BrokerMsg where
+instance ProtocolEncoding ErrorType BrokerMsg where
   type Tag BrokerMsg = BrokerMsgTag
   encodeProtocol v = \case
     IDS (QIK rcvId sndId srvDh) -> e (IDS_, ' ', rcvId, sndId, srvDh)
@@ -1081,6 +1095,13 @@ instance ProtocolEncoding BrokerMsg where
     ERR_ -> ERR <$> _smpP
     PONG_ -> pure PONG
 
+  fromProtocolError = \case
+    PECmdSyntax -> CMD SYNTAX
+    PECmdUnknown -> CMD UNKNOWN
+    PESession -> SESSION
+    PEBlock -> BLOCK
+  {-# INLINE fromProtocolError #-}
+
   checkCredentials (_, _, queueId, _) cmd = case cmd of
     -- IDS response should not have queue ID
     IDS _ -> Right cmd
@@ -1099,12 +1120,12 @@ _smpP :: Encoding a => Parser a
 _smpP = A.space *> smpP
 
 -- | Parse SMP protocol commands and broker messages
-parseProtocol :: ProtocolEncoding msg => Version -> ByteString -> Either ErrorType msg
+parseProtocol :: forall err msg. ProtocolEncoding err msg => Version -> ByteString -> Either err msg
 parseProtocol v s =
   let (tag, params) = B.break (== ' ') s
    in case decodeTag tag of
-        Just cmd -> parse (protocolP v cmd) (CMD SYNTAX) params
-        Nothing -> Left $ CMD UNKNOWN
+        Just cmd -> parse (protocolP v cmd) (fromProtocolError @err @msg $ PECmdSyntax) params
+        Nothing -> Left $ fromProtocolError @err @msg $ PECmdUnknown
 
 checkParty :: forall t p p'. (PartyI p, PartyI p') => t p' -> Either String (t p)
 checkParty c = case testEquality (sParty @p) (sParty @p') of
@@ -1169,7 +1190,7 @@ tPut th trs
     tPutBatch :: [Either TransportError ()] -> NonEmpty ByteString -> IO [Either TransportError ()]
     tPutBatch rs ts = do
       let (n, s, ts_) = encodeBatch 0 "" ts
-      r <- if n == 0 then largeMsg else replicate n <$> tPutLog (lenEncode n `B.cons` s)
+      r <- if n == 0 then largeMsg else replicate n <$> tPutLog (tEncodeBatch n s)
       let rs' = rs <> r
       case ts_ of
         Just ts' -> tPutBatch rs' ts'
@@ -1192,41 +1213,49 @@ tPut th trs
               else case L.nonEmpty ts_ of
                 Just ts' -> encodeBatch n' s' ts'
                 _ -> (n', s', Nothing)
-    tEncode (sig, tr) = smpEncode (C.signatureBytes sig) <> tr
 
-encodeTransmission :: ProtocolEncoding c => Version -> ByteString -> Transmission c -> ByteString
+tEncode :: C.CryptoSignature s => (s, ByteString) -> ByteString
+tEncode (sig, t) = smpEncode (C.signatureBytes sig) <> t
+
+tEncodeBatch :: Int -> ByteString -> ByteString
+tEncodeBatch n s = lenEncode n `B.cons` s
+
+encodeTransmission :: ProtocolEncoding e c => Version -> ByteString -> Transmission c -> ByteString
 encodeTransmission v sessionId (CorrId corrId, queueId, command) =
   smpEncode (sessionId, corrId, queueId) <> encodeProtocol v command
 
 -- | Receive and parse transmission from the TCP transport (ignoring any trailing padding).
 tGetParse :: Transport c => THandle c -> IO (NonEmpty (Either TransportError RawTransmission))
-tGetParse th
-  | batch th = either ((:| []) . Left) id <$> runExceptT getBatch
-  | otherwise = (:| []) . (parse transmissionP TEBadBlock =<<) <$> tGetBlock th
+tGetParse th = eitherList (tParse $ batch th) <$> tGetBlock th
+
+tParse :: Bool -> ByteString -> NonEmpty (Either TransportError RawTransmission)
+tParse batch s
+  | batch = eitherList (L.map (\(Large t) -> tParse1 t)) ts
+  | otherwise = [tParse1 s]
   where
-    getBatch :: ExceptT TransportError IO (NonEmpty (Either TransportError RawTransmission))
-    getBatch = do
-      s <- ExceptT $ tGetBlock th
-      ts <- liftEither $ parse smpP TEBadBlock s
-      pure $ L.map (\(Large t) -> parse transmissionP TEBadBlock t) ts
+    tParse1 = parse transmissionP TEBadBlock
+    ts = parse smpP TEBadBlock s
+
+eitherList :: (a -> NonEmpty (Either e b)) -> Either e a -> NonEmpty (Either e b)
+eitherList = either (\e -> [Left e])
 
 -- | Receive client and server transmissions (determined by `cmd` type).
-tGet :: forall cmd c m. (ProtocolEncoding cmd, Transport c, MonadIO m) => THandle c -> m (NonEmpty (SignedTransmission cmd))
-tGet th@THandle {sessionId, thVersion = v} = liftIO (tGetParse th) >>= mapM decodeParseValidate
+tGet :: forall err cmd c. (ProtocolEncoding err cmd, Transport c) => THandle c -> IO (NonEmpty (SignedTransmission err cmd))
+tGet th@THandle {sessionId, thVersion = v} = L.map (tDecodeParseValidate sessionId v) <$> tGetParse th
+
+tDecodeParseValidate :: forall err cmd. ProtocolEncoding err cmd => SessionId -> Version -> Either TransportError RawTransmission -> SignedTransmission err cmd
+tDecodeParseValidate sessionId v = \case
+  Right RawTransmission {signature, signed, sessId, corrId, entityId, command}
+    | sessId == sessionId ->
+      let decodedTransmission = (,corrId,entityId,command) <$> C.decodeSignature signature
+       in either (const $ tError corrId) (tParseValidate signed) decodedTransmission
+    | otherwise -> (Nothing, "", (CorrId corrId, "", Left $ fromProtocolError @err @cmd PESession))
+  Left _ -> tError ""
   where
-    decodeParseValidate :: Either TransportError RawTransmission -> m (SignedTransmission cmd)
-    decodeParseValidate = \case
-      Right RawTransmission {signature, signed, sessId, corrId, entityId, command}
-        | sessId == sessionId ->
-          let decodedTransmission = (,corrId,entityId,command) <$> C.decodeSignature signature
-           in either (const $ tError corrId) (tParseValidate signed) decodedTransmission
-        | otherwise -> pure (Nothing, "", (CorrId corrId, "", Left SESSION))
-      Left _ -> tError ""
+    tError :: ByteString -> SignedTransmission err cmd
+    tError corrId = (Nothing, "", (CorrId corrId, "", Left $ fromProtocolError @err @cmd PEBlock))
 
-    tError :: ByteString -> m (SignedTransmission cmd)
-    tError corrId = pure (Nothing, "", (CorrId corrId, "", Left BLOCK))
-
-    tParseValidate :: ByteString -> SignedRawTransmission -> m (SignedTransmission cmd)
-    tParseValidate signed t@(sig, corrId, entityId, command) = do
-      let cmd = parseProtocol v command >>= checkCredentials t
-      pure (sig, signed, (CorrId corrId, entityId, cmd))
+    tParseValidate :: ByteString -> SignedRawTransmission -> SignedTransmission err cmd
+    tParseValidate signed t@(sig, corrId, entityId, command) =
+      let cmd = parseProtocol @err @cmd v command >>= checkCredentials t
+       in (sig, signed, (CorrId corrId, entityId, cmd))

src/Simplex/Messaging/Server.hs

@@ -260,13 +260,13 @@ cancelSub sub =
 
 receive :: Transport c => THandle c -> Client -> M ()
 receive th Client {rcvQ, sndQ, activeAt} = forever $ do
-  ts <- L.toList <$> tGet th
+  ts <- L.toList <$> liftIO (tGet th)
   atomically . writeTVar activeAt =<< liftIO getSystemTime
   as <- partitionEithers <$> mapM cmdAction ts
   write sndQ $ fst as
   write rcvQ $ snd as
   where
-    cmdAction :: SignedTransmission Cmd -> M (Either (Transmission BrokerMsg) (Maybe QueueRec, Transmission Cmd))
+    cmdAction :: SignedTransmission ErrorType Cmd -> M (Either (Transmission BrokerMsg) (Maybe QueueRec, Transmission Cmd))
     cmdAction (sig, signed, (corrId, queueId, cmdOrError)) =
       case cmdOrError of
         Left e -> pure $ Left (corrId, queueId, ERR e)

src/Simplex/Messaging/Server/StoreLog.hs

@@ -38,7 +38,7 @@ import qualified Data.Map.Strict as M
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..), ServerQueueStatus (..))
-import Simplex.Messaging.Transport (trimCR)
+import Simplex.Messaging.Transport.Buffer (trimCR)
 import System.Directory (doesFileExist)
 import System.IO
 

src/Simplex/Messaging/Transport.hs

@@ -56,9 +56,6 @@ module Simplex.Messaging.Transport
     transportErrorP,
     sendHandshake,
     getHandshake,
-
-    -- * Trim trailing CR
-    trimCR,
   )
 where
 
@@ -86,6 +83,7 @@ import qualified Paths_simplexmq as SMQ
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (dropPrefix, parse, parseRead1, sumTypeJSON)
+import Simplex.Messaging.Transport.Buffer
 import Simplex.Messaging.Util (bshow, catchAll, catchAll_)
 import Simplex.Messaging.Version
 import Test.QuickCheck (Arbitrary (..))
@@ -152,8 +150,7 @@ data TLS = TLS
   { tlsContext :: T.Context,
     tlsPeer :: TransportPeer,
     tlsUniq :: ByteString,
-    buffer :: TVar ByteString,
-    getLock :: TMVar ()
+    tlsBuffer :: TBuffer
   }
 
 connectTLS :: T.TLSParams p => Maybe HostName -> Bool -> p -> Socket -> IO T.Context
@@ -169,9 +166,8 @@ getTLS :: TransportPeer -> T.Context -> IO TLS
 getTLS tlsPeer cxt = withTlsUnique tlsPeer cxt newTLS
   where
     newTLS tlsUniq = do
-      buffer <- newTVarIO ""
-      getLock <- newTMVarIO ()
-      pure TLS {tlsContext = cxt, tlsPeer, tlsUniq, buffer, getLock}
+      tlsBuffer <- atomically newTBuffer
+      pure TLS {tlsContext = cxt, tlsPeer, tlsUniq, tlsBuffer}
 
 withTlsUnique :: TransportPeer -> T.Context -> (ByteString -> IO c) -> IO c
 withTlsUnique peer cxt f =
@@ -209,52 +205,22 @@ instance Transport TLS where
   closeConnection tls = closeTLS $ tlsContext tls
 
   cGet :: TLS -> Int -> IO ByteString
-  cGet TLS {tlsContext, buffer, getLock} n =
-    E.bracket_
-      (atomically $ takeTMVar getLock)
-      (atomically $ putTMVar getLock ())
-      $ do
-        b <- readChunks =<< readTVarIO buffer
-        let (s, b') = B.splitAt n b
-        atomically $ writeTVar buffer $! b'
-        pure s
-    where
-      readChunks :: ByteString -> IO ByteString
-      readChunks b
-        | B.length b >= n = pure b
-        | otherwise =
-          T.recvData tlsContext >>= \case
-            -- https://hackage.haskell.org/package/tls-1.6.0/docs/Network-TLS.html#v:recvData
-            "" -> ioe_EOF
-            s -> readChunks $ b <> s
+  cGet TLS {tlsContext, tlsBuffer} n = do
+    s <- getBuffered tlsBuffer n (T.recvData tlsContext)
+    -- https://hackage.haskell.org/package/tls-1.6.0/docs/Network-TLS.html#v:recvData
+    if B.length s == n then pure s else ioe_EOF
 
   cPut :: TLS -> ByteString -> IO ()
   cPut tls = T.sendData (tlsContext tls) . BL.fromStrict
 
   getLn :: TLS -> IO ByteString
-  getLn TLS {tlsContext, buffer, getLock} = do
-    E.bracket_
-      (atomically $ takeTMVar getLock)
-      (atomically $ putTMVar getLock ())
-      $ do
-        b <- readChunks =<< readTVarIO buffer
-        let (s, b') = B.break (== '\n') b
-        atomically $ writeTVar buffer $! B.drop 1 b' -- drop '\n' we made a break at
-        pure $ trimCR s
+  getLn TLS {tlsContext, tlsBuffer} = do
+    getLnBuffered tlsBuffer (T.recvData tlsContext) `E.catch` handleEOF
     where
-      readChunks :: ByteString -> IO ByteString
-      readChunks b
-        | B.elem '\n' b = pure b
-        | otherwise = readChunks . (b <>) =<< T.recvData tlsContext `E.catch` handleEOF
       handleEOF = \case
         T.Error_EOF -> E.throwIO TEBadBlock
         e -> E.throwIO e
 
--- | Trim trailing CR from ByteString.
-trimCR :: ByteString -> ByteString
-trimCR "" = ""
-trimCR s = if B.last s == '\r' then B.init s else s
-
 -- * SMP transport
 
 -- | The handle for SMP encrypted transport connection over Transport .

src/Simplex/Messaging/Transport/Buffer.hs

@@ -0,0 +1,61 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Simplex.Messaging.Transport.Buffer where
+
+import Control.Concurrent.STM
+import qualified Control.Exception as E
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+
+data TBuffer = TBuffer
+  { buffer :: TVar ByteString,
+    getLock :: TMVar ()
+  }
+
+newTBuffer :: STM TBuffer
+newTBuffer = do
+  buffer <- newTVar ""
+  getLock <- newTMVar ()
+  pure TBuffer {buffer, getLock}
+
+withBufferLock :: TBuffer -> IO a -> IO a
+withBufferLock TBuffer {getLock} =
+  E.bracket_
+    (atomically $ takeTMVar getLock)
+    (atomically $ putTMVar getLock ())
+
+getBuffered :: TBuffer -> Int -> IO ByteString -> IO ByteString
+getBuffered tb@TBuffer {buffer} n getChunk = withBufferLock tb $ do
+  b <- readChunks =<< readTVarIO buffer
+  let (s, b') = B.splitAt n b
+  atomically $ writeTVar buffer $! b'
+  pure s
+  where
+    readChunks :: ByteString -> IO ByteString
+    readChunks b
+      | B.length b >= n = pure b
+      | otherwise =
+        getChunk >>= \case
+          "" -> pure b
+          s -> readChunks $ b <> s
+
+-- This function is only used in test and needs to be improved before it can be used in production,
+-- it will never complete if TLS connection is closed before there is newline.
+getLnBuffered :: TBuffer -> IO ByteString -> IO ByteString
+getLnBuffered tb@TBuffer {buffer} getChunk = withBufferLock tb $ do
+  b <- readChunks =<< readTVarIO buffer
+  let (s, b') = B.break (== '\n') b
+  atomically $ writeTVar buffer $! B.drop 1 b' -- drop '\n' we made a break at
+  pure $ trimCR s
+  where
+    readChunks :: ByteString -> IO ByteString
+    readChunks b
+      | B.elem '\n' b = pure b
+      | otherwise = readChunks . (b <>) =<< getChunk
+
+-- | Trim trailing CR from ByteString.
+trimCR :: ByteString -> ByteString
+trimCR "" = ""
+trimCR s = if B.last s == '\r' then B.init s else s

src/Simplex/Messaging/Transport/HTTP2.hs

@@ -1,20 +1,32 @@
+{-# LANGUAGE NamedFieldPuns #-}
+
 module Simplex.Messaging.Transport.HTTP2 where
 
+import Control.Concurrent.STM
 import qualified Control.Exception as E
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
 import Data.Default (def)
+import Data.Maybe (fromMaybe)
 import Foreign (mallocBytes)
 import Network.HPACK (BufferSize)
 import Network.HTTP2.Client (Config (..), defaultPositionReadMaker, freeSimpleConfig)
+import qualified Network.HTTP2.Client as HC
+import qualified Network.HTTP2.Server as HS
 import qualified Network.TLS as T
 import qualified Network.TLS.Extra as TE
-import Simplex.Messaging.Transport (TLS, Transport (cGet, cPut))
+import Simplex.Messaging.Transport (SessionId, TLS (tlsUniq), Transport (cGet, cPut))
+import Simplex.Messaging.Transport.Buffer
 import qualified System.TimeManager as TI
 
-withTlsConfig :: TLS -> BufferSize -> (Config -> IO ()) -> IO ()
-withTlsConfig c sz = E.bracket (allocTlsConfig c sz) freeSimpleConfig
+defaultHTTP2BufferSize :: BufferSize
+defaultHTTP2BufferSize = 32768
 
-allocTlsConfig :: TLS -> BufferSize -> IO Config
-allocTlsConfig c sz = do
+withHTTP2 :: BufferSize -> (Config -> SessionId -> IO ()) -> TLS -> IO ()
+withHTTP2 sz run c = E.bracket (allocHTTP2Config c sz) freeSimpleConfig (`run` tlsUniq c)
+
+allocHTTP2Config :: TLS -> BufferSize -> IO Config
+allocHTTP2Config c sz = do
   buf <- mallocBytes sz
   tm <- TI.initialize $ 30 * 1000000
   pure
@@ -34,3 +46,35 @@ http2TLSParams =
       T.supportedCiphers = TE.ciphersuite_strong_det,
       T.supportedSecureRenegotiation = False
     }
+
+data HTTP2Body = HTTP2Body
+  { bodyHead :: ByteString,
+    bodySize :: Int,
+    bodyPart :: Maybe (Int -> IO ByteString),
+    bodyBuffer :: TBuffer
+  }
+
+class HTTP2BodyChunk a where
+  getBodyChunk :: a -> IO ByteString
+  getBodeSize :: a -> Maybe Int
+
+instance HTTP2BodyChunk HC.Response where
+  getBodyChunk = HC.getResponseBodyChunk
+  {-# INLINE getBodyChunk #-}
+  getBodeSize = HC.responseBodySize
+  {-# INLINE getBodeSize #-}
+
+instance HTTP2BodyChunk HS.Request where
+  getBodyChunk = HS.getRequestBodyChunk
+  {-# INLINE getBodyChunk #-}
+  getBodeSize = HS.requestBodySize
+  {-# INLINE getBodeSize #-}
+
+getHTTP2Body :: HTTP2BodyChunk a => a -> Int -> IO HTTP2Body
+getHTTP2Body r n = do
+  bodyBuffer <- atomically newTBuffer
+  let getPart n' = getBuffered bodyBuffer n' $ getBodyChunk r
+  bodyHead <- getPart n
+  let bodySize = fromMaybe 0 $ getBodeSize r
+      bodyPart = if bodySize > n && B.length bodyHead == n then Just getPart else Nothing
+  pure HTTP2Body {bodyHead, bodySize, bodyPart, bodyBuffer}

src/Simplex/Messaging/Transport/HTTP2/Client.hs

@@ -10,24 +10,32 @@ import Control.Exception (IOException)
 import qualified Control.Exception as E
 import Control.Monad.Except
 import Data.ByteString.Char8 (ByteString)
-import qualified Data.ByteString.Char8 as B
-import Data.Maybe (isNothing)
+import Data.Time (UTCTime, getCurrentTime)
 import qualified Data.X509.CertificateStore as XS
-import Network.HPACK (HeaderTable)
+import Network.HPACK (BufferSize)
 import Network.HTTP2.Client (ClientConfig (..), Request, Response)
 import qualified Network.HTTP2.Client as H
 import Network.Socket (HostName, ServiceName)
 import qualified Network.TLS as T
 import Numeric.Natural (Natural)
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Encoding.String
+import Simplex.Messaging.Transport (SessionId)
 import Simplex.Messaging.Transport.Client (TransportClientConfig (..), TransportHost (..), runTLSTransportClient)
-import Simplex.Messaging.Transport.HTTP2 (http2TLSParams, withTlsConfig)
+import Simplex.Messaging.Transport.HTTP2
 import UnliftIO.STM
 import UnliftIO.Timeout
 
 data HTTP2Client = HTTP2Client
   { action :: Maybe (Async ()),
-    connected :: TVar Bool,
-    host :: HostName,
+    sessionId :: SessionId,
+    sessionTs :: UTCTime,
+    client_ :: HClient
+  }
+
+data HClient = HClient
+  { connected :: TVar Bool,
+    host :: TransportHost,
     port :: ServiceName,
     config :: HTTP2ClientConfig,
     reqQ :: TBQueue (Request, TMVar HTTP2Response)
@@ -35,15 +43,15 @@ data HTTP2Client = HTTP2Client
 
 data HTTP2Response = HTTP2Response
   { response :: Response,
-    respBody :: ByteString,
-    respTrailers :: Maybe HeaderTable
+    respBody :: HTTP2Body
   }
 
 data HTTP2ClientConfig = HTTP2ClientConfig
   { qSize :: Natural,
     connTimeout :: Int,
     transportConfig :: TransportClientConfig,
-    caStoreFile :: FilePath,
+    bufferSize :: BufferSize,
+    bodyHeadSize :: Int,
     suportedTLSParams :: T.Supported
   }
   deriving (Show)
@@ -54,73 +62,69 @@ defaultHTTP2ClientConfig =
     { qSize = 64,
       connTimeout = 10000000,
       transportConfig = TransportClientConfig Nothing Nothing True,
-      caStoreFile = "/etc/ssl/cert.pem",
+      bufferSize = defaultHTTP2BufferSize,
+      bodyHeadSize = 16384,
       suportedTLSParams = http2TLSParams
     }
 
-data HTTP2ClientError = HCResponseTimeout | HCNetworkError | HCNetworkError1 | HCIOError IOException
+data HTTP2ClientError = HCResponseTimeout | HCNetworkError | HCIOError IOException
   deriving (Show)
 
-getHTTP2Client :: HostName -> ServiceName -> HTTP2ClientConfig -> IO () -> IO (Either HTTP2ClientError HTTP2Client)
-getHTTP2Client host port config@HTTP2ClientConfig {transportConfig, connTimeout, caStoreFile, suportedTLSParams} disconnected =
+getHTTP2Client :: HostName -> ServiceName -> Maybe XS.CertificateStore -> HTTP2ClientConfig -> IO () -> IO (Either HTTP2ClientError HTTP2Client)
+getHTTP2Client host port = getVerifiedHTTP2Client Nothing (THDomainName host) port Nothing
+
+getVerifiedHTTP2Client :: Maybe ByteString -> TransportHost -> ServiceName -> Maybe C.KeyHash -> Maybe XS.CertificateStore -> HTTP2ClientConfig -> IO () -> IO (Either HTTP2ClientError HTTP2Client)
+getVerifiedHTTP2Client proxyUsername host port keyHash caStore config@HTTP2ClientConfig {transportConfig, bufferSize, bodyHeadSize, connTimeout, suportedTLSParams} disconnected =
   (atomically mkHTTPS2Client >>= runClient)
     `E.catch` \(e :: IOException) -> pure . Left $ HCIOError e
   where
-    mkHTTPS2Client :: STM HTTP2Client
+    mkHTTPS2Client :: STM HClient
     mkHTTPS2Client = do
       connected <- newTVar False
       reqQ <- newTBQueue $ qSize config
-      pure HTTP2Client {action = Nothing, connected, host, port, config, reqQ}
+      pure HClient {connected, host, port, config, reqQ}
 
-    runClient :: HTTP2Client -> IO (Either HTTP2ClientError HTTP2Client)
+    runClient :: HClient -> IO (Either HTTP2ClientError HTTP2Client)
     runClient c = do
       cVar <- newEmptyTMVarIO
-      caStore <- XS.readCertificateStore caStoreFile
-      when (isNothing caStore) . putStrLn $ "Error loading CertificateStore from " <> caStoreFile
       action <-
         async $
-          runHTTP2Client suportedTLSParams caStore transportConfig host port (client c cVar)
+          runHTTP2Client suportedTLSParams caStore transportConfig bufferSize proxyUsername host port keyHash (client c cVar)
             `E.finally` atomically (putTMVar cVar $ Left HCNetworkError)
-      conn_ <- connTimeout `timeout` atomically (takeTMVar cVar)
-      pure $ case conn_ of
-        Just (Right ()) -> Right c {action = Just action}
+      c_ <- connTimeout `timeout` atomically (takeTMVar cVar)
+      pure $ case c_ of
+        Just (Right c') -> Right c' {action = Just action}
         Just (Left e) -> Left e
-        Nothing -> Left HCNetworkError1
+        Nothing -> Left HCNetworkError
 
-    client :: HTTP2Client -> TMVar (Either HTTP2ClientError ()) -> (Request -> (Response -> IO ()) -> IO ()) -> IO ()
-    client c cVar sendReq = do
+    client :: HClient -> TMVar (Either HTTP2ClientError HTTP2Client) -> SessionId -> H.Client ()
+    client c cVar sessionId sendReq = do
+      sessionTs <- getCurrentTime
+      let c' = HTTP2Client {action = Nothing, client_ = c, sessionId, sessionTs}
       atomically $ do
         writeTVar (connected c) True
-        putTMVar cVar $ Right ()
-      process c sendReq `E.finally` disconnected
+        putTMVar cVar (Right c')
+      process c' sendReq `E.finally` disconnected
 
-    process :: HTTP2Client -> (Request -> (Response -> IO ()) -> IO ()) -> IO ()
-    process HTTP2Client {reqQ} sendReq = forever $ do
+    process :: HTTP2Client -> H.Client ()
+    process HTTP2Client {client_ = HClient {reqQ}} sendReq = forever $ do
       (req, respVar) <- atomically $ readTBQueue reqQ
       sendReq req $ \r -> do
-        let writeResp respBody respTrailers = atomically $ putTMVar respVar HTTP2Response {response = r, respBody, respTrailers}
-        respBody <- getResponseBody r ""
-        respTrailers <- H.getResponseTrailers r
-        writeResp respBody respTrailers
-
-    getResponseBody :: Response -> ByteString -> IO ByteString
-    getResponseBody r s =
-      H.getResponseBodyChunk r >>= \chunk ->
-        if B.null chunk then pure s else getResponseBody r $ s <> chunk
+        respBody <- getHTTP2Body r bodyHeadSize
+        atomically $ putTMVar respVar HTTP2Response {response = r, respBody}
 
 -- | Disconnects client from the server and terminates client threads.
 closeHTTP2Client :: HTTP2Client -> IO ()
 closeHTTP2Client = mapM_ uninterruptibleCancel . action
 
 sendRequest :: HTTP2Client -> Request -> IO (Either HTTP2ClientError HTTP2Response)
-sendRequest HTTP2Client {reqQ, config} req = do
+sendRequest HTTP2Client {client_ = HClient {config, reqQ}} req = do
   resp <- newEmptyTMVarIO
   atomically $ writeTBQueue reqQ (req, resp)
   maybe (Left HCResponseTimeout) Right <$> (connTimeout config `timeout` atomically (takeTMVar resp))
 
-runHTTP2Client :: T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> HostName -> ServiceName -> ((Request -> (Response -> IO ()) -> IO ()) -> IO ()) -> IO ()
-runHTTP2Client tlsParams caStore tcConfig host port client =
-  runTLSTransportClient tlsParams caStore tcConfig Nothing (THDomainName host) port Nothing $ \c ->
-    withTlsConfig c 16384 (`run` client)
+runHTTP2Client :: T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> BufferSize -> Maybe ByteString -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (SessionId -> H.Client ()) -> IO ()
+runHTTP2Client tlsParams caStore tcConfig bufferSize proxyUsername host port keyHash client =
+  runTLSTransportClient tlsParams caStore tcConfig proxyUsername host port keyHash $ withHTTP2 bufferSize run
   where
-    run = H.run $ ClientConfig "https" (B.pack host) 20
+    run cfg = H.run (ClientConfig "https" (strEncode host) 20) cfg . client

src/Simplex/Messaging/Transport/HTTP2/Server.hs

@@ -6,22 +6,23 @@ module Simplex.Messaging.Transport.HTTP2.Server where
 import Control.Concurrent.Async (Async, async, uninterruptibleCancel)
 import Control.Concurrent.STM
 import Control.Monad
-import Data.ByteString (ByteString)
-import qualified Data.ByteString.Char8 as B
-import Network.HPACK (HeaderTable)
-import Network.HTTP2.Server (Aux, PushPromise, Request, Response)
+import Network.HPACK (BufferSize)
+import Network.HTTP2.Server (Request, Response)
 import qualified Network.HTTP2.Server as H
 import Network.Socket
 import qualified Network.TLS as T
 import Numeric.Natural (Natural)
-import Simplex.Messaging.Transport.HTTP2 (withTlsConfig)
+import Simplex.Messaging.Transport (SessionId)
+import Simplex.Messaging.Transport.HTTP2
 import Simplex.Messaging.Transport.Server (loadSupportedTLSServerParams, runTransportServer)
 
-type HTTP2ServerFunc = (Request -> (Response -> IO ()) -> IO ())
+type HTTP2ServerFunc = SessionId -> Request -> (Response -> IO ()) -> IO ()
 
 data HTTP2ServerConfig = HTTP2ServerConfig
   { qSize :: Natural,
     http2Port :: ServiceName,
+    bufferSize :: BufferSize,
+    bodyHeadSize :: Int,
     serverSupported :: T.Supported,
     caCertificateFile :: FilePath,
     privateKeyFile :: FilePath,
@@ -31,9 +32,9 @@ data HTTP2ServerConfig = HTTP2ServerConfig
   deriving (Show)
 
 data HTTP2Request = HTTP2Request
-  { request :: Request,
-    reqBody :: ByteString,
-    reqTrailers :: Maybe HeaderTable,
+  { sessionId :: SessionId,
+    request :: Request,
+    reqBody :: HTTP2Body,
     sendResponse :: Response -> IO ()
   }
 
@@ -42,30 +43,24 @@ data HTTP2Server = HTTP2Server
     reqQ :: TBQueue HTTP2Request
   }
 
+-- This server is for testing only, it processes all requests in a single queue.
 getHTTP2Server :: HTTP2ServerConfig -> IO HTTP2Server
-getHTTP2Server HTTP2ServerConfig {qSize, http2Port, serverSupported, caCertificateFile, certificateFile, privateKeyFile, logTLSErrors} = do
+getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, serverSupported, caCertificateFile, certificateFile, privateKeyFile, logTLSErrors} = do
   tlsServerParams <- loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile privateKeyFile
   started <- newEmptyTMVarIO
   reqQ <- newTBQueueIO qSize
   action <- async $
-    runHTTP2Server started http2Port tlsServerParams logTLSErrors $ \r sendResponse -> do
-      reqBody <- getRequestBody r ""
-      reqTrailers <- H.getRequestTrailers r
-      atomically $ writeTBQueue reqQ HTTP2Request {request = r, reqBody, reqTrailers, sendResponse}
+    runHTTP2Server started http2Port bufferSize tlsServerParams logTLSErrors $ \sessionId r sendResponse -> do
+      reqBody <- getHTTP2Body r bodyHeadSize
+      atomically $ writeTBQueue reqQ HTTP2Request {sessionId, request = r, reqBody, sendResponse}
   void . atomically $ takeTMVar started
   pure HTTP2Server {action, reqQ}
-  where
-    getRequestBody :: Request -> ByteString -> IO ByteString
-    getRequestBody r s =
-      H.getRequestBodyChunk r >>= \chunk ->
-        if B.null chunk then pure s else getRequestBody r $ s <> chunk
 
 closeHTTP2Server :: HTTP2Server -> IO ()
 closeHTTP2Server = uninterruptibleCancel . action
 
-runHTTP2Server :: TMVar Bool -> ServiceName -> T.ServerParams -> Bool -> HTTP2ServerFunc -> IO ()
-runHTTP2Server started port serverParams logTLSErrors http2Server =
-  runTransportServer started port serverParams logTLSErrors $ \c -> withTlsConfig c 16384 (`H.run` server)
+runHTTP2Server :: TMVar Bool -> ServiceName -> BufferSize -> T.ServerParams -> Bool -> HTTP2ServerFunc -> IO ()
+runHTTP2Server started port bufferSize serverParams logTLSErrors http2Server =
+  runTransportServer started port serverParams logTLSErrors $ withHTTP2 bufferSize run
   where
-    server :: Request -> Aux -> (Response -> [PushPromise] -> IO ()) -> IO ()
-    server req _aux sendResp = http2Server req (`sendResp` [])
+    run cfg sessId = H.run cfg $ \req _aux sendResp -> http2Server sessId req (`sendResp` [])

src/Simplex/Messaging/Transport/WebSockets.hs

@@ -19,9 +19,9 @@ import Simplex.Messaging.Transport
     TransportPeer (..),
     closeTLS,
     smpBlockSize,
-    trimCR,
     withTlsUnique,
   )
+import Simplex.Messaging.Transport.Buffer (trimCR)
 
 data WS = WS
   { wsPeer :: TransportPeer,

tests/AgentTests/FunctionalAPITests.hs

@@ -25,6 +25,7 @@ module AgentTests.FunctionalAPITests
   )
 where
 
+import AgentTests.ConnectionRequestTests (connReqData, queueAddr, testE2ERatchetParams)
 import Control.Concurrent (killThread, threadDelay)
 import Control.Monad
 import Control.Monad.Except (ExceptT, runExceptT, throwError)
@@ -45,8 +46,9 @@ import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.Store (UserId)
 import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), TransportSessionMode (TSMEntity, TSMUser), defaultClientConfig)
+import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolServer (..))
+import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolServer (..), supportedSMPClientVRange)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
@@ -96,8 +98,10 @@ runRight action =
 
 functionalAPITests :: ATransport -> Spec
 functionalAPITests t = do
-  describe "Establishing duplex connection" $
+  describe "Establishing duplex connection" $ do
     testMatrix2 t runAgentClientTest
+    it "should connect when server with multiple identities is stored" $
+      withSmpServer t testServerMultipleIdentities
   describe "Establishing duplex connection v2, different Ratchet versions" $
     testRatchetMatrix2 t runAgentClientTest
   describe "Establish duplex connection via contact address" $
@@ -1010,6 +1014,40 @@ testTwoUsers = do
     hasClients :: HasCallStack => AgentClient -> Int -> ExceptT AgentErrorType IO ()
     hasClients c n = liftIO $ M.size <$> readTVarIO (smpClients c) `shouldReturn` n
 
+testServerMultipleIdentities :: HasCallStack => IO ()
+testServerMultipleIdentities = do
+  alice <- getSMPAgentClient agentCfg initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
+  runRight_ $ do
+    (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing
+    aliceId <- joinConnection bob 1 True cReq "bob's connInfo"
+    ("", _, CONF confId _ "bob's connInfo") <- get alice
+    allowConnection alice bobId confId "alice's connInfo"
+    get alice ##> ("", bobId, CON)
+    get bob ##> ("", aliceId, INFO "alice's connInfo")
+    get bob ##> ("", aliceId, CON)
+    exchangeGreetings alice bobId bob aliceId
+    -- this saves queue with second server identity
+    Left (BROKER _ NETWORK) <- runExceptT $ joinConnection bob 1 True secondIdentityCReq "bob's connInfo"
+    disconnectAgentClient bob
+    bob' <- liftIO $ getSMPAgentClient agentCfg {database = testDB2} initAgentServers
+    subscribeConnection bob' aliceId
+    exchangeGreetingsMsgId 6 alice bobId bob' aliceId
+  where
+    secondIdentityCReq :: ConnectionRequestUri 'CMInvitation
+    secondIdentityCReq =
+      CRInvitationUri
+        connReqData
+          { crSmpQueues =
+              [ SMPQueueUri
+                  supportedSMPClientVRange
+                  queueAddr
+                    { smpServer = SMPServer "localhost" "5001" (C.KeyHash "\215m\248\251")
+                    }
+              ]
+          }
+        testE2ERatchetParams
+
 exchangeGreetings :: HasCallStack => AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
 exchangeGreetings = exchangeGreetingsMsgId 4
 

tests/CoreTests/CryptoTests.hs

@@ -4,10 +4,15 @@
 module CoreTests.CryptoTests (cryptoTests) where
 
 import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Either (isRight)
+import Data.Int (Int64)
 import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
+import qualified Data.Text.Lazy as LT
+import qualified Data.Text.Lazy.Encoding as LE
 import qualified Simplex.Messaging.Crypto as C
+import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Test.Hspec
 import Test.Hspec.QuickCheck (modifyMaxSuccess)
 import Test.QuickCheck
@@ -36,24 +41,65 @@ cryptoTests = do
     it "unpad should fail on shorter string" $ do
       C.unPad "\000\003abc" `shouldBe` Right "abc"
       C.unPad "\000\003ab" `shouldBe` Left C.CryptoInvalidMsgError
+  modifyMaxSuccess (const 10000) . describe "lazy padding / unpadding" $ do
+    it "should pad / unpad lazy bytestrings" . property $ \(s, paddedLen) ->
+      let b = LE.encodeUtf8 $ LT.pack s
+          len = LB.length b
+          padded = LC.pad b len paddedLen
+       in if len <= paddedLen - 8
+            then (padded >>= LC.unPad) == Right b
+            else padded == Left C.CryptoLargeMsgError
+    it "pad should support large string" $ do
+      LC.pad "abc" 3 11 `shouldBe` Right "\000\000\000\000\000\000\000\003abc"
+      LC.pad "abc" 3 10 `shouldBe` Left C.CryptoLargeMsgError
+      let s = LB.replicate 100000 'a'
+      (LC.pad s 100000 100100 >>= LC.unPad) `shouldBe` Right s
+      (LC.pad s 100000 100008 >>= LC.unPad) `shouldBe` Right s
+      (LC.pad s 100000 100007 >>= LC.unPad) `shouldBe` Left C.CryptoLargeMsgError
+    it "pad should truncate string if a shorter length is passed and unpad incorrectly when longer length is passed" $ do
+      let s = LB.replicate 10000 'a'
+      (LC.pad s 9000 10100 >>= LC.unPad) `shouldBe` Right (LB.take 9000 s)
+      (LC.pad s 11000 11100 >>= LC.unPad) `shouldBe` Right (s <> LB.replicate 92 '#') -- 92 = pad size, it is not truncated in this case
+    it "unpad should fail on invalid string" $ do
+      LC.unPad "\000\000\000\000\000\000\000\000" `shouldBe` Right ""
+      LC.unPad "\000\000" `shouldBe` Left C.CryptoInvalidMsgError
+      LC.unPad "" `shouldBe` Left C.CryptoInvalidMsgError
+    it "unpad won't fail on shorter string" $ do
+      LC.unPad "\000\000\000\000\000\000\000\003abc" `shouldBe` Right "abc"
+      LC.unPad "\000\000\000\000\000\000\000\003ab" `shouldBe` Right "ab"
+    it "should pad / unpad file" testPadUnpadFile
   describe "Ed signatures" $ do
     describe "Ed25519" $ testSignature C.SEd25519
     describe "Ed448" $ testSignature C.SEd448
-  describe "DH X25519 + cryptobox" $
-    testDHCryptoBox
+  describe "DH X25519 + cryptobox" testDHCryptoBox
+  describe "secretbox" testSecretBox
+  describe "lazy secretbox" $ do
+    testLazySecretBox
+    testLazySecretBoxFile
   describe "X509 key encoding" $ do
     describe "Ed25519" $ testEncoding C.SEd25519
     describe "Ed448" $ testEncoding C.SEd448
     describe "X25519" $ testEncoding C.SX25519
     describe "X448" $ testEncoding C.SX448
 
+testPadUnpadFile :: IO ()
+testPadUnpadFile = do
+  let f = "tests/tmp/testpad"
+      paddedLen = 1024 * 1024
+      len = 1000000
+      s = LB.replicate len 'a'
+  Right s' <- pure $ LC.pad s len paddedLen
+  LB.writeFile (f <> ".padded") s'
+  Right s'' <- LC.unPad <$> LB.readFile (f <> ".padded")
+  s'' `shouldBe` s
+
 testSignature :: (C.AlgorithmI a, C.SignatureAlgorithm a) => C.SAlgorithm a -> Spec
 testSignature alg = it "should sign / verify string" . ioProperty $ do
   (k, pk) <- C.generateSignatureKeyPair alg
   pure $ \s -> let b = encodeUtf8 $ T.pack s in C.verify k (C.sign pk b) b
 
 testDHCryptoBox :: Spec
-testDHCryptoBox = it "should encrypt / decrypt string" . ioProperty $ do
+testDHCryptoBox = it "should encrypt / decrypt string with asymmetric DH keys" . ioProperty $ do
   (sk, spk) <- C.generateKeyPair'
   (rk, rpk) <- C.generateKeyPair'
   nonce <- C.randomCbNonce
@@ -64,6 +110,44 @@ testDHCryptoBox = it "should encrypt / decrypt string" . ioProperty $ do
         plain = C.cbDecrypt (C.dh' sk rpk) nonce =<< cipher
      in isRight cipher && cipher /= plain && Right b == plain
 
+testSecretBox :: Spec
+testSecretBox = it "should encrypt / decrypt string with a random symmetric key" . ioProperty $ do
+  k <- C.randomSbKey
+  nonce <- C.randomCbNonce
+  pure $ \(s, pad) ->
+    let b = encodeUtf8 $ T.pack s
+        pad' = min (abs pad) 100000
+        paddedLen = B.length b + pad' + 2
+        cipher = C.sbEncrypt k nonce b paddedLen
+        plain = C.sbDecrypt k nonce =<< cipher
+     in isRight cipher && cipher /= plain && Right b == plain
+
+testLazySecretBox :: Spec
+testLazySecretBox = it "should lazily encrypt / decrypt string with a random symmetric key" . ioProperty $ do
+  k <- C.randomSbKey
+  nonce <- C.randomCbNonce
+  pure $ \(s, pad) ->
+    let b = LE.encodeUtf8 $ LT.pack s
+        len = LB.length b
+        pad' = min (abs pad) 100000
+        paddedLen = len + pad' + 8
+        cipher = LC.sbEncrypt k nonce b len paddedLen
+        plain = LC.sbDecrypt k nonce =<< cipher
+     in isRight cipher && cipher /= plain && Right b == plain
+
+testLazySecretBoxFile :: Spec
+testLazySecretBoxFile = it "should lazily encrypt / decrypt file with a random symmetric key" $ do
+  k <- C.randomSbKey
+  nonce <- C.randomCbNonce
+  let f = "tests/tmp/testsecretbox"
+      paddedLen = 4 * 1024 * 1024
+      len = 4 * 1000 * 1000 :: Int64
+      s = LC.fastReplicate len 'a'
+  Right s' <- pure $ LC.sbEncrypt k nonce s len paddedLen
+  LB.writeFile (f <> ".encrypted") s'
+  Right s'' <- LC.sbDecrypt k nonce <$> LB.readFile (f <> ".encrypted")
+  s'' `shouldBe` s
+
 testEncoding :: (C.AlgorithmI a) => C.SAlgorithm a -> Spec
 testEncoding alg = it "should encode / decode key" . ioProperty $ do
   (k, pk) <- C.generateKeyPair alg

tests/CoreTests/ProtocolErrorTests.hs

@@ -6,7 +6,7 @@ module CoreTests.ProtocolErrorTests where
 import qualified Data.ByteString.Char8 as B
 import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
-import Simplex.Messaging.Agent.Protocol (AgentErrorType (..))
+import Simplex.Messaging.Agent.Protocol (AgentErrorType (..), BrokerErrorType (..))
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll)
 import Test.Hspec
@@ -17,12 +17,14 @@ protocolErrorTests :: Spec
 protocolErrorTests = modifyMaxSuccess (const 1000) $ do
   describe "errors parsing / serializing" $ do
     it "should parse SMP protocol errors" . property $ \(err :: AgentErrorType) ->
-      errServerHasSpaces err
+      errHasSpaces err
         || parseAll strP (strEncode err) == Right err
     it "should parse SMP agent errors" . property $ \(err :: AgentErrorType) ->
-      errServerHasSpaces err
+      errHasSpaces err
         || parseAll strP (strEncode err) == Right err
   where
-    errServerHasSpaces = \case
-      BROKER srv _ -> ' ' `B.elem` encodeUtf8 (T.pack srv)
+    errHasSpaces = \case
+      BROKER srv (RESPONSE e) -> hasSpaces srv || hasSpaces e
+      BROKER srv _ -> hasSpaces srv
       _ -> False
+    hasSpaces s = ' ' `B.elem` encodeUtf8 (T.pack s)

tests/NtfClient.hs

@@ -41,8 +41,7 @@ import Simplex.Messaging.Notifications.Transport
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
-import Simplex.Messaging.Transport.HTTP2 (http2TLSParams)
-import Simplex.Messaging.Transport.HTTP2.Client
+import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..), http2TLSParams)
 import Simplex.Messaging.Transport.HTTP2.Server
 import Test.Hspec
 import UnliftIO.Async
@@ -87,7 +86,7 @@ ntfServerCfg =
       apnsConfig =
         defaultAPNSPushClientConfig
           { apnsPort = apnsTestPort,
-            http2cfg = defaultHTTP2ClientConfig {caStoreFile = "tests/fixtures/ca.crt"}
+            caStoreFile = "tests/fixtures/ca.crt"
           },
       inactiveClientExpiration = Just defaultInactiveClientExpiration,
       storeLogFile = Nothing,
@@ -175,6 +174,8 @@ apnsMockServerConfig =
   HTTP2ServerConfig
     { qSize = 1,
       http2Port = apnsTestPort,
+      bufferSize = 16384,
+      bodyHeadSize = 16384,
       serverSupported = http2TLSParams,
       caCertificateFile = "tests/fixtures/ca.crt",
       privateKeyFile = "tests/fixtures/server.key",
@@ -210,16 +211,16 @@ getAPNSMockServer config@HTTP2ServerConfig {qSize} = do
   pure APNSMockServer {action, apnsQ, http2Server}
   where
     runAPNSMockServer apnsQ HTTP2Server {reqQ} = forever $ do
-      HTTP2Request {reqBody, sendResponse} <- atomically $ readTBQueue reqQ
+      HTTP2Request {reqBody = HTTP2Body {bodyHead}, sendResponse} <- atomically $ readTBQueue reqQ
       let sendApnsResponse = \case
             APNSRespOk -> sendResponse $ H.responseNoBody N.ok200 []
             APNSRespError status reason ->
               sendResponse . H.responseBuilder status [] . lazyByteString $ J.encode APNSErrorResponse {reason}
-      case J.decodeStrict' reqBody of
+      case J.decodeStrict' bodyHead of
         Just notification ->
           atomically $ writeTBQueue apnsQ APNSMockRequest {notification, sendApnsResponse}
         _ -> do
-          putStrLn $ "runAPNSMockServer J.decodeStrict' error, reqBody: " <> show reqBody
+          putStrLn $ "runAPNSMockServer J.decodeStrict' error, reqBody: " <> show bodyHead
           sendApnsResponse $ APNSRespError N.badRequest400 "bad_request_body"
 
 closeAPNSMockServer :: APNSMockServer -> IO ()

tests/NtfServerTests.hs

@@ -64,16 +64,16 @@ ntfSyntaxTests (ATransport t) = do
       Expectation
     command >#> response = withAPNSMockServer $ \_ -> ntfServerTest t command `shouldReturn` response
 
-pattern RespNtf :: CorrId -> QueueId -> NtfResponse -> SignedTransmission NtfResponse
+pattern RespNtf :: CorrId -> QueueId -> NtfResponse -> SignedTransmission ErrorType NtfResponse
 pattern RespNtf corrId queueId command <- (_, _, (corrId, queueId, Right command))
 
-sendRecvNtf :: forall c e. (Transport c, NtfEntityI e) => THandle c -> (Maybe C.ASignature, ByteString, ByteString, NtfCommand e) -> IO (SignedTransmission NtfResponse)
+sendRecvNtf :: forall c e. (Transport c, NtfEntityI e) => THandle c -> (Maybe C.ASignature, ByteString, ByteString, NtfCommand e) -> IO (SignedTransmission ErrorType NtfResponse)
 sendRecvNtf h@THandle {thVersion, sessionId} (sgn, corrId, qId, cmd) = do
   let t = encodeTransmission thVersion sessionId (CorrId corrId, qId, cmd)
   Right () <- tPut1 h (sgn, t)
   tGet1 h
 
-signSendRecvNtf :: forall c e. (Transport c, NtfEntityI e) => THandle c -> C.APrivateSignKey -> (ByteString, ByteString, NtfCommand e) -> IO (SignedTransmission NtfResponse)
+signSendRecvNtf :: forall c e. (Transport c, NtfEntityI e) => THandle c -> C.APrivateSignKey -> (ByteString, ByteString, NtfCommand e) -> IO (SignedTransmission ErrorType NtfResponse)
 signSendRecvNtf h@THandle {thVersion, sessionId} pk (corrId, qId, cmd) = do
   let t = encodeTransmission thVersion sessionId (CorrId corrId, qId, cmd)
   Right () <- tPut1 h (Just $ C.sign pk t, t)

tests/ServerTests.hs

@@ -63,7 +63,7 @@ serverTests t@(ATransport t') = do
     testMsgExpireOnInterval t'
     testMsgNOTExpireOnInterval t'
 
-pattern Resp :: CorrId -> QueueId -> BrokerMsg -> SignedTransmission BrokerMsg
+pattern Resp :: CorrId -> QueueId -> BrokerMsg -> SignedTransmission ErrorType BrokerMsg
 pattern Resp corrId queueId command <- (_, _, (corrId, queueId, Right command))
 
 pattern Ids :: RecipientId -> SenderId -> RcvPublicDhKey -> BrokerMsg
@@ -72,13 +72,13 @@ pattern Ids rId sId srvDh <- IDS (QIK rId sId srvDh)
 pattern Msg :: MsgId -> MsgBody -> BrokerMsg
 pattern Msg msgId body <- MSG RcvMessage {msgId, msgBody = EncRcvMsgBody body}
 
-sendRecv :: forall c p. (Transport c, PartyI p) => THandle c -> (Maybe C.ASignature, ByteString, ByteString, Command p) -> IO (SignedTransmission BrokerMsg)
+sendRecv :: forall c p. (Transport c, PartyI p) => THandle c -> (Maybe C.ASignature, ByteString, ByteString, Command p) -> IO (SignedTransmission ErrorType BrokerMsg)
 sendRecv h@THandle {thVersion, sessionId} (sgn, corrId, qId, cmd) = do
   let t = encodeTransmission thVersion sessionId (CorrId corrId, qId, cmd)
   Right () <- tPut1 h (sgn, t)
   tGet1 h
 
-signSendRecv :: forall c p. (Transport c, PartyI p) => THandle c -> C.APrivateSignKey -> (ByteString, ByteString, Command p) -> IO (SignedTransmission BrokerMsg)
+signSendRecv :: forall c p. (Transport c, PartyI p) => THandle c -> C.APrivateSignKey -> (ByteString, ByteString, Command p) -> IO (SignedTransmission ErrorType BrokerMsg)
 signSendRecv h@THandle {thVersion, sessionId} pk (corrId, qId, cmd) = do
   let t = encodeTransmission thVersion sessionId (CorrId corrId, qId, cmd)
   Right () <- tPut1 h (Just $ C.sign pk t, t)
@@ -89,9 +89,9 @@ tPut1 h t = do
   [r] <- tPut h [t]
   pure r
 
-tGet1 :: (ProtocolEncoding cmd, Transport c, MonadIO m, MonadFail m) => THandle c -> m (SignedTransmission cmd)
+tGet1 :: (ProtocolEncoding err cmd, Transport c, MonadIO m, MonadFail m) => THandle c -> m (SignedTransmission err cmd)
 tGet1 h = do
-  [r] <- tGet h
+  [r] <- liftIO $ tGet h
   pure r
 
 (#==) :: (HasCallStack, Eq a, Show a) => (a, a) -> String -> Assertion
@@ -428,7 +428,7 @@ testSwitchSub (ATransport t) =
       Resp "bcda" _ ok3 <- signSendRecv rh2 rKey ("bcda", rId, ACK mId3)
       (ok3, OK) #== "accepts ACK from the 2nd TCP connection"
 
-      1000 `timeout` tGet @BrokerMsg rh1 >>= \case
+      1000 `timeout` tGet @ErrorType @BrokerMsg rh1 >>= \case
         Nothing -> return ()
         Just _ -> error "nothing else is delivered to the 1st TCP connection"
 
@@ -869,14 +869,14 @@ testMessageNotifications (ATransport t) =
       Resp "5a" _ OK <- signSendRecv rh rKey ("5a", rId, ACK mId2)
       (dec mId2 msg2, Right "hello again") #== "delivered from queue again"
       Resp "" _ (NMSG _ _) <- tGet1 nh2
-      1000 `timeout` tGet @BrokerMsg nh1 >>= \case
+      1000 `timeout` tGet @ErrorType @BrokerMsg nh1 >>= \case
         Nothing -> pure ()
         Just _ -> error "nothing else should be delivered to the 1st notifier's TCP connection"
       Resp "6" _ OK <- signSendRecv rh rKey ("6", rId, NDEL)
       Resp "7" _ OK <- signSendRecv sh sKey ("7", sId, _SEND' "hello there")
       Resp "" _ (Msg mId3 msg3) <- tGet1 rh
       (dec mId3 msg3, Right "hello there") #== "delivered from queue again"
-      1000 `timeout` tGet @BrokerMsg nh2 >>= \case
+      1000 `timeout` tGet @ErrorType @BrokerMsg nh2 >>= \case
         Nothing -> pure ()
         Just _ -> error "nothing else should be delivered to the 2nd notifier's TCP connection"
 
@@ -895,7 +895,7 @@ testMsgExpireOnSend t =
         testSMPClient @c $ \rh -> do
           Resp "3" _ (Msg mId msg) <- signSendRecv rh rKey ("3", rId, SUB)
           (dec mId msg, Right "hello (should NOT expire)") #== "delivered"
-          1000 `timeout` tGet @BrokerMsg rh >>= \case
+          1000 `timeout` tGet @ErrorType @BrokerMsg rh >>= \case
             Nothing -> return ()
             Just _ -> error "nothing else should be delivered"
 
@@ -911,7 +911,7 @@ testMsgExpireOnInterval t =
         threadDelay 2500000
         testSMPClient @c $ \rh -> do
           Resp "2" _ OK <- signSendRecv rh rKey ("2", rId, SUB)
-          1000 `timeout` tGet @BrokerMsg rh >>= \case
+          1000 `timeout` tGet @ErrorType @BrokerMsg rh >>= \case
             Nothing -> return ()
             Just _ -> error "nothing should be delivered"
 
@@ -929,7 +929,7 @@ testMsgNOTExpireOnInterval t =
         testSMPClient @c $ \rh -> do
           Resp "2" _ (Msg mId msg) <- signSendRecv rh rKey ("2", rId, SUB)
           (dec mId msg, Right "hello (should NOT expire)") #== "delivered"
-          1000 `timeout` tGet @BrokerMsg rh >>= \case
+          1000 `timeout` tGet @ErrorType @BrokerMsg rh >>= \case
             Nothing -> return ()
             Just _ -> error "nothing else should be delivered"