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