transport: do not retain last received TLS record

src/Simplex/Messaging/Server.hs

@@ -45,6 +45,7 @@ module Simplex.Messaging.Server
   )
 where
 
+import Control.Applicative ((<|>))
 import Control.Concurrent.STM (throwSTM)
 import qualified Control.Exception as E
 import Control.Logger.Simple
@@ -197,6 +198,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
         : sigIntHandlerThread
         : map runServer transports
             <> expireMessagesThread_ cfg
+            <> expireClientsThread_ cfg
             <> serverStatsThread_ cfg
             <> prometheusMetricsThread_ cfg
             <> controlPortThread_ cfg
@@ -474,6 +476,27 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
     expireMessagesThread_ ServerConfig {messageExpiration = Just msgExp} = [expireMessagesThread msgExp]
     expireMessagesThread_ _ = []
 
+    expireClientsThread_ :: ServerConfig s -> [M s ()]
+    expireClientsThread_ ServerConfig {inactiveClientExpiration = Just expCfg} = [expireClientsThread expCfg]
+    expireClientsThread_ _ = []
+
+    expireClientsThread :: ExpirationConfig -> M s ()
+    expireClientsThread expCfg = do
+      labelMyThread "expireClients"
+      srv <- asks server
+      liftIO $ forever $ do
+        threadDelay' $ checkInterval expCfg * 1000000
+        old <- expireBeforeEpoch expCfg
+        getServerClients srv >>= mapM_ (expireClient srv old)
+      where
+        expireClient srv old c@Client {rcvActiveAt, sndActiveAt, closeTransport} = do
+          ts <- max <$> readTVarIO rcvActiveAt <*> readTVarIO sndActiveAt
+          when (systemSeconds ts < old) $ whenM (noSubscriptions srv c) closeTransport
+        noSubscriptions srv Client {clientId} =
+          not <$> anyM [hasSubs (subscribers srv), hasSubs (ntfSubscribers srv)]
+          where
+            hasSubs ServerSubscribers {subClients} = IS.member clientId <$> readTVarIO subClients
+
     expireMessagesThread :: ExpirationConfig -> M s ()
     expireMessagesThread ExpirationConfig {checkInterval, ttl} = do
       ms <- asks msgStore
@@ -1067,7 +1090,7 @@ runClientTransport h@THandle {params = thParams@THandleParams {sessionId}} = do
   ts <- liftIO getSystemTime
   nextClientId <- asks clientSeq
   clientId <- atomically $ stateTVar nextClientId $ \next -> (next, next + 1)
-  c <- liftIO $ newClient clientId q thParams ts
+  c <- liftIO $ newClient clientId q thParams ts (closeConnection $ connection h)
   runClientThreads c `finally` clientDisconnected c
   where
     runClientThreads :: Client s -> M s ()
@@ -1075,17 +1098,8 @@ runClientTransport h@THandle {params = thParams@THandleParams {sessionId}} = do
       s <- asks server
       ms <- asks msgStore
       whenM (liftIO $ insertServerClient c s) $ do
-        expCfg <- asks $ inactiveClientExpiration . config
-        th <- newMVar h -- put TH under a fair lock to interleave messages and command responses
         labelMyThread . B.unpack $ "client $" <> encode sessionId
-        raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, client s ms c, receive h ms c] <> disconnectThread_ c s expCfg
-    disconnectThread_ :: Client s -> Server s -> Maybe ExpirationConfig -> [M s ()]
-    disconnectThread_ c s (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c s)]
-    disconnectThread_ _ _ _ = []
-    noSubscriptions Client {clientId} s =
-      not <$> anyM [hasSubs (subscribers s), hasSubs (ntfSubscribers s)]
-      where
-        hasSubs ServerSubscribers {subClients} = IS.member clientId <$> readTVarIO subClients
+        raceAny_ [liftIO $ send h c, client s ms c, receive h ms c]
 
 controlPortAuth :: Handle -> Maybe BasicAuth -> Maybe BasicAuth -> TVar CPClientRole -> BasicAuth -> IO ()
 controlPortAuth h user admin role auth = do
@@ -1191,36 +1205,38 @@ receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ,
             GET -> incStat $ msgGetAuth stats
             _ -> pure ()
 
