transport: add send-queueing timeout

CHANGELOG.md

@@ -1,28 +1,3 @@
-# 5.7.1
-
-SMP agent:
-- increase timeout for TLS connection via SOCKS
-
-# 5.7.0
-
-Version 5.7.0.4
-
-_Please note_: the earliest SimpleX Chat clients supported by this version of the servers is 5.5.3 (released on February 11, 2024).
-
-SMP server:
-- increase max SMP protocol version to 7 (support for deniable authenticators).
-
-NTF server:
-- increase max NTF protocol version to 2 (support for deniable authenticators).
-
-XFTP server:
-- version handshake using ALPN.
-
-SMP agent:
-- increase timeouts for XFTP files.
-- don't send commands after timeout.
-- PQ encryption support.
-
 # 5.6.2
 
 Version 5.6.2.2.

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.7.1.0
+version: 5.7.0.2
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.7.1.0
+version:        5.7.0.2
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/Messaging/Client.hs

@@ -700,7 +700,7 @@ sendBatch c@ProtocolClient {client_ = PClient {rcvConcurrency, sndQ}} b = do
 
 -- | Send Protocol command
 sendProtocolCommand :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
-sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THandleParams {batch, blockSize}} pKey entId cmd =
+sendProtocolCommand c@ProtocolClient {client_ = PClient {tcpTimeout, sndQ}, thParams = THandleParams {batch, blockSize}} pKey entId cmd =
   ExceptT $ uncurry sendRecv =<< mkTransmission c (pKey, entId, cmd)
   where
     -- two separate "atomically" needed to avoid blocking
@@ -711,9 +711,12 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THand
         | B.length s > blockSize - 2 -> pure . Left $ PCETransportError TELargeMsg
         | otherwise -> do
             active <- newTVarIO True
-            atomically (writeTBQueue sndQ (active, s))
-            response <$> getResponse c active r
+            timeout tcpSendTimeout (atomically $ writeTBQueue sndQ (active, s)) >>= \case
+              Nothing -> pure $ Left PCEResponseTimeout
+              Just () -> response <$> getResponse c active r
         where
+          -- TODO: move to configuration
+          tcpSendTimeout = tcpTimeout * 3 -- conservative timeout, allowing some asymmetry in uplink
           s
             | batch = tEncodeBatch1 t
             | otherwise = tEncode t

src/Simplex/Messaging/Compression.hs

@@ -4,9 +4,20 @@
 module Simplex.Messaging.Compression where
 
 import qualified Codec.Compression.Zstd as Z1
+import qualified Codec.Compression.Zstd.FFI as Z
+import Control.Monad (forM)
+import Control.Monad.Except
+import Control.Monad.IO.Class
 import Data.ByteString (ByteString)
 import qualified Data.ByteString as B
+import qualified Data.ByteString.Unsafe as B
+import Data.Either (fromRight)
+import Data.List.NonEmpty (NonEmpty)
+import Foreign
+import Foreign.C.Types
+import GHC.IO (unsafePerformIO)
 import Simplex.Messaging.Encoding
+import UnliftIO.Exception (bracket)
 
 data Compressed
   = -- | Short messages are left intact to skip copying and FFI festivities.
@@ -31,15 +42,49 @@ instance Encoding Compressed where
       '1' -> Compressed <$> smpP
       x -> fail $ "unknown Compressed tag: " <> show x
 
+-- | Compress as single chunk using stack-allocated context.
 compress1 :: ByteString -> Compressed
 compress1 bs
   | B.length bs <= maxLengthPassthrough = Passthrough bs
   | otherwise = Compressed . Large $ Z1.compress compressionLevel bs
 
-decompress1 :: Compressed -> Either String ByteString
-decompress1 = \case
-  Passthrough bs -> Right bs
-  Compressed (Large bs) -> case Z1.decompress bs of
-    Z1.Error e -> Left e
-    Z1.Skip -> Right mempty
-    Z1.Decompress bs' -> Right bs'
+type CompressCtx = (Ptr Z.CCtx, Ptr CChar, CSize)
+
+withCompressCtx :: CSize -> (CompressCtx -> IO a) -> IO a
+withCompressCtx scratchSize action =
+  bracket Z.createCCtx Z.freeCCtx $ \cctx ->
+    allocaBytes (fromIntegral scratchSize) $ \scratchPtr ->
+      action (cctx, scratchPtr, scratchSize)
+
+-- | Compress bytes, falling back to Passthrough in case of some internal error.
+compress :: CompressCtx -> ByteString -> IO Compressed
+compress ctx bs = fromRight (Passthrough bs) <$> compress_ ctx bs
+
+compress_ :: CompressCtx -> ByteString -> IO (Either String Compressed)
+compress_ (cctx, scratchPtr, scratchSize) bs
+  | B.length bs <= maxLengthPassthrough = pure . Right $ Passthrough bs
+  | otherwise =
+      B.unsafeUseAsCStringLen bs $ \(sourcePtr, sourceSize) -> runExceptT $ do
+        -- should not fail, unless input buffer is too short
+        dstSize <- ExceptT $ Z.checkError $ Z.compressCCtx cctx scratchPtr scratchSize sourcePtr (fromIntegral sourceSize) compressionLevel
+        liftIO $ Compressed . Large <$> B.packCStringLen (scratchPtr, fromIntegral dstSize)
+
+type DecompressCtx = (Ptr Z.DCtx, Ptr CChar, CSize)
+
+withDecompressCtx :: Int -> (DecompressCtx -> IO a) -> IO a
+withDecompressCtx maxUnpackedSize action =
+  bracket Z.createDCtx Z.freeDCtx $ \dctx ->
+    allocaBytes maxUnpackedSize $ \scratchPtr ->
+      action (dctx, scratchPtr, fromIntegral maxUnpackedSize)
+
+decompress :: DecompressCtx -> Compressed -> IO (Either String ByteString)
+decompress (dctx, scratchPtr, scratchSize) = \case
+  Passthrough bs -> pure $ Right bs
+  Compressed (Large bs) ->
+    B.unsafeUseAsCStringLen bs $ \(sourcePtr, sourceSize) -> do
+      res <- Z.checkError $ Z.decompressDCtx dctx scratchPtr scratchSize sourcePtr (fromIntegral sourceSize)
+      forM res $ \dstSize -> B.packCStringLen (scratchPtr, fromIntegral dstSize)
+
+decompressBatch :: Int -> NonEmpty Compressed -> NonEmpty (Either String ByteString)
+decompressBatch maxUnpackedSize items = unsafePerformIO $ withDecompressCtx maxUnpackedSize $ forM items . decompress
+{-# NOINLINE decompressBatch #-} -- prevent double-evaluation under unsafePerformIO

src/Simplex/Messaging/Notifications/Transport.hs

@@ -47,7 +47,7 @@ currentClientNTFVersion :: VersionNTF
 currentClientNTFVersion = VersionNTF 1
 
 currentServerNTFVersion :: VersionNTF
-currentServerNTFVersion = VersionNTF 2
+currentServerNTFVersion = VersionNTF 1
 
 supportedClientNTFVRange :: VersionRangeNTF
 supportedClientNTFVRange = mkVersionRange initialNTFVersion currentClientNTFVersion

src/Simplex/Messaging/Transport.hs

@@ -153,7 +153,7 @@ currentClientSMPRelayVersion :: VersionSMP
 currentClientSMPRelayVersion = VersionSMP 6
 
 currentServerSMPRelayVersion :: VersionSMP
-currentServerSMPRelayVersion = VersionSMP 7
+currentServerSMPRelayVersion = VersionSMP 6
 
 -- minimal supported protocol version is 4
 -- TODO remove code that supports sending commands without batching

src/Simplex/Messaging/Transport/Client.hs

@@ -19,7 +19,7 @@ module Simplex.Messaging.Transport.Client
     TransportHost (..),
     TransportHosts (..),
     TransportHosts_ (..),
-    validateCertificateChain,
+    validateCertificateChain
   )
 where
 
@@ -52,7 +52,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll, parseString)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.KeepAlive
-import Simplex.Messaging.Util (bshow, catchAll, tshow, (<$?>))
+import Simplex.Messaging.Util (bshow, (<$?>), catchAll, tshow)
 import System.IO.Error
 import System.Timeout (timeout)
 import Text.Read (readMaybe)
@@ -143,19 +143,14 @@ runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy,
   serverCert <- newEmptyTMVarIO
   let hostName = B.unpack $ strEncode host
       clientParams = mkTLSClientParams tlsParams caStore_ hostName port keyHash clientCredentials alpn serverCert
-      (connectTCP, tlsTimeout) = case socksProxy of
-        -- We use a much larger timeout for connections via SOCKS proxy, to allow the circuits created
-        -- in the socket connection that would otherwise timeout to be used in the next connection attempt.
-        -- Using standard timeout results in permanent timeout for the clients using SOCKS in cases
-        -- when SOCKS proxy is very slow (bad network, congestion in underlying network, etc.),
-        -- because SOCKS proxy destroys circuits when the last session using them is closed.
-        Just proxy -> (connectSocksClient proxy proxyUsername (hostAddr host), tcpConnectTimeout * 10)
-        _ -> (connectTCPClient hostName, tcpConnectTimeout)
+      connectTCP = case socksProxy of
+        Just proxy -> connectSocksClient proxy proxyUsername $ hostAddr host
+        _ -> connectTCPClient hostName
   c <- do
     sock <- connectTCP port
     mapM_ (setSocketKeepAlive sock) tcpKeepAlive `catchAll` \e -> logError ("Error setting TCP keep-alive" <> tshow e)
     let tCfg = clientTransportConfig cfg
-    tlsTimeout `timeout` connectTLS (Just hostName) tCfg clientParams sock >>= \case
+    tcpConnectTimeout `timeout` connectTLS (Just hostName) tCfg clientParams sock >>= \case
       Nothing -> do
         close sock
         logError "connection timed out"