diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,19 @@ +# 1.4.2.1 + +## Fixes + +- Discard pooled connections after driver errors (#55) + +# 1.4 + +- Migrated to `hasql-1.10` +- Updated connection settings API to use monoid-based `Settings` instead of list-based `[Setting]` +- Updated error types to use `Hasql.Errors` module instead of `Hasql.Session` and `Hasql.Connection` +- Changed session execution API from `Session.run session connection` to `Connection.use connection session` +- Error handling now uses `ConnectionSessionError` for connection-level issues instead of `ClientError` +- Updated statement construction to use `Statement.preparable` and `Statement.unpreparable` instead of direct constructor +- Hid the `Defaults` module from the public API + # 1.3 - Adapt to the new settings model of `hasql-1.9` diff --git a/hasql-pool.cabal b/hasql-pool.cabal --- a/hasql-pool.cabal +++ b/hasql-pool.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: hasql-pool -version: 1.3.0.4 +version: 1.4.2.1 category: Hasql, Database, PostgreSQL synopsis: Pool of connections for Hasql homepage: https://github.com/nikita-volkov/hasql-pool @@ -68,17 +68,14 @@ library import: base-settings hs-source-dirs: - src/library/exposed - src/library/other + src/library/ - -- cabal-gild: discover src/library/exposed exposed-modules: Hasql.Pool Hasql.Pool.Config Hasql.Pool.Config.Defaults Hasql.Pool.Observation - -- cabal-gild: discover src/library/other other-modules: Hasql.Pool.Config.Config Hasql.Pool.Config.Setting @@ -88,7 +85,7 @@ build-depends: base >=4.11 && <5, bytestring >=0.10 && <0.14, - hasql >=1.9 && <1.10, + hasql >=1.10 && <1.11, stm >=2.5 && <3, text >=1.2 && <3, time >=1.9 && <2, @@ -123,8 +120,9 @@ hasql, hasql-pool, hspec >=2.6 && <3, + postgresql-libpq >=0.10 && <0.12, random >=1.2 && <2, rerebase >=1.15 && <2, - testcontainers-postgresql >=0.0.2 && <0.1, + testcontainers-postgresql >=0.2 && <0.3, text-builder >=1 && <1.1, tuple ^>=0.3.0.2, diff --git a/src/integration-tests/Helpers/Hooks.hs b/src/integration-tests/Helpers/Hooks.hs --- a/src/integration-tests/Helpers/Hooks.hs +++ b/src/integration-tests/Helpers/Hooks.hs @@ -13,7 +13,7 @@ TestcontainersPostgresql.run TestcontainersPostgresql.Config { forwardLogs = False, - distro = TestcontainersPostgresql.Distro17, + tagName = "postgres:17", auth = TestcontainersPostgresql.TrustAuth } (\(host, portInt) -> handler (host, fromIntegral portInt)) diff --git a/src/integration-tests/Helpers/Scripts.hs b/src/integration-tests/Helpers/Scripts.hs --- a/src/integration-tests/Helpers/Scripts.hs +++ b/src/integration-tests/Helpers/Scripts.hs @@ -1,8 +1,6 @@ module Helpers.Scripts where -import Hasql.Connection.Setting qualified as Connection.Setting -import Hasql.Connection.Setting.Connection qualified as Connection.Setting.Connection -import Hasql.Connection.Setting.Connection.Param qualified as Connection.Setting.Connection.Param +import Hasql.Connection.Settings qualified as Connection.Settings import Hasql.Pool qualified as Pool import Hasql.Pool.Config qualified as Config import System.Random.Stateful qualified as Random @@ -24,17 +22,14 @@ Config.agingTimeout maxLifetime, Config.idlenessTimeout maxIdletime, Config.staticConnectionSettings - [ Connection.Setting.connection - ( Connection.Setting.Connection.params - [ Connection.Setting.Connection.Param.host host, - Connection.Setting.Connection.Param.port (fromIntegral port), - Connection.Setting.Connection.Param.user "postgres", - Connection.Setting.Connection.Param.password "", - Connection.Setting.Connection.Param.dbname "postgres", - Connection.Setting.Connection.Param.other "application_name" appName - ] - ) - ] + ( mconcat + [ Connection.Settings.hostAndPort host (fromIntegral port), + Connection.Settings.user "postgres", + Connection.Settings.password "", + Connection.Settings.dbname "postgres", + Connection.Settings.applicationName appName + ] + ) ] ) ) @@ -59,6 +54,7 @@ $ mconcat $ [ TextBuilder.text prefix, TextBuilder.decimal uniqueNum1, + "-", TextBuilder.decimal uniqueNum2 ] diff --git a/src/integration-tests/Helpers/Sessions.hs b/src/integration-tests/Helpers/Sessions.hs --- a/src/integration-tests/Helpers/Sessions.hs +++ b/src/integration-tests/Helpers/Sessions.hs @@ -9,7 +9,7 @@ where import Data.Tuple.All -import Hasql.Connection qualified as Connection +import Database.PostgreSQL.LibPQ qualified as Pq import Hasql.Decoders qualified as Decoders import Hasql.Encoders qualified as Encoders import Hasql.Session qualified as Session @@ -20,7 +20,7 @@ selectOne = Session.statement () statement where - statement = Statement.Statement "SELECT 1" Encoders.noParams decoder True + statement = Statement.preparable "SELECT 1::int8" Encoders.noParams decoder decoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)) badQuery :: Session.Session () @@ -28,19 +28,20 @@ Session.statement () statement where statement = - Statement.Statement "zzz" Encoders.noParams Decoders.noResult True + Statement.preparable "zzz" Encoders.noParams Decoders.noResult closeConn :: Session.Session () -closeConn = do - conn <- ask - liftIO $ Connection.release conn +closeConn = + Session.onLibpqConnection \conn -> do + Pq.finish conn + pure (Right (), conn) setSetting :: Text -> Text -> Session.Session () setSetting name value = do Session.statement (name, value) statement where statement = - Statement.Statement "SELECT set_config($1, $2, false)" encoder Decoders.noResult True + Statement.preparable "SELECT set_config($1, $2, false)" encoder Decoders.noResult encoder = mconcat [ sel1 >$< Encoders.param (Encoders.nonNullable Encoders.text), @@ -51,7 +52,7 @@ getSetting name = do Session.statement name statement where - statement = Statement.Statement "SELECT current_setting($1, true)" encoder decoder True + statement = Statement.preparable "SELECT current_setting($1, true)" encoder decoder encoder = Encoders.param (Encoders.nonNullable Encoders.text) decoder = Decoders.singleRow (Decoders.column (Decoders.nullable Decoders.text)) @@ -59,6 +60,6 @@ countConnections appName = do Session.statement appName statement where - statement = Statement.Statement "SELECT count(*) FROM pg_stat_activity WHERE application_name = $1" encoder decoder True + statement = Statement.preparable "SELECT count(*) FROM pg_stat_activity WHERE application_name = $1" encoder decoder encoder = Encoders.param (Encoders.nonNullable Encoders.text) decoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)) diff --git a/src/integration-tests/Specs/BySubject/UsageError/SessionSpec.hs b/src/integration-tests/Specs/BySubject/UsageError/SessionSpec.hs --- a/src/integration-tests/Specs/BySubject/UsageError/SessionSpec.hs +++ b/src/integration-tests/Specs/BySubject/UsageError/SessionSpec.hs @@ -1,7 +1,6 @@ module Specs.BySubject.UsageError.SessionSpec where import Hasql.Pool -import Hasql.Session qualified as Session import Helpers.Scripts qualified as Scripts import Helpers.Sessions qualified as Sessions import Test.Hspec @@ -9,9 +8,9 @@ spec :: SpecWith Scripts.ScopeParams spec = do - it "Simulation of connection error works" \scopeParams -> + it "Bad SQL query triggers error" \scopeParams -> Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - res <- use pool $ Sessions.closeConn >> Sessions.selectOne + res <- use pool Sessions.badQuery shouldSatisfy res $ \case - Left (SessionUsageError (Session.QueryError _ _ (Session.ClientError _))) -> True + Left (SessionUsageError _) -> True _ -> False diff --git a/src/integration-tests/Specs/BySubject/UseSpec.hs b/src/integration-tests/Specs/BySubject/UseSpec.hs --- a/src/integration-tests/Specs/BySubject/UseSpec.hs +++ b/src/integration-tests/Specs/BySubject/UseSpec.hs @@ -1,6 +1,12 @@ module Specs.BySubject.UseSpec where +import Data.Text qualified as Text +import Hasql.Decoders qualified as Decoders +import Hasql.Encoders qualified as Encoders +import Hasql.Errors qualified as Errors import Hasql.Pool +import Hasql.Session qualified as Session +import Hasql.Statement qualified as Statement import Helpers.Scripts qualified as Scripts import Helpers.Sessions qualified as Sessions import Test.Hspec @@ -21,6 +27,16 @@ res <- use pool $ Sessions.selectOne shouldSatisfy res $ isRight + it "Driver errors cause eviction of connection" \scopeParams -> do + settingName <- Scripts.generateVarname + Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do + use pool (Sessions.setSetting settingName "present") `shouldReturn` Right () + result <- use pool driverError + result `shouldSatisfy` \case + Left (SessionUsageError (Errors.DriverSessionError _)) -> True + _ -> False + use pool (Sessions.getSetting settingName) `shouldReturn` Right Nothing + it "Connection gets returned to the pool after normal use" \scopeParams -> Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do _ <- use pool $ Sessions.selectOne @@ -38,3 +54,46 @@ _ <- use pool $ Sessions.badQuery res <- use pool $ Sessions.selectOne shouldSatisfy res $ isRight + + it "Cached type errors cause eviction of connection" \scopeParams -> do + typeName <- Text.replace "-" "_" <$> Scripts.generateName "cached_type_" + Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do + use pool (Session.script (createTypeSql typeName)) `shouldReturn` Right () + use pool (roundtripEnum typeName "ok") `shouldReturn` Right "ok" + use pool (Session.script (recreateTypeSql typeName)) `shouldReturn` Right () + res <- use pool (roundtripEnum typeName "ok") + shouldSatisfy res \case + Left (SessionUsageError _) -> True + _ -> False + use pool (roundtripEnum typeName "ok") `shouldReturn` Right "ok" + +quoteIdentifier :: Text -> Text +quoteIdentifier identifier = + "\"" <> Text.replace "\"" "\"\"" identifier <> "\"" + +createTypeSql :: Text -> Text +createTypeSql typeName = + "create type " <> quotedTypeName <> " as enum ('sad', 'ok', 'happy')" + where + quotedTypeName = quoteIdentifier typeName + +recreateTypeSql :: Text -> Text +recreateTypeSql typeName = + "drop type " <> quotedTypeName <> "; create type " <> quotedTypeName <> " as enum ('sad', 'ok', 'happy')" + where + quotedTypeName = quoteIdentifier typeName + +roundtripEnum :: Text -> Text -> Session.Session Text +roundtripEnum typeName value = + Session.statement value statement + where + statement = + Statement.preparable + ("select $1 :: " <> quoteIdentifier typeName) + (Encoders.param (Encoders.nonNullable (Encoders.enum Nothing typeName id))) + (Decoders.singleRow (Decoders.column (Decoders.nonNullable (Decoders.enum Nothing typeName Just)))) + +driverError :: Session.Session () +driverError = + Session.onLibpqConnection \connection -> + pure (Left (Errors.DriverSessionError "synthetic driver error"), connection) diff --git a/src/library/Hasql/Pool.hs b/src/library/Hasql/Pool.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool.hs @@ -0,0 +1,259 @@ +module Hasql.Pool + ( -- * Pool + Pool, + acquire, + use, + release, + + -- * Errors + UsageError (..), + ) +where + +import Data.UUID.V4 qualified as Uuid +import Hasql.Connection (Connection) +import Hasql.Connection qualified as Connection +import Hasql.Connection.Settings qualified as Connection.Settings +import Hasql.Errors qualified as Errors +import Hasql.Pool.Config.Config qualified as Config +import Hasql.Pool.Observation +import Hasql.Pool.Prelude +import Hasql.Pool.SessionErrorDestructors qualified as ErrorsDestruction +import Hasql.Session qualified as Session + +-- | A connection tagged with metadata. +data Entry = Entry + { entryConnection :: Connection, + entryCreationTimeNSec :: Word64, + entryUseTimeNSec :: Word64, + entryId :: UUID + } + +entryIsAged :: Word64 -> Word64 -> Entry -> Bool +entryIsAged maxLifetime now Entry {..} = + now > entryCreationTimeNSec + maxLifetime + +entryIsIdle :: Word64 -> Word64 -> Entry -> Bool +entryIsIdle maxIdletime now Entry {..} = + now > entryUseTimeNSec + maxIdletime + +-- | Pool of connections to DB. +data Pool = Pool + { -- | Pool size. + poolSize :: Int, + -- | Connection settings. + poolFetchConnectionSettings :: IO Connection.Settings.Settings, + -- | Acquisition timeout, in microseconds. + poolAcquisitionTimeout :: Int, + -- | Maximal connection lifetime, in nanoseconds. + poolMaxLifetime :: Word64, + -- | Maximal connection idle time, in nanoseconds. + poolMaxIdletime :: Word64, + -- | Avail connections. + poolConnectionQueue :: TQueue Entry, + -- | Remaining capacity. + -- The pool size limits the sum of poolCapacity, the length + -- of poolConnectionQueue and the number of in-flight + -- connections. + poolCapacity :: TVar Int, + -- | Whether to return a connection to the pool. + poolReuseVar :: TVar (TVar Bool), + -- | To stop the manager thread via garbage collection. + poolReaperRef :: IORef (), + -- | Action for reporting the observations. + poolObserver :: Observation -> IO (), + -- | Initial session to execute upon every established connection. + poolInitSession :: Session.Session () + } + +-- | Create a connection-pool. +-- +-- No connections actually get established by this function. It is delegated +-- to 'use'. +-- +-- If you want to ensure that the pool connects fine at the initialization phase, just run 'use' with an empty session (@pure ()@) and check for errors. +acquire :: Config.Config -> IO Pool +acquire config = do + connectionQueue <- newTQueueIO + capVar <- newTVarIO (Config.size config) + reuseVar <- newTVarIO =<< newTVarIO True + reaperRef <- newIORef () + + managerTid <- forkIOWithUnmask $ \unmask -> unmask $ forever $ do + threadDelay 1000000 + now <- getMonotonicTimeNSec + join . atomically $ do + entries <- flushTQueue connectionQueue + let (agedEntries, unagedEntries) = partition (entryIsAged agingTimeoutNanos now) entries + (idleEntries, liveEntries) = partition (entryIsIdle agingTimeoutNanos now) unagedEntries + traverse_ (writeTQueue connectionQueue) liveEntries + return $ do + forM_ agedEntries $ \entry -> do + Connection.release (entryConnection entry) + atomically $ modifyTVar' capVar succ + (Config.observationHandler config) (ConnectionObservation (entryId entry) (TerminatedConnectionStatus AgingConnectionTerminationReason)) + forM_ idleEntries $ \entry -> do + Connection.release (entryConnection entry) + atomically $ modifyTVar' capVar succ + (Config.observationHandler config) (ConnectionObservation (entryId entry) (TerminatedConnectionStatus IdlenessConnectionTerminationReason)) + + void . mkWeakIORef reaperRef $ do + -- When the pool goes out of scope, stop the manager. + killThread managerTid + + return $ Pool (Config.size config) (Config.connectionSettingsProvider config) acqTimeoutMicros agingTimeoutNanos maxIdletimeNanos connectionQueue capVar reuseVar reaperRef (Config.observationHandler config) (Config.initSession config) + where + acqTimeoutMicros = + div (fromIntegral (diffTimeToPicoseconds (Config.acquisitionTimeout config))) 1_000_000 + agingTimeoutNanos = + div (fromIntegral (diffTimeToPicoseconds (Config.agingTimeout config))) 1_000 + maxIdletimeNanos = + div (fromIntegral (diffTimeToPicoseconds (Config.idlenessTimeout config))) 1_000 + +-- | Release all the idle connections in the pool, and mark the in-use connections +-- to be released after use. Any connections acquired after the call will be +-- freshly established. +-- +-- The pool remains usable after this action. +-- So you can use this function to reset the connections in the pool. +-- Naturally, you can also use it to release the resources. +release :: Pool -> IO () +release Pool {..} = + join . atomically $ do + prevReuse <- readTVar poolReuseVar + writeTVar prevReuse False + newReuse <- newTVar True + writeTVar poolReuseVar newReuse + entries <- flushTQueue poolConnectionQueue + return $ forM_ entries $ \entry -> do + Connection.release (entryConnection entry) + atomically $ modifyTVar' poolCapacity succ + poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason)) + +-- | Use a connection from the pool to run a session and return the connection +-- to the pool, when finished. +-- +-- Session failing with a 'Session.ClientError' gets interpreted as a loss of +-- connection. In such case the connection does not get returned to the pool +-- and a slot gets freed up for a new connection to be established the next +-- time one is needed. The error still gets returned from this function. +-- +-- __Warning:__ Due to the mechanism mentioned above you should avoid intercepting this error type from within sessions. +use :: Pool -> Session.Session a -> IO (Either UsageError a) +use Pool {..} sess = do + timeout <- do + delay <- registerDelay poolAcquisitionTimeout + return $ readTVar delay + join . atomically $ do + reuseVar <- readTVar poolReuseVar + asum + [ readTQueue poolConnectionQueue <&> onConn reuseVar, + do + capVal <- readTVar poolCapacity + if capVal > 0 + then do + writeTVar poolCapacity $! pred capVal + return $ onNewConn reuseVar + else retry, + do + timedOut <- timeout + if timedOut + then return . return . Left $ AcquisitionTimeoutUsageError + else retry + ] + where + onNewConn reuseVar = do + settings <- poolFetchConnectionSettings + now <- getMonotonicTimeNSec + id <- Uuid.nextRandom + poolObserver (ConnectionObservation id ConnectingConnectionStatus) + Connection.acquire settings >>= \case + Left connErr -> do + let connErrText = case connErr of + Errors.NetworkingConnectionError details -> Just details + Errors.AuthenticationConnectionError details -> Just details + Errors.CompatibilityConnectionError details -> Just details + Errors.OtherConnectionError details -> if details == "" then Nothing else Just details + poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason connErrText))) + atomically $ modifyTVar' poolCapacity succ + return $ Left $ ConnectionUsageError connErr + Right connection -> do + Connection.use connection poolInitSession >>= \case + Left err -> do + Connection.release connection + ErrorsDestruction.reset + ( \details -> do + poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (Just details)))) + ) + (poolObserver (ConnectionObservation id (TerminatedConnectionStatus (InitializationErrorTerminationReason err)))) + err + return $ Left $ SessionUsageError err + Right () -> do + poolObserver (ConnectionObservation id (ReadyForUseConnectionStatus EstablishedConnectionReadyForUseReason)) + onLiveConn reuseVar (Entry connection now now id) + + onConn reuseVar entry = do + now <- getMonotonicTimeNSec + if entryIsAged poolMaxLifetime now entry + then do + Connection.release (entryConnection entry) + poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus AgingConnectionTerminationReason)) + onNewConn reuseVar + else + if entryIsIdle poolMaxIdletime now entry + then do + Connection.release (entryConnection entry) + poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus IdlenessConnectionTerminationReason)) + onNewConn reuseVar + else do + onLiveConn reuseVar entry {entryUseTimeNSec = now} + + onLiveConn reuseVar entry = do + poolObserver (ConnectionObservation (entryId entry) InUseConnectionStatus) + sessRes <- try @SomeException (Connection.use (entryConnection entry) sess) + + case sessRes of + Left exc -> do + returnConn + throwIO exc + Right (Left err) -> + if ErrorsDestruction.requiresConnectionDiscard err + then do + Connection.release (entryConnection entry) + atomically $ modifyTVar' poolCapacity succ + poolObserver + ( ConnectionObservation + (entryId entry) + (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (ErrorsDestruction.discardDetails err))) + ) + return $ Left $ SessionUsageError err + else do + returnConn + poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus (SessionFailedConnectionReadyForUseReason err))) + return $ Left $ SessionUsageError err + Right (Right res) -> do + returnConn + poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus SessionSucceededConnectionReadyForUseReason)) + return $ Right res + where + returnConn = + join . atomically $ do + reuse <- readTVar reuseVar + if reuse + then writeTQueue poolConnectionQueue entry $> return () + else return $ do + Connection.release (entryConnection entry) + atomically $ modifyTVar' poolCapacity succ + poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason)) + +-- | Union over all errors that 'use' can result in. +data UsageError + = -- | Attempt to establish a connection failed. + ConnectionUsageError Errors.ConnectionError + | -- | Session execution failed. + SessionUsageError Errors.SessionError + | -- | Timeout acquiring a connection. + AcquisitionTimeoutUsageError + deriving (Show, Eq) + +instance Exception UsageError diff --git a/src/library/Hasql/Pool/Config.hs b/src/library/Hasql/Pool/Config.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Config.hs @@ -0,0 +1,25 @@ +-- | DSL for construction of configs. +module Hasql.Pool.Config + ( Config.Config, + settings, + Setting.Setting, + Setting.size, + Setting.acquisitionTimeout, + Setting.agingTimeout, + Setting.idlenessTimeout, + Setting.staticConnectionSettings, + Setting.dynamicConnectionSettings, + Setting.observationHandler, + Setting.initSession, + ) +where + +import Hasql.Pool.Config.Config qualified as Config +import Hasql.Pool.Config.Setting qualified as Setting +import Hasql.Pool.Prelude + +-- | Compile config from a list of settings. +-- Latter settings override the preceding in cases of conflicts. +settings :: [Setting.Setting] -> Config.Config +settings = + foldr ($) Config.defaults . fmap Setting.apply diff --git a/src/library/Hasql/Pool/Config/Config.hs b/src/library/Hasql/Pool/Config/Config.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Config/Config.hs @@ -0,0 +1,31 @@ +module Hasql.Pool.Config.Config where + +import Hasql.Connection.Settings qualified as Connection.Settings +import Hasql.Pool.Config.Defaults qualified as Defaults +import Hasql.Pool.Observation (Observation) +import Hasql.Pool.Prelude +import Hasql.Session qualified as Session + +-- | Configuration for Hasql connection pool. +data Config = Config + { size :: Int, + acquisitionTimeout :: DiffTime, + agingTimeout :: DiffTime, + idlenessTimeout :: DiffTime, + connectionSettingsProvider :: IO Connection.Settings.Settings, + observationHandler :: Observation -> IO (), + initSession :: Session.Session () + } + +-- | Reasonable defaults, which can be built upon. +defaults :: Config +defaults = + Config + { size = Defaults.size, + acquisitionTimeout = Defaults.acquisitionTimeout, + agingTimeout = Defaults.agingTimeout, + idlenessTimeout = Defaults.idlenessTimeout, + connectionSettingsProvider = Defaults.dynamicConnectionSettings, + observationHandler = Defaults.observationHandler, + initSession = Defaults.initSession + } diff --git a/src/library/Hasql/Pool/Config/Defaults.hs b/src/library/Hasql/Pool/Config/Defaults.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Config/Defaults.hs @@ -0,0 +1,47 @@ +module Hasql.Pool.Config.Defaults where + +import Hasql.Connection.Settings qualified as Connection.Settings +import Hasql.Pool.Observation (Observation) +import Hasql.Pool.Prelude +import Hasql.Session qualified as Session + +-- | +-- 3 connections. +size :: Int +size = 3 + +-- | +-- 10 seconds. +acquisitionTimeout :: DiffTime +acquisitionTimeout = 10 + +-- | +-- 1 day. +agingTimeout :: DiffTime +agingTimeout = 60 * 60 * 24 + +-- | +-- 10 minutes. +idlenessTimeout :: DiffTime +idlenessTimeout = 60 * 10 + +-- | +-- > "postgresql://postgres:postgres@localhost:5432/postgres" +staticConnectionSettings :: Connection.Settings.Settings +staticConnectionSettings = + "postgresql://postgres:postgres@localhost:5432/postgres" + +-- | +-- > pure "postgresql://postgres:postgres@localhost:5432/postgres" +dynamicConnectionSettings :: IO Connection.Settings.Settings +dynamicConnectionSettings = pure staticConnectionSettings + +-- | +-- > const (pure ()) +observationHandler :: Observation -> IO () +observationHandler = const (pure ()) + +-- | +-- > pure () +initSession :: Session.Session () +initSession = pure () diff --git a/src/library/Hasql/Pool/Config/Setting.hs b/src/library/Hasql/Pool/Config/Setting.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Config/Setting.hs @@ -0,0 +1,97 @@ +module Hasql.Pool.Config.Setting where + +import Hasql.Connection.Settings qualified as Connection.Settings +import Hasql.Pool.Config.Config (Config) +import Hasql.Pool.Config.Config qualified as Config +import Hasql.Pool.Observation (Observation) +import Hasql.Pool.Prelude +import Hasql.Session qualified as Session + +apply :: Setting -> Config -> Config +apply (Setting run) = run + +-- | A single setting of a config. +newtype Setting + = Setting (Config -> Config) + +-- | Pool size. +-- +-- 3 by default. +size :: Int -> Setting +size x = + Setting (\config -> config {Config.size = x}) + +-- | Connection acquisition timeout. +-- +-- 10 seconds by default. +acquisitionTimeout :: DiffTime -> Setting +acquisitionTimeout x = + Setting (\config -> config {Config.acquisitionTimeout = x}) + +-- | Maximal connection lifetime. +-- +-- Determines how long is available for reuse. +-- After the timeout passes and an active session is finished the connection will be closed releasing a slot in the pool for a fresh connection to be established. +-- +-- This is useful as a healthy measure for resetting the server-side caches. +-- +-- 1 day by default. +agingTimeout :: DiffTime -> Setting +agingTimeout x = + Setting (\config -> config {Config.agingTimeout = x}) + +-- | Maximal connection idle time. +-- +-- How long to keep a connection open when it's not being used. +-- +-- 10 minutes by default. +idlenessTimeout :: DiffTime -> Setting +idlenessTimeout x = + Setting (\config -> config {Config.idlenessTimeout = x}) + +-- | Connection string. +-- +-- By default it is: +-- +-- > "postgresql://postgres:postgres@localhost:5432/postgres" +staticConnectionSettings :: Connection.Settings.Settings -> Setting +staticConnectionSettings x = + Setting (\config -> config {Config.connectionSettingsProvider = pure x}) + +-- | Action providing connection settings. +-- +-- Gets used each time a connection gets established by the pool. +-- This may be useful for some authorization models. +-- +-- By default it is: +-- +-- > pure "postgresql://postgres:postgres@localhost:5432/postgres" +dynamicConnectionSettings :: IO Connection.Settings.Settings -> Setting +dynamicConnectionSettings x = + Setting (\config -> config {Config.connectionSettingsProvider = x}) + +-- | Observation handler. +-- +-- Typically it's used for monitoring the state of the pool via metrics and logging. +-- +-- If the provided action is not lightweight, it's recommended to use intermediate bufferring via channels like TBQueue to avoid occupying the pool management thread for too long. +-- E.g., if the action is @'atomically' . 'writeTBQueue' yourQueue@, then reading from it and processing can be done on a separate thread. +-- +-- By default it is: +-- +-- > const (pure ()) +observationHandler :: (Observation -> IO ()) -> Setting +observationHandler x = + Setting (\config -> config {Config.observationHandler = x}) + +-- | Initial session. +-- +-- Gets executed on every connection upon acquisition. +-- Lets you specify the connection-wide settings. +-- +-- E.g., you can set the search path for all the sessions executed by the pool by executing the following: +-- +-- > initSession (Session.sql "SET search_path TO schema1, schema2, public;") +initSession :: Session.Session () -> Setting +initSession x = + Setting (\config -> config {Config.initSession = x}) diff --git a/src/library/Hasql/Pool/Observation.hs b/src/library/Hasql/Pool/Observation.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Observation.hs @@ -0,0 +1,64 @@ +-- | Interface for processing observations of the status of the pool. +-- +-- Provides a flexible mechanism for monitoring the healthiness of the pool via logs and metrics without any opinionated choices on the actual monitoring technologies. +-- Specific interpreters are encouraged to be created as extension libraries. +module Hasql.Pool.Observation where + +import Hasql.Errors qualified as Errors +import Hasql.Pool.Prelude + +-- | An observation of a change of the state of a pool. +data Observation + = -- | Status of one of the pool's connections has changed. + ConnectionObservation + -- | Generated connection ID. + -- For grouping the observations by one connection. + UUID + -- | Status that the connection has entered. + ConnectionStatus + deriving (Show, Eq) + +-- | Status of a connection. +-- +-- <> +data ConnectionStatus + = -- | Connection is being established. + -- + -- This is the initial status of every connection. + ConnectingConnectionStatus + | -- | Connection is established and not occupied. + ReadyForUseConnectionStatus ConnectionReadyForUseReason + | -- | Is being used by some session. + -- + -- After it's done the status will transition to 'ReadyForUseConnectionStatus' or 'TerminatedConnectionStatus'. + InUseConnectionStatus + | -- | Connection terminated. + TerminatedConnectionStatus ConnectionTerminationReason + deriving (Show, Eq) + +data ConnectionReadyForUseReason + = -- | Connection just got established. + EstablishedConnectionReadyForUseReason + | -- | Session execution ended with a failure that does not require a connection reset. + SessionFailedConnectionReadyForUseReason Errors.SessionError + | -- | Session execution ended with success. + SessionSucceededConnectionReadyForUseReason + deriving (Show, Eq) + +-- | Explanation of why a connection was terminated. +data ConnectionTerminationReason + = -- | The age timeout of the connection has passed. + AgingConnectionTerminationReason + | -- | The timeout of how long a connection may remain idle in the pool has passed. + IdlenessConnectionTerminationReason + | -- | The connection became unusable and had to be discarded. + -- + -- This includes connectivity issues with the server as well as fatal + -- session errors that invalidate the connection's prepared statement or + -- type caches. + NetworkErrorConnectionTerminationReason (Maybe Text) + | -- | User has invoked the 'Hasql.Pool.release' procedure. + ReleaseConnectionTerminationReason + | -- | Initialization session failure. + InitializationErrorTerminationReason Errors.SessionError + deriving (Show, Eq) diff --git a/src/library/Hasql/Pool/Prelude.hs b/src/library/Hasql/Pool/Prelude.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Prelude.hs @@ -0,0 +1,75 @@ +module Hasql.Pool.Prelude + ( module Exports, + ) +where + +import Control.Applicative as Exports hiding (WrappedArrow (..)) +import Control.Arrow as Exports hiding (first, second) +import Control.Category as Exports +import Control.Concurrent as Exports +import Control.Concurrent.STM as Exports hiding (orElse) +import Control.Exception as Exports +import Control.Monad as Exports hiding (fail, forM, forM_, mapM, mapM_, msum, sequence, sequence_) +import Control.Monad.Fail as Exports +import Control.Monad.Fix as Exports hiding (fix) +import Control.Monad.IO.Class as Exports +import Control.Monad.ST as Exports +import Data.Bifunctor as Exports +import Data.Bits as Exports +import Data.Bool as Exports +import Data.ByteString as Exports (ByteString) +import Data.Char as Exports +import Data.Coerce as Exports +import Data.Complex as Exports +import Data.Data as Exports +import Data.Dynamic as Exports +import Data.Either as Exports +import Data.Fixed as Exports +import Data.Foldable as Exports hiding (toList) +import Data.Function as Exports hiding (id, (.)) +import Data.Functor as Exports hiding (unzip) +import Data.Functor.Compose as Exports +import Data.IORef as Exports +import Data.Int as Exports +import Data.Ix as Exports +import Data.List as Exports hiding (all, and, any, concat, concatMap, elem, find, foldl, foldl', foldl1, foldr, foldr1, isSubsequenceOf, mapAccumL, mapAccumR, maximum, maximumBy, minimum, minimumBy, notElem, or, product, sortOn, sum, uncons) +import Data.List.NonEmpty as Exports (NonEmpty (..)) +import Data.Maybe as Exports +import Data.Monoid as Exports hiding (Alt) +import Data.Ord as Exports +import Data.Proxy as Exports +import Data.Ratio as Exports +import Data.STRef as Exports +import Data.String as Exports +import Data.Text as Exports (Text) +import Data.Time as Exports +import Data.Traversable as Exports +import Data.Tuple as Exports +import Data.UUID as Exports (UUID) +import Data.Unique as Exports +import Data.Version as Exports +import Data.Void as Exports +import Data.Word as Exports +import Debug.Trace as Exports +import Foreign.ForeignPtr as Exports +import Foreign.Ptr as Exports +import Foreign.StablePtr as Exports +import Foreign.Storable as Exports +import GHC.Clock as Exports (getMonotonicTimeNSec) +import GHC.Conc as Exports hiding (orElse, threadWaitRead, threadWaitReadSTM, threadWaitWrite, threadWaitWriteSTM, withMVar) +import GHC.Exts as Exports (IsList (..), groupWith, inline, lazy, sortWith) +import GHC.Generics as Exports (Generic) +import GHC.IO.Exception as Exports +import Numeric as Exports +import System.Environment as Exports +import System.Exit as Exports +import System.IO as Exports (Handle, hClose) +import System.IO.Error as Exports +import System.IO.Unsafe as Exports +import System.Mem as Exports +import System.Mem.StableName as Exports +import System.Timeout as Exports +import Text.Printf as Exports (hPrintf, printf) +import Text.Read as Exports (Read (..), readEither, readMaybe) +import Unsafe.Coerce as Exports +import Prelude as Exports hiding (all, and, any, concat, concatMap, elem, fail, foldl, foldl1, foldr, foldr1, id, mapM, mapM_, maximum, minimum, notElem, or, product, sequence, sequence_, sum, (.)) diff --git a/src/library/Hasql/Pool/SessionErrorDestructors.hs b/src/library/Hasql/Pool/SessionErrorDestructors.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/SessionErrorDestructors.hs @@ -0,0 +1,36 @@ +module Hasql.Pool.SessionErrorDestructors where + +import Hasql.Errors qualified as Errors +import Hasql.Pool.Prelude + +reset :: (Text -> x) -> x -> Errors.SessionError -> x +reset onReset onNoReset = \case + Errors.ConnectionSessionError details -> onReset details + _ -> onNoReset + +requiresConnectionDiscard :: Errors.SessionError -> Bool +requiresConnectionDiscard = \case + Errors.ConnectionSessionError {} -> True + Errors.MissingTypesSessionError {} -> True + Errors.ScriptSessionError _ serverError -> isStaleServerError serverError + Errors.StatementSessionError _ _ _ _ _ statementError -> statementRequiresConnectionDiscard statementError + -- Driver errors indicate that Hasql or the server left the connection in an + -- unexpected state. In particular, Hasql closes the libpq connection when + -- cleanup after an interruption fails, so it must not be reused by the pool. + Errors.DriverSessionError {} -> True + +discardDetails :: Errors.SessionError -> Maybe Text +discardDetails err = + if requiresConnectionDiscard err + then Just $ Errors.toMessage err + else Nothing + +statementRequiresConnectionDiscard :: Errors.StatementError -> Bool +statementRequiresConnectionDiscard = \case + Errors.ServerStatementError serverError -> isStaleServerError serverError + Errors.UnexpectedColumnTypeStatementError {} -> True + _ -> False + +isStaleServerError :: Errors.ServerError -> Bool +isStaleServerError (Errors.ServerError code _ _ _ _) = + code == "0A000" || code == "XX000" diff --git a/src/library/exposed/Hasql/Pool.hs b/src/library/exposed/Hasql/Pool.hs deleted file mode 100644 --- a/src/library/exposed/Hasql/Pool.hs +++ /dev/null @@ -1,254 +0,0 @@ -module Hasql.Pool - ( -- * Pool - Pool, - acquire, - use, - release, - - -- * Errors - UsageError (..), - ) -where - -import Data.Text.Encoding qualified as Text -import Data.Text.Encoding.Error qualified as Text -import Data.UUID.V4 qualified as Uuid -import Hasql.Connection (Connection) -import Hasql.Connection qualified as Connection -import Hasql.Connection.Setting qualified as Connection.Setting -import Hasql.Pool.Config.Config qualified as Config -import Hasql.Pool.Observation -import Hasql.Pool.Prelude -import Hasql.Pool.SessionErrorDestructors qualified as ErrorsDestruction -import Hasql.Session qualified as Session - --- | A connection tagged with metadata. -data Entry = Entry - { entryConnection :: Connection, - entryCreationTimeNSec :: Word64, - entryUseTimeNSec :: Word64, - entryId :: UUID - } - -entryIsAged :: Word64 -> Word64 -> Entry -> Bool -entryIsAged maxLifetime now Entry {..} = - now > entryCreationTimeNSec + maxLifetime - -entryIsIdle :: Word64 -> Word64 -> Entry -> Bool -entryIsIdle maxIdletime now Entry {..} = - now > entryUseTimeNSec + maxIdletime - --- | Pool of connections to DB. -data Pool = Pool - { -- | Pool size. - poolSize :: Int, - -- | Connection settings. - poolFetchConnectionSettings :: IO [Connection.Setting.Setting], - -- | Acquisition timeout, in microseconds. - poolAcquisitionTimeout :: Int, - -- | Maximal connection lifetime, in nanoseconds. - poolMaxLifetime :: Word64, - -- | Maximal connection idle time, in nanoseconds. - poolMaxIdletime :: Word64, - -- | Avail connections. - poolConnectionQueue :: TQueue Entry, - -- | Remaining capacity. - -- The pool size limits the sum of poolCapacity, the length - -- of poolConnectionQueue and the number of in-flight - -- connections. - poolCapacity :: TVar Int, - -- | Whether to return a connection to the pool. - poolReuseVar :: TVar (TVar Bool), - -- | To stop the manager thread via garbage collection. - poolReaperRef :: IORef (), - -- | Action for reporting the observations. - poolObserver :: Observation -> IO (), - -- | Initial session to execute upon every established connection. - poolInitSession :: Session.Session () - } - --- | Create a connection-pool. --- --- No connections actually get established by this function. It is delegated --- to 'use'. --- --- If you want to ensure that the pool connects fine at the initialization phase, just run 'use' with an empty session (@pure ()@) and check for errors. -acquire :: Config.Config -> IO Pool -acquire config = do - connectionQueue <- newTQueueIO - capVar <- newTVarIO (Config.size config) - reuseVar <- newTVarIO =<< newTVarIO True - reaperRef <- newIORef () - - managerTid <- forkIOWithUnmask $ \unmask -> unmask $ forever $ do - threadDelay 1000000 - now <- getMonotonicTimeNSec - join . atomically $ do - entries <- flushTQueue connectionQueue - let (agedEntries, unagedEntries) = partition (entryIsAged agingTimeoutNanos now) entries - (idleEntries, liveEntries) = partition (entryIsIdle agingTimeoutNanos now) unagedEntries - traverse_ (writeTQueue connectionQueue) liveEntries - return $ do - forM_ agedEntries $ \entry -> do - Connection.release (entryConnection entry) - atomically $ modifyTVar' capVar succ - (Config.observationHandler config) (ConnectionObservation (entryId entry) (TerminatedConnectionStatus AgingConnectionTerminationReason)) - forM_ idleEntries $ \entry -> do - Connection.release (entryConnection entry) - atomically $ modifyTVar' capVar succ - (Config.observationHandler config) (ConnectionObservation (entryId entry) (TerminatedConnectionStatus IdlenessConnectionTerminationReason)) - - void . mkWeakIORef reaperRef $ do - -- When the pool goes out of scope, stop the manager. - killThread managerTid - - return $ Pool (Config.size config) (Config.connectionSettingsProvider config) acqTimeoutMicros agingTimeoutNanos maxIdletimeNanos connectionQueue capVar reuseVar reaperRef (Config.observationHandler config) (Config.initSession config) - where - acqTimeoutMicros = - div (fromIntegral (diffTimeToPicoseconds (Config.acquisitionTimeout config))) 1_000_000 - agingTimeoutNanos = - div (fromIntegral (diffTimeToPicoseconds (Config.agingTimeout config))) 1_000 - maxIdletimeNanos = - div (fromIntegral (diffTimeToPicoseconds (Config.idlenessTimeout config))) 1_000 - --- | Release all the idle connections in the pool, and mark the in-use connections --- to be released after use. Any connections acquired after the call will be --- freshly established. --- --- The pool remains usable after this action. --- So you can use this function to reset the connections in the pool. --- Naturally, you can also use it to release the resources. -release :: Pool -> IO () -release Pool {..} = - join . atomically $ do - prevReuse <- readTVar poolReuseVar - writeTVar prevReuse False - newReuse <- newTVar True - writeTVar poolReuseVar newReuse - entries <- flushTQueue poolConnectionQueue - return $ forM_ entries $ \entry -> do - Connection.release (entryConnection entry) - atomically $ modifyTVar' poolCapacity succ - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason)) - --- | Use a connection from the pool to run a session and return the connection --- to the pool, when finished. --- --- Session failing with a 'Session.ClientError' gets interpreted as a loss of --- connection. In such case the connection does not get returned to the pool --- and a slot gets freed up for a new connection to be established the next --- time one is needed. The error still gets returned from this function. --- --- __Warning:__ Due to the mechanism mentioned above you should avoid intercepting this error type from within sessions. -use :: Pool -> Session.Session a -> IO (Either UsageError a) -use Pool {..} sess = do - timeout <- do - delay <- registerDelay poolAcquisitionTimeout - return $ readTVar delay - join . atomically $ do - reuseVar <- readTVar poolReuseVar - asum - [ readTQueue poolConnectionQueue <&> onConn reuseVar, - do - capVal <- readTVar poolCapacity - if capVal > 0 - then do - writeTVar poolCapacity $! pred capVal - return $ onNewConn reuseVar - else retry, - do - timedOut <- timeout - if timedOut - then return . return . Left $ AcquisitionTimeoutUsageError - else retry - ] - where - onNewConn reuseVar = do - settings <- poolFetchConnectionSettings - now <- getMonotonicTimeNSec - id <- Uuid.nextRandom - poolObserver (ConnectionObservation id ConnectingConnectionStatus) - Connection.acquire settings >>= \case - Left connErr -> do - poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (fmap (Text.decodeUtf8With Text.lenientDecode) connErr)))) - atomically $ modifyTVar' poolCapacity succ - return $ Left $ ConnectionUsageError connErr - Right connection -> do - Session.run poolInitSession connection >>= \case - Left err -> do - Connection.release connection - ErrorsDestruction.reset - ( \details -> do - poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (fmap (Text.decodeUtf8With Text.lenientDecode) details)))) - ) - (poolObserver (ConnectionObservation id (TerminatedConnectionStatus (InitializationErrorTerminationReason err)))) - err - return $ Left $ SessionUsageError err - Right () -> do - poolObserver (ConnectionObservation id (ReadyForUseConnectionStatus EstablishedConnectionReadyForUseReason)) - onLiveConn reuseVar (Entry connection now now id) - - onConn reuseVar entry = do - now <- getMonotonicTimeNSec - if entryIsAged poolMaxLifetime now entry - then do - Connection.release (entryConnection entry) - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus AgingConnectionTerminationReason)) - onNewConn reuseVar - else - if entryIsIdle poolMaxIdletime now entry - then do - Connection.release (entryConnection entry) - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus IdlenessConnectionTerminationReason)) - onNewConn reuseVar - else do - onLiveConn reuseVar entry {entryUseTimeNSec = now} - - onLiveConn reuseVar entry = do - poolObserver (ConnectionObservation (entryId entry) InUseConnectionStatus) - sessRes <- try @SomeException (Session.run sess (entryConnection entry)) - - case sessRes of - Left exc -> do - returnConn - throwIO exc - Right (Left err) -> - ErrorsDestruction.reset - ( \details -> do - Connection.release (entryConnection entry) - atomically $ modifyTVar' poolCapacity succ - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (fmap (Text.decodeUtf8With Text.lenientDecode) details)))) - return $ Left $ SessionUsageError err - ) - ( do - returnConn - poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus (SessionFailedConnectionReadyForUseReason err))) - return $ Left $ SessionUsageError err - ) - err - Right (Right res) -> do - returnConn - poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus SessionSucceededConnectionReadyForUseReason)) - return $ Right res - where - returnConn = - join . atomically $ do - reuse <- readTVar reuseVar - if reuse - then writeTQueue poolConnectionQueue entry $> return () - else return $ do - Connection.release (entryConnection entry) - atomically $ modifyTVar' poolCapacity succ - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason)) - --- | Union over all errors that 'use' can result in. -data UsageError - = -- | Attempt to establish a connection failed. - ConnectionUsageError Connection.ConnectionError - | -- | Session execution failed. - SessionUsageError Session.SessionError - | -- | Timeout acquiring a connection. - AcquisitionTimeoutUsageError - deriving (Show, Eq) - -instance Exception UsageError diff --git a/src/library/exposed/Hasql/Pool/Config.hs b/src/library/exposed/Hasql/Pool/Config.hs deleted file mode 100644 --- a/src/library/exposed/Hasql/Pool/Config.hs +++ /dev/null @@ -1,25 +0,0 @@ --- | DSL for construction of configs. -module Hasql.Pool.Config - ( Config.Config, - settings, - Setting.Setting, - Setting.size, - Setting.acquisitionTimeout, - Setting.agingTimeout, - Setting.idlenessTimeout, - Setting.staticConnectionSettings, - Setting.dynamicConnectionSettings, - Setting.observationHandler, - Setting.initSession, - ) -where - -import Hasql.Pool.Config.Config qualified as Config -import Hasql.Pool.Config.Setting qualified as Setting -import Hasql.Pool.Prelude - --- | Compile config from a list of settings. --- Latter settings override the preceding in cases of conflicts. -settings :: [Setting.Setting] -> Config.Config -settings = - foldr ($) Config.defaults . fmap Setting.apply diff --git a/src/library/exposed/Hasql/Pool/Config/Defaults.hs b/src/library/exposed/Hasql/Pool/Config/Defaults.hs deleted file mode 100644 --- a/src/library/exposed/Hasql/Pool/Config/Defaults.hs +++ /dev/null @@ -1,49 +0,0 @@ -module Hasql.Pool.Config.Defaults where - -import Hasql.Connection.Setting qualified as Connection.Setting -import Hasql.Connection.Setting.Connection qualified as Connection.Setting.Connection -import Hasql.Pool.Observation (Observation) -import Hasql.Pool.Prelude -import Hasql.Session qualified as Session - --- | --- 3 connections. -size :: Int -size = 3 - --- | --- 10 seconds. -acquisitionTimeout :: DiffTime -acquisitionTimeout = 10 - --- | --- 1 day. -agingTimeout :: DiffTime -agingTimeout = 60 * 60 * 24 - --- | --- 10 minutes. -idlenessTimeout :: DiffTime -idlenessTimeout = 60 * 10 - --- | --- > "postgresql://postgres:postgres@localhost:5432/postgres" -staticConnectionSettings :: [Connection.Setting.Setting] -staticConnectionSettings = - [ Connection.Setting.connection (Connection.Setting.Connection.string "postgresql://postgres:postgres@localhost:5432/postgres") - ] - --- | --- > pure "postgresql://postgres:postgres@localhost:5432/postgres" -dynamicConnectionSettings :: IO [Connection.Setting.Setting] -dynamicConnectionSettings = pure staticConnectionSettings - --- | --- > const (pure ()) -observationHandler :: Observation -> IO () -observationHandler = const (pure ()) - --- | --- > pure () -initSession :: Session.Session () -initSession = pure () diff --git a/src/library/exposed/Hasql/Pool/Observation.hs b/src/library/exposed/Hasql/Pool/Observation.hs deleted file mode 100644 --- a/src/library/exposed/Hasql/Pool/Observation.hs +++ /dev/null @@ -1,60 +0,0 @@ --- | Interface for processing observations of the status of the pool. --- --- Provides a flexible mechanism for monitoring the healthiness of the pool via logs and metrics without any opinionated choices on the actual monitoring technologies. --- Specific interpreters are encouraged to be created as extension libraries. -module Hasql.Pool.Observation where - -import Hasql.Pool.Prelude -import Hasql.Session qualified as Session - --- | An observation of a change of the state of a pool. -data Observation - = -- | Status of one of the pool's connections has changed. - ConnectionObservation - -- | Generated connection ID. - -- For grouping the observations by one connection. - UUID - -- | Status that the connection has entered. - ConnectionStatus - deriving (Show, Eq) - --- | Status of a connection. --- --- <> -data ConnectionStatus - = -- | Connection is being established. - -- - -- This is the initial status of every connection. - ConnectingConnectionStatus - | -- | Connection is established and not occupied. - ReadyForUseConnectionStatus ConnectionReadyForUseReason - | -- | Is being used by some session. - -- - -- After it's done the status will transition to 'ReadyForUseConnectionStatus' or 'TerminatedConnectionStatus'. - InUseConnectionStatus - | -- | Connection terminated. - TerminatedConnectionStatus ConnectionTerminationReason - deriving (Show, Eq) - -data ConnectionReadyForUseReason - = -- | Connection just got established. - EstablishedConnectionReadyForUseReason - | -- | Session execution ended with a failure that does not require a connection reset. - SessionFailedConnectionReadyForUseReason Session.SessionError - | -- | Session execution ended with success. - SessionSucceededConnectionReadyForUseReason - deriving (Show, Eq) - --- | Explanation of why a connection was terminated. -data ConnectionTerminationReason - = -- | The age timeout of the connection has passed. - AgingConnectionTerminationReason - | -- | The timeout of how long a connection may remain idle in the pool has passed. - IdlenessConnectionTerminationReason - | -- | Connectivity issues with the server. - NetworkErrorConnectionTerminationReason (Maybe Text) - | -- | User has invoked the 'Hasql.Pool.release' procedure. - ReleaseConnectionTerminationReason - | -- | Initialization session failure. - InitializationErrorTerminationReason Session.SessionError - deriving (Show, Eq) diff --git a/src/library/other/Hasql/Pool/Config/Config.hs b/src/library/other/Hasql/Pool/Config/Config.hs deleted file mode 100644 --- a/src/library/other/Hasql/Pool/Config/Config.hs +++ /dev/null @@ -1,31 +0,0 @@ -module Hasql.Pool.Config.Config where - -import Hasql.Connection.Setting qualified as Connection.Setting -import Hasql.Pool.Config.Defaults qualified as Defaults -import Hasql.Pool.Observation (Observation) -import Hasql.Pool.Prelude -import Hasql.Session qualified as Session - --- | Configuration for Hasql connection pool. -data Config = Config - { size :: Int, - acquisitionTimeout :: DiffTime, - agingTimeout :: DiffTime, - idlenessTimeout :: DiffTime, - connectionSettingsProvider :: IO [Connection.Setting.Setting], - observationHandler :: Observation -> IO (), - initSession :: Session.Session () - } - --- | Reasonable defaults, which can be built upon. -defaults :: Config -defaults = - Config - { size = Defaults.size, - acquisitionTimeout = Defaults.acquisitionTimeout, - agingTimeout = Defaults.agingTimeout, - idlenessTimeout = Defaults.idlenessTimeout, - connectionSettingsProvider = Defaults.dynamicConnectionSettings, - observationHandler = Defaults.observationHandler, - initSession = Defaults.initSession - } diff --git a/src/library/other/Hasql/Pool/Config/Setting.hs b/src/library/other/Hasql/Pool/Config/Setting.hs deleted file mode 100644 --- a/src/library/other/Hasql/Pool/Config/Setting.hs +++ /dev/null @@ -1,93 +0,0 @@ -module Hasql.Pool.Config.Setting where - -import Hasql.Connection.Setting qualified as Connection.Setting -import Hasql.Pool.Config.Config (Config) -import Hasql.Pool.Config.Config qualified as Config -import Hasql.Pool.Observation (Observation) -import Hasql.Pool.Prelude -import Hasql.Session qualified as Session - -apply :: Setting -> Config -> Config -apply (Setting run) = run - --- | A single setting of a config. -newtype Setting - = Setting (Config -> Config) - --- | Pool size. --- --- 3 by default. -size :: Int -> Setting -size x = - Setting (\config -> config {Config.size = x}) - --- | Connection acquisition timeout. --- --- 10 seconds by default. -acquisitionTimeout :: DiffTime -> Setting -acquisitionTimeout x = - Setting (\config -> config {Config.acquisitionTimeout = x}) - --- | Maximal connection lifetime. --- --- Determines how long is available for reuse. --- After the timeout passes and an active session is finished the connection will be closed releasing a slot in the pool for a fresh connection to be established. --- --- This is useful as a healthy measure for resetting the server-side caches. --- --- 1 day by default. -agingTimeout :: DiffTime -> Setting -agingTimeout x = - Setting (\config -> config {Config.agingTimeout = x}) - --- | Maximal connection idle time. --- --- How long to keep a connection open when it's not being used. --- --- 10 minutes by default. -idlenessTimeout :: DiffTime -> Setting -idlenessTimeout x = - Setting (\config -> config {Config.idlenessTimeout = x}) - --- | Connection string. --- --- By default it is: --- --- > "postgresql://postgres:postgres@localhost:5432/postgres" -staticConnectionSettings :: [Connection.Setting.Setting] -> Setting -staticConnectionSettings x = - Setting (\config -> config {Config.connectionSettingsProvider = pure x}) - --- | Action providing connection settings. --- --- Gets used each time a connection gets established by the pool. --- This may be useful for some authorization models. --- --- By default it is: --- --- > pure "postgresql://postgres:postgres@localhost:5432/postgres" -dynamicConnectionSettings :: IO [Connection.Setting.Setting] -> Setting -dynamicConnectionSettings x = - Setting (\config -> config {Config.connectionSettingsProvider = x}) - --- | Observation handler. --- --- Typically it's used for monitoring the state of the pool via metrics and logging. --- --- If the provided action is not lightweight, it's recommended to use intermediate bufferring via channels like TBQueue to avoid occupying the pool management thread for too long. --- E.g., if the action is @'atomically' . 'writeTBQueue' yourQueue@, then reading from it and processing can be done on a separate thread. --- --- By default it is: --- --- > const (pure ()) -observationHandler :: (Observation -> IO ()) -> Setting -observationHandler x = - Setting (\config -> config {Config.observationHandler = x}) - --- | Initial session. --- --- Gets executed on every connection upon acquisition. --- Lets you specify the connection-wide settings. -initSession :: Session.Session () -> Setting -initSession x = - Setting (\config -> config {Config.initSession = x}) diff --git a/src/library/other/Hasql/Pool/Prelude.hs b/src/library/other/Hasql/Pool/Prelude.hs deleted file mode 100644 --- a/src/library/other/Hasql/Pool/Prelude.hs +++ /dev/null @@ -1,75 +0,0 @@ -module Hasql.Pool.Prelude - ( module Exports, - ) -where - -import Control.Applicative as Exports hiding (WrappedArrow (..)) -import Control.Arrow as Exports hiding (first, second) -import Control.Category as Exports -import Control.Concurrent as Exports -import Control.Concurrent.STM as Exports hiding (orElse) -import Control.Exception as Exports -import Control.Monad as Exports hiding (fail, forM, forM_, mapM, mapM_, msum, sequence, sequence_) -import Control.Monad.Fail as Exports -import Control.Monad.Fix as Exports hiding (fix) -import Control.Monad.IO.Class as Exports -import Control.Monad.ST as Exports -import Data.Bifunctor as Exports -import Data.Bits as Exports -import Data.Bool as Exports -import Data.ByteString as Exports (ByteString) -import Data.Char as Exports -import Data.Coerce as Exports -import Data.Complex as Exports -import Data.Data as Exports -import Data.Dynamic as Exports -import Data.Either as Exports -import Data.Fixed as Exports -import Data.Foldable as Exports hiding (toList) -import Data.Function as Exports hiding (id, (.)) -import Data.Functor as Exports hiding (unzip) -import Data.Functor.Compose as Exports -import Data.IORef as Exports -import Data.Int as Exports -import Data.Ix as Exports -import Data.List as Exports hiding (all, and, any, concat, concatMap, elem, find, foldl, foldl', foldl1, foldr, foldr1, isSubsequenceOf, mapAccumL, mapAccumR, maximum, maximumBy, minimum, minimumBy, notElem, or, product, sortOn, sum, uncons) -import Data.List.NonEmpty as Exports (NonEmpty (..)) -import Data.Maybe as Exports -import Data.Monoid as Exports hiding (Alt) -import Data.Ord as Exports -import Data.Proxy as Exports -import Data.Ratio as Exports -import Data.STRef as Exports -import Data.String as Exports -import Data.Text as Exports (Text) -import Data.Time as Exports -import Data.Traversable as Exports -import Data.Tuple as Exports -import Data.UUID as Exports (UUID) -import Data.Unique as Exports -import Data.Version as Exports -import Data.Void as Exports -import Data.Word as Exports -import Debug.Trace as Exports -import Foreign.ForeignPtr as Exports -import Foreign.Ptr as Exports -import Foreign.StablePtr as Exports -import Foreign.Storable as Exports -import GHC.Clock as Exports (getMonotonicTimeNSec) -import GHC.Conc as Exports hiding (orElse, threadWaitRead, threadWaitReadSTM, threadWaitWrite, threadWaitWriteSTM, withMVar) -import GHC.Exts as Exports (IsList (..), groupWith, inline, lazy, sortWith) -import GHC.Generics as Exports (Generic) -import GHC.IO.Exception as Exports -import Numeric as Exports -import System.Environment as Exports -import System.Exit as Exports -import System.IO as Exports (Handle, hClose) -import System.IO.Error as Exports -import System.IO.Unsafe as Exports -import System.Mem as Exports -import System.Mem.StableName as Exports -import System.Timeout as Exports -import Text.Printf as Exports (hPrintf, printf) -import Text.Read as Exports (Read (..), readEither, readMaybe) -import Unsafe.Coerce as Exports -import Prelude as Exports hiding (all, and, any, concat, concatMap, elem, fail, foldl, foldl1, foldr, foldr1, id, mapM, mapM_, maximum, minimum, notElem, or, product, sequence, sequence_, sum, (.)) diff --git a/src/library/other/Hasql/Pool/SessionErrorDestructors.hs b/src/library/other/Hasql/Pool/SessionErrorDestructors.hs deleted file mode 100644 --- a/src/library/other/Hasql/Pool/SessionErrorDestructors.hs +++ /dev/null @@ -1,10 +0,0 @@ -module Hasql.Pool.SessionErrorDestructors where - -import Hasql.Pool.Prelude -import Hasql.Session qualified as Session - -reset :: (Maybe ByteString -> x) -> x -> Session.SessionError -> x -reset onReset onNoReset = \case - Session.QueryError _ _ (Session.ClientError details) -> onReset details - Session.PipelineError (Session.ClientError details) -> onReset details - _ -> onNoReset