Merge remote-tracking branch 'origin/master' into ab/async-subs

package.yaml

@@ -87,6 +87,9 @@ flags:
     manual: True
     default: True
 
+# cpp-options:
+#   - -Dslow_servers
+
 when:
   - condition: flag(swift)
     cpp-options:

simplexmq.cabal

@@ -172,6 +172,7 @@ library
       Simplex.Messaging.Server.MsgStore
       Simplex.Messaging.Server.MsgStore.STM
       Simplex.Messaging.Server.QueueStore
+      Simplex.Messaging.Server.QueueStore.QueueInfo
       Simplex.Messaging.Server.QueueStore.STM
       Simplex.Messaging.Server.Stats
       Simplex.Messaging.Server.StoreLog

src/Simplex/FileTransfer/Client.hs

@@ -52,7 +52,7 @@ import Simplex.Messaging.Protocol
     RecipientId,
     SenderId,
   )
-import Simplex.Messaging.Transport (ALPN, THandleAuth (..), THandleParams (..), TransportError (..), TransportPeer (..), supportedParameters)
+import Simplex.Messaging.Transport (ALPN, HandshakeError (..), THandleAuth (..), THandleParams (..), TransportError (..), TransportPeer (..), supportedParameters)
 import Simplex.Messaging.Transport.Client (TransportClientConfig, TransportHost, alpn)
 import Simplex.Messaging.Transport.HTTP2
 import Simplex.Messaging.Transport.HTTP2.Client
@@ -98,6 +98,12 @@ defaultXFTPClientConfig =
       clientALPN = Just supportedXFTPhandshakes
     }
 
+http2XFTPClientError :: HTTP2ClientError -> XFTPClientError
+http2XFTPClientError = \case
+  HCResponseTimeout -> PCEResponseTimeout
+  HCNetworkError -> PCENetworkError
+  HCIOError e -> PCEIOError e
+
 getXFTPClient :: TransportSession FileResponse -> XFTPClientConfig -> (XFTPClient -> IO ()) -> IO (Either XFTPClientError XFTPClient)
 getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {clientALPN, xftpNetworkConfig, serverVRange} disconnected = runExceptT $ do
   let username = proxyUsername transportSession