-send :: Transport c => MVar (THandleSMP c 'TServer) -> Client s -> IO ()
-send th c@Client {sndQ, msgQ, clientTHParams = THandleParams {sessionId}} = do
+-- Messages are sent one per transmission, interleaved with command responses.
+send :: Transport c => THandleSMP c 'TServer -> Client s -> IO ()
+send h c@Client {sndQ, msgQ, clientTHParams = THandleParams {sessionId}} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
-  forever $ atomically (readTBQueue sndQ) >>= sendTransmissions
+  sendLoop []
   where
+    sendLoop :: [Transmission BrokerMsg] -> IO ()
+    sendLoop = \case
+      [] ->
+        atomically (Left <$> readTBQueue sndQ <|> Right <$> readTBQueue msgQ) >>= \case
+          Left ts -> sendTransmissions ts >>= sendLoop
+          Right msgs -> sendLoop $ L.toList msgs
+      msg : msgs -> do
+        tSend h c [msg]
+        atomically (tryReadTBQueue sndQ) >>= \case
+          Just ts -> sendTransmissions ts >>= sendLoop . (msgs <>)
+          Nothing -> sendLoop msgs
     -- If the request had batched subscriptions
     -- this will reply SOKs to all SUBs in the first batched transmission,
     -- to reduce client timeouts.
     -- After that all messages will be sent in separate transmissions,
     -- without any client response timeouts, and allowing them to interleave
     -- with other requests responses.
-    sendTransmissions :: (NonEmpty (Transmission BrokerMsg), [Transmission BrokerMsg]) -> IO ()
-    sendTransmissions (ts, []) = tSend th c ts
+    sendTransmissions :: (NonEmpty (Transmission BrokerMsg), [Transmission BrokerMsg]) -> IO [Transmission BrokerMsg]
+    sendTransmissions (ts, []) = [] <$ tSend h c ts
     sendTransmissions (ts, msg : msgs)
-      | length ts <= 4 = do -- up to 4 SOKs can be in one block with MSG (see testBatchSubResponses test)
-          tSend th c $ ts <> [msg]
-          mapM_ (atomically . writeTBQueue msgQ) $ L.nonEmpty msgs
-      | otherwise = do
-          tSend th c ts
-          atomically $ writeTBQueue msgQ (msg :| msgs)
+      | length ts <= 4 = msgs <$ tSend h c (ts <> [msg]) -- up to 4 SOKs can be in one block with MSG (see testBatchSubResponses test)
+      | otherwise = (msg : msgs) <$ tSend h c ts
 
-sendMsg :: Transport c => MVar (THandleSMP c 'TServer) -> Client s -> IO ()
-sendMsg th c@Client {msgQ, clientTHParams = THandleParams {sessionId}} = do
-  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " sendMsg"
-  forever $ atomically (readTBQueue msgQ) >>= mapM_ (\t -> tSend th c [t])
-
-tSend :: Transport c => MVar (THandleSMP c 'TServer) -> Client s -> NonEmpty (Transmission BrokerMsg) -> IO ()
-tSend th Client {sndActiveAt} ts = do
-  withMVar th $ \h@THandle {params} ->
-    void . tPut h $ L.map (\t -> Right (Nothing, encodeTransmission params t)) ts
+tSend :: Transport c => THandleSMP c 'TServer -> Client s -> NonEmpty (Transmission BrokerMsg) -> IO ()
+tSend h@THandle {params} Client {sndActiveAt} ts = do
+  void . tPut h $ L.map (\t -> Right (Nothing, encodeTransmission params t)) ts
   atomically . (writeTVar sndActiveAt $!) =<< liftIO getSystemTime
 
 disconnectTransport :: Transport c => THandle v c 'TServer -> TVar SystemTime -> TVar SystemTime -> ExpirationConfig -> IO Bool -> IO ()

src/Simplex/Messaging/Server/Env/STM.hs

@@ -465,7 +465,8 @@ data Client s = Client
     connected :: TVar Bool,
     createdAt :: SystemTime,
     rcvActiveAt :: TVar SystemTime,
-    sndActiveAt :: TVar SystemTime
+    sndActiveAt :: TVar SystemTime,
+    closeTransport :: IO ()
   }
 
 type VerifiedTransmission s = (Maybe (StoreQueue s, QueueRec), Transmission Cmd)
@@ -520,8 +521,8 @@ newServerSubscribers = do
   pendingEvents <- newTVarIO IM.empty
   pure ServerSubscribers {subQ, queueSubscribers, serviceSubscribers, totalServiceSubs, subClients, pendingEvents}
 
-newClient :: ClientId -> Natural -> THandleParams SMPVersion 'TServer -> SystemTime -> IO (Client s)
-newClient clientId qSize clientTHParams createdAt = do
+newClient :: ClientId -> Natural -> THandleParams SMPVersion 'TServer -> SystemTime -> IO () -> IO (Client s)
+newClient clientId qSize clientTHParams createdAt closeTransport = do
   subscriptions <- TM.emptyIO
   ntfSubscriptions <- TM.emptyIO
   serviceSubscribed <- newTVarIO False
@@ -556,7 +557,8 @@ newClient clientId qSize clientTHParams createdAt = do
         connected,
         createdAt,
         rcvActiveAt,
-        sndActiveAt
+        sndActiveAt,
+        closeTransport
       }
 
 newSubscription :: SubscriptionThread -> STM Sub

src/Simplex/Messaging/Transport.hs

@@ -77,6 +77,7 @@ module Simplex.Messaging.Transport
     ALPN,
     connectTLS,
     closeTLS,
+    recvTLS,
     defaultSupportedParams,
     defaultSupportedParamsHTTPS,
     withTlsUnique,
@@ -375,6 +376,11 @@ withTlsUnique cxt f =
       STServer -> T.getPeerFinished
       STClient -> T.getFinished
 
+-- tls stores the receive record state lazily, so until the next record arrives it retains the last one (up to 16KB).
+-- Reading the negotiated cipher forces it.
+recvTLS :: T.Context -> IO ByteString
+recvTLS cxt = T.recvData cxt <* T.contextGetInformation cxt
+
 closeTLS :: T.Context -> IO ()
 closeTLS ctx =
   T.bye ctx -- sometimes socket was closed before 'TLS.bye' so we catch the 'Broken pipe' error here
@@ -437,7 +443,7 @@ instance Transport TLS where
   -- this function may return less than requested number of bytes
   cGet :: TLS p -> Int -> IO ByteString
   cGet TLS {tlsContext, tlsBuffer, tlsTransportConfig = TransportConfig {transportTimeout = t_}} n =
-    getBuffered tlsBuffer n t_ (T.recvData tlsContext)
+    getBuffered tlsBuffer n t_ (recvTLS tlsContext)
 
   cPut :: TLS p -> ByteString -> IO ()
   cPut TLS {tlsContext, tlsTransportConfig = TransportConfig {transportTimeout = t_}} =
@@ -445,7 +451,7 @@ instance Transport TLS where
 
   getLn :: TLS p -> IO ByteString
   getLn TLS {tlsContext, tlsBuffer} = do
-    getLnBuffered tlsBuffer (T.recvData tlsContext) `E.catches` [E.Handler handleTlsEOF, E.Handler handleEOF]
+    getLnBuffered tlsBuffer (recvTLS tlsContext) `E.catches` [E.Handler handleTlsEOF, E.Handler handleEOF]
     where
       handleTlsEOF = \case
         T.PostHandshake T.Error_EOF -> E.throwIO TEBadBlock

src/Simplex/Messaging/Transport/WebSockets.hs

@@ -27,6 +27,7 @@ import Simplex.Messaging.Transport
     STransportPeer (..),
     TransportPeerI (..),
     closeTLS,
+    recvTLS,
     smpBlockSize,
     withTlsUnique,
   )
@@ -106,7 +107,7 @@ makeTLSContextStream cxt =
   S.makeStream readStream writeStream
   where
     readStream :: IO (Maybe ByteString)
-    readStream = (Just <$> T.recvData cxt) `E.catches` [E.Handler handleTlsEOF, E.Handler handleEOF]
+    readStream = (Just <$> recvTLS cxt) `E.catches` [E.Handler handleTlsEOF, E.Handler handleEOF]
       where
         handleTlsEOF = \case
           T.PostHandshake T.Error_EOF -> pure Nothing

tests/ServerTests.hs

@@ -106,6 +106,7 @@ serverTests = do
     testMsgExpireOnInterval
     testMsgNOTExpireOnInterval
   describe "Blocking queues" $ testBlockMessageQueue
+  describe "Inactive clients" testInactiveClientExpiration
   describe "Short links" $ do
     testInvQueueLinkData
     testContactQueueLinkData
@@ -1563,6 +1564,23 @@ testMsgExpireOnInterval =
             Nothing -> return ()
             Just _ -> error "nothing should be delivered"
 
+testInactiveClientExpiration :: SpecWith (ASrvTransport, AStoreType)
+testInactiveClientExpiration =
+  it "should disconnect inactive clients without subscriptions" $ \(ATransport (t :: TProxy c 'TServer), msType) -> do
+    g <- C.newRandom
+    (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
+    (dhPub, _ :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
+    let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
+    withSmpServerConfigOn (ATransport t) cfg' testPort $ \_ ->
+      testSMPClient @c $ \rh -> testSMPClient @c $ \h -> do
+        Resp "1" NoEntity (Ids _ _ _) <- signSendRecv rh rKey ("1", NoEntity, New rPub dhPub)
+        threadDelay 2500000
+        try (timeout 2000000 $ tGet1 h) >>= \case
+          Left (_ :: SomeException) -> pure ()
+          Right r -> unexpected (r :: Maybe (Transmission (Either ErrorType BrokerMsg)))
+        Resp "2" NoEntity PONG <- sendRecv rh (Nothing, "2", NoEntity, PING)
+        pure ()
+
 testMsgNOTExpireOnInterval :: SpecWith (ASrvTransport, AStoreType)
 testMsgNOTExpireOnInterval =
   it "should block and unblock message queues" $ \(ATransport (t :: TProxy c 'TServer), msType) -> do