hasql-pool 0.4.3.1 → 1.5.1.1

raw patch · 31 files changed

Files

• CHANGELOG.md +164/−0
• Setup.hs +0/−2
• diagrams-output/connection-status-model.png binary
• hasql-pool.cabal +124/−46
• library/Hasql/Pool.hs +0/−73
• library/Hasql/Pool/Prelude.hs +0/−14
• 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

@@ -0,0 +1,164 @@+# 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.++# v0.8.0.5++Fix connections not returning to the pool on exceptions.++# v0.8.0.2++Fixed Windows build.++# 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.++Acquisition timeout added.++Breaking changes in API:++- Removed `PoolIsReleasedUsageError`+- `acquire` extended with the acquisition timeout parameter+- `acquireDynamically` extended with the acquisition timeout parameter++# v0.7.2++Added support for dynamic connection configuration ([issue #11](https://github.com/nikita-volkov/hasql-pool/issues/11)).++# v0.7.1.2++Fixed connections not being released if they were in use during the call to `release`.++# v0.7.1++Added `Exception` for `UsageError`.++# v0.7++Simplified the implementation a lot by removing the notion of timeout.++Breaking:+- Removed the `Settings` type+- Changed the signature of `acquire`++# v0.6++Moved away from "resource-pool" and fixed the handling of lost connections.++Breaking:++- Changed the suffix of `UsageError` constructors from `Error` to `UsageError`+- Added `PoolIsReleasedUsageError`

@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain

binary file changed (absent → 16434 bytes)

@@ -1,56 +1,134 @@-name:-  hasql-pool-version:-  0.4.3.1-category:-  Hasql, Database, PostgreSQL-synopsis:-  A pool of connections for Hasql-homepage:-  https://github.com/nikita-volkov/hasql-pool -bug-reports:-  https://github.com/nikita-volkov/hasql-pool/issues -author:-  Nikita Volkov <nikita.y.volkov@mail.ru>-maintainer:-  Nikita Volkov <nikita.y.volkov@mail.ru>-copyright:-  (c) 2015, Nikita Volkov-license:-  MIT-license-file:-  LICENSE-build-type:-  Simple-cabal-version:-  >=1.10+cabal-version: 3.0+name: hasql-pool+version: 1.5.1.1+category: Hasql, Database, PostgreSQL+synopsis: Pool of connections for Hasql+homepage: https://github.com/nikita-volkov/hasql-pool+bug-reports: https://github.com/nikita-volkov/hasql-pool/issues+author: Nikita Volkov <nikita.y.volkov@mail.ru>+maintainer: Nikita Volkov <nikita.y.volkov@mail.ru>+copyright: (c) 2015, Nikita Volkov+license: MIT+license-file: LICENSE+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+  type: git+  location: https://github.com/nikita-volkov/hasql-pool +common base-settings+  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-  ghc-options:-  default-extensions:-    Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples-  default-language:-    Haskell2010-  other-modules:-    Hasql.Pool.Prelude+    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:-    -- resources:-    resource-pool >= 0.2 && < 0.3,-    -- database:-    hasql >= 0.19 && < 0.20 || >= 1 && < 1.2,-    -- data:-    time >= 1.5 && < 2,-    -- general:-    base-prelude >= 1 && < 2+    base >=4.11 && <5,+    bytestring >=0.10 && <0.14,+    hasql ^>=2.0,+    pqi >=1.0 && <1.2,+    stm >=2.5 && <3,+    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: 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,+    testcontainers-postgresql >=0.2 && <0.3,+    text-builder >=1 && <1.1,+    tuple ^>=0.3.0.2,

@@ -1,73 +0,0 @@-module Hasql.Pool-(-  Pool,-  Settings(..),-  acquire,-  release,-  UsageError(..),-  use,-)-where--import Hasql.Pool.Prelude-import qualified Hasql.Connection-import qualified Hasql.Session-import qualified Data.Pool----- |--- A pool of connections to DB.-newtype Pool =-  Pool (Data.Pool.Pool (Either Hasql.Connection.ConnectionError Hasql.Connection.Connection))---- |--- Settings of the connection pool. Consist of:--- --- * Pool-size.--- --- * Timeout.   --- An amount of time for which an unused resource is kept open.--- The smallest acceptable value is 0.5 seconds.--- --- * Connection settings.--- -type Settings =-  (Int, NominalDiffTime, Hasql.Connection.Settings)---- |--- Given the pool-size, timeout and connection settings--- create a connection-pool.-acquire :: Settings -> IO Pool-acquire (size, timeout, connectionSettings) =-  fmap Pool $-  Data.Pool.createPool acquire release stripes timeout size-  where-    acquire =-      Hasql.Connection.acquire connectionSettings-    release =-      either (const (pure ())) Hasql.Connection.release-    stripes =-      1---- |--- Release the connection-pool.-release :: Pool -> IO ()-release (Pool pool) =-  Data.Pool.destroyAllResources pool---- |--- A union over the connection establishment error and the session error.-data UsageError =-  ConnectionError !Hasql.Connection.ConnectionError |-  SessionError !Hasql.Session.Error-  deriving (Show, Eq)---- |--- Use a connection from the pool to run a session and--- return the connection to the pool, when finished.-use :: Pool -> Hasql.Session.Session a -> IO (Either UsageError a)-use (Pool pool) session =-  fmap (either (Left . ConnectionError) (either (Left . SessionError) Right)) $-  Data.Pool.withResource pool $-  traverse $ -  Hasql.Session.run session

@@ -1,14 +0,0 @@-module Hasql.Pool.Prelude-( -  module Exports,-)-where----- base-prelude---------------------------import BasePrelude as Exports hiding (assert, left, right, isLeft, isRight, error)---- time---------------------------import Data.Time as Exports

@@ -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

@@ -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))

@@ -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+      ]

@@ -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))

@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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"

@@ -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

@@ -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)+    )

@@ -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

@@ -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

@@ -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)

@@ -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

@@ -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

@@ -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+    }

@@ -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

@@ -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)

@@ -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})

@@ -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)

@@ -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

@@ -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"