Merge branch 'master' into ep/journal-mode-wal

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

@@ -31,6 +31,7 @@ module Simplex.Messaging.Agent.Store.SQLite
     createSQLiteStore,
     connectSQLiteStore,
     closeSQLiteStore,
+    openSQLiteStore,
     sqlString,
     execSQL,
     upMigration, -- used in tests
@@ -269,13 +270,13 @@ import Simplex.Messaging.Parsers (blobFieldParser, dropPrefix, fromTextField_, s
 import Simplex.Messaging.Protocol
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Transport.Client (TransportHost)
-import Simplex.Messaging.Util (bshow, eitherToMaybe, groupOn, ($>>=), (<$$>))
+import Simplex.Messaging.Util (bshow, eitherToMaybe, groupOn, ifM, ($>>=), (<$$>))
 import Simplex.Messaging.Version
 import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
 import System.Exit (exitFailure)
 import System.FilePath (takeDirectory)
 import System.IO (hFlush, stdout)
-import UnliftIO.Exception (onException)
+import UnliftIO.Exception (onException, bracketOnError)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
@@ -379,10 +380,12 @@ confirmOrExit s = do
 connectSQLiteStore :: FilePath -> String -> IO SQLiteStore
 connectSQLiteStore dbFilePath dbKey = do
   dbNew <- not <$> doesFileExist dbFilePath
-  dbConn <- dbBusyLoop $ connectDB dbFilePath dbKey
-  dbConnVar <- newTMVarIO dbConn
-  dbEncrypted <- newTVarIO . not $ null dbKey
-  pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection = dbConnVar, dbNew}
+  dbConn <- dbBusyLoop (connectDB dbFilePath dbKey)
+  atomically $ do
+    dbConnection <- newTMVar dbConn
+    dbEncrypted <- newTVar . not $ null dbKey
+    dbClosed <- newTVar False
+    pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection, dbNew, dbClosed}
 
 connectDB :: FilePath -> String -> IO DB.Connection
 connectDB path key = do
@@ -406,7 +409,25 @@ connectDB path key = do
         |]
 
 closeSQLiteStore :: SQLiteStore -> IO ()
-closeSQLiteStore st = atomically (takeTMVar $ dbConnection st) >>= DB.close
+closeSQLiteStore st@SQLiteStore {dbClosed} =
+  ifM (readTVarIO dbClosed) (putStrLn "closeSQLiteStore: already closed") $
+    withConnection st $ \conn -> do
+      DB.close conn
+      atomically $ writeTVar dbClosed True
+
+openSQLiteStore :: SQLiteStore -> String -> IO ()
+openSQLiteStore SQLiteStore {dbConnection, dbFilePath, dbClosed} key =
+  ifM (readTVarIO dbClosed) open (putStrLn "closeSQLiteStore: already opened")
+  where
+    open =
+      bracketOnError
+        (atomically $ takeTMVar dbConnection)
+        (atomically . tryPutTMVar dbConnection)
+        $ \DB.Connection {slow} -> do
+          DB.Connection {conn} <- connectDB dbFilePath key
+          atomically $ do
+            putTMVar dbConnection DB.Connection {conn, slow}
+            writeTVar dbClosed False
 
 sqlString :: String -> Text
 sqlString s = quote <> T.replace quote "''" (T.pack s) <> quote

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

@@ -27,6 +27,7 @@ data SQLiteStore = SQLiteStore
   { dbFilePath :: FilePath,
     dbEncrypted :: TVar Bool,
     dbConnection :: TMVar DB.Connection,
+    dbClosed :: TVar Bool,
     dbNew :: Bool
   }
 

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

@@ -25,6 +25,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Transport (SessionId)
 import Simplex.Messaging.Transport.Client (TransportClientConfig (..), TransportHost (..), runTLSTransportClient)
 import Simplex.Messaging.Transport.HTTP2
+import Simplex.Messaging.Transport (TLS)
 import UnliftIO.STM
 import UnliftIO.Timeout
 
@@ -78,7 +79,17 @@ getHTTP2Client :: HostName -> ServiceName -> Maybe XS.CertificateStore -> HTTP2C
 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 =
