call sendData with an empty bytestring

cabal.project

@@ -2,5 +2,5 @@ packages: .
 
 source-repository-package
     type: git
-    location: git://github.com/simplex-chat/aeson.git
+    location: https://github.com/simplex-chat/aeson.git
     tag: 3eb66f9a68f103b5f1489382aad89f5712a64db7

package.yaml

@@ -45,7 +45,7 @@ dependencies:
   - iso8601-time == 0.1.*
   - memory == 0.15.*
   - mtl == 2.2.*
-  - network == 3.1.*
+  - network == 3.1.2.*
   - network-transport == 0.5.*
   - QuickCheck == 2.14.*
   - random >= 1.1 && < 1.3

simplexmq.cabal

@@ -41,7 +41,6 @@ library
       Simplex.Messaging.Agent.Store.SQLite.Migrations
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220320_server_ips
       Simplex.Messaging.Client
       Simplex.Messaging.Crypto
       Simplex.Messaging.Crypto.Ratchet
@@ -58,6 +57,7 @@ library
       Simplex.Messaging.Server.StoreLog
       Simplex.Messaging.Transport
       Simplex.Messaging.Transport.Client
+      Simplex.Messaging.Transport.KeepAlive
       Simplex.Messaging.Transport.Server
       Simplex.Messaging.Transport.WebSockets
       Simplex.Messaging.Util
@@ -92,7 +92,7 @@ library
     , iso8601-time ==0.1.*
     , memory ==0.15.*
     , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
     , network-transport ==0.5.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
@@ -143,7 +143,7 @@ executable smp-agent
     , iso8601-time ==0.1.*
     , memory ==0.15.*
     , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
     , network-transport ==0.5.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
@@ -196,7 +196,7 @@ executable smp-server
     , iso8601-time ==0.1.*
     , memory ==0.15.*
     , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
     , network-transport ==0.5.*
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
@@ -265,7 +265,7 @@ test-suite smp-server-test
     , iso8601-time ==0.1.*
     , memory ==0.15.*
     , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
     , network-transport ==0.5.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*

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

@@ -26,7 +26,6 @@ import Database.SQLite.Simple.QQ (sql)
 import qualified Database.SQLite3 as SQLite3
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
-import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220320_server_ips
 
 data Migration = Migration {name :: String, up :: Text}
   deriving (Show)
@@ -34,8 +33,7 @@ data Migration = Migration {name :: String, up :: Text}
 schemaMigrations :: [(String, Query)]
 schemaMigrations =
   [ ("20220101_initial", m20220101_initial),
-    ("20220301_snd_queue_keys", m20220301_snd_queue_keys),
-    ("20220320_server_ips", m20220320_server_ips)
+    ("20220301_snd_queue_keys", m20220301_snd_queue_keys)
   ]
 
 -- | The list of migrations in ascending order by date

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

@@ -1,14 +0,0 @@
-{-# LANGUAGE QuasiQuotes #-}
-
-module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220320_server_ips where
-
-import Database.SQLite.Simple (Query)
-import Database.SQLite.Simple.QQ (sql)
-
-m20220320_server_ips :: Query
-m20220320_server_ips =
-  [sql|
-UPDATE servers SET host = '178.79.168.175' WHERE host = 'smp8.simplex.im';
-UPDATE servers SET host = '178.79.169.107' WHERE host = 'smp9.simplex.im';
-UPDATE servers SET host = '45.33.54.229' WHERE host = 'smp10.simplex.im';
-|]

src/Simplex/Messaging/Client.hs

@@ -110,7 +110,7 @@ smpDefaultConfig =
     { qSize = 64,
       defaultTransport = ("5223", transport @TLS),
       tcpTimeout = 4_000_000,
-      smpPing = 30_000_000
+      smpPing = 3_600_000_000 -- 1 hour
     }
 
 data Request = Request

src/Simplex/Messaging/Transport.hs

@@ -80,6 +80,7 @@ import qualified Network.TLS.Extra as TE
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (dropPrefix, parse, parseRead1, sumTypeJSON)
+import Simplex.Messaging.Transport.KeepAlive
 import Simplex.Messaging.Util (bshow)
 import Simplex.Messaging.Version
 import Test.QuickCheck (Arbitrary (..))
@@ -147,7 +148,8 @@ data TLS = TLS
     tlsPeer :: TransportPeer,
     tlsUniq :: ByteString,
     buffer :: TVar ByteString,
-    getLock :: TMVar ()
+    getLock :: TMVar (),
+    keepAlive :: Maybe KeepAliveThread
   }
 
 connectTLS :: T.TLSParams p => p -> Socket -> IO T.Context
