From d09adb5cd6deabc5f7f071953345f7373567783c Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com> Date: Mon, 21 Aug 2023 09:06:51 +0100 Subject: [PATCH] agent: wait for db (Revert Revert db error busy treatments (#796) (#808)) (#828) * Revert "Revert "db error busy treatments (#796)" (#808)" This reverts commit 82aec2cd8f7b4033dbf08d5de33ced216f574bbb. * update test * combine pragmas --- simplexmq.cabal | 1 + src/Simplex/Messaging/Agent/Store/SQLite.hs | 82 +++++-------------- .../Messaging/Agent/Store/SQLite/Common.hs | 71 ++++++++++++++++ .../Agent/Store/SQLite/Migrations.hs | 43 +++++----- tests/AgentTests/SchemaDump.hs | 6 +- tests/ServerTests.hs | 4 +- 6 files changed, 118 insertions(+), 89 deletions(-) create mode 100644 src/Simplex/Messaging/Agent/Store/SQLite/Common.hs diff --git a/simplexmq.cabal b/simplexmq.cabal index ad211c8e8..cf02c13c1 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -62,6 +62,7 @@ library Simplex.Messaging.Agent.Server Simplex.Messaging.Agent.Store Simplex.Messaging.Agent.Store.SQLite + Simplex.Messaging.Agent.Store.SQLite.Common Simplex.Messaging.Agent.Store.SQLite.DB Simplex.Messaging.Agent.Store.SQLite.Migrations Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 96ff2f883..cf137d984 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -203,7 +203,6 @@ module Simplex.Messaging.Agent.Store.SQLite -- * utilities withConnection, - withConnection', withTransaction, withTransactionCtx, firstRow, @@ -212,7 +211,6 @@ module Simplex.Messaging.Agent.Store.SQLite ) where -import Control.Concurrent (threadDelay) import Control.Concurrent.STM (stateTVar) import Control.Monad.Except import Crypto.Random (ChaChaDRG, randomBytesGenerate) @@ -235,7 +233,7 @@ import Data.Ord (Down (..)) import Data.Text (Text) import qualified Data.Text as T import Data.Text.Encoding (decodeLatin1, encodeUtf8) -import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, diffUTCTime, getCurrentTime) +import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, getCurrentTime) import Data.Word (Word32) import Database.SQLite.Simple (FromRow (..), NamedParam (..), Only (..), Query (..), SQLError, ToRow (..), field, (:.) (..)) import qualified Database.SQLite.Simple as SQL @@ -252,6 +250,7 @@ import Simplex.FileTransfer.Types import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.RetryInterval (RI2State (..)) import Simplex.Messaging.Agent.Store +import Simplex.Messaging.Agent.Store.SQLite.Common import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB import Simplex.Messaging.Agent.Store.SQLite.Migrations (DownMigration (..), MTRError, Migration (..), MigrationsToRun (..), mtrErrorDescription) import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations @@ -265,25 +264,18 @@ 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, diffToMilliseconds, eitherToMaybe, groupOn, ($>>=), (<$$>)) +import Simplex.Messaging.Util (bshow, eitherToMaybe, groupOn, ($>>=), (<$$>)) 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 (bracket, onException) +import UnliftIO.Exception (onException) import qualified UnliftIO.Exception as E import UnliftIO.STM -- * SQLite Store implementation -data SQLiteStore = SQLiteStore - { dbFilePath :: FilePath, - dbEncrypted :: TVar Bool, - dbConnection :: TMVar DB.Connection, - dbNew :: Bool - } - data MigrationError = MEUpgrade {upMigrations :: [UpMigration]} | MEDowngrade {downMigrations :: [String]} @@ -340,35 +332,35 @@ createSQLiteStore dbFilePath dbKey migrations confirmMigrations = do Left e -> closeSQLiteStore st $> Left e migrateSchema :: SQLiteStore -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError ()) -migrateSchema st migrations confirmMigrations = withConnection' st $ \db -> do - Migrations.initialize db - Migrations.get db migrations >>= \case +migrateSchema st migrations confirmMigrations = do + Migrations.initialize st + Migrations.get st migrations >>= \case Left e -> do when (confirmMigrations == MCConsole) $ confirmOrExit ("Database state error: " <> mtrErrorDescription e) pure . Left $ MigrationError e Right MTRNone -> pure $ Right () Right ms@(MTRUp ums) - | dbNew st -> Migrations.run db ms $> Right () + | dbNew st -> Migrations.run st ms $> Right () | otherwise -> case confirmMigrations of - MCYesUp -> run db ms - MCYesUpDown -> run db ms - MCConsole -> confirm err >> run db ms + MCYesUp -> run ms + MCYesUpDown -> run ms + MCConsole -> confirm err >> run ms MCError -> pure $ Left err where err = MEUpgrade $ map upMigration ums -- "The app has a newer version than the database.\nConfirm to back up and upgrade using these migrations: " <> intercalate ", " (map name ums) Right ms@(MTRDown dms) -> case confirmMigrations of - MCYesUpDown -> run db ms - MCConsole -> confirm err >> run db ms + MCYesUpDown -> run ms + MCConsole -> confirm err >> run ms MCYesUp -> pure $ Left err MCError -> pure $ Left err where err = MEDowngrade $ map downName dms where confirm err = confirmOrExit $ migrationErrorDescription err - run db ms = do + run ms = do let f = dbFilePath st copyFile f (f <> ".bak") - Migrations.run db ms + Migrations.run st ms pure $ Right () confirmOrExit :: String -> IO () @@ -382,9 +374,10 @@ confirmOrExit s = do connectSQLiteStore :: FilePath -> String -> IO SQLiteStore connectSQLiteStore dbFilePath dbKey = do dbNew <- not <$> doesFileExist dbFilePath - dbConnection <- newTMVarIO =<< connectDB dbFilePath dbKey + dbConn <- dbBusyLoop $ connectDB dbFilePath dbKey + dbConnVar <- newTMVarIO dbConn dbEncrypted <- newTVarIO . not $ null dbKey - pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection, dbNew} + pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection = dbConnVar, dbNew} connectDB :: FilePath -> String -> IO DB.Connection connectDB path key = do @@ -398,6 +391,7 @@ connectDB path key = do unless (null key) . exec $ "PRAGMA key = " <> sqlString key <> ";" exec . fromQuery $ [sql| + PRAGMA busy_timeout = 100; PRAGMA foreign_keys = ON; -- PRAGMA trusted_schema = OFF; PRAGMA secure_delete = ON; @@ -445,44 +439,6 @@ handleSQLError err e | SQL.sqlError e == SQL.ErrorConstraint = err | otherwise = SEInternal $ bshow e -withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a -withConnection SQLiteStore {dbConnection} = - bracket - (atomically $ takeTMVar dbConnection) - (atomically . putTMVar dbConnection) - -withConnection' :: SQLiteStore -> (SQL.Connection -> IO a) -> IO a -withConnection' SQLiteStore {dbConnection} a = - bracket - (atomically $ takeTMVar dbConnection) - (atomically . putTMVar dbConnection) - (a . DB.conn) - -withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a -withTransaction = withTransactionCtx Nothing - -withTransactionCtx :: forall a. Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a -withTransactionCtx ctx_ st action = withConnection st $ loop 500 3_000_000 - where - loop :: Int -> Int -> DB.Connection -> IO a - loop t tLim db@DB.Connection {conn} = - transactionWithCtx `E.catch` \(e :: SQLError) -> - if tLim > t && SQL.sqlError e == SQL.ErrorBusy - then do - threadDelay t - loop (t * 9 `div` 8) (tLim - t) db - else E.throwIO e - where - transactionWithCtx = case ctx_ of - Nothing -> SQL.withImmediateTransaction conn (action db) - Just ctx -> do - t1 <- getCurrentTime - r <- SQL.withImmediateTransaction conn (action db) - t2 <- getCurrentTime - putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx - putStrLn $ "withTransactionCtx end :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1) - pure r - createUserRecord :: DB.Connection -> IO UserId createUserRecord db = do DB.execute_ db "INSERT INTO users DEFAULT VALUES" diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Common.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Common.hs new file mode 100644 index 000000000..ef2c688aa --- /dev/null +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Common.hs @@ -0,0 +1,71 @@ +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Simplex.Messaging.Agent.Store.SQLite.Common + ( SQLiteStore (..), + withConnection, + withConnection', + withTransaction, + withTransaction', + withTransactionCtx, + dbBusyLoop, + ) +where + +import Control.Concurrent (threadDelay) +import Data.Time.Clock (diffUTCTime, getCurrentTime) +import Database.SQLite.Simple (SQLError) +import qualified Database.SQLite.Simple as SQL +import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB +import Simplex.Messaging.Util (diffToMilliseconds) +import UnliftIO.Exception (bracket) +import qualified UnliftIO.Exception as E +import UnliftIO.STM + +data SQLiteStore = SQLiteStore + { dbFilePath :: FilePath, + dbEncrypted :: TVar Bool, + dbConnection :: TMVar DB.Connection, + dbNew :: Bool + } + +withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a +withConnection SQLiteStore {dbConnection} = + bracket + (atomically $ takeTMVar dbConnection) + (atomically . putTMVar dbConnection) + +withConnection' :: SQLiteStore -> (SQL.Connection -> IO a) -> IO a +withConnection' st action = withConnection st $ action . DB.conn + +withTransaction :: SQLiteStore -> (DB.Connection -> IO a) -> IO a +withTransaction = withTransactionCtx Nothing + +withTransaction' :: SQLiteStore -> (SQL.Connection -> IO a) -> IO a +withTransaction' st action = withTransaction st $ action . DB.conn + +withTransactionCtx :: Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a +withTransactionCtx ctx_ st action = withConnection st $ dbBusyLoop . transactionWithCtx + where + transactionWithCtx db@DB.Connection {conn} = case ctx_ of + Nothing -> SQL.withImmediateTransaction conn $ action db + Just ctx -> do + t1 <- getCurrentTime + r <- SQL.withImmediateTransaction conn $ action db + t2 <- getCurrentTime + putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx + putStrLn $ "withTransactionCtx end :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1) + pure r + +dbBusyLoop :: forall a. IO a -> IO a +dbBusyLoop action = loop 500 3000000 + where + loop :: Int -> Int -> IO a + loop t tLim = + action `E.catch` \(e :: SQLError) -> + if tLim > t && SQL.sqlError e == SQL.ErrorBusy + then do + threadDelay t + loop (t * 9 `div` 8) (tLim - t) + else E.throwIO e diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs index da97e5d1d..d84d8d2fe 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs @@ -42,6 +42,7 @@ import Database.SQLite.Simple.QQ (sql) import qualified Database.SQLite3 as SQLite3 import GHC.Generics (Generic) import Simplex.Messaging.Agent.Protocol (extraSMPServerHosts) +import Simplex.Messaging.Agent.Store.SQLite.Common 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.M20220322_notifications @@ -107,44 +108,42 @@ app = sortOn name $ map migration schemaMigrations where migration (name, up, down) = Migration {name, up = fromQuery up, down = fromQuery <$> down} -get :: Connection -> [Migration] -> IO (Either MTRError MigrationsToRun) -get db migrations = migrationsToRun migrations <$> getCurrent db +get :: SQLiteStore -> [Migration] -> IO (Either MTRError MigrationsToRun) +get st migrations = migrationsToRun migrations <$> withTransaction' st getCurrent getCurrent :: Connection -> IO [Migration] getCurrent db = map toMigration <$> DB.query_ db "SELECT name, down FROM migrations ORDER BY name ASC;" where toMigration (name, down) = Migration {name, up = "", down} -run :: Connection -> MigrationsToRun -> IO () -run db = \case +run :: SQLiteStore -> MigrationsToRun -> IO () +run st = \case MTRUp [] -> pure () - MTRUp ms -> mapM_ runUp ms >> execSQL "VACUUM;" + MTRUp ms -> mapM_ runUp ms >> withConnection' st (`execSQL` "VACUUM;") MTRDown ms -> mapM_ runDown $ reverse ms MTRNone -> pure () where - runUp Migration {name, up, down} = do - when (name == "m20220811_onion_hosts") updateServers - DB.withImmediateTransaction db $ insert >> execSQL up + runUp Migration {name, up, down} = withTransaction' st $ \db -> do + when (name == "m20220811_onion_hosts") $ updateServers db + insert db >> execSQL db up where - insert = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime - updateServers = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') -> - DB.withImmediateTransaction db $ - let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost) - in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h) - runDown DownMigration {downName, downQuery} = do - DB.withImmediateTransaction db $ do - execSQL downQuery - DB.execute db "DELETE FROM migrations WHERE name = ?" (Only downName) - execSQL = SQLite3.exec $ DB.connectionHandle db + insert db = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime + updateServers db = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') -> + let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost) + in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h) + runDown DownMigration {downName, downQuery} = withTransaction' st $ \db -> do + execSQL db downQuery + DB.execute db "DELETE FROM migrations WHERE name = ?" (Only downName) + execSQL db = SQLite3.exec $ DB.connectionHandle db -initialize :: Connection -> IO () -initialize db = do +initialize :: SQLiteStore -> IO () +initialize st = withTransaction' st $ \db -> do cs :: [Text] <- map fromOnly <$> DB.query_ db "SELECT name FROM pragma_table_info('migrations')" case cs of - [] -> createMigrations + [] -> createMigrations db _ -> when ("down" `notElem` cs) $ DB.execute_ db "ALTER TABLE migrations ADD COLUMN down TEXT" where - createMigrations = + createMigrations db = DB.execute_ db [sql| diff --git a/tests/AgentTests/SchemaDump.hs b/tests/AgentTests/SchemaDump.hs index 2453b7f6d..3bea2188b 100644 --- a/tests/AgentTests/SchemaDump.hs +++ b/tests/AgentTests/SchemaDump.hs @@ -50,14 +50,14 @@ testSchemaMigrations = do putStrLn $ "down migration " <> name m let downMigr = fromJust $ toDownMigration m schema <- getSchema testDB testSchema - withConnection' st (`Migrations.run` MTRUp [m]) + Migrations.run st $ MTRUp [m] schema' <- getSchema testDB testSchema schema' `shouldNotBe` schema - withConnection' st (`Migrations.run` MTRDown [downMigr]) + Migrations.run st $ MTRDown [downMigr] unless (name m `elem` skipComparisonForDownMigrations) $ do schema'' <- getSchema testDB testSchema schema'' `shouldBe` schema - withConnection' st (`Migrations.run` MTRUp [m]) + Migrations.run st $ MTRUp [m] schema''' <- getSchema testDB testSchema schema''' `shouldBe` schema' diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 5c995ce6f..7a3402f61 100644 --- a/tests/ServerTests.hs +++ b/tests/ServerTests.hs @@ -558,7 +558,9 @@ testWithStoreLog at@(ATransport t) = (sId2, rId2, rKey2, dhShared2) <- createAndSecureQueue h sPub2 atomically $ writeTVar senderId2 sId2 - Resp "cdab" _ OK <- signSendRecv h sKey2 ("cdab", sId2, _SEND "hello too") + signSendRecv h sKey2 ("cdab", sId2, _SEND "hello too") >>= \case + Resp "cdab" _ OK -> pure () + r -> print $ "unexpected response " <> show r Resp "" _ (Msg mId2 msg2) <- tGet1 h (decryptMsgV3 dhShared2 mId2 msg2, Right "hello too") #== "delivered from queue 2"