+getVerifiedHTTP2Client proxyUsername host port keyHash caStore config disconnected = getVerifiedHTTP2ClientWith config host port disconnected setup
+  where
+    setup = runHTTP2Client (suportedTLSParams config) caStore (transportConfig config) (bufferSize config) proxyUsername host port keyHash
+
+attachHTTP2Client :: HTTP2ClientConfig -> TransportHost -> ServiceName -> IO () -> Int -> TLS -> IO (Either HTTP2ClientError HTTP2Client)
+attachHTTP2Client config host port disconnected bufferSize tls = getVerifiedHTTP2ClientWith config host port disconnected setup
+  where
+    setup = runHTTP2ClientWith bufferSize host ($ tls)
+
+getVerifiedHTTP2ClientWith :: HTTP2ClientConfig -> TransportHost -> ServiceName -> IO () -> ((SessionId -> H.Client HTTP2Response) -> IO HTTP2Response) -> IO (Either HTTP2ClientError HTTP2Client)
+getVerifiedHTTP2ClientWith config host port disconnected setup =
   (atomically mkHTTPS2Client >>= runClient)
     `E.catch` \(e :: IOException) -> pure . Left $ HCIOError e
   where
@@ -91,11 +102,8 @@ getVerifiedHTTP2Client proxyUsername host port keyHash caStore config@HTTP2Clien
     runClient :: HClient -> IO (Either HTTP2ClientError HTTP2Client)
     runClient c = do
       cVar <- newEmptyTMVarIO
-      action <-
-        async $
-          runHTTP2Client suportedTLSParams caStore transportConfig bufferSize proxyUsername host port keyHash (client c cVar)
-            `E.finally` atomically (putTMVar cVar $ Left HCNetworkError)
-      c_ <- connTimeout `timeout` atomically (takeTMVar cVar)
+      action <- async $ setup (client c cVar) `E.finally` atomically (putTMVar cVar $ Left HCNetworkError)
+      c_ <- connTimeout config `timeout` atomically (takeTMVar cVar)
       pure $ case c_ of
         Just (Right c') -> Right c' {action = Just action}
         Just (Left e) -> Left e
@@ -114,7 +122,7 @@ getVerifiedHTTP2Client proxyUsername host port keyHash caStore config@HTTP2Clien
     process HTTP2Client {client_ = HClient {reqQ}} sendReq = forever $ do
       (req, respVar) <- atomically $ readTBQueue reqQ
       sendReq req $ \r -> do
-        respBody <- getHTTP2Body r bodyHeadSize
+        respBody <- getHTTP2Body r (bodyHeadSize config)
         let resp = HTTP2Response {response = r, respBody}
         atomically $ putTMVar respVar resp
         pure resp
@@ -147,8 +155,12 @@ http2RequestTimeout :: HTTP2ClientConfig -> Maybe Int -> Int
 http2RequestTimeout HTTP2ClientConfig {connTimeout} = maybe connTimeout (connTimeout +)
 
 runHTTP2Client :: forall a. T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> BufferSize -> Maybe ByteString -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (SessionId -> H.Client a) -> IO a
-runHTTP2Client tlsParams caStore tcConfig bufferSize proxyUsername host port keyHash client =
-  runTLSTransportClient tlsParams caStore tcConfig proxyUsername host port keyHash $ withHTTP2 bufferSize run
+runHTTP2Client tlsParams caStore tcConfig bufferSize proxyUsername host port keyHash = runHTTP2ClientWith bufferSize host setup
+  where
+    setup = runTLSTransportClient tlsParams caStore tcConfig proxyUsername host port keyHash
+
+runHTTP2ClientWith :: forall a. BufferSize -> TransportHost -> ((TLS -> IO a) -> IO a) -> (SessionId -> H.Client a) -> IO a
+runHTTP2ClientWith bufferSize host setup client = setup $ withHTTP2 bufferSize run
   where
     run :: H.Config -> SessionId -> IO a
     run cfg = H.run (ClientConfig "https" (strEncode host) 20) cfg . client

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

@@ -12,7 +12,7 @@ import qualified Network.HTTP2.Server as H
 import Network.Socket
 import qualified Network.TLS as T
 import Numeric.Natural (Natural)
-import Simplex.Messaging.Transport (SessionId)
+import Simplex.Messaging.Transport (SessionId, TLS)
 import Simplex.Messaging.Transport.HTTP2
 import Simplex.Messaging.Transport.Server (TransportServerConfig (..), loadSupportedTLSServerParams, runTransportServer)
 
@@ -60,7 +60,11 @@ closeHTTP2Server :: HTTP2Server -> IO ()
 closeHTTP2Server = uninterruptibleCancel . action
 
 runHTTP2Server :: TMVar Bool -> ServiceName -> BufferSize -> T.ServerParams -> TransportServerConfig -> HTTP2ServerFunc -> IO ()
-runHTTP2Server started port bufferSize serverParams transportConfig http2Server =
-  runTransportServer started port serverParams transportConfig $ withHTTP2 bufferSize run
+runHTTP2Server started port bufferSize serverParams transportConfig = runHTTP2ServerWith bufferSize setup
+  where
+    setup = runTransportServer started port serverParams transportConfig
+
+runHTTP2ServerWith :: BufferSize -> ((TLS -> IO ()) -> a) -> (SessionId -> Request -> (Response -> IO ()) -> IO ()) -> a
+runHTTP2ServerWith bufferSize setup http2Server = setup $ withHTTP2 bufferSize run
   where
     run cfg sessId = H.run cfg $ \req _aux sendResp -> http2Server sessId req (`sendResp` [])

src/Simplex/Messaging/Util.hs

@@ -3,13 +3,11 @@
 
 module Simplex.Messaging.Util where
 
-import Control.Concurrent (threadDelay)
 import qualified Control.Exception as E
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Data.Bifunctor (first)
-import qualified Data.ByteString as BW
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Int (Int64)
@@ -21,7 +19,6 @@ import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With)
 import Data.Time (NominalDiffTime)
 import GHC.Conc