@@ -163,7 +165,7 @@ getTLS tlsPeer cxt = withTlsUnique tlsPeer cxt newTLS
     newTLS tlsUniq = do
       buffer <- newTVarIO ""
       getLock <- newTMVarIO ()
-      pure TLS {tlsContext = cxt, tlsPeer, tlsUniq, buffer, getLock}
+      pure TLS {tlsContext = cxt, tlsPeer, tlsUniq, buffer, getLock, keepAlive = Nothing}
 
 withTlsUnique :: TransportPeer -> T.Context -> (ByteString -> IO c) -> IO c
 withTlsUnique peer cxt f =
@@ -195,12 +197,17 @@ instance Transport TLS where
   transportName _ = "TLS"
   transportPeer = tlsPeer
   getServerConnection = getTLS TServer
-  getClientConnection = getTLS TClient
+  getClientConnection cxt = do
+    tls <- getTLS TClient cxt
+    keepAlive <- Just <$> startKeepAlive (tlsContext tls)
+    pure tls {keepAlive}
   tlsUnique = tlsUniq
-  closeConnection tls = closeTLS $ tlsContext tls
+  closeConnection TLS {tlsContext, keepAlive} = do
+    mapM_ stopKeepAlive keepAlive 
+    closeTLS tlsContext
 
   cGet :: TLS -> Int -> IO ByteString
-  cGet TLS {tlsContext, buffer, getLock} n =
+  cGet TLS {tlsContext, buffer, getLock, keepAlive} n =
     E.bracket_
       (atomically $ takeTMVar getLock)
       (atomically $ putTMVar getLock ())
@@ -213,16 +220,21 @@ instance Transport TLS where
       readChunks :: ByteString -> IO ByteString
       readChunks b
         | B.length b >= n = pure b
-        | otherwise = readChunks . (b <>) =<< T.recvData tlsContext `E.catch` handleEOF
+        | otherwise = do
+          chunk <- T.recvData tlsContext `E.catch` handleEOF
+          mapM_ touchKeepAlive keepAlive
+          readChunks $ b <> chunk
       handleEOF = \case
         T.Error_EOF -> E.throwIO TEBadBlock
         e -> E.throwIO e
 
   cPut :: TLS -> ByteString -> IO ()
-  cPut tls = T.sendData (tlsContext tls) . BL.fromStrict
+  cPut TLS {tlsContext, keepAlive} s = do
+    mapM_ touchKeepAlive keepAlive
+    T.sendData tlsContext $ BL.fromStrict s
 
   getLn :: TLS -> IO ByteString
-  getLn TLS {tlsContext, buffer, getLock} = do
+  getLn TLS {tlsContext, buffer, getLock, keepAlive} = do
     E.bracket_
       (atomically $ takeTMVar getLock)
       (atomically $ putTMVar getLock ())
@@ -236,9 +248,9 @@ instance Transport TLS where
       readChunks b
         | B.elem '\n' b = pure b
         | otherwise = readChunks . (b <>) =<< T.recvData tlsContext `E.catch` handleEOF
-      handleEOF = \case
+      handleEOF e = mapM_ stopKeepAlive keepAlive >> case e of
         T.Error_EOF -> E.throwIO TEBadBlock
-        e -> E.throwIO e
+        _ -> E.throwIO e
 
 -- | Trim trailing CR from ByteString.
 trimCR :: ByteString -> ByteString

src/Simplex/Messaging/Transport/Client.hs

@@ -20,6 +20,7 @@ import Network.Socket
 import qualified Network.TLS as T
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.KeepAlive
 import System.IO.Error
 import UnliftIO.Exception (IOException)
 import qualified UnliftIO.Exception as E
@@ -51,6 +52,7 @@ startTCPClient host port clientParams = withSocketsDo $ resolve >>= tryOpen err
     open addr = do
       sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
       connect sock $ addrAddress addr
+      setSocketKeepAlive sock defaultKeepAlive
       ctx <- connectTLS clientParams sock
       getClientConnection ctx
 
@@ -75,7 +77,7 @@ validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain sc@[_,
     x509validate = XV.validate X.HashSHA256 hooks checks certStore cache serviceID cc
       where
         hooks = XV.defaultHooks
-        checks = XV.defaultChecks {XV.checkFQHN = False}
+        checks = XV.defaultChecks
         certStore = XS.makeCertificateStore sc
         cache = XV.exceptionValidationCache [] -- we manually check fingerprint only of the identity certificate (ca.crt)
         serviceID = (host, port)

src/Simplex/Messaging/Transport/KeepAlive.hs

@@ -0,0 +1,72 @@
+{-# LANGUAGE CApiFFI #-}
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Simplex.Messaging.Transport.KeepAlive where
+
+import Control.Concurrent
+import Control.Concurrent.STM
+import Control.Monad
+import Data.Time.Clock.System (SystemTime (..), getSystemTime)
+import Foreign.C (CInt (..))
+import Network.Socket
+import qualified Network.TLS as T
+
+foreign import capi "netinet/tcp.h value TCP_KEEPCNT" tcpKeepCnt :: CInt
+
+foreign import capi "netinet/tcp.h value TCP_KEEPINTVL" tcpKeepIntvl :: CInt
+
+#if defined(darwin_HOST_OS)
+foreign import capi "netinet/tcp.h value TCP_KEEPALIVE" tcpKeepIdle :: CInt
+foreign import capi "netinet/in.h value IPPROTO_TCP" solTcp :: CInt
+#else
+foreign import capi "netinet/tcp.h value TCP_KEEPIDLE" tcpKeepIdle :: CInt
+foreign import capi "netinet/tcp.h value SOL_TCP" solTcp :: CInt
+#endif
+
+data KeepAliveOpts = KeepAliveOpts
+  { keepCnt :: Int,
+    keepIdle :: Int,
+    keepIntvl :: Int
+  }
+
+defaultKeepAlive :: KeepAliveOpts
+defaultKeepAlive =
+  KeepAliveOpts
+    { keepCnt = 4,
+      keepIdle = 60,
+      keepIntvl = 30
+    }
+
+setSocketKeepAlive :: Socket -> KeepAliveOpts -> IO ()
+setSocketKeepAlive sock KeepAliveOpts {keepCnt, keepIdle, keepIntvl} = do
+  setSocketOption sock KeepAlive 1
+  setSocketOption sock (SockOpt solTcp tcpKeepCnt) keepCnt
+  setSocketOption sock (SockOpt solTcp tcpKeepIdle) keepIdle
+  setSocketOption sock (SockOpt solTcp tcpKeepIntvl) keepIntvl
+
+data KeepAliveThread = KeepAliveThread
+  { threadId :: ThreadId,
+    dataTs :: TVar SystemTime
+  }
+
+startKeepAlive :: T.Context -> IO KeepAliveThread
+startKeepAlive cxt = do
+  dataTs <- newTVarIO =<< getSystemTime
+  threadId <- forkIO . forever $ do
+    threadDelay 30000000
+    ts' <- getSystemTime
+    doPing <- atomically $ do
+      ts <- readTVar dataTs
+      let ping = systemSeconds ts' - systemSeconds ts >= 30
+      when ping $ writeTVar dataTs ts'
+      pure ping
+    when doPing $ putStrLn "*** ping ***" >> T.sendData cxt ""
+  pure KeepAliveThread {threadId, dataTs}
+
+touchKeepAlive :: KeepAliveThread -> IO ()
+touchKeepAlive KeepAliveThread {dataTs} = atomically . writeTVar dataTs =<< getSystemTime
+
+stopKeepAlive :: KeepAliveThread -> IO ()
+stopKeepAlive KeepAliveThread {threadId} = killThread threadId

stack.yaml

@@ -36,6 +36,7 @@ packages:
 #
 extra-deps:
   - cryptostore-0.2.1.0@sha256:9896e2984f36a1c8790f057fd5ce3da4cbcaf8aa73eb2d9277916886978c5b19,3881
+  - network-3.1.2.7@sha256:e3d78b13db9512aeb106e44a334ab42b7aa48d26c097299084084cb8be5c5568,4888
   - simple-logger-0.1.0@sha256:be8ede4bd251a9cac776533bae7fb643369ebd826eb948a9a18df1a8dd252ff8,1079
   - tls-1.5.7@sha256:1cc30253a9696b65a9cafc0317fbf09f7dcea15e3a145ed6c9c0e28c632fa23a,6991
 # below dependancies are to update Aeson to 2.0.3

tests/AgentTests.hs

@@ -468,7 +468,7 @@ syntaxTests t = do
           "a",
           "JOIN https://simpex.chat/invitation#/?smp=smp%3A%2F%2F"
             <> urlEncode True "LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI="
-            <> "%40127.0.0.1%3A5001%2F3456-w%3D%3D%23"
+            <> "%40localhost%3A5001%2F3456-w%3D%3D%23"
             <> urlEncode True sampleDhKey
             <> "&v=1"
             <> "&e2e=v%3D1%26x3dh%3DMEIwBQYDK2VvAzkAmKuSYeQ_m0SixPDS8Wq8VBaTS1cW-Lp0n0h4Diu-kUpR-qXx4SDJ32YGEFoGFGSbGPry5Ychr6U%3D%2CMEIwBQYDK2VvAzkAmKuSYeQ_m0SixPDS8Wq8VBaTS1cW-Lp0n0h4Diu-kUpR-qXx4SDJ32YGEFoGFGSbGPry5Ychr6U%3D"

tests/SMPAgentClient.hs

@@ -157,7 +157,7 @@ cfg :: AgentConfig
 cfg =
   defaultAgentConfig
     { tcpPort = agentTestPort,
-      initialSMPServers = L.fromList ["smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@127.0.0.1:5001"],
+      initialSMPServers = L.fromList ["smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"],
       tbqSize = 1,
       dbFile = testDB,
       smpCfg =
@@ -174,7 +174,7 @@ cfg =
 
 withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m () -> (ThreadId -> m a) -> m a
 withSmpAgentThreadOn_ t (port', smpPort', db') afterProcess =
-  let cfg' = cfg {tcpPort = port', dbFile = db', initialSMPServers = L.fromList [SMPServer "127.0.0.1" smpPort' testKeyHash]}
+  let cfg' = cfg {tcpPort = port', dbFile = db', initialSMPServers = L.fromList [SMPServer "localhost" smpPort' testKeyHash]}
    in serverBracket
         (\started -> runSMPAgentBlocking t started cfg')
         afterProcess

tests/SMPClient.hs

@@ -29,7 +29,7 @@ import UnliftIO.STM (TMVar, atomically, newEmptyTMVarIO, takeTMVar)
 import UnliftIO.Timeout (timeout)
 
 testHost :: HostName
-testHost = "127.0.0.1"
+testHost = "localhost"
 
 testPort :: ServiceName
 testPort = "5001"