hasql-pool 0.8.0.7 → 1.5.1.1
raw patch · 31 files changed
Files
- CHANGELOG.md +121/−9
- diagrams-output/connection-status-model.png binary
- hasql-pool.cabal +95/−11
- library/Hasql/Pool.hs +0/−171
- library/Hasql/Pool/Prelude.hs +0/−79
- src/integration-tests/Helpers/Adapters.hs +30/−0
- src/integration-tests/Helpers/Hooks.hs +19/−0
- src/integration-tests/Helpers/Scripts.hs +166/−0
- src/integration-tests/Helpers/Sessions.hs +81/−0
- src/integration-tests/Main.hs +1/−0
- src/integration-tests/Specs/Config/AgingTimeoutSpec.hs +32/−0
- src/integration-tests/Specs/Config/IdlenessTimeoutSpec.hs +62/−0
- src/integration-tests/Specs/Config/InitSessionSpec.hs +59/−0
- src/integration-tests/Specs/Helpers/Sessions/CountConnectionsSpec.hs +14/−0
- src/integration-tests/Specs/Helpers/Sessions/GetSettingSpec.hs +36/−0
- src/integration-tests/Specs/InterruptionSpec.hs +163/−0
- src/integration-tests/Specs/ReleaseSpec.hs +16/−0
- src/integration-tests/Specs/SpecHook.hs +15/−0
- src/integration-tests/Specs/UsageError/AcquisitionTimeoutSpec.hs +45/−0
- src/integration-tests/Specs/UsageError/SessionSpec.hs +16/−0
- src/integration-tests/Specs/UseSpec.hs +252/−0
- src/library/Hasql/Pool.hs +511/−0
- src/library/Hasql/Pool/Config.hs +28/−0
- src/library/Hasql/Pool/Config/Config.hs +34/−0
- src/library/Hasql/Pool/Config/Defaults.hs +53/−0
- src/library/Hasql/Pool/Config/ReuseOrder.hs +17/−0
- src/library/Hasql/Pool/Config/Setting.hs +111/−0
- src/library/Hasql/Pool/Observation.hs +64/−0
- src/library/Hasql/Pool/Prelude.hs +75/−0
- src/library/Hasql/Pool/SessionErrorDestructors.hs +30/−0
- test/Main.hs +0/−147
CHANGELOG.md view
@@ -1,16 +1,128 @@-# 0.8.0.7+# v1.5.1.1 +## Fixes++- After the server closes an idle pooled connection (a Postgres restart,+ `pg_terminate_backend`, a server-side idle timeout), the first session on+ it can fail with `UnexpectedRowCountStatementError` instead of a connection+ error. Hasql maps the extra result that reports the closed connection to+ that error, and a `singleRow` statement that returns two rows fails the+ same way on a healthy connection, so the error alone cannot justify a+ discard: the dead connection got returned to the pool and failed one more+ session before it was discarded, so after a database restart every pooled+ connection failed two sessions. Now, when a session fails with an error+ that does not require a discard, `use` also checks the status that the+ adapter reports for the connection and discards it when it is marked bad.++# v1.5.1.0++## Fixes++- Fix pool capacity leak when `use` gets interrupted during connection+ establishment, connection refresh, or an observation (#60)+- Fix pool state corruption from throwing observation handlers; handlers+ are now contained (see the updated `observationHandler` docs)+- Fix background reaper dying permanently from a failed connection+ release or observation+- Fix `release` leaking flushed entries on failure+- Fix waiters waiting forever on a garbage-collected pool++## Added++- `reuseOrder` setting (#59): LIFO reuse lets the pool shrink to the+ connections the load needs and keeps caches warm; FIFO stays the default+- One background timer for acquisition timeouts instead of one+ uncancelable timer per waiting request (#58)++## Changed++- Connections are handed to waiting requests in arrival order; a request+ no longer takes a connection while others wait (#58)++# v1.5.0.1++- Allow newer pqi.++# v1.5.0.0++## Breaking++- Migrate to `hasql-2`, replacing `postgresql-libpq` with the `pqi` connection-adapter interface. `acquire` now takes a `Pqi.Adapter` as its first argument; pick one from an adapter package such as `pqi-ffi` or `pqi-native`.++# v1.4.2.3++## Fixes++- Fix publishing.++# v1.4.2.2++## Fixes++- Fix pool capacity leak when `initSession` fails (#56)+- Fix background reaper never passively evicting idle connections: it used the connection-lifetime timeout instead of the idleness timeout when checking for idle connections++# v1.4.2.1++## Fixes++- Discard pooled connections after driver errors (#55)++# v1.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++# v1.3++- Adapt to the new settings model of `hasql-1.9`++# v1.2++- Migrated to `hasql-1.7`+- Changed references to `QueryError` in observations to `SessionError`++# v1.1++- `ReadyForUseConnectionStatus` got extended with the `ConnectionReadyForUseReason` details.+- `initSession` setting added.++# v1++- Optional observability event stream added. Provides a flexible mechanism for monitoring the healthiness of the pool via logs and metrics.+- Configuration got isolated into a DSL, which will allow to provide new configurations without breaking backward compatibility.++# v0.10.1++- Avoid releasing connections on exceptions thrown in session++# v0.9++- Maximal lifetime added for connections. Allows to refresh the connections in time cleaning up the resources.++Breaking:++- The acquisition timeout is now non-optional.+- Moved to `DiffTime` for timeouts.++# v0.8.0.7+ Fix excessive connections during releases due to race conditions. -# 0.8.0.5+# v0.8.0.5 Fix connections not returning to the pool on exceptions. -# 0.8.0.2+# v0.8.0.2 Fixed Windows build. -# 0.8+# v0.8 `release` became reusable. You can use it to destroy the whole pool (same as before), but now also you can use it to reset the connections. @@ -22,19 +134,19 @@ - `acquire` extended with the acquisition timeout parameter - `acquireDynamically` extended with the acquisition timeout parameter -# 0.7.2+# v0.7.2 Added support for dynamic connection configuration ([issue #11](https://github.com/nikita-volkov/hasql-pool/issues/11)). -# 0.7.1.2+# v0.7.1.2 Fixed connections not being released if they were in use during the call to `release`. -# 0.7.1+# v0.7.1 Added `Exception` for `UsageError`. -# 0.7+# v0.7 Simplified the implementation a lot by removing the notion of timeout. @@ -42,7 +154,7 @@ - Removed the `Settings` type - Changed the signature of `acquire` -# 0.6+# v0.6 Moved away from "resource-pool" and fixed the handling of lost connections.
+ diagrams-output/connection-status-model.png view
binary file changed (absent → 16434 bytes)
hasql-pool.cabal view
@@ -1,8 +1,6 @@ cabal-version: 3.0- name: hasql-pool-version: 0.8.0.7-+version: 1.5.1.1 category: Hasql, Database, PostgreSQL synopsis: Pool of connections for Hasql homepage: https://github.com/nikita-volkov/hasql-pool@@ -12,39 +10,125 @@ copyright: (c) 2015, Nikita Volkov license: MIT license-file: LICENSE-extra-source-files: CHANGELOG.md+extra-source-files:+ CHANGELOG.md+ diagrams-output/*.png +extra-doc-files:+ diagrams-output/*.png+ source-repository head type: git- location: git://github.com/nikita-volkov/hasql-pool.git+ location: https://github.com/nikita-volkov/hasql-pool common base-settings- default-extensions: BangPatterns, BlockArguments, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DerivingVia, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, InstanceSigs, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, StrictData, TemplateHaskell, TupleSections, TypeApplications, TypeFamilies, TypeOperators, UnboxedTuples default-language: Haskell2010+ default-extensions:+ ApplicativeDo+ Arrows+ BangPatterns+ BlockArguments+ ConstraintKinds+ DataKinds+ DefaultSignatures+ DeriveDataTypeable+ DeriveFoldable+ DeriveFunctor+ DeriveGeneric+ DeriveTraversable+ DerivingVia+ EmptyDataDecls+ FlexibleContexts+ FlexibleInstances+ FunctionalDependencies+ GADTs+ GeneralizedNewtypeDeriving+ ImportQualifiedPost+ LambdaCase+ LiberalTypeSynonyms+ MultiParamTypeClasses+ MultiWayIf+ NoImplicitPrelude+ NoMonomorphismRestriction+ NumericUnderscores+ OverloadedStrings+ PatternGuards+ QuasiQuotes+ RankNTypes+ RecordWildCards+ RoleAnnotations+ ScopedTypeVariables+ StandaloneDeriving+ StrictData+ TupleSections+ TypeApplications+ TypeFamilies+ TypeOperators library import: base-settings- hs-source-dirs: library+ hs-source-dirs:+ src/library/+ exposed-modules: Hasql.Pool+ Hasql.Pool.Config+ Hasql.Pool.Config.Defaults+ Hasql.Pool.Observation+ other-modules:+ Hasql.Pool.Config.Config+ Hasql.Pool.Config.ReuseOrder+ Hasql.Pool.Config.Setting Hasql.Pool.Prelude+ Hasql.Pool.SessionErrorDestructors+ build-depends: base >=4.11 && <5,- hasql >=1.6.0.1 && <1.7,+ bytestring >=0.10 && <0.14,+ hasql ^>=2.0,+ pqi >=1.0 && <1.2, stm >=2.5 && <3,- transformers >=0.5 && <0.7,+ text >=1.2 && <3,+ time >=1.9 && <2,+ uuid >=1.3 && <2, test-suite test import: base-settings type: exitcode-stdio-1.0- hs-source-dirs: test+ hs-source-dirs: src/integration-tests main-is: Main.hs+ other-modules:+ Helpers.Adapters+ Helpers.Hooks+ Helpers.Scripts+ Helpers.Sessions+ Specs.Config.AgingTimeoutSpec+ Specs.Config.IdlenessTimeoutSpec+ Specs.Config.InitSessionSpec+ Specs.Helpers.Sessions.CountConnectionsSpec+ Specs.Helpers.Sessions.GetSettingSpec+ Specs.InterruptionSpec+ Specs.ReleaseSpec+ Specs.SpecHook+ Specs.UsageError.AcquisitionTimeoutSpec+ Specs.UsageError.SessionSpec+ Specs.UseSpec+ ghc-options: -threaded+ build-tool-depends:+ hspec-discover:hspec-discover ^>=2.11.12+ build-depends: async >=2.2 && <3, hasql, hasql-pool, hspec >=2.6 && <3,+ pqi >=1.0 && <1.2,+ pqi-ffi ^>=1.0,+ pqi-native ^>=1.0,+ random >=1.2 && <2, rerebase >=1.15 && <2,- stm >=2.5 && <3,+ testcontainers-postgresql >=0.2 && <0.3,+ text-builder >=1 && <1.1,+ tuple ^>=0.3.0.2,
− library/Hasql/Pool.hs
@@ -1,171 +0,0 @@-module Hasql.Pool- ( -- * Pool- Pool,- acquire,- acquireDynamically,- release,- use,-- -- * Errors- UsageError (..),- )-where--import Hasql.Connection (Connection)-import qualified Hasql.Connection as Connection-import Hasql.Pool.Prelude-import Hasql.Session (Session)-import qualified Hasql.Session as Session---- | Pool of connections to DB.-data Pool = Pool- { -- | Connection settings.- poolFetchConnectionSettings :: IO Connection.Settings,- -- | Acquisition timeout, in microseconds.- poolAcquisitionTimeout :: Maybe Int,- -- | Avail connections.- poolConnectionQueue :: TQueue Connection,- -- | 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.- poolReuse :: TVar (TVar Bool)- }---- | Create a connection-pool.------ No connections actually get established by this function. It is delegated--- to 'use'.-acquire ::- -- | Pool size.- Int ->- -- | Connection acquisition timeout in microseconds.- Maybe Int ->- -- | Connection settings.- Connection.Settings ->- IO Pool-acquire poolSize timeout connectionSettings =- acquireDynamically poolSize timeout (pure connectionSettings)---- | Create a connection-pool.------ In difference to 'acquire' new settings get fetched each time a connection--- is created. This may be useful for some security models.------ No connections actually get established by this function. It is delegated--- to 'use'.-acquireDynamically ::- -- | Pool size.- Int ->- -- | Connection acquisition timeout in microseconds.- Maybe Int ->- -- | Action fetching connection settings.- IO Connection.Settings ->- IO Pool-acquireDynamically poolSize timeout fetchConnectionSettings = do- Pool fetchConnectionSettings timeout- <$> newTQueueIO- <*> newTVarIO poolSize- <*> (newTVarIO =<< newTVarIO True)---- | 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 poolReuse- writeTVar prevReuse False- newReuse <- newTVar True- writeTVar poolReuse newReuse- conns <- flushTQueue poolConnectionQueue- return $ forM_ conns $ \conn -> do- Connection.release conn- atomically $ modifyTVar' poolCapacity succ---- | 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.-use :: Pool -> Session.Session a -> IO (Either UsageError a)-use Pool {..} sess = do- timeout <- case poolAcquisitionTimeout of- Just delta -> do- delay <- registerDelay delta- return $ readTVar delay- Nothing ->- return $ return False- join . atomically $ do- reuseVar <- readTVar poolReuse- 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- connRes <- Connection.acquire settings- case connRes of- Left connErr -> do- atomically $ modifyTVar' poolCapacity succ- return $ Left $ ConnectionUsageError connErr- Right conn -> onConn reuseVar conn- onConn reuseVar conn = do- sessRes <-- catch (Session.run sess conn) $ \(err :: SomeException) -> do- Connection.release conn- atomically $ modifyTVar' poolCapacity succ- throw err-- case sessRes of- Left err -> case err of- Session.QueryError _ _ (Session.ClientError _) -> do- atomically $ modifyTVar' poolCapacity succ- return $ Left $ SessionUsageError err- _ -> do- returnConn- return $ Left $ SessionUsageError err- Right res -> do- returnConn- return $ Right res- where- returnConn =- join . atomically $ do- reuse <- readTVar reuseVar- if reuse- then writeTQueue poolConnectionQueue conn $> return ()- else return $ do- Connection.release conn- atomically $ modifyTVar' poolCapacity succ---- | 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.QueryError- | -- | Timeout acquiring a connection.- AcquisitionTimeoutUsageError- deriving (Show, Eq)--instance Exception UsageError
− library/Hasql/Pool/Prelude.hs
@@ -1,79 +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 Control.Monad.Trans.Class as Exports-import Control.Monad.Trans.Cont as Exports hiding (callCC, shift)-import Control.Monad.Trans.Except as Exports (Except, ExceptT (ExceptT), except, mapExcept, mapExceptT, runExcept, runExceptT, withExcept, withExceptT)-import Control.Monad.Trans.Maybe as Exports-import Control.Monad.Trans.Reader as Exports (Reader, ReaderT (ReaderT), mapReader, mapReaderT, runReader, runReaderT, withReader, withReaderT)-import Control.Monad.Trans.State.Strict as Exports (State, StateT (StateT), evalState, evalStateT, execState, execStateT, mapState, mapStateT, runState, runStateT, withState, withStateT)-import Control.Monad.Trans.Writer.Strict as Exports (Writer, WriterT (..), execWriter, execWriterT, mapWriter, mapWriterT, runWriter)-import Data.Bifunctor as Exports-import Data.Bits as Exports-import Data.Bool as Exports-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-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.Traversable as Exports-import Data.Tuple as Exports-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.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.ParserCombinators.ReadP as Exports (ReadP, ReadS, readP_to_S, readS_to_P)-import Text.ParserCombinators.ReadPrec as Exports (ReadPrec, readP_to_Prec, readPrec_to_P, readPrec_to_S, readS_to_Prec)-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, (.))
+ src/integration-tests/Helpers/Adapters.hs view
@@ -0,0 +1,30 @@+module Helpers.Adapters+ ( adapters,+ byAdapter,+ hook,+ )+where++import Pqi qualified+import Pqi.Ffi qualified+import Pqi.Native qualified+import Prelude+import Test.Hspec++adapters :: [Pqi.Adapter]+adapters =+ [ Pqi.Ffi.adapter,+ Pqi.Native.adapter+ ]++-- | Run the given spec-building function once per available Pqi adapter,+-- nesting each run under a @describe@ named after the adapter.+byAdapter :: (Pqi.Adapter -> Spec) -> Spec+byAdapter f =+ for_ adapters \adapter ->+ describe (toList (Pqi.name adapter)) (f adapter)++hook :: SpecWith Pqi.Adapter -> Spec+hook hookedSpec =+ byAdapter \adapter ->+ mapSubject (const adapter) hookedSpec
+ src/integration-tests/Helpers/Hooks.hs view
@@ -0,0 +1,19 @@+-- | Hooks for Hspec.+module Helpers.Hooks where++import Data.Bool+import Prelude hiding (Handler)+import TestcontainersPostgresql qualified++-- | Testing action in the scope of the host name and port of a running fresh isolated postgres server.+type Handler = (Text, Word16) -> IO ()++postgres17 :: Handler -> IO ()+postgres17 handler =+ TestcontainersPostgresql.run+ TestcontainersPostgresql.Config+ { forwardLogs = False,+ tagName = "postgres:17",+ auth = TestcontainersPostgresql.TrustAuth+ }+ (\(host, portInt) -> handler (host, fromIntegral portInt))
+ src/integration-tests/Helpers/Scripts.hs view
@@ -0,0 +1,166 @@+module Helpers.Scripts where++import Hasql.Connection.Settings qualified as Connection.Settings+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as Config+import Hasql.Pool.Observation (Observation)+import Hasql.Session qualified as Session+import Pqi qualified+import Prelude+import System.Random.Stateful qualified as Random+import TextBuilder qualified++-- |+-- Parameters provided by the scope.+-- Adapter, host and port of a running isolated postgres server.+type ScopeParams = (Pqi.Adapter, Text, Word16)++onTaggedPool :: Int -> DiffTime -> DiffTime -> DiffTime -> Text -> ScopeParams -> (Pool.Pool -> IO ()) -> IO ()+onTaggedPool poolSize acqTimeout maxLifetime maxIdletime appName (adapter, host, port) =+ bracket+ ( Pool.acquire+ adapter+ ( Config.settings+ [ Config.size poolSize,+ Config.acquisitionTimeout acqTimeout,+ Config.agingTimeout maxLifetime,+ Config.idlenessTimeout maxIdletime,+ Config.staticConnectionSettings+ ( mconcat+ [ Connection.Settings.hostAndPort host (fromIntegral port),+ Connection.Settings.user "postgres",+ Connection.Settings.password "",+ Connection.Settings.dbname "postgres",+ Connection.Settings.applicationName appName+ ]+ )+ ]+ )+ )+ Pool.release++onAutotaggedPool :: Int -> DiffTime -> DiffTime -> DiffTime -> ScopeParams -> (Text -> Pool.Pool -> IO ()) -> IO ()+onAutotaggedPool poolSize acqTimeout maxLifetime maxIdletime scopeParams cont = do+ -- Generate app name+ appName <- generateName "hasql-pool-test-"+ onTaggedPool poolSize acqTimeout maxLifetime maxIdletime appName scopeParams (cont appName)++onTaggedPoolWithObservationHandler :: Int -> DiffTime -> DiffTime -> DiffTime -> (Observation -> IO ()) -> Text -> ScopeParams -> (Pool.Pool -> IO ()) -> IO ()+onTaggedPoolWithObservationHandler poolSize acqTimeout maxLifetime maxIdletime observationHandler appName (adapter, host, port) =+ bracket+ ( Pool.acquire+ adapter+ ( Config.settings+ [ Config.size poolSize,+ Config.acquisitionTimeout acqTimeout,+ Config.agingTimeout maxLifetime,+ Config.idlenessTimeout maxIdletime,+ Config.observationHandler observationHandler,+ Config.staticConnectionSettings+ ( mconcat+ [ Connection.Settings.hostAndPort host (fromIntegral port),+ Connection.Settings.user "postgres",+ Connection.Settings.password "",+ Connection.Settings.dbname "postgres",+ Connection.Settings.applicationName appName+ ]+ )+ ]+ )+ )+ Pool.release++onAutotaggedPoolWithObservationHandler :: Int -> DiffTime -> DiffTime -> DiffTime -> (Observation -> IO ()) -> ScopeParams -> (Text -> Pool.Pool -> IO ()) -> IO ()+onAutotaggedPoolWithObservationHandler poolSize acqTimeout maxLifetime maxIdletime observationHandler scopeParams cont = do+ appName <- generateName "hasql-pool-test-"+ onTaggedPoolWithObservationHandler poolSize acqTimeout maxLifetime maxIdletime observationHandler appName scopeParams (cont appName)++onTaggedPoolWithInitSession :: Int -> DiffTime -> DiffTime -> DiffTime -> Session.Session () -> Text -> ScopeParams -> (Pool.Pool -> IO ()) -> IO ()+onTaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession appName (adapter, host, port) =+ bracket+ ( Pool.acquire+ adapter+ ( Config.settings+ [ Config.size poolSize,+ Config.acquisitionTimeout acqTimeout,+ Config.agingTimeout maxLifetime,+ Config.idlenessTimeout maxIdletime,+ Config.initSession initSession,+ Config.staticConnectionSettings+ ( mconcat+ [ Connection.Settings.hostAndPort host (fromIntegral port),+ Connection.Settings.user "postgres",+ Connection.Settings.password "",+ Connection.Settings.dbname "postgres",+ Connection.Settings.applicationName appName+ ]+ )+ ]+ )+ )+ Pool.release++onAutotaggedPoolWithInitSession :: Int -> DiffTime -> DiffTime -> DiffTime -> Session.Session () -> ScopeParams -> (Text -> Pool.Pool -> IO ()) -> IO ()+onAutotaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession (adapter, host, port) cont = do+ appName <- generateName "hasql-pool-test-"+ onTaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession appName (adapter, host, port) (cont appName)++onTaggedPoolWithReuseOrder :: Int -> DiffTime -> DiffTime -> DiffTime -> Config.ReuseOrder -> Text -> ScopeParams -> (Pool.Pool -> IO ()) -> IO ()+onTaggedPoolWithReuseOrder poolSize acqTimeout maxLifetime maxIdletime reuseOrder appName (adapter, host, port) =+ bracket+ ( Pool.acquire+ adapter+ ( Config.settings+ [ Config.size poolSize,+ Config.acquisitionTimeout acqTimeout,+ Config.agingTimeout maxLifetime,+ Config.idlenessTimeout maxIdletime,+ Config.reuseOrder reuseOrder,+ Config.staticConnectionSettings+ ( mconcat+ [ Connection.Settings.hostAndPort host (fromIntegral port),+ Connection.Settings.user "postgres",+ Connection.Settings.password "",+ Connection.Settings.dbname "postgres",+ Connection.Settings.applicationName appName+ ]+ )+ ]+ )+ )+ Pool.release++onAutotaggedPoolWithReuseOrder :: Int -> DiffTime -> DiffTime -> DiffTime -> Config.ReuseOrder -> ScopeParams -> (Text -> Pool.Pool -> IO ()) -> IO ()+onAutotaggedPoolWithReuseOrder poolSize acqTimeout maxLifetime maxIdletime reuseOrder scopeParams cont = do+ appName <- generateName "hasql-pool-test-"+ onTaggedPoolWithReuseOrder poolSize acqTimeout maxLifetime maxIdletime reuseOrder appName scopeParams (cont appName)++onDefaultTaggedPool :: ScopeParams -> (Text -> Pool.Pool -> IO ()) -> IO ()+onDefaultTaggedPool =+ onAutotaggedPool 3 10 1_800 1_800++generateName :: Text -> IO Text+generateName prefix = do+ uniqueNum1 <- Random.uniformWord64 Random.globalStdGen+ uniqueNum2 <- Random.uniformWord64 Random.globalStdGen+ pure+ $ TextBuilder.toText+ $ mconcat+ $ [ TextBuilder.text prefix,+ TextBuilder.decimal uniqueNum1,+ "-",+ TextBuilder.decimal uniqueNum2+ ]++generateVarname :: IO Text+generateVarname = do+ uniqueNum1 <- Random.uniformWord64 Random.globalStdGen+ uniqueNum2 <- Random.uniformWord64 Random.globalStdGen+ pure+ $ TextBuilder.toText+ $ mconcat+ $ [ "testing.v",+ TextBuilder.decimal uniqueNum1,+ "v",+ TextBuilder.decimal uniqueNum2+ ]
+ src/integration-tests/Helpers/Sessions.hs view
@@ -0,0 +1,81 @@+module Helpers.Sessions+ ( selectOne,+ badQuery,+ closeConn,+ setSetting,+ getSetting,+ countConnections,+ sleep,+ backendPid,+ )+where++import Data.Tuple.All+import Hasql.Decoders qualified as Decoders+import Hasql.Encoders qualified as Encoders+import Hasql.Session qualified as Session+import Hasql.Statement qualified as Statement+import Pqi qualified+import Prelude++selectOne :: Session.Session Int64+selectOne =+ Session.statement () statement+ where+ statement = Statement.preparable "SELECT 1::int8" Encoders.noParams decoder+ decoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8))++badQuery :: Session.Session ()+badQuery =+ Session.statement () statement+ where+ statement =+ Statement.preparable "zzz" Encoders.noParams Decoders.noResult++closeConn :: Session.Session ()+closeConn =+ Session.onLibpqConnection \conn -> do+ Pqi.finish conn+ pure (Right (), conn)++setSetting :: Text -> Text -> Session.Session ()+setSetting name value = do+ Session.statement (name, value) statement+ where+ statement =+ Statement.preparable "SELECT set_config($1, $2, false)" encoder Decoders.noResult+ encoder =+ mconcat+ [ sel1 >$< Encoders.param (Encoders.nonNullable Encoders.text),+ sel2 >$< Encoders.param (Encoders.nonNullable Encoders.text)+ ]++getSetting :: Text -> Session.Session (Maybe Text)+getSetting name = do+ Session.statement name statement+ where+ 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))++countConnections :: Text -> Session.Session Int64+countConnections appName = do+ Session.statement appName statement+ where+ 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))++sleep :: Double -> Session.Session ()+sleep seconds =+ Session.statement seconds statement+ where+ statement = Statement.preparable "SELECT pg_sleep($1)" encoder Decoders.noResult+ encoder = Encoders.param (Encoders.nonNullable Encoders.float8)++backendPid :: Session.Session Int32+backendPid =+ Session.statement () statement+ where+ statement = Statement.preparable "SELECT pg_backend_pid()" Encoders.noParams decoder+ decoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int4))
+ src/integration-tests/Main.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}
+ src/integration-tests/Specs/Config/AgingTimeoutSpec.hs view
@@ -0,0 +1,32 @@+module Specs.Config.AgingTimeoutSpec where++import Hasql.Pool+import Helpers.Scripts qualified as Scripts+import Helpers.Sessions qualified as Sessions+import Prelude+import Test.Hspec++spec :: SpecWith Scripts.ScopeParams+spec = do+ it "Actively times out old connections" \scopeParams -> do+ Scripts.onAutotaggedPool 3 10 1_800 1_800 scopeParams \_appName1 pool1 -> do+ Scripts.onAutotaggedPool 3 10 0.5 1_800 scopeParams \appName2 pool2 -> do+ res <- use pool2 $ Sessions.selectOne+ res `shouldBe` Right 1+ res2 <- use pool1 $ Sessions.countConnections appName2+ res2 `shouldBe` Right 1+ threadDelay 1_000_000 -- 1s+ res3 <- use pool1 $ Sessions.countConnections appName2+ res3 `shouldBe` Right 0++ it "Passively times out old connections" \scopeParams -> do+ -- 0.5s connection lifetime+ Scripts.onAutotaggedPool 1 10 0.5 1_800 scopeParams \_ pool -> do+ varName <- Scripts.generateVarname+ res <- use pool $ Sessions.setSetting varName "hello world"+ res `shouldBe` Right ()+ res2 <- use pool $ Sessions.getSetting varName+ res2 `shouldBe` Right (Just "hello world")+ threadDelay 1_000_000 -- 1s+ res3 <- use pool $ Sessions.getSetting varName+ res3 `shouldBe` Right Nothing
+ src/integration-tests/Specs/Config/IdlenessTimeoutSpec.hs view
@@ -0,0 +1,62 @@+module Specs.Config.IdlenessTimeoutSpec where++import Hasql.Pool+import Hasql.Pool.Config qualified as Config+import Helpers.Scripts qualified as Scripts+import Helpers.Sessions qualified as Sessions+import Prelude+import Test.Hspec++spec :: SpecWith Scripts.ScopeParams+spec = do+ it "Times out old connections (maxIdletime)" \scopeParams -> do+ -- 0.5s connection idle time+ Scripts.onAutotaggedPool 1 10 1_800 0.5 scopeParams \_ pool -> do+ varName <- Scripts.generateVarname+ res <- use pool $ Sessions.setSetting varName "hello world"+ res `shouldBe` Right ()+ res2 <- use pool $ Sessions.getSetting varName+ res2 `shouldBe` Right (Just "hello world")+ -- busy sleep, to keep connection alive+ forM_ [1 :: Int .. 10] $ \_ -> do+ r <- use pool $ Sessions.selectOne+ r `shouldBe` Right 1+ threadDelay 100_000 -- 0.1s+ res3 <- use pool $ Sessions.getSetting varName+ res3 `shouldBe` Right (Just "hello world")+ -- idle sleep, connection times out+ threadDelay 1_000_000 -- 1s+ res4 <- use pool $ Sessions.getSetting varName+ res4 `shouldBe` Right Nothing++ it "Passively times out idle connections" \scopeParams -> do+ -- 0.5s connection idle time, large lifetime, so only idleness can explain a passive close.+ Scripts.onAutotaggedPool 3 10 1_800 1_800 scopeParams \_appName1 pool1 -> do+ Scripts.onAutotaggedPool 3 10 1_800 0.5 scopeParams \appName2 pool2 -> do+ res <- use pool2 $ Sessions.selectOne+ res `shouldBe` Right 1+ res2 <- use pool1 $ Sessions.countConnections appName2+ res2 `shouldBe` Right 1+ -- Give the background reaper (1s tick) a chance to passively evict+ -- the now-idle connection, without ever calling `use pool2` again+ -- (which would trigger the separate active idleness check).+ threadDelay 1_500_000 -- 1.5s+ res3 <- use pool1 $ Sessions.countConnections appName2+ res3 `shouldBe` Right 0++ it "Times out the connections that aren't needed with LIFO" \scopeParams -> do+ -- 0.5s connection idle time+ Scripts.onAutotaggedPoolWithReuseOrder 2 10 1_800 0.5 Config.LIFO scopeParams \appName pool -> do+ held <- newEmptyMVar+ done <- newEmptyMVar+ -- Establish two connections.+ _ <- forkIO (void (use pool (liftIO (putMVar held () *> takeMVar done))))+ takeMVar held+ use pool Sessions.selectOne `shouldReturn` Right 1+ putMVar done ()+ -- Keep using the pool one pool request at a time, which needs only one+ -- connection, for longer than the idle time and the reaper's 1s tick.+ forM_ [1 :: Int .. 15] \_ -> do+ use pool Sessions.selectOne `shouldReturn` Right 1+ threadDelay 100_000 -- 0.1s+ use pool (Sessions.countConnections appName) `shouldReturn` Right 1
+ src/integration-tests/Specs/Config/InitSessionSpec.hs view
@@ -0,0 +1,59 @@+module Specs.Config.InitSessionSpec where++import Hasql.Pool+import Helpers.Scripts qualified as Scripts+import Helpers.Sessions qualified as Sessions+import Prelude+import Test.Hspec++spec :: SpecWith Scripts.ScopeParams+spec = do+ it "Persists after exceptions thrown in session" \scopeParams -> do+ Scripts.onAutotaggedPool 1 10 60 60 scopeParams \_ pool -> do+ varName <- Scripts.generateVarname++ res <- use pool do+ Sessions.setSetting varName "1"+ Sessions.getSetting varName+ shouldBe res (Right (Just "1"))++ try @SomeException do+ use pool do+ liftIO do+ throwIO (userError "Intentional error for testing")++ res <- use pool do+ Sessions.getSetting varName+ shouldBe res (Right (Just "1"))++ it "Persists after bad query" \scopeParams -> do+ Scripts.onAutotaggedPool 1 10 60 60 scopeParams \_ pool -> do+ varName <- Scripts.generateVarname++ res <- use pool do+ Sessions.setSetting varName "1"+ Sessions.getSetting varName+ shouldBe res (Right (Just "1"))++ use pool do+ Sessions.badQuery++ res <- use pool do+ Sessions.getSetting varName+ shouldBe res (Right (Just "1"))++ -- https://github.com/nikita-volkov/hasql-pool/issues/56+ it "Does not exhaust the pool capacity when it fails" \scopeParams -> do+ -- Pool of size 1 whose init session always fails, with a short+ -- acquisition timeout so that a leaked capacity slot shows up as+ -- an AcquisitionTimeoutUsageError instead of hanging the test.+ Scripts.onAutotaggedPoolWithInitSession 1 1 60 60 Sessions.badQuery scopeParams \_ pool -> do+ res1 <- use pool Sessions.selectOne+ res1 `shouldSatisfy` \case+ Left (SessionUsageError _) -> True+ _ -> False++ res2 <- use pool Sessions.selectOne+ res2 `shouldSatisfy` \case+ Left (SessionUsageError _) -> True+ _ -> False
+ src/integration-tests/Specs/Helpers/Sessions/CountConnectionsSpec.hs view
@@ -0,0 +1,14 @@+module Specs.Helpers.Sessions.CountConnectionsSpec where++import Hasql.Pool+import Helpers.Scripts qualified as Scripts+import Helpers.Sessions qualified as Sessions+import Prelude+import Test.Hspec++spec :: SpecWith Scripts.ScopeParams+spec = do+ it "Counts active connections" \scopeParams -> do+ Scripts.onAutotaggedPool 3 10 1_800 1_800 scopeParams \appName pool -> do+ res <- use pool $ Sessions.countConnections appName+ res `shouldBe` Right 1
+ src/integration-tests/Specs/Helpers/Sessions/GetSettingSpec.hs view
@@ -0,0 +1,36 @@+module Specs.Helpers.Sessions.GetSettingSpec where++import Hasql.Pool+import Helpers.Scripts qualified as Scripts+import Helpers.Sessions qualified as Sessions+import Prelude+import Test.Hspec++spec :: SpecWith Scripts.ScopeParams+spec = do+ it "Getting and setting session variables works" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ varName <- Scripts.generateVarname+ res <- use pool $ Sessions.getSetting varName+ res `shouldBe` Right Nothing+ res <- use pool $ do+ Sessions.setSetting varName "hello world"+ Sessions.getSetting varName+ res `shouldBe` Right (Just "hello world")++ it "Session variables stay set when a connection gets reused" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ varName <- Scripts.generateVarname+ res <- use pool $ Sessions.setSetting varName "hello world"+ res `shouldBe` Right ()+ res2 <- use pool $ Sessions.getSetting varName+ res2 `shouldBe` Right (Just "hello world")++ it "Releasing the pool resets session variables" \scopeParams -> do+ varName <- Scripts.generateVarname+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ res <- use pool $ Sessions.setSetting varName "hello world"+ res `shouldBe` Right ()+ release pool+ res <- use pool $ Sessions.getSetting varName+ res `shouldBe` Right Nothing
+ src/integration-tests/Specs/InterruptionSpec.hs view
@@ -0,0 +1,163 @@+module Specs.InterruptionSpec where++import Control.Exception (SomeException, try)+import Hasql.Pool+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as Config+import Hasql.Pool.Observation+import Helpers.Scripts qualified as Scripts+import Helpers.Sessions qualified as Sessions+import Prelude+import Test.Hspec++-- | Regression tests for the pool capacity getting lost when a pool request+-- gets interrupted anywhere between the enrollment and the session+-- (https://github.com/nikita-volkov/hasql-pool/issues/60), and for the pool+-- surviving failing observation handlers.+spec :: SpecWith Scripts.ScopeParams+spec = do+ it "Restores the pool capacity when the pool request is interrupted while observing the connection attempt" \scopeParams ->+ onGatedPool connectingObs 1 10 1_800 1_800 scopeParams \gate pool -> do+ interruptAt gate (\pool -> void (use pool Sessions.selectOne)) pool+ use pool Sessions.selectOne `shouldReturn` Right 1++ it "Restores the pool capacity when the pool request is interrupted while observing the idle-eviction" \scopeParams ->+ onGatedPool idleEvictionObs 1 10 1_800 0.3 scopeParams \gate pool -> do+ -- Establish a connection and let it go idle.+ _ <- use pool Sessions.selectOne+ threadDelay 600_000 -- 0.6s+ interruptAt gate (\pool -> void (use pool Sessions.selectOne)) pool+ use pool Sessions.selectOne `shouldReturn` Right 1++ it "Preserves the dequeued connection when the pool request is interrupted while observing its use" \scopeParams ->+ onGatedPool inUseObs 1 10 1_800 1_800 scopeParams \gate pool -> do+ -- Establish a connection and return it to the pool.+ _ <- use pool Sessions.selectOne+ interruptAt gate (\pool -> void (use pool Sessions.selectOne)) pool+ use pool Sessions.selectOne `shouldReturn` Right 1++ it "Succeeds the session when the observation handler throws synchronously during the connection attempt" \scopeParams ->+ Scripts.onAutotaggedPoolWithObservationHandler 1 10 1_800 1_800 (throwAt connectingObs) scopeParams \_ pool -> do+ use pool Sessions.selectOne `shouldReturn` Right 1++ it "Succeeds the session when the observation handler throws synchronously during the idle-eviction" \scopeParams ->+ Scripts.onAutotaggedPoolWithObservationHandler 1 10 1_800 0.3 (throwAt idleEvictionObs) scopeParams \_ pool -> do+ _ <- use pool Sessions.selectOne+ threadDelay 600_000 -- 0.6s+ use pool Sessions.selectOne `shouldReturn` Right 1++ it "Succeeds the session when the observation handler throws synchronously while observing the use" \scopeParams ->+ Scripts.onAutotaggedPoolWithObservationHandler 1 10 1_800 1_800 (throwAt inUseObs) scopeParams \_ pool -> do+ use pool Sessions.selectOne `shouldReturn` Right 1++ it "The reaper survives a throwing observation handler and keeps reaping on later ticks" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ counterPool ->+ Scripts.onAutotaggedPoolWithObservationHandler 1 10 1_800 0.3 (throwAt idleEvictionObs) scopeParams \appName pool -> do+ -- Establish a connection and let it go idle past the reaper's tick.+ _ <- use pool Sessions.selectOne+ threadDelay 1_600_000 -- 1.6s+ -- The connection got reaped and the reaper is still alive.+ use counterPool (Sessions.countConnections appName) `shouldReturn` Right 0+ use pool Sessions.selectOne `shouldReturn` Right 1+ -- It keeps reaping on the later ticks.+ threadDelay 1_600_000 -- 1.6s+ use counterPool (Sessions.countConnections appName) `shouldReturn` Right 0++ it "The pool remains usable after the release with a throwing observation handler" \scopeParams ->+ Scripts.onAutotaggedPoolWithObservationHandler 1 10 1_800 1_800 (throwAt releaseObs) scopeParams \_ pool -> do+ _ <- use pool Sessions.selectOne+ release pool+ use pool Sessions.selectOne `shouldReturn` Right 1++ it "Resolves the pool requests waiting on a garbage-collected pool" \scopeParams -> do+ (adapter, host, port) <- pure scopeParams+ result <- newEmptyMVar+ -- Inside this block the pool is only reachable from the waiting pool+ -- request. Once collected, the finalizer resolves it, and if it is not+ -- collectible, the acquisition timeout does. Either way it does not wait+ -- forever.+ do+ pool <-+ Pool.acquire+ adapter+ ( Config.settings+ [ -- No capacity: every pool request waits.+ Config.size 0,+ Config.acquisitionTimeout 0.5+ ]+ )+ _ <- forkIO (putMVar result =<< use pool Sessions.selectOne)+ threadDelay 100_000 -- 0.1s, to let the pool request start waiting+ pure ()+ performGC+ performGC+ res <- takeMVar result+ res `shouldBe` Left AcquisitionTimeoutUsageError++-- | Runs the action in a pool request interrupted at the gated observation.+-- The pool request is expected to die with the interruption.+interruptAt :: Gate -> (Pool.Pool -> IO ()) -> Pool.Pool -> IO ()+interruptAt gate action pool = do+ armGate gate+ done <- newEmptyMVar+ thread <- forkIO do+ _ <- try @SomeException (action pool)+ putMVar done ()+ takeMVar (gateEntered gate)+ killThread thread+ disarmGate gate+ takeMVar done++-- | Runs the continuation in a pool with a gated observation handler.+onGatedPool :: (Observation -> Bool) -> Int -> DiffTime -> DiffTime -> DiffTime -> Scripts.ScopeParams -> (Gate -> Pool.Pool -> IO ()) -> IO ()+onGatedPool matches poolSize acqTimeout maxLifetime maxIdletime scopeParams cont = do+ gate <- newGate matches+ Scripts.onAutotaggedPoolWithObservationHandler poolSize acqTimeout maxLifetime maxIdletime (gateHandler gate) scopeParams \appName pool -> cont gate pool++-- | A gate for stopping a pool request at a chosen observation and letting the+-- test interrupt it there. While the gate is armed, the pool request stops at+-- the first matching observation, waiting to be released or interrupted.+data Gate = Gate+ { gateMatches :: Observation -> Bool,+ gateEntered :: MVar (),+ gateProceed :: MVar (),+ gateArmed :: IORef Bool+ }++newGate :: (Observation -> Bool) -> IO Gate+newGate matches = Gate matches <$> newEmptyMVar <*> newEmptyMVar <*> newIORef False++armGate, disarmGate :: Gate -> IO ()+armGate gate = writeIORef (gateArmed gate) True+disarmGate gate = writeIORef (gateArmed gate) False++-- | The observation handler stopping at the gate.+gateHandler :: Gate -> Observation -> IO ()+gateHandler gate observation =+ when (gateMatches gate observation) do+ armed <- readIORef (gateArmed gate)+ when armed do+ putMVar (gateEntered gate) ()+ takeMVar (gateProceed gate)++-- | The observations the tests stop or throw at.+connectingObs, idleEvictionObs, inUseObs, releaseObs :: Observation -> Bool+connectingObs = \case+ ConnectionObservation _ ConnectingConnectionStatus -> True+ _ -> False+idleEvictionObs = \case+ ConnectionObservation _ (TerminatedConnectionStatus IdlenessConnectionTerminationReason) -> True+ _ -> False+inUseObs = \case+ ConnectionObservation _ InUseConnectionStatus -> True+ _ -> False+releaseObs = \case+ ConnectionObservation _ (TerminatedConnectionStatus ReleaseConnectionTerminationReason) -> True+ _ -> False++-- | An observation handler that throws synchronously at the matching+-- observations, and passes all the other observations through.+throwAt :: (Observation -> Bool) -> Observation -> IO ()+throwAt matches observation =+ when (matches observation) do+ error "Synchronous exception from the observation handler"
+ src/integration-tests/Specs/ReleaseSpec.hs view
@@ -0,0 +1,16 @@+module Specs.ReleaseSpec where++import Hasql.Pool+import Helpers.Scripts qualified as Scripts+import Helpers.Sessions qualified as Sessions+import Prelude+import Test.Hspec++spec :: SpecWith Scripts.ScopeParams+spec = do+ it "The pool remains usable after release" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ _ <- use pool $ Sessions.selectOne+ release pool+ res <- use pool $ Sessions.selectOne+ shouldSatisfy res $ isRight
+ src/integration-tests/Specs/SpecHook.hs view
@@ -0,0 +1,15 @@+-- Docs: https://hspec.github.io/hspec-discover.html+module Specs.SpecHook where++import Helpers.Adapters qualified as Adapters+import Helpers.Hooks qualified as Hooks+import Helpers.Scripts qualified as Scripts+import Test.Hspec++hook :: SpecWith Scripts.ScopeParams -> Spec+hook hookedSpec =+ Adapters.hook+ ( aroundAllWith+ (\action adapter -> Hooks.postgres17 \(host, port) -> action (adapter, host, port))+ (parallel hookedSpec)+ )
+ src/integration-tests/Specs/UsageError/AcquisitionTimeoutSpec.hs view
@@ -0,0 +1,45 @@+module Specs.UsageError.AcquisitionTimeoutSpec where++import Control.Concurrent.Async (forConcurrently, race)+import Hasql.Pool+import Helpers.Scripts qualified as Scripts+import Helpers.Sessions qualified as Sessions+import Prelude+import Test.Hspec++spec :: SpecWith Scripts.ScopeParams+spec = do+ it "Gets produced on timeout" \scopeParams ->+ -- 1ms timeout+ Scripts.onAutotaggedPool 1 0.001 1_800 1_800 scopeParams \_ pool -> do+ sleeping <- newEmptyMVar+ t0 <- getCurrentTime+ res <-+ race+ ( use pool+ $ liftIO+ $ do+ putMVar sleeping ()+ -- 1s+ threadDelay 1_000_000+ )+ ( do+ takeMVar sleeping+ use pool $ Sessions.selectOne+ )+ t1 <- getCurrentTime+ res `shouldBe` Right (Left AcquisitionTimeoutUsageError)+ -- 0.5s+ diffUTCTime t1 t0 `shouldSatisfy` (< 0.5)++ it "Gets produced for every waiting pool request, and the pool remains usable" \scopeParams ->+ -- 100ms timeout+ Scripts.onAutotaggedPool 1 0.1 1_800 1_800 scopeParams \_ pool -> do+ held <- newEmptyMVar+ done <- newEmptyMVar+ _ <- forkIO (void (use pool (liftIO (putMVar held () *> takeMVar done))))+ takeMVar held+ results <- forConcurrently [1 .. 3 :: Int] \_ -> use pool Sessions.selectOne+ results `shouldBe` replicate 3 (Left AcquisitionTimeoutUsageError)+ putMVar done ()+ use pool Sessions.selectOne `shouldReturn` Right 1
+ src/integration-tests/Specs/UsageError/SessionSpec.hs view
@@ -0,0 +1,16 @@+module Specs.UsageError.SessionSpec where++import Hasql.Pool+import Helpers.Scripts qualified as Scripts+import Helpers.Sessions qualified as Sessions+import Prelude+import Test.Hspec++spec :: SpecWith Scripts.ScopeParams+spec = do+ it "Bad SQL query triggers error" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ res <- use pool Sessions.badQuery+ shouldSatisfy res $ \case+ Left (SessionUsageError _) -> True+ _ -> False
+ src/integration-tests/Specs/UseSpec.hs view
@@ -0,0 +1,252 @@+module Specs.UseSpec where++import Control.Concurrent.Async (race)+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.Pool.Config qualified as Config+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 Prelude+import Test.Hspec++spec :: SpecWith Scripts.ScopeParams+spec = do+ it "Releases a spot in the pool when there is a query error" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ use pool Sessions.badQuery `shouldNotReturn` (Right ())+ use pool Sessions.selectOne `shouldReturn` (Right 1)++ it "Connection errors cause eviction of connection" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ _ <- use pool $ Sessions.closeConn >> Sessions.selectOne+ _ <- use pool $ Sessions.closeConn >> Sessions.selectOne+ _ <- use pool $ Sessions.closeConn >> Sessions.selectOne+ 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++ -- After a server restart or @pg_terminate_backend@ the first statement on+ -- the closed connection can fail with an 'Errors.UnexpectedRowCountStatementError'+ -- instead of a 'Errors.ConnectionSessionError'. The pool must not hand the+ -- connection out again.+ it "Connections closed by the server cause eviction of connection" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \appName pool -> do+ use pool Sessions.selectOne `shouldReturn` Right 1+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ adminPool ->+ use adminPool (terminateBackends appName) `shouldReturn` Right [True]+ _ <- use pool Sessions.selectOne+ use pool Sessions.selectOne `shouldReturn` Right 1++ it "Unexpected row counts on a healthy connection don't cause eviction of connection" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ Right pid <- use pool backendPid+ result <- use pool selectTwoRowsAsSingleRow+ result `shouldSatisfy` \case+ Left (SessionUsageError (Errors.StatementSessionError _ _ _ _ _ Errors.UnexpectedRowCountStatementError {})) -> True+ _ -> False+ use pool backendPid `shouldReturn` Right pid++ 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+ _ <- use pool $ Sessions.selectOne+ _ <- use pool $ Sessions.selectOne+ _ <- use pool $ Sessions.selectOne+ res <- use pool $ Sessions.selectOne+ shouldSatisfy res $ isRight++ it "Connection gets returned to the pool after non-connection error" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ _ <- use pool $ Sessions.badQuery+ _ <- use pool $ Sessions.badQuery+ _ <- use pool $ Sessions.badQuery+ _ <- use pool $ Sessions.badQuery+ res <- use pool $ Sessions.selectOne+ shouldSatisfy res $ isRight++ -- https://github.com/nikita-volkov/hasql-pool/issues/38+ --+ -- When a session is interrupted by an asynchronous exception (e.g., a+ -- caller-side timeout racing the query, as simulated here via `race`)+ -- while it is genuinely blocked waiting on the server's response, the+ -- underlying libpq connection is left mid-command: the query was sent,+ -- but its result was never read. `onLiveConn` in Hasql.Pool.use still+ -- unconditionally returns such a connection to the pool (the `Left exc`+ -- branch calls `returnConn` for all exceptions, not just synchronous+ -- ones), so the next `use` call hands out a connection whose protocol+ -- state is desynced from libpq's expectations. This is a plausible root+ -- cause of the "connection pointer is NULL" reports: two independent+ -- consumers of hasql-pool end up driving the same libpq connection state+ -- machine without coordination.+ it "Does not return a connection to the pool when the session is interrupted by an asynchronous exception" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ started <- newEmptyMVar+ _ <-+ race+ ( use pool do+ liftIO $ putMVar started ()+ Sessions.sleep 2+ )+ ( do+ takeMVar started+ -- Give the query time to actually reach the server and for+ -- the client to start blocking on the socket read, as+ -- opposed to being cancelled while still sending.+ threadDelay 200_000+ )+ res <- use pool Sessions.selectOne+ res `shouldSatisfy` isRight++ it "Gives a freed connection to a pool request waiting for it rather than to a later one" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ order <- newIORef []+ held <- newEmptyMVar+ done <- newEmptyMVar+ finished <- newEmptyMVar+ let record name = use pool (liftIO (atomicModifyIORef' order \names -> (names <> [name], ())))+ -- Requests a connection again as soon as it returns one, before the+ -- waiting pool request gets to run.+ _ <- forkIO do+ _ <- use pool (liftIO (putMVar held () *> takeMVar done))+ _ <- record "later"+ putMVar finished ()+ takeMVar held+ _ <- forkIO (record "waiting" *> putMVar finished ())+ -- Give the waiting pool request time to start waiting.+ threadDelay 100_000+ putMVar done ()+ replicateM_ 2 (takeMVar finished)+ readIORef order `shouldReturn` ["waiting", "later"]++ it "Loses no capacity when a pool request waiting for a connection is interrupted" \scopeParams ->+ Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do+ held <- newEmptyMVar+ done <- newEmptyMVar+ _ <- forkIO (void (use pool (liftIO (putMVar held () *> takeMVar done))))+ takeMVar held+ waiting <- forkIO (void (use pool Sessions.selectOne))+ -- Give the pool request time to start waiting.+ threadDelay 100_000+ killThread waiting+ putMVar done ()+ use pool Sessions.selectOne `shouldReturn` Right 1++ it "Reuses the most recently returned connection with LIFO" \scopeParams ->+ Scripts.onAutotaggedPoolWithReuseOrder 2 10 1_800 1_800 Config.LIFO scopeParams \_ pool -> do+ (first, second) <- returnTwoConnections pool+ first `shouldNotBe` second+ use pool Sessions.backendPid `shouldReturn` Right second++ it "Reuses the least recently returned connection by default" \scopeParams ->+ Scripts.onAutotaggedPool 2 10 1_800 1_800 scopeParams \_ pool -> do+ (first, second) <- returnTwoConnections pool+ first `shouldNotBe` second+ use pool Sessions.backendPid `shouldReturn` Right first++ 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))))++terminateBackends :: Text -> Session.Session [Bool]+terminateBackends appName =+ Session.statement appName statement+ where+ statement =+ Statement.unpreparable+ "select pg_terminate_backend(pid, 10000) from pg_stat_activity where application_name = $1"+ (Encoders.param (Encoders.nonNullable Encoders.text))+ (Decoders.rowList (Decoders.column (Decoders.nonNullable Decoders.bool)))++backendPid :: Session.Session Int32+backendPid =+ Session.statement () statement+ where+ statement =+ Statement.preparable+ "select pg_backend_pid()"+ Encoders.noParams+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int4)))++selectTwoRowsAsSingleRow :: Session.Session Int64+selectTwoRowsAsSingleRow =+ Session.statement () statement+ where+ statement =+ Statement.preparable+ "select * from (values (1 :: int8), (2 :: int8)) as t"+ Encoders.noParams+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))++driverError :: Session.Session ()+driverError =+ Session.onLibpqConnection \connection ->+ pure (Left (Errors.DriverSessionError "synthetic driver error"), connection)++-- | Establish two connections, return the first before the second, and give+-- their backend PIDs.+returnTwoConnections :: Pool -> IO (Int32, Int32)+returnTwoConnections pool = do+ held <- newEmptyMVar+ done <- newEmptyMVar+ returned <- newEmptyMVar+ _ <- forkIO do+ res <- use pool (Sessions.backendPid <* liftIO (putMVar held () *> takeMVar done))+ putMVar returned res+ takeMVar held+ res <- use pool do+ second <- Sessions.backendPid+ first <- liftIO do+ putMVar done ()+ takeMVar returned+ pure (first, second)+ case res of+ Right (Right first, second) -> pure (first, second)+ _ -> fail ("Unexpected result: " <> show res)
+ src/library/Hasql/Pool.hs view
@@ -0,0 +1,511 @@+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.Config.ReuseOrder (ReuseOrder (..))+import Hasql.Pool.Observation+import Hasql.Pool.Prelude+import Hasql.Pool.SessionErrorDestructors qualified as ErrorsDestruction+import Hasql.Session qualified as Session+import Pqi qualified++-- | 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++-- | A pool request waiting for a connection.+data Waiter = Waiter+ { -- | When the pool request times out, in nanoseconds of the monotonic clock.+ waiterDeadline :: Word64,+ -- | What the pool request gets.+ waiterGiven :: TMVar Given,+ -- | Whether the pool request stopped waiting, so that it gets passed over.+ waiterGone :: TVar Bool+ }++-- | What a waiting pool request gets: an idle connection, or the slot of a+-- closed one to establish a connection in, with the reuse flag at the time,+-- or a timeout.+data Given+ = Given (TVar Bool) (Maybe Entry)+ | TimedOut++-- | What a pool request holds after the enrollment transaction commits and+-- has to give back on any failure.+data Holding+ = -- | An entry dequeued from the idle connections.+ EntryHolding (TVar Bool) Entry+ | -- | A capacity unit taken from the pool.+ SlotHolding (TVar Bool)+ | -- | A spot in the waiting queue.+ WaitingHolding Waiter++-- | Pool of connections to DB.+data Pool = Pool+ { -- | Pool size.+ poolSize :: Int,+ -- | Adapter used to establish connections.+ poolAdapter :: Pqi.Adapter,+ -- | Connection settings.+ poolFetchConnectionSettings :: IO Connection.Settings.Settings,+ -- | Acquisition timeout, in nanoseconds.+ poolAcquisitionTimeout :: Word64,+ -- | Maximal connection lifetime, in nanoseconds.+ poolMaxLifetime :: Word64,+ -- | Maximal connection idle time, in nanoseconds.+ poolMaxIdletime :: Word64,+ -- | Avail connections, in the order they get reused.+ poolConnectionQueue :: TQueue Entry,+ -- | Which end of the queue returned connections go to.+ poolReuseOrder :: ReuseOrder,+ -- | Remaining capacity.+ -- The pool size limits the sum of poolCapacity, the length+ -- of poolConnectionQueue and the number of in-flight+ -- connections.+ poolCapacity :: TVar Int,+ -- | Pool requests waiting for a connection, the longest waiting first.+ -- Freed connections and slots go to them before anyone else,+ -- so that only the pool request given one wakes up.+ poolWaiting :: TQueue Waiter,+ -- | Whether to return a connection to the pool.+ poolReuseVar :: TVar (TVar Bool),+ -- | To stop the manager and expirer threads via garbage collection.+ poolReaperRef :: IORef (),+ -- | Action for reporting the observations.+ poolObserver :: Observation -> IO (),+ -- | Initial session to execute upon every established connection.+ poolInitSession :: Session.Session ()+ }++-- | Contain the failures of the user-supplied observation handler. Synchronous+-- exceptions from it are discarded, asynchronous ones are honored, so that+-- neither affects the pool's state or the session result, and the pool's+-- threads stay killable.+containObserver :: (Observation -> IO ()) -> Observation -> IO ()+containObserver observer observation =+ catch @SomeException (observer observation) \exc ->+ case fromException exc of+ Just SomeAsyncException {} -> throwIO exc+ Nothing -> return ()++-- | 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.+--+-- The 'Pqi.Adapter' determines which connection implementation the pool uses, e.g. an FFI adapter backed by @postgresql-libpq@, or a pure Haskell one. Pick one from an adapter package such as @pqi-ffi@ or @pqi-native@.+acquire :: Pqi.Adapter -> Config.Config -> IO Pool+acquire adapter config = do+ connectionQueue <- newTQueueIO+ capVar <- newTVarIO (Config.size config)+ reuseVar <- newTVarIO =<< newTVarIO True+ waiting <- newTQueueIO+ reaperRef <- newIORef ()+ let observer = containObserver (Config.observationHandler config)++ 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 maxIdletimeNanos now) unagedEntries+ traverse_ (writeTQueue connectionQueue) liveEntries+ return $ do+ forM_ agedEntries $ reap observer reuseVar waiting capVar AgingConnectionTerminationReason+ forM_ idleEntries $ reap observer reuseVar waiting capVar IdlenessConnectionTerminationReason++ expirerTid <- forkIOWithUnmask $ \unmask -> unmask $ expire waiting++ void . mkWeakIORef reaperRef $ do+ -- When the pool goes out of scope, resolve the pool requests still+ -- waiting, and stop the manager and the expirer.+ atomically $ drainWaiting waiting+ killThread managerTid+ killThread expirerTid++ return $ Pool (Config.size config) adapter (Config.connectionSettingsProvider config) acqTimeoutNanos agingTimeoutNanos maxIdletimeNanos connectionQueue (Config.reuseOrder config) capVar waiting reuseVar reaperRef observer (Config.initSession config)+ where+ acqTimeoutNanos =+ div (fromIntegral (diffTimeToPicoseconds (Config.acquisitionTimeout config))) 1_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 ->+ mask \restore ->+ catch @SomeException+ ( do+ restore $ Connection.release (entryConnection entry)+ restore $ poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason))+ atomically $ freeSlot poolReuseVar poolWaiting poolCapacity+ )+ \exc -> do+ atomically $ freeSlot poolReuseVar poolWaiting poolCapacity+ case fromException exc of+ Just SomeAsyncException {} -> throwIO exc+ Nothing -> return ()++-- | Use a connection from the pool to run a session and return the connection+-- to the pool, when finished.+--+-- Session failing with a 'Errors.ConnectionSessionError' or leaving the+-- connection in the 'Pqi.ConnectionBad' state 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.+--+-- When no connection is available, the pool requests wait for one in the+-- order they came, and a later pool request doesn't take a connection while+-- others wait.+use :: forall a. Pool -> Session.Session a -> IO (Either UsageError a)+use Pool {..} sess = do+ now <- getMonotonicTimeNSec+ mask \unmask -> do+ holding <- atomically $ do+ reuseVar <- readTVar poolReuseVar+ waited <- someoneWaiting poolWaiting+ -- Behind the pool requests already waiting, which get what's freed first.+ if waited+ then enqueue now <&> WaitingHolding+ else+ asum+ [ readTQueue poolConnectionQueue <&> EntryHolding reuseVar,+ do+ capVal <- readTVar poolCapacity+ if capVal > 0+ then do+ writeTVar poolCapacity $! pred capVal+ return $ SlotHolding reuseVar+ else retry,+ enqueue now <&> WaitingHolding+ ]+ case holding of+ EntryHolding reuseVar entry -> onEntry unmask reuseVar entry+ SlotHolding reuseVar -> onSlot unmask reuseVar+ WaitingHolding waiter -> do+ given <- atomically (takeTMVar (waiterGiven waiter)) `onException` abandon waiter+ case given of+ TimedOut -> return $ Left AcquisitionTimeoutUsageError+ Given reuseVar Nothing -> onSlot unmask reuseVar+ Given reuseVar (Just entry) -> onEntry unmask reuseVar entry+ where+ enqueue :: Word64 -> STM Waiter+ enqueue now = do+ waiter <- Waiter (now + poolAcquisitionTimeout) <$> newEmptyTMVar <*> newTVar False+ writeTQueue poolWaiting waiter+ return waiter++ abandon :: Waiter -> IO ()+ abandon waiter = do+ given <- atomically $ do+ writeTVar (waiterGone waiter) True+ tryTakeTMVar (waiterGiven waiter)+ forM_ given \case+ Given _ Nothing -> atomically returnSlot+ Given reuseVar (Just entry) -> returnConn reuseVar entry+ TimedOut -> return ()++ returnSlot :: STM ()+ returnSlot = freeSlot poolReuseVar poolWaiting poolCapacity++ -- \| Holding one entry dequeued from the idle connections. If it is still+ -- good, run the session on it; if it expired, replace it with a fresh+ -- connection on the same slot. All the way, the holding is given back on+ -- interruption.+ onEntry :: (forall x. IO x -> IO x) -> TVar Bool -> Entry -> IO (Either UsageError a)+ onEntry unmask reuseVar entry = do+ now <- getMonotonicTimeNSec+ if entryIsAged poolMaxLifetime now entry+ then renewEntry unmask reuseVar entry AgingConnectionTerminationReason+ else+ if entryIsIdle poolMaxIdletime now entry+ then renewEntry unmask reuseVar entry IdlenessConnectionTerminationReason+ else onLiveConn unmask reuseVar entry {entryUseTimeNSec = now}++ -- \| Holding an expired entry. Release its connection and establish a+ -- fresh one on the same slot. On interruption the slot gets given back:+ -- the released connection cannot return to the pool.+ renewEntry :: (forall x. IO x -> IO x) -> TVar Bool -> Entry -> ConnectionTerminationReason -> IO (Either UsageError a)+ renewEntry unmask reuseVar entry reason = do+ outcome <-+ onException+ ( do+ Connection.release (entryConnection entry)+ unmask $ poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus reason))+ establish unmask+ )+ (atomically returnSlot)+ case outcome of+ Left err -> return $ Left err+ Right newEntry -> onLiveConn unmask reuseVar newEntry++ -- \| Holding one capacity unit. Establish a connection on it and run the+ -- session. On interruption the slot gets given back.+ onSlot :: (forall x. IO x -> IO x) -> TVar Bool -> IO (Either UsageError a)+ onSlot unmask reuseVar = do+ outcome <- onException (establish unmask) (atomically returnSlot)+ case outcome of+ Left err -> return $ Left err+ Right entry -> onLiveConn unmask reuseVar entry++ -- Run with one capacity unit held: establish the connection and run the+ -- init session on it. On any failure the slot gets given back and the+ -- error gets returned. On interruption the slot gets given back and the+ -- exception gets rethrown.+ establish :: (forall x. IO x -> IO x) -> IO (Either UsageError Entry)+ establish unmask = do+ settings <- unmask poolFetchConnectionSettings+ now <- getMonotonicTimeNSec+ id <- Uuid.nextRandom+ unmask $ poolObserver (ConnectionObservation id ConnectingConnectionStatus)+ unmask (Connection.acquire poolAdapter 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+ unmask $ poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason connErrText)))+ atomically returnSlot+ return $ Left $ ConnectionUsageError connErr+ Right connection ->+ unmask (Connection.use connection poolInitSession) >>= \case+ Left err -> do+ Connection.release connection+ unmask+ $ ErrorsDestruction.reset+ (\details -> poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (Just details)))))+ (poolObserver (ConnectionObservation id (TerminatedConnectionStatus (InitializationErrorTerminationReason err))))+ err+ atomically returnSlot+ return $ Left $ SessionUsageError err+ Right () -> do+ unmask $ poolObserver (ConnectionObservation id (ReadyForUseConnectionStatus EstablishedConnectionReadyForUseReason))+ return $ Right (Entry connection now now id)++ -- \| Holding one entry ready for use. Run the session on it, return the+ -- entry to the pool, and return the session result. Interrupted, the+ -- session exception gets rethrown after the entry is back in the pool.+ onLiveConn :: (forall x. IO x -> IO x) -> TVar Bool -> Entry -> IO (Either UsageError a)+ onLiveConn unmask reuseVar entry = do+ sessRes <-+ onException+ ( unmask $ do+ poolObserver (ConnectionObservation (entryId entry) InUseConnectionStatus)+ try @SomeException (Connection.use (entryConnection entry) sess)+ )+ (returnConn reuseVar entry)++ case sessRes of+ Left exc -> do+ returnConn reuseVar entry+ throwIO exc+ Right (Left err) -> do+ discard <-+ if ErrorsDestruction.requiresConnectionDiscard err+ then return True+ else connectionIsLost (entryConnection entry)+ if discard+ then do+ Connection.release (entryConnection entry)+ atomically returnSlot+ unmask+ $ poolObserver+ ( ConnectionObservation+ (entryId entry)+ (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (Just (Errors.toMessage err))))+ )+ return $ Left $ SessionUsageError err+ else do+ returnConn reuseVar entry+ unmask $ poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus (SessionFailedConnectionReadyForUseReason err)))+ return $ Left $ SessionUsageError err+ Right (Right res) -> do+ returnConn reuseVar entry+ unmask $ poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus SessionSucceededConnectionReadyForUseReason))+ return $ Right res++ -- Back to the pool request waiting longest or to the idle connections,+ -- unless the pool was released meanwhile.+ returnConn reuseVar entry =+ join . atomically $ do+ reuse <- readTVar reuseVar+ if reuse+ then do+ nextWaiter poolWaiting >>= \case+ Just waiter -> do+ currentReuseVar <- readTVar poolReuseVar+ putTMVar (waiterGiven waiter) (Given currentReuseVar (Just entry))+ Nothing ->+ case poolReuseOrder of+ LIFO -> unGetTQueue poolConnectionQueue entry+ FIFO -> writeTQueue poolConnectionQueue entry+ return $ return ()+ else return $ do+ Connection.release (entryConnection entry)+ atomically returnSlot+ poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason))++-- | Whether the adapter has marked the connection as lost.+--+-- When the server closes an idle connection, e.g. on restart or with+-- @pg_terminate_backend@, the next statement on it can fail with an error+-- that doesn't say so: 'Errors.UnexpectedRowCountStatementError', for the+-- extra result that reports the closed connection. A statement returning more+-- rows than its decoder expects fails the same way on a healthy connection, so+-- the error can't tell the two apart. The connection status can: the adapter+-- has set it to 'Pqi.ConnectionBad' by then.+connectionIsLost :: Connection -> IO Bool+connectionIsLost connection =+ Connection.use connection checkStatus <&> \case+ Left _ -> True+ Right lost -> lost+ where+ checkStatus =+ Session.onLibpqConnection \pqConnection -> do+ status <- Pqi.status pqConnection+ return (Right (status == Pqi.ConnectionBad), pqConnection)++-- | Whether a pool request is waiting. The ones that stopped waiting get+-- dropped.+someoneWaiting :: TQueue Waiter -> STM Bool+someoneWaiting waiting =+ tryPeekTQueue waiting >>= \case+ Nothing -> return False+ Just waiter ->+ readTVar (waiterGone waiter) >>= \case+ True -> readTQueue waiting *> someoneWaiting waiting+ False -> return True++-- | Take the pool request waiting longest, passing over the ones that stopped+-- waiting.+nextWaiter :: TQueue Waiter -> STM (Maybe Waiter)+nextWaiter waiting =+ tryReadTQueue waiting >>= \case+ Nothing -> return Nothing+ Just waiter ->+ readTVar (waiterGone waiter) >>= \case+ True -> nextWaiter waiting+ False -> return (Just waiter)++-- | Give the slot of a closed connection to the pool request waiting longest,+-- or back to the pool if no pool request waits.+freeSlot :: TVar (TVar Bool) -> TQueue Waiter -> TVar Int -> STM ()+freeSlot reuseVarVar waiting capVar =+ nextWaiter waiting >>= \case+ Just waiter -> do+ reuseVar <- readTVar reuseVarVar+ putTMVar (waiterGiven waiter) (Given reuseVar Nothing)+ Nothing -> modifyTVar' capVar succ++-- | Time out the pool requests waiting too long. All wait as long, so the pool+-- request waiting longest times out first: one thread sleeps until its+-- deadline, rather than every waiting pool request registering a timer, which+-- GHC can't cancel.+expire :: TQueue Waiter -> IO ()+expire waiting = forever $ do+ deadline <- atomically $ do+ waited <- someoneWaiting waiting+ if waited then waiterDeadline <$> peekTQueue waiting else retry+ now <- getMonotonicTimeNSec+ -- In microseconds, at most an hour at a time.+ when (deadline > now) $ threadDelay (fromIntegral (min 3_600_000_000 (div (deadline - now) 1_000 + 1)))+ atomically . timeOut =<< getMonotonicTimeNSec+ where+ timeOut now =+ tryPeekTQueue waiting >>= \case+ Just waiter | waiterDeadline waiter <= now -> do+ _ <- readTQueue waiting+ gone <- readTVar (waiterGone waiter)+ unless gone $ putTMVar (waiterGiven waiter) TimedOut+ timeOut now+ _ -> return ()++-- | Resolve the pool requests still waiting: with the pool gone, nothing will+-- ever be given to them.+drainWaiting :: TQueue Waiter -> STM ()+drainWaiting waiting =+ tryReadTQueue waiting >>= \case+ Nothing -> return ()+ Just waiter -> do+ gone <- readTVar (waiterGone waiter)+ unless gone $ void $ tryPutTMVar (waiterGiven waiter) TimedOut+ drainWaiting waiting++-- | Terminate one expired connection on the reaper's tick: release it, report+-- the termination, and give its slot back. The slot gets given back even when+-- the release or the observation fails, so no capacity is lost, and a failing+-- entry neither skips the other entries of the tick nor kills the reaper.+reap :: (Observation -> IO ()) -> TVar (TVar Bool) -> TQueue Waiter -> TVar Int -> ConnectionTerminationReason -> Entry -> IO ()+reap observer reuseVar waiting capVar reason entry =+ mask \restore ->+ catch @SomeException+ ( do+ restore $ Connection.release (entryConnection entry)+ restore $ observer (ConnectionObservation (entryId entry) (TerminatedConnectionStatus reason))+ atomically $ freeSlot reuseVar waiting capVar+ )+ \exc -> do+ atomically $ freeSlot reuseVar waiting capVar+ case fromException exc of+ Just SomeAsyncException {} -> throwIO exc+ Nothing -> return ()++-- | 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
+ src/library/Hasql/Pool/Config.hs view
@@ -0,0 +1,28 @@+-- | 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,+ Setting.reuseOrder,+ ReuseOrder.ReuseOrder (..),+ )+where++import Hasql.Pool.Config.Config qualified as Config+import Hasql.Pool.Config.ReuseOrder qualified as ReuseOrder+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
+ src/library/Hasql/Pool/Config/Config.hs view
@@ -0,0 +1,34 @@+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.Config.ReuseOrder (ReuseOrder)+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 (),+ reuseOrder :: ReuseOrder+ }++-- | 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,+ reuseOrder = Defaults.reuseOrder+ }
+ src/library/Hasql/Pool/Config/Defaults.hs view
@@ -0,0 +1,53 @@+module Hasql.Pool.Config.Defaults where++import Hasql.Connection.Settings qualified as Connection.Settings+import Hasql.Pool.Config.ReuseOrder (ReuseOrder (..))+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 ()++-- |+-- The idle connection returned longest ago first.+reuseOrder :: ReuseOrder+reuseOrder = FIFO
+ src/library/Hasql/Pool/Config/ReuseOrder.hs view
@@ -0,0 +1,17 @@+module Hasql.Pool.Config.ReuseOrder where++import Hasql.Pool.Prelude++-- | Which of the idle connections a pool request gets.+data ReuseOrder+ = -- | Last in, first out: the one returned most recently.+ --+ -- The pool keeps reusing the few connections the load needs, so the rest+ -- time out as idle, and the ones in use keep their caches warm.+ LIFO+ | -- | First in, first out: the one returned longest ago.+ --+ -- Pool requests cycle through all the connections, so steady load keeps+ -- every connection open.+ FIFO+ deriving (Show, Eq)
+ src/library/Hasql/Pool/Config/Setting.hs view
@@ -0,0 +1,111 @@+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.Config.ReuseOrder (ReuseOrder (..))+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.+--+-- An observation handler must not throw. Synchronous exceptions from it are discarded; asynchronous exceptions are honored. Its failure never affects the session result or the pool's state.+--+-- 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})++-- | Which of the idle connections a pool request gets.+--+-- With 'LIFO', the pool shrinks to the connections the load needs: the rest+-- time out as idle (see 'idlenessTimeout'). With 'FIFO', steady load keeps+-- every connection open.+--+-- 'FIFO' by default.+reuseOrder :: ReuseOrder -> Setting+reuseOrder x =+ Setting (\config -> config {Config.reuseOrder = x})
+ src/library/Hasql/Pool/Observation.hs view
@@ -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.+--+-- <<diagrams-output/connection-status-model.png>>+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)
+ src/library/Hasql/Pool/Prelude.hs view
@@ -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 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, (.))+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
+ src/library/Hasql/Pool/SessionErrorDestructors.hs view
@@ -0,0 +1,30 @@+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++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"
− test/Main.hs
@@ -1,147 +0,0 @@-module Main where--import Control.Concurrent.Async (race)-import qualified Data.ByteString.Char8 as B8-import qualified Hasql.Connection as Connection-import qualified Hasql.Decoders as Decoders-import qualified Hasql.Encoders as Encoders-import Hasql.Pool-import qualified Hasql.Session as Session-import qualified Hasql.Statement as Statement-import qualified System.Environment-import Test.Hspec-import Prelude--main = do- connectionSettings <- getConnectionSettings- hspec . describe "" $ do- it "Releases a spot in the pool when there is a query error" $ do- pool <- acquire 1 Nothing connectionSettings- use pool badQuerySession `shouldNotReturn` (Right ())- use pool selectOneSession `shouldReturn` (Right 1)- it "Simulation of connection error works" $ do- pool <- acquire 3 Nothing connectionSettings- res <- use pool $ closeConnSession >> selectOneSession- shouldSatisfy res $ \case- Left (SessionUsageError (Session.QueryError _ _ (Session.ClientError _))) -> True- _ -> False- it "Connection errors cause eviction of connection" $ do- pool <- acquire 3 Nothing connectionSettings- res <- use pool $ closeConnSession >> selectOneSession- res <- use pool $ closeConnSession >> selectOneSession- res <- use pool $ closeConnSession >> selectOneSession- res <- use pool $ selectOneSession- shouldSatisfy res $ isRight- it "Connection gets returned to the pool after normal use" $ do- pool <- acquire 3 Nothing connectionSettings- res <- use pool $ selectOneSession- res <- use pool $ selectOneSession- res <- use pool $ selectOneSession- res <- use pool $ selectOneSession- res <- use pool $ selectOneSession- shouldSatisfy res $ isRight- it "Connection gets returned to the pool after non-connection error" $ do- pool <- acquire 3 Nothing connectionSettings- res <- use pool $ badQuerySession- res <- use pool $ badQuerySession- res <- use pool $ badQuerySession- res <- use pool $ badQuerySession- res <- use pool $ selectOneSession- shouldSatisfy res $ isRight- it "The pool remains usable after release" $ do- pool <- acquire 1 Nothing connectionSettings- res <- use pool $ selectOneSession- release pool- res <- use pool $ selectOneSession- shouldSatisfy res $ isRight- it "Getting and setting session variables works" $ do- pool <- acquire 1 Nothing connectionSettings- res <- use pool $ getSettingSession "testing.foo"- res `shouldBe` Right Nothing- res <- use pool $ do- setSettingSession "testing.foo" "hello world"- getSettingSession "testing.foo"- res `shouldBe` Right (Just "hello world")- it "Session variables stay set when a connection gets reused" $ do- pool <- acquire 1 Nothing connectionSettings- res <- use pool $ setSettingSession "testing.foo" "hello world"- res `shouldBe` Right ()- res2 <- use pool $ getSettingSession "testing.foo"- res2 `shouldBe` Right (Just "hello world")- it "Releasing the pool resets session variables" $ do- pool <- acquire 1 Nothing connectionSettings- res <- use pool $ setSettingSession "testing.foo" "hello world"- res `shouldBe` Right ()- release pool- res <- use pool $ getSettingSession "testing.foo"- res `shouldBe` Right Nothing- it "Times out connection acquisition" $ do- pool <- acquire 1 (Just 1000) connectionSettings -- 1ms timeout- sleeping <- newEmptyMVar- t0 <- getCurrentTime- res <-- race- ( use pool $- liftIO $ do- putMVar sleeping ()- threadDelay 1000000 -- 1s- )- ( do- takeMVar sleeping- use pool $ selectOneSession- )- t1 <- getCurrentTime- res `shouldBe` Right (Left AcquisitionTimeoutUsageError)- diffUTCTime t1 t0 `shouldSatisfy` (< 0.5) -- 0.5s--getConnectionSettings :: IO Connection.Settings-getConnectionSettings =- B8.unwords . catMaybes- <$> sequence- [ setting "host" $ defaultEnv "POSTGRES_HOST" "localhost",- setting "port" $ defaultEnv "POSTGRES_PORT" "5432",- setting "user" $ defaultEnv "POSTGRES_USER" "postgres",- setting "password" $ maybeEnv "POSTGRES_PASSWORD",- setting "dbname" $ defaultEnv "POSTGRES_DBNAME" "postgres"- ]- where- maybeEnv env = fmap B8.pack <$> System.Environment.lookupEnv env- defaultEnv env val = Just . fromMaybe val <$> maybeEnv env- setting label getEnv = do- val <- getEnv- return $ (\v -> label <> "=" <> v) <$> val--selectOneSession :: Session.Session Int64-selectOneSession =- Session.statement () statement- where- statement = Statement.Statement "SELECT 1" Encoders.noParams decoder True- decoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8))--badQuerySession :: Session.Session ()-badQuerySession =- Session.statement () statement- where- statement = Statement.Statement "zzz" Encoders.noParams Decoders.noResult True--closeConnSession :: Session.Session ()-closeConnSession = do- conn <- ask- liftIO $ Connection.release conn--setSettingSession :: Text -> Text -> Session.Session ()-setSettingSession name value = do- Session.statement (name, value) statement- where- statement = Statement.Statement "SELECT set_config($1, $2, false)" encoder Decoders.noResult True- encoder =- contramap fst (Encoders.param (Encoders.nonNullable Encoders.text))- <> contramap snd (Encoders.param (Encoders.nonNullable Encoders.text))--getSettingSession :: Text -> Session.Session (Maybe Text)-getSettingSession name = do- Session.statement name statement- where- statement = Statement.Statement "SELECT current_setting($1, true)" encoder decoder True- encoder = Encoders.param (Encoders.nonNullable Encoders.text)- decoder = Decoders.singleRow (Decoders.column (Decoders.nullable Decoders.text))