@@ -116,8 +122,7 @@ getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {clientALPN,
   logDebug $ "Client negotiated handshake protocol: " <> tshow sessionALPN
   thParams@THandleParams {thVersion} <- case sessionALPN of
     Just "xftp/1" -> xftpClientHandshakeV1 serverVRange keyHash http2Client thParams0
-    Nothing -> pure thParams0
-    _ -> throwError $ PCETransportError TEVersion
+    _ -> pure thParams0
   logDebug $ "Client negotiated protocol: " <> tshow thVersion
   let c = XFTPClient {http2Client, thParams, transportSession, config}
   atomically $ writeTVar clientVar $ Just c
@@ -135,15 +140,15 @@ xftpClientHandshakeV1 serverVRange keyHash@(C.KeyHash kh) c@HTTP2Client {session
     getServerHandshake = do
       let helloReq = H.requestNoBody "POST" "/" []
       HTTP2Response {respBody = HTTP2Body {bodyHead = shsBody}} <-
-        liftError' (const $ PCEResponseError HANDSHAKE) $ sendRequest c helloReq Nothing
-      liftHS . smpDecode =<< liftHS (C.unPad shsBody)
+        liftError' http2XFTPClientError $ sendRequest c helloReq Nothing
+      liftTransportErr (TEHandshake PARSE) . smpDecode =<< liftTransportErr TEBadBlock (C.unPad shsBody)
     processServerHandshake :: XFTPServerHandshake -> ExceptT XFTPClientError IO (VersionRangeXFTP, C.PublicKeyX25519)
     processServerHandshake XFTPServerHandshake {xftpVersionRange, sessionId = serverSessId, authPubKey = serverAuth} = do
-      unless (sessionId == serverSessId) $ throwError $ PCEResponseError SESSION
+      unless (sessionId == serverSessId) $ throwError $ PCETransportError TEBadSession
       case xftpVersionRange `compatibleVRange` serverVRange of
         Nothing -> throwError $ PCETransportError TEVersion
         Just (Compatible vr) ->
-          fmap (vr,) . liftHS $ do
+          fmap (vr,) . liftTransportErr (TEHandshake BAD_AUTH) $ do
             let (X.CertificateChain cert, exact) = serverAuth
             case cert of
               [_leaf, ca] | XV.Fingerprint kh == XV.getFingerprint ca X.HashSHA256 -> pure ()
@@ -152,11 +157,11 @@ xftpClientHandshakeV1 serverVRange keyHash@(C.KeyHash kh) c@HTTP2Client {session
             C.x509ToPublic (pubKey, []) >>= C.pubKey
     sendClientHandshake :: XFTPClientHandshake -> ExceptT XFTPClientError IO ()
     sendClientHandshake chs = do
-      chs' <- liftHS $ C.pad (smpEncode chs) xftpBlockSize
+      chs' <- liftTransportErr TELargeMsg $ C.pad (smpEncode chs) xftpBlockSize
       let chsReq = H.requestBuilder "POST" "/" [] $ byteString chs'
-      HTTP2Response {respBody = HTTP2Body {bodyHead}} <- liftError' (const $ PCEResponseError HANDSHAKE) $ sendRequest c chsReq Nothing
-      unless (B.null bodyHead) $ throwError $ PCEResponseError HANDSHAKE
-    liftHS = liftEitherWith (const $ PCEResponseError HANDSHAKE)
+      HTTP2Response {respBody = HTTP2Body {bodyHead}} <- liftError' http2XFTPClientError $ sendRequest c chsReq Nothing
+      unless (B.null bodyHead) $ throwError $ PCETransportError TEBadBlock
+    liftTransportErr e = liftEitherWith (const $ PCETransportError e)
 
 closeXFTPClient :: XFTPClient -> IO ()
 closeXFTPClient XFTPClient {http2Client} = closeHTTP2Client http2Client

src/Simplex/FileTransfer/Server.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
@@ -67,6 +68,9 @@ import Simplex.Messaging.Version
 import System.Exit (exitFailure)
 import System.FilePath ((</>))
 import System.IO (hPrint, hPutStrLn, universalNewlineMode)
+#ifdef slow_servers
+import System.Random (getStdRandom, randomR)
+#endif
 import UnliftIO
 import UnliftIO.Concurrent (threadDelay)
 import UnliftIO.Directory (doesFileExist, removeFile, renameFile)
@@ -138,6 +142,9 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
           let authPubKey = (chain, C.signX509 serverSignKey $ C.publicToX509 k)
           let hs = XFTPServerHandshake {xftpVersionRange = xftpServerVRange, sessionId, authPubKey}
           shs <- encodeXftp hs
+#ifdef slow_servers
+          lift randomDelay
+#endif
           liftIO . sendResponse $ H.responseBuilder N.ok200 [] shs
           pure Nothing
         processClientHandshake pk = do
@@ -151,6 +158,9 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
               let auth = THAuthServer {serverPrivKey = pk, sessSecret' = Nothing}
                   thParams = thParams0 {thAuth = Just auth, thVersion = v, thServerVRange = vr}
               atomically $ TM.insert sessionId (HandshakeAccepted thParams) sessions
+#ifdef slow_servers
+              lift randomDelay
+#endif
               liftIO . sendResponse $ H.responseNoBody N.ok200 []
               pure Nothing
             Nothing -> throwError HANDSHAKE
@@ -315,6 +325,9 @@ processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHea
   where
     sendXFTPResponse (corrId, fId, resp) serverFile_ = do
       let t_ = xftpEncodeTransmission thParams (corrId, fId, resp)
+#ifdef slow_servers
+      randomDelay
+#endif
       liftIO $ sendResponse $ H.responseStreaming N.ok200 [] $ streamBody t_
       where
         streamBody t_ send done = do
@@ -329,6 +342,15 @@ processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHea
                 withFile filePath ReadMode $ \h -> sendEncFile h send sbState fileSize
           done
 
+#ifdef slow_servers
+randomDelay :: M ()
+randomDelay = do
+  d <- asks $ responseDelay . config
+  when (d > 0) $ do
+    pc <- getStdRandom (randomR (-200, 200))
+    threadDelay $ (d * (1000 + pc)) `div` 1000
+#endif
+
 data VerificationResult = VRVerified XFTPRequest | VRFailed
 
 verifyXFTPTransmission :: Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> XFTPFileId -> FileCmd -> M VerificationResult

src/Simplex/FileTransfer/Server/Env.hs

@@ -69,7 +69,8 @@ data XFTPServerConfig = XFTPServerConfig
     logStatsStartTime :: Int64,
     serverStatsLogFile :: FilePath,
     serverStatsBackupFile :: Maybe FilePath,
-    transportConfig :: TransportServerConfig
+    transportConfig :: TransportServerConfig,
+    responseDelay :: Int
   }
 
 defaultInactiveClientExpiration :: ExpirationConfig

src/Simplex/FileTransfer/Server/Main.hs

@@ -185,7 +185,8 @@ xftpServerCLI cfgPath logPath = do
                 defaultTransportServerConfig
                   { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini,
                     alpn = Just supportedXFTPhandshakes
-                  }
+                  },
+              responseDelay = 0
             }
 
 data CliCommand

src/Simplex/Messaging/Agent.hs

@@ -70,6 +70,7 @@ module Simplex.Messaging.Agent
     sendMessages,
     sendMessagesB,
     ackMessage,
+    getConnectionQueueInfo,
     switchConnection,
     abortConnectionSwitch,
     synchronizeRatchet,
@@ -82,8 +83,6 @@ module Simplex.Messaging.Agent
     testProtocolServer,
     setNtfServers,
     setNetworkConfig,
-    getNetworkConfig,
-    getNetworkConfig',
     setUserNetworkInfo,
     reconnectAllServers,
     registerNtfToken,
@@ -176,6 +175,7 @@ import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Parsers (parse)
 import Simplex.Messaging.Protocol (BrokerMsg, Cmd (..), EntityId, ErrorType (AUTH), MsgBody, MsgFlags (..), NtfServer, ProtoServerWithAuth, ProtocolTypeI (..), SMPMsgMeta, SParty (..), SProtocolType (..), SndPublicAuthKey, SubscriptionMode (..), UserProtocol, VersionSMPC, XFTPServerWithAuth)
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (SMPVersion, THandleParams (sessionId))
@@ -371,6 +371,10 @@ ackMessage :: AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AE
 ackMessage c = withAgentEnv c .:. ackMessage' c
 {-# INLINE ackMessage #-}
 
+getConnectionQueueInfo :: AgentClient -> ConnId -> AE QueueInfo
+getConnectionQueueInfo c = withAgentEnv c . getConnectionQueueInfo' c
+{-# INLINE getConnectionQueueInfo #-}
+
 -- | Switch connection to the new receive queue
 switchConnection :: AgentClient -> ConnId -> AE ConnectionStats
 switchConnection c = withAgentEnv c . switchConnection' c
@@ -428,11 +432,6 @@ setNetworkConfig c@AgentClient {useNetworkConfig} cfg' = do
       else True <$ (writeTVar useNetworkConfig $! (slowNetworkConfig cfg', cfg'))
   when changed $ reconnectAllServers c
 
--- returns fast network config
-getNetworkConfig :: AgentClient -> IO NetworkConfig
-getNetworkConfig = getNetworkConfig'
-{-# INLINE getNetworkConfig #-}
-
 setUserNetworkInfo :: AgentClient -> UserNetworkInfo -> IO ()
 setUserNetworkInfo c@AgentClient {userNetworkInfo, userNetworkUpdated} ni = withAgentEnv' c $ do
   ts' <- liftIO getCurrentTime
@@ -1510,6 +1509,16 @@ ackMessage' c connId msgId rcptInfo_ = withConnLock c connId "ackMessage" $ do
             withStore' c $ \db -> deleteDeliveredSndMsg db connId $ InternalId sndMsgId
           _ -> pure ()
 
+getConnectionQueueInfo' :: AgentClient -> ConnId -> AM QueueInfo
+getConnectionQueueInfo' c connId = do
+  SomeConn _ conn <- withStore c (`getConn` connId)
+  case conn of
+    DuplexConnection _ (rq :| _) _ -> getQueueInfo c rq
+    RcvConnection _ rq -> getQueueInfo c rq
+    ContactConnection _ rq -> getQueueInfo c rq
+    SndConnection {} -> throwE $ CONN SIMPLEX
+    NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionQueueInfo': NewConnection"
+
 switchConnection' :: AgentClient -> ConnId -> AM ConnectionStats
 switchConnection' c connId =
   withConnLock c connId "switchConnection" $

src/Simplex/Messaging/Agent/Client.hs

@@ -56,6 +56,7 @@ module Simplex.Messaging.Agent.Client
     disableQueueNotifications,
     disableQueuesNtfs,
     sendAgentMessage,
+    getQueueInfo,
     agentNtfRegisterToken,
     agentNtfVerifyToken,
     agentNtfCheckToken,
@@ -238,6 +239,7 @@ import Simplex.Messaging.Protocol
     sameSrvAddr',
   )
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Session
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
@@ -1573,6 +1575,11 @@ sendAgentMessage c sq@SndQueue {userId, server, sndId, sndPrivateKey} msgFlags a
   msg <- agentCbEncrypt sq Nothing $ smpEncode clientMsg
   sendOrProxySMPMessage c userId server "<MSG>" (Just sndPrivateKey) sndId msgFlags msg
 
+getQueueInfo :: AgentClient -> RcvQueue -> AM QueueInfo
+getQueueInfo c rq@RcvQueue {rcvId, rcvPrivateKey} =
+  withSMPClient c rq "QUE" $ \smp ->
+    getSMPQueueInfo smp rcvPrivateKey rcvId
+
 agentNtfRegisterToken :: AgentClient -> NtfToken -> NtfPublicAuthKey -> C.PublicKeyX25519 -> AM (NtfTokenId, C.PublicKeyX25519)
 agentNtfRegisterToken c NtfToken {deviceToken, ntfServer, ntfPrivKey} ntfPubKey pubDhKey =
   withClient c (0, ntfServer, Nothing) "TNEW" $ \ntf -> ntfRegisterToken ntf ntfPrivKey (NewNtfTkn deviceToken ntfPubKey pubDhKey)

src/Simplex/Messaging/Client.hs

@@ -59,6 +59,7 @@ module Simplex.Messaging.Client
     connectSMPProxiedRelay,
     proxySMPMessage,
     forwardSMPMessage,
+    getSMPQueueInfo,
     sendProtocolCommand,
 
     -- * Supporting types and client configuration
@@ -128,6 +129,7 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, enumJSON)
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
@@ -289,6 +291,32 @@ data SMPProxyFallback
   | SPFProhibit -- prohibit direct connection to destination relay.
   deriving (Eq, Show)
 
+instance StrEncoding SMPProxyMode where
+  strEncode = \case
+    SPMAlways -> "always"
+    SPMUnknown -> "unknown"
+    SPMUnprotected -> "unprotected"
+    SPMNever -> "never"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "always" -> pure SPMAlways
+      "unknown" -> pure SPMUnknown
+      "unprotected" -> pure SPMUnprotected
+      "never" -> pure SPMNever
+      _ -> fail "Invalid SMP proxy mode"
+
+instance StrEncoding SMPProxyFallback where
+  strEncode = \case
+    SPFAllow -> "yes"
+    SPFAllowProtected -> "protected"
+    SPFProhibit -> "no"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "yes" -> pure SPFAllow
+      "protected" -> pure SPFAllowProtected
+      "no" -> pure SPFProhibit
+      _ -> fail "Invalid SMP proxy fallback mode"
+
 defaultNetworkConfig :: NetworkConfig
 defaultNetworkConfig =
   NetworkConfig
@@ -920,6 +948,12 @@ forwardSMPMessage c@ProtocolClient {thParams, client_ = PClient {clientCorrId =
       pure fwdResponse
     r -> throwE $ unexpectedResponse r
 
+getSMPQueueInfo :: SMPClient -> C.APrivateAuthKey -> QueueId -> ExceptT SMPClientError IO QueueInfo
+getSMPQueueInfo c pKey qId =
+  sendSMPCommand c (Just pKey) qId QUE >>= \case
+    INFO info -> pure info
+    r -> throwE $ unexpectedResponse r
+
 okSMPCommand :: PartyI p => Command p -> SMPClient -> C.APrivateAuthKey -> QueueId -> ExceptT SMPClientError IO ()
 okSMPCommand cmd c pKey qId =
   sendSMPCommand c (Just pKey) qId cmd >>= \case

src/Simplex/Messaging/Protocol.hs

@@ -156,6 +156,7 @@ module Simplex.Messaging.Protocol
     sameSrvAddr,
     sameSrvAddr',
     noAuthSrv,
+    toMsgInfo,
 
     -- * TCP transport functions
     TransportBatch (..),
@@ -197,8 +198,8 @@ import qualified Data.List.NonEmpty as L
 import Data.Maybe (isJust, isNothing)
 import Data.String
 import qualified Data.Text as T
-import Data.Text.Encoding (encodeUtf8)
-import Data.Time.Clock.System (SystemTime (..))
+import Data.Text.Encoding (decodeLatin1, encodeUtf8)
+import Data.Time.Clock.System (SystemTime (..), systemToUTCTime)
 import Data.Type.Equality
 import Data.Word (Word16)
 import qualified Data.X509 as X
@@ -210,6 +211,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
+import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.ServiceScheme
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (TransportHost, TransportHosts (..))
@@ -386,6 +388,7 @@ data Command (p :: Party) where
   ACK :: MsgId -> Command Recipient
   OFF :: Command Recipient
   DEL :: Command Recipient
+  QUE :: Command Recipient
   -- SMP sender commands
   -- SEND v1 has to be supported for encoding/decoding
   -- SEND :: MsgBody -> Command Sender
@@ -463,6 +466,7 @@ data BrokerMsg where
   RRES :: EncFwdResponse -> BrokerMsg -- relay to proxy
   PRES :: EncResponse -> BrokerMsg -- proxy to client
   END :: BrokerMsg
+  INFO :: QueueInfo -> BrokerMsg
   OK :: BrokerMsg
   ERR :: ErrorType -> BrokerMsg
   PONG :: BrokerMsg
@@ -505,6 +509,13 @@ data Message
         msgTs :: SystemTime
       }
 
+toMsgInfo :: Message -> MsgInfo
+toMsgInfo = \case
+  Message {msgId, msgTs} -> msgInfo msgId msgTs MTMessage
+  MessageQuota {msgId, msgTs} -> msgInfo msgId msgTs MTQuota
+  where
+    msgInfo msgId msgTs msgType = MsgInfo {msgId = decodeLatin1 $ B64.encode msgId, msgTs = systemToUTCTime msgTs, msgType}
+
 messageId :: Message -> MsgId
 messageId = \case
   Message {msgId} -> msgId
@@ -652,6 +663,7 @@ data CommandTag (p :: Party) where
   ACK_ :: CommandTag Recipient
   OFF_ :: CommandTag Recipient
   DEL_ :: CommandTag Recipient
+  QUE_ :: CommandTag Recipient
   SEND_ :: CommandTag Sender
   PING_ :: CommandTag Sender
   PRXY_ :: CommandTag ProxiedClient
@@ -674,6 +686,7 @@ data BrokerMsgTag
   | RRES_
   | PRES_
   | END_
+  | INFO_
   | OK_
   | ERR_
   | PONG_
@@ -698,6 +711,7 @@ instance PartyI p => Encoding (CommandTag p) where
     ACK_ -> "ACK"
     OFF_ -> "OFF"
     DEL_ -> "DEL"
+    QUE_ -> "QUE"
     SEND_ -> "SEND"
     PING_ -> "PING"
     PRXY_ -> "PRXY"
@@ -717,6 +731,7 @@ instance ProtocolMsgTag CmdTag where
     "ACK" -> Just $ CT SRecipient ACK_
     "OFF" -> Just $ CT SRecipient OFF_
     "DEL" -> Just $ CT SRecipient DEL_
+    "QUE" -> Just $ CT SRecipient QUE_
     "SEND" -> Just $ CT SSender SEND_
     "PING" -> Just $ CT SSender PING_
     "PRXY" -> Just $ CT SProxiedClient PRXY_
@@ -742,6 +757,7 @@ instance Encoding BrokerMsgTag where
     RRES_ -> "RRES"
     PRES_ -> "PRES"
     END_ -> "END"
+    INFO_ -> "INFO"
     OK_ -> "OK"
     ERR_ -> "ERR"
     PONG_ -> "PONG"
@@ -757,6 +773,7 @@ instance ProtocolMsgTag BrokerMsgTag where
     "RRES" -> Just RRES_
     "PRES" -> Just PRES_
     "END" -> Just END_
+    "INFO" -> Just INFO_
     "OK" -> Just OK_
     "ERR" -> Just ERR_
     "PONG" -> Just PONG_
@@ -1275,6 +1292,7 @@ instance PartyI p => ProtocolEncoding SMPVersion ErrorType (Command p) where
     ACK msgId -> e (ACK_, ' ', msgId)
     OFF -> e OFF_
     DEL -> e DEL_
+    QUE -> e QUE_
     SEND flags msg -> e (SEND_, ' ', flags, ' ', Tail msg)
     PING -> e PING_
     NSUB -> e NSUB_
@@ -1340,6 +1358,7 @@ instance ProtocolEncoding SMPVersion ErrorType Cmd where
         ACK_ -> ACK <$> _smpP
         OFF_ -> pure OFF
         DEL_ -> pure DEL
+        QUE_ -> pure QUE
     CT SSender tag ->
       Cmd SSender <$> case tag of
         SEND_ -> SEND <$> _smpP <*> (unTail <$> _smpP)
@@ -1368,6 +1387,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
     RRES (EncFwdResponse encBlock) -> e (RRES_, ' ', Tail encBlock)
     PRES (EncResponse encBlock) -> e (PRES_, ' ', Tail encBlock)
     END -> e END_
+    INFO info -> e (INFO_, ' ', info)
     OK -> e OK_
     ERR err -> e (ERR_, ' ', err)
     PONG -> e PONG_
@@ -1388,6 +1408,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
     RRES_ -> RRES <$> (EncFwdResponse . unTail <$> _smpP)
     PRES_ -> PRES <$> (EncResponse . unTail <$> _smpP)
     END_ -> pure END
+    INFO_ -> INFO <$> _smpP
     OK_ -> pure OK
     ERR_ -> ERR <$> _smpP
     PONG_ -> pure PONG

src/Simplex/Messaging/Server.hs

@@ -59,7 +59,7 @@ import Data.List (intercalate, mapAccumR)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
-import Data.Maybe (catMaybes, fromMaybe, isNothing)
+import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
@@ -82,6 +82,7 @@ import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.QueueStore
+import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Server.QueueStore.STM as QS
 import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.Server.StoreLog
@@ -713,6 +714,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                 let own = isOwnServer a srv
                 inc own pRequests
                 inc own $ if temporaryClientError e then pErrorsConnect else pErrorsOther
+                logError $ "Error connecting: " <> decodeLatin1 (strEncode $ host srv) <> " " <> tshow e
                 pure . ERR $ smpProxyError e
             where
               proxyResp smp =
@@ -791,6 +793,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
           NDEL -> deleteQueueNotifier_ st
           OFF -> suspendQueue_ st
           DEL -> delQueueAndMsgs st
+          QUE -> withQueue getQueueInfo
       where
         createQueue :: QueueStore -> RcvPublicAuthKey -> RcvPublicDhKey -> SubscriptionMode -> M (Transmission BrokerMsg)
         createQueue st recipientKey dhKey subMode = time "NEW" $ do
@@ -1162,6 +1165,26 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
             Right q -> updateDeletedStats q $> ok
             Left e -> pure $ err e
 
+        getQueueInfo :: QueueRec -> M (Transmission BrokerMsg)
+        getQueueInfo QueueRec {senderKey, notifier} = do
+          q@MsgQueue {size} <- getStoreMsgQueue "getQueueInfo" queueId
+          info <- atomically $ do
+            qiSub <- TM.lookup queueId subscriptions >>= mapM mkQSub
+            qiSize <- readTVar size
+            qiMsg <- toMsgInfo <$$> tryPeekMsg q
+            pure QueueInfo {qiSnd = isJust senderKey, qiNtf = isJust notifier, qiSub, qiSize, qiMsg}
+          pure (corrId, queueId, INFO info)
+          where
+            mkQSub sub = do
+              Sub {subThread, delivered} <- readTVar sub
+              let qSubThread = case subThread of
+                    NoSub -> QNoSub
+                    SubPending -> QSubPending
+                    SubThread _ -> QSubThread
+                    ProhibitSub -> QProhibitSub
+              qDelivered <- decodeLatin1 . encode <$$> tryReadTMVar delivered
+              pure QSub {qSubThread, qDelivered}
+          
         ok :: Transmission BrokerMsg
         ok = (corrId, queueId, OK)
 

src/Simplex/Messaging/Server/QueueStore/QueueInfo.hs

@@ -0,0 +1,55 @@
+{-# LANGUAGE TemplateHaskell #-}
+
+module Simplex.Messaging.Server.QueueStore.QueueInfo where
+
+import qualified Data.Aeson as J
+import qualified Data.Aeson.TH as JQ
+import qualified Data.Attoparsec.ByteString.Char8 as A
+import qualified Data.ByteString.Lazy.Char8 as LB
+import Data.Text (Text)
+import Data.Time.Clock (UTCTime)
+import Simplex.Messaging.Encoding
+import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, enumJSON)
+import Simplex.Messaging.Util ((<$?>))
+
+data QueueInfo = QueueInfo
+  { qiSnd :: Bool,
+    qiNtf :: Bool,
+    qiSub :: Maybe QSub,
+    qiSize :: Int,
+    qiMsg :: Maybe MsgInfo
+  }
+  deriving (Eq, Show)
+
+data QSub = QSub
+  { qSubThread :: QSubThread,
+    qDelivered :: Maybe Text
+  }
+  deriving (Eq, Show)
+
+data QSubThread = QNoSub | QSubPending | QSubThread | QProhibitSub
+  deriving (Eq, Show)
+
+data MsgInfo = MsgInfo
+  { msgId :: Text,
+    msgTs :: UTCTime,
+    msgType :: MsgType
+  }
+  deriving (Eq, Show)
+
+data MsgType = MTMessage | MTQuota
+  deriving (Eq, Show)
+
+$(JQ.deriveJSON (enumJSON $ dropPrefix "Q") ''QSubThread)
+
+$(JQ.deriveJSON defaultJSON ''QSub)
+
+$(JQ.deriveJSON (enumJSON $ dropPrefix "MT") ''MsgType)
+
+$(JQ.deriveJSON defaultJSON ''MsgInfo)
+
+$(JQ.deriveJSON defaultJSON ''QueueInfo)
+
+instance Encoding QueueInfo where
+  smpEncode = LB.toStrict . J.encode
+  smpP = J.eitherDecodeStrict <$?> A.takeByteString

src/Simplex/RemoteControl/Invitation.hs

@@ -123,7 +123,7 @@ instance StrEncoding RCSignedInvitation where
     idsig <- requiredP sigs "idsig" $ parseAll strP
     pure RCSignedInvitation {invitation, ssig, idsig}
 
-signInvitation :: C.PrivateKey C.Ed25519 -> C.PrivateKey C.Ed25519 -> RCInvitation -> RCSignedInvitation
+signInvitation :: C.PrivateKey 'C.Ed25519 -> C.PrivateKey 'C.Ed25519 -> RCInvitation -> RCSignedInvitation
 signInvitation sKey idKey invitation = RCSignedInvitation {invitation, ssig, idsig}
   where
     uri = strEncode invitation

tests/AgentTests/FunctionalAPITests.hs

@@ -59,6 +59,7 @@ import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
 import Data.Bifunctor (first)
+import qualified Data.ByteString.Base64 as B64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight)
@@ -68,6 +69,7 @@ import Data.List.NonEmpty (NonEmpty)
 import qualified Data.Map as M
 import Data.Maybe (isJust, isNothing)
 import qualified Data.Set as S
+import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Type.Equality (testEquality, (:~:) (Refl))
@@ -94,6 +96,7 @@ import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolS
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
+import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion, supportedSMPHandshakes)
 import Simplex.Messaging.Util (diffToMicroseconds)
 import Simplex.Messaging.Version (VersionRange (..))
@@ -109,8 +112,11 @@ type AEntityTransmission e = (ACorrId, ConnId, ACommand 'Agent e)
 
 -- deriving instance Eq (ValidFileDescription p)
 
+shouldRespond :: (HasCallStack, MonadUnliftIO m, Eq a, Show a) => m a -> a -> m ()
+a `shouldRespond` r = withFrozenCallStack $ withTimeout a (`shouldBe` r)
+
 (##>) :: (HasCallStack, MonadUnliftIO m) => m (AEntityTransmission e) -> AEntityTransmission e -> m ()
-a ##> t = withTimeout a (`shouldBe` t)
+a ##> t = a `shouldRespond` t
 
 (=##>) :: (Show a, HasCallStack, MonadUnliftIO m) => m a -> (HasCallStack => a -> Bool) -> m ()
 a =##> p =
@@ -241,7 +247,7 @@ mkVersionRange :: Word16 -> Word16 -> VersionRange v
 mkVersionRange v1 v2 = V.mkVersionRange (Version v1) (Version v2)
 
 runRight_ :: (Eq e, Show e, HasCallStack) => ExceptT e IO () -> Expectation
-runRight_ action = runExceptT action `shouldReturn` Right ()
+runRight_ action = withFrozenCallStack $ runExceptT action `shouldReturn` Right ()
 
 runRight :: (Show e, HasCallStack) => ExceptT e IO a -> IO a
 runRight action =
@@ -457,6 +463,9 @@ functionalAPITests t = do
     it "should wait for user network" testWaitForUserNetwork
     it "should not reset online to offline if happens too quickly" testDoNotResetOnlineToOffline
     it "should resume multiple threads" testResumeMultipleThreads
+  describe "SMP queue info" $ do
+    it "server should respond with queue and subscription information" $
+      withSmpServer t testServerQueueInfo
 
 testBasicAuth :: ATransport -> Bool -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> IO Int
 testBasicAuth t allowNewQueues srv@(srvAuth, srvVersion) clnt1 clnt2 = do
@@ -2734,6 +2743,91 @@ testResumeMultipleThreads = do
   where
     aCfg = agentCfg {userOfflineDelay = 0}
 
+testServerQueueInfo :: IO ()
+testServerQueueInfo = do
+  withAgentClients2 $ \alice bob -> runRight_ $ do
+    (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
+    liftIO $ threadDelay 200000
+    checkEmptyQ alice bobId False
+    aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
+    ("", _, CONF confId _ "bob's connInfo") <- get alice
+    liftIO $ threadDelay 200000
+    checkEmptyQ alice bobId False
+    allowConnection alice bobId confId "alice's connInfo"
+    get alice ##> ("", bobId, CON)
+    get bob ##> ("", aliceId, INFO "alice's connInfo")
+    get bob ##> ("", aliceId, CON)
+    liftIO $ threadDelay 200000
+    checkEmptyQ alice bobId True
+    checkEmptyQ bob aliceId True
+    let msgId = 4
+    (msgId', PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello"
+    liftIO $ msgId' `shouldBe` msgId
+    get alice ##> ("", bobId, SENT msgId)
+    liftIO $ threadDelay 200000
+    Just srvMsgId <- checkMsgQ bob aliceId 1
+    get bob =##> \case
+      ("", c, MSG MsgMeta {integrity = MsgOk, broker = (smId, _), recipient = (mId, _), pqEncryption = PQEncOn} _ "hello") ->
+        c == aliceId && decodeLatin1 (B64.encode smId) == srvMsgId && mId == msgId
+      _ -> False
+    ackMessage bob aliceId msgId Nothing
+    liftIO $ threadDelay 200000
+    checkEmptyQ bob aliceId True
+    (msgId1, PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello 1"
+    get alice ##> ("", bobId, SENT msgId1)
+    Just _ <- checkMsgQ bob aliceId 1
+    (msgId2, PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello 2"
+    get alice ##> ("", bobId, SENT msgId2)
+    (msgId3, PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello 3"
+    get alice ##> ("", bobId, SENT msgId3)
+    (msgId4, PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello 4"
+    get alice ##> ("", bobId, SENT msgId4)
+    Just _ <- checkMsgQ bob aliceId 4
+    (msgId5, PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello: quota exceeded"
+    liftIO $ threadDelay 200000
+    Just _ <- checkMsgQ bob aliceId 5
+    get bob =##> \case ("", c, Msg' mId PQEncOn "hello 1") -> c == aliceId && mId == msgId1; _ -> False
+    ackMessage bob aliceId msgId1 Nothing
+    liftIO $ threadDelay 200000
+    Just _ <- checkMsgQ bob aliceId 4
+    get bob =##> \case ("", c, Msg' mId PQEncOn "hello 2") -> c == aliceId && mId == msgId2; _ -> False
+    ackMessage bob aliceId msgId2 Nothing
+    get bob =##> \case ("", c, Msg' mId PQEncOn "hello 3") -> c == aliceId && mId == msgId3; _ -> False
+    ackMessage bob aliceId msgId3 Nothing
+    liftIO $ threadDelay 200000
+    Just _ <- checkMsgQ bob aliceId 2
+    get bob =##> \case ("", c, Msg' mId PQEncOn "hello 4") -> c == aliceId && mId == msgId4; _ -> False
+    ackMessage bob aliceId msgId4 Nothing
+    liftIO $ threadDelay 200000
+    Just _ <- checkMsgQ bob aliceId 1 -- the one that did not fit now accepted
+    get alice ##> ("", bobId, QCONT)
+    get alice ##> ("", bobId, SENT msgId5)
+    liftIO $ threadDelay 200000
+    Just _srvMsgId <- checkQ bob aliceId True (Just QNoSub) 1 (Just MTMessage)
+    get bob =##> \case ("", c, Msg' mId PQEncOn "hello: quota exceeded") -> c == aliceId && mId == msgId5 + 1; _ -> False
+    ackMessage bob aliceId (msgId5 + 1) Nothing
+    liftIO $ threadDelay 200000
+    checkEmptyQ bob aliceId True
+    pure ()
+  where
+    checkEmptyQ c cId qiSnd' = do
+      r <- checkQ c cId qiSnd' (Just QSubThread) 0 Nothing
+      liftIO $ r `shouldBe` Nothing
+    checkMsgQ c cId qiSize' = do
+      r <- checkQ c cId True (Just QNoSub) qiSize' (Just MTMessage)
+      liftIO $ isJust r `shouldBe` True
+      pure r
+    checkQ c cId qiSnd' qiSubThread_ qiSize' msgType_ = do
+      QueueInfo {qiSnd, qiNtf, qiSub, qiSize, qiMsg} <- getConnectionQueueInfo c cId
+      liftIO $ do
+        qiSnd `shouldBe` qiSnd'
+        qiNtf `shouldBe` False
+        qSubThread <$> qiSub `shouldBe` qiSubThread_
+        qiSize `shouldBe` qiSize'
+        msgId_ <- forM qiMsg $ \MsgInfo {msgId, msgType} -> msgId <$ (Just msgType `shouldBe` msgType_)
+        qDelivered <$> qiSub `shouldBe` Just msgId_
+        pure msgId_
+
 noNetworkDelay :: AgentClient -> IO ()
 noNetworkDelay a = do
   d <- waitNetwork a

tests/CLITests.hs

@@ -77,13 +77,13 @@ smpServerTest storeLog basicAuth = do
   let certPath = cfgPath </> "server.crt"
   oldCrt@X.Certificate {} <-
     XF.readSignedObject certPath >>= \case
-      [cert] -> pure . X.signedObject $ X.getSigned cert
+      [cert'] -> pure . X.signedObject $ X.getSigned cert'
       _ -> error "bad crt format"
   r' <- lines <$> capture_ (withArgs ["cert"] $ (100000 `timeout` smpServerCLI cfgPath logPath) `catchAll_` pure (Just ()))
   r' `shouldContain` ["Generated new server credentials"]
   newCrt <-
     XF.readSignedObject certPath >>= \case
-      [cert] -> pure . X.signedObject $ X.getSigned cert
+      [cert'] -> pure . X.signedObject $ X.getSigned cert'
       _ -> error "bad crt format after cert"
   X.certSignatureAlg oldCrt `shouldBe` X.certSignatureAlg newCrt
   X.certSubjectDN oldCrt `shouldBe` X.certSubjectDN newCrt

tests/XFTPAgent.hs

@@ -46,7 +46,11 @@ import XFTPClient
 
 xftpAgentTests :: Spec
 xftpAgentTests = around_ testBracket . describe "agent XFTP API" $ do
-  it "should send and receive file" testXFTPAgentSendReceive
+  it "should send and receive file" $ withXFTPServer testXFTPAgentSendReceive
+  -- uncomment CPP option slow_servers and run hpack to run this test
+  xit "should send and receive file with slow server responses" $
+    withXFTPServerCfg testXFTPServerConfig {responseDelay = 500000} $
+      \_ -> testXFTPAgentSendReceive
   it "should send and receive with encrypted local files" testXFTPAgentSendReceiveEncrypted
   it "should send and receive large file with a redirect" testXFTPAgentSendReceiveRedirect
   it "should send and receive small file without a redirect" testXFTPAgentSendReceiveNoRedirect
@@ -100,7 +104,7 @@ checkProgress (prev, expected) (progress, total) loop
   | otherwise = pure ()
 
 testXFTPAgentSendReceive :: HasCallStack => IO ()
-testXFTPAgentSendReceive = withXFTPServer $ do
+testXFTPAgentSendReceive = do
   filePath <- createRandomFile
   -- send file, delete snd file internally
   (rfd1, rfd2) <- withAgent 1 agentCfg initAgentServers testDB $ \sndr -> runRight $ do

tests/XFTPClient.hs

@@ -124,7 +124,8 @@ testXFTPServerConfig_ alpn =
       logStatsStartTime = 0,
       serverStatsLogFile = "tests/tmp/xftp-server-stats.daily.log",
       serverStatsBackupFile = Nothing,
-      transportConfig = defaultTransportServerConfig {alpn}
+      transportConfig = defaultTransportServerConfig {alpn},
+      responseDelay = 0
     }
 
 testXFTPClientConfig :: XFTPClientConfig