-import Numeric (showHex)
 import UnliftIO.Async
 import qualified UnliftIO.Exception as UE
 

tests/AgentTests/SQLiteTests.hs

@@ -6,6 +6,7 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE QuasiQuotes #-}
 {-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 
@@ -13,9 +14,11 @@ module AgentTests.SQLiteTests (storeTests) where
 
 import Control.Concurrent.Async (concurrently_)
 import Control.Concurrent.STM
+import Control.Exception (SomeException)
 import Control.Monad (replicateM_)
 import Crypto.Random (drgNew)
 import Data.ByteString.Char8 (ByteString)
+import Data.List (isInfixOf)
 import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
 import Data.Time
@@ -51,11 +54,14 @@ withStore2 = before connect2 . after (removeStore . fst)
       pure (s1, s2)
 
 createStore :: IO SQLiteStore
-createStore = do
+createStore = createEncryptedStore ""
+
+createEncryptedStore :: String -> IO SQLiteStore
+createEncryptedStore key = do
   -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous
   -- IO operations on multiple similarly named files; error seems to be environment specific
   r <- randomIO :: IO Word32
-  Right st <- createSQLiteStore (testDB <> show r) "" Migrations.app MCError
+  Right st <- createSQLiteStore (testDB <> show r) key Migrations.app MCError
   pure st
 
 removeStore :: SQLiteStore -> IO ()
@@ -104,6 +110,9 @@ storeTests = do
           testCreateRcvMsg
           testCreateSndMsg
           testCreateRcvAndSndMsgs
+  describe "open/close store" $ do
+    it "should close and re-open" testCloseReopenStore
+    it "should close and re-open encrypted store" testCloseReopenEncryptedStore
 
 testConcurrentWrites :: SpecWith (SQLiteStore, SQLiteStore)
 testConcurrentWrites =
@@ -504,3 +513,43 @@ testCreateRcvAndSndMsgs =
       testCreateRcvMsg_ db 2 "rcv_hash_2" connId rcvQueue1 $ mkRcvMsgData (InternalId 4) (InternalRcvId 3) 3 "3" "rcv_hash_3"
       testCreateSndMsg_ db "snd_hash_1" connId $ mkSndMsgData (InternalId 5) (InternalSndId 2) "snd_hash_2"
       testCreateSndMsg_ db "snd_hash_2" connId $ mkSndMsgData (InternalId 6) (InternalSndId 3) "snd_hash_3"
+
+testCloseReopenStore :: IO ()
+testCloseReopenStore = do
+  st <- createStore
+  hasMigrations st
+  closeSQLiteStore st
+  closeSQLiteStore st
+  errorGettingMigrations st
+  openSQLiteStore st ""
+  openSQLiteStore st ""
+  hasMigrations st
+  closeSQLiteStore st
+  errorGettingMigrations st
+  openSQLiteStore st ""
+  hasMigrations st
+
+testCloseReopenEncryptedStore :: IO ()
+testCloseReopenEncryptedStore = do
+  let key = "test_key"
+  st <- createEncryptedStore key
+  hasMigrations st
+  closeSQLiteStore st
+  closeSQLiteStore st
+  errorGettingMigrations st
+  openSQLiteStore st key
+  openSQLiteStore st key
+  hasMigrations st
+  closeSQLiteStore st
+  errorGettingMigrations st
+  openSQLiteStore st key
+  hasMigrations st
+
+getMigrations :: SQLiteStore -> IO Bool
+getMigrations st = not . null <$> withTransaction st (Migrations.getCurrent . DB.conn)
+
+hasMigrations :: SQLiteStore -> Expectation
+hasMigrations st = getMigrations st `shouldReturn` True
+
+errorGettingMigrations :: SQLiteStore -> Expectation
+errorGettingMigrations st = getMigrations st `shouldThrow` \(e :: SomeException) -> "ErrorMisuse" `isInfixOf` show e