diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -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` diff --git a/Setup.hs b/Setup.hs deleted file mode 100644 --- a/Setup.hs +++ /dev/null @@ -1,2 +0,0 @@ -import Distribution.Simple -main = defaultMain diff --git a/diagrams-output/connection-status-model.png b/diagrams-output/connection-status-model.png new file mode 100644 Binary files /dev/null and b/diagrams-output/connection-status-model.png differ diff --git a/hasql-pool.cabal b/hasql-pool.cabal --- a/hasql-pool.cabal +++ b/hasql-pool.cabal @@ -1,75 +1,134 @@ -name: - hasql-pool -version: - 0.5.2.2 -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 -maintainer: - Nikita Volkov -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 +maintainer: Nikita Volkov +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 - Hasql.Pool.ResourcePool + src/library/ + exposed-modules: Hasql.Pool - build-depends: - -- resources: - resource-pool >= 0.2 && < 0.3, - -- database: - hasql >= 1.3 && < 1.6, - -- data: - time >= 1.5 && < 2, - -- general: - base-prelude >= 1 && < 2 + Hasql.Pool.Config + Hasql.Pool.Config.Defaults + Hasql.Pool.Observation + other-modules: + Hasql.Pool.Config.Config + Hasql.Pool.Config.ReuseOrder + Hasql.Pool.Config.Setting + Hasql.Pool.Prelude + Hasql.Pool.SessionErrorDestructors + build-depends: + base >=4.11 && <5, + 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 - type: - exitcode-stdio-1.0 - hs-source-dirs: - test - main-is: - Main.hs - 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 + 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: - base-prelude, + async >=2.2 && <3, hasql, hasql-pool, - hspec >= 2.6 && < 3 + 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, diff --git a/library/Hasql/Pool.hs b/library/Hasql/Pool.hs deleted file mode 100644 --- a/library/Hasql/Pool.hs +++ /dev/null @@ -1,75 +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 as ResourcePool -import qualified Hasql.Pool.ResourcePool as ResourcePool - - --- | --- A pool of connections to DB. -newtype Pool = - Pool (ResourcePool.Pool (Either Hasql.Connection.ConnectionError Hasql.Connection.Connection)) - deriving (Show) - --- | --- 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 $ - ResourcePool.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) = - ResourcePool.destroyAllResources pool - --- | --- A union over the connection establishment error and the session error. -data UsageError = - ConnectionError Hasql.Connection.ConnectionError | - SessionError Hasql.Session.QueryError - 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)) $ - ResourcePool.withResourceOnEither pool $ - traverse $ - Hasql.Session.run session diff --git a/library/Hasql/Pool/Prelude.hs b/library/Hasql/Pool/Prelude.hs deleted file mode 100644 --- a/library/Hasql/Pool/Prelude.hs +++ /dev/null @@ -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 diff --git a/library/Hasql/Pool/ResourcePool.hs b/library/Hasql/Pool/ResourcePool.hs deleted file mode 100644 --- a/library/Hasql/Pool/ResourcePool.hs +++ /dev/null @@ -1,21 +0,0 @@ -{-| -Extras for the resource-pool library. --} -module Hasql.Pool.ResourcePool -where - -import Hasql.Pool.Prelude -import Data.Pool - - -withResourceOnEither :: Pool resource -> (resource -> IO (Either failure success)) -> IO (Either failure success) -withResourceOnEither pool act = mask_ $ do - (resource, localPool) <- takeResource pool - failureOrSuccess <- act resource `onException` destroyResource pool localPool resource - case failureOrSuccess of - Right success -> do - putResource localPool resource - return (Right success) - Left failure -> do - destroyResource pool localPool resource - return (Left failure) diff --git a/src/integration-tests/Helpers/Adapters.hs b/src/integration-tests/Helpers/Adapters.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Helpers/Adapters.hs @@ -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 diff --git a/src/integration-tests/Helpers/Hooks.hs b/src/integration-tests/Helpers/Hooks.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Helpers/Hooks.hs @@ -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)) diff --git a/src/integration-tests/Helpers/Scripts.hs b/src/integration-tests/Helpers/Scripts.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Helpers/Scripts.hs @@ -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 + ] diff --git a/src/integration-tests/Helpers/Sessions.hs b/src/integration-tests/Helpers/Sessions.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Helpers/Sessions.hs @@ -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)) diff --git a/src/integration-tests/Main.hs b/src/integration-tests/Main.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Main.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF hspec-discover #-} diff --git a/src/integration-tests/Specs/Config/AgingTimeoutSpec.hs b/src/integration-tests/Specs/Config/AgingTimeoutSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/Config/AgingTimeoutSpec.hs @@ -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 diff --git a/src/integration-tests/Specs/Config/IdlenessTimeoutSpec.hs b/src/integration-tests/Specs/Config/IdlenessTimeoutSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/Config/IdlenessTimeoutSpec.hs @@ -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 diff --git a/src/integration-tests/Specs/Config/InitSessionSpec.hs b/src/integration-tests/Specs/Config/InitSessionSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/Config/InitSessionSpec.hs @@ -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 diff --git a/src/integration-tests/Specs/Helpers/Sessions/CountConnectionsSpec.hs b/src/integration-tests/Specs/Helpers/Sessions/CountConnectionsSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/Helpers/Sessions/CountConnectionsSpec.hs @@ -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 diff --git a/src/integration-tests/Specs/Helpers/Sessions/GetSettingSpec.hs b/src/integration-tests/Specs/Helpers/Sessions/GetSettingSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/Helpers/Sessions/GetSettingSpec.hs @@ -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 diff --git a/src/integration-tests/Specs/InterruptionSpec.hs b/src/integration-tests/Specs/InterruptionSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/InterruptionSpec.hs @@ -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" diff --git a/src/integration-tests/Specs/ReleaseSpec.hs b/src/integration-tests/Specs/ReleaseSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/ReleaseSpec.hs @@ -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 diff --git a/src/integration-tests/Specs/SpecHook.hs b/src/integration-tests/Specs/SpecHook.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/SpecHook.hs @@ -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) + ) diff --git a/src/integration-tests/Specs/UsageError/AcquisitionTimeoutSpec.hs b/src/integration-tests/Specs/UsageError/AcquisitionTimeoutSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/UsageError/AcquisitionTimeoutSpec.hs @@ -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 diff --git a/src/integration-tests/Specs/UsageError/SessionSpec.hs b/src/integration-tests/Specs/UsageError/SessionSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/UsageError/SessionSpec.hs @@ -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 diff --git a/src/integration-tests/Specs/UseSpec.hs b/src/integration-tests/Specs/UseSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/UseSpec.hs @@ -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) diff --git a/src/library/Hasql/Pool.hs b/src/library/Hasql/Pool.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool.hs @@ -0,0 +1,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 diff --git a/src/library/Hasql/Pool/Config.hs b/src/library/Hasql/Pool/Config.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Config.hs @@ -0,0 +1,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 diff --git a/src/library/Hasql/Pool/Config/Config.hs b/src/library/Hasql/Pool/Config/Config.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Config/Config.hs @@ -0,0 +1,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 + } diff --git a/src/library/Hasql/Pool/Config/Defaults.hs b/src/library/Hasql/Pool/Config/Defaults.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Config/Defaults.hs @@ -0,0 +1,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 diff --git a/src/library/Hasql/Pool/Config/ReuseOrder.hs b/src/library/Hasql/Pool/Config/ReuseOrder.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Config/ReuseOrder.hs @@ -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) diff --git a/src/library/Hasql/Pool/Config/Setting.hs b/src/library/Hasql/Pool/Config/Setting.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Config/Setting.hs @@ -0,0 +1,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}) diff --git a/src/library/Hasql/Pool/Observation.hs b/src/library/Hasql/Pool/Observation.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Observation.hs @@ -0,0 +1,64 @@ +-- | Interface for processing observations of the status of the pool. +-- +-- Provides a flexible mechanism for monitoring the healthiness of the pool via logs and metrics without any opinionated choices on the actual monitoring technologies. +-- Specific interpreters are encouraged to be created as extension libraries. +module Hasql.Pool.Observation where + +import Hasql.Errors qualified as Errors +import Hasql.Pool.Prelude + +-- | An observation of a change of the state of a pool. +data Observation + = -- | Status of one of the pool's connections has changed. + ConnectionObservation + -- | Generated connection ID. + -- For grouping the observations by one connection. + UUID + -- | Status that the connection has entered. + ConnectionStatus + deriving (Show, Eq) + +-- | Status of a connection. +-- +-- <> +data ConnectionStatus + = -- | Connection is being established. + -- + -- This is the initial status of every connection. + ConnectingConnectionStatus + | -- | Connection is established and not occupied. + ReadyForUseConnectionStatus ConnectionReadyForUseReason + | -- | Is being used by some session. + -- + -- After it's done the status will transition to 'ReadyForUseConnectionStatus' or 'TerminatedConnectionStatus'. + InUseConnectionStatus + | -- | Connection terminated. + TerminatedConnectionStatus ConnectionTerminationReason + deriving (Show, Eq) + +data ConnectionReadyForUseReason + = -- | Connection just got established. + EstablishedConnectionReadyForUseReason + | -- | Session execution ended with a failure that does not require a connection reset. + SessionFailedConnectionReadyForUseReason Errors.SessionError + | -- | Session execution ended with success. + SessionSucceededConnectionReadyForUseReason + deriving (Show, Eq) + +-- | Explanation of why a connection was terminated. +data ConnectionTerminationReason + = -- | The age timeout of the connection has passed. + AgingConnectionTerminationReason + | -- | The timeout of how long a connection may remain idle in the pool has passed. + IdlenessConnectionTerminationReason + | -- | The connection became unusable and had to be discarded. + -- + -- This includes connectivity issues with the server as well as fatal + -- session errors that invalidate the connection's prepared statement or + -- type caches. + NetworkErrorConnectionTerminationReason (Maybe Text) + | -- | User has invoked the 'Hasql.Pool.release' procedure. + ReleaseConnectionTerminationReason + | -- | Initialization session failure. + InitializationErrorTerminationReason Errors.SessionError + deriving (Show, Eq) diff --git a/src/library/Hasql/Pool/Prelude.hs b/src/library/Hasql/Pool/Prelude.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Prelude.hs @@ -0,0 +1,75 @@ +module Hasql.Pool.Prelude + ( module Exports, + ) +where + +import Control.Applicative as Exports hiding (WrappedArrow (..)) +import Control.Arrow as Exports hiding (first, second) +import Control.Category as Exports +import Control.Concurrent as Exports +import Control.Concurrent.STM as Exports hiding (orElse) +import Control.Exception as Exports +import Control.Monad as Exports hiding (fail, forM, forM_, mapM, mapM_, msum, sequence, sequence_) +import Control.Monad.Fail as Exports +import Control.Monad.Fix as Exports hiding (fix) +import Control.Monad.IO.Class as Exports +import Control.Monad.ST as Exports +import Data.Bifunctor as Exports +import Data.Bits as Exports +import Data.Bool as Exports +import Data.ByteString as Exports (ByteString) +import Data.Char as Exports +import Data.Coerce as Exports +import Data.Complex as Exports +import Data.Data as Exports +import Data.Dynamic as Exports +import Data.Either as Exports +import Data.Fixed as Exports +import Data.Foldable as Exports hiding (toList) +import Data.Function as Exports hiding (id, (.)) +import Data.Functor as Exports hiding (unzip) +import Data.Functor.Compose as Exports +import Data.IORef as Exports +import Data.Int as Exports +import Data.Ix as Exports +import Data.List as Exports hiding (all, and, any, concat, concatMap, elem, find, foldl, foldl', foldl1, foldr, foldr1, isSubsequenceOf, mapAccumL, mapAccumR, maximum, maximumBy, minimum, minimumBy, notElem, or, product, sortOn, sum, uncons) +import Data.List.NonEmpty as Exports (NonEmpty (..)) +import Data.Maybe as Exports +import Data.Monoid as Exports hiding (Alt) +import Data.Ord as Exports +import Data.Proxy as Exports +import Data.Ratio as Exports +import Data.STRef as Exports +import Data.String as Exports +import Data.Text as Exports (Text) +import Data.Time as Exports +import Data.Traversable as Exports +import Data.Tuple as Exports +import Data.UUID as Exports (UUID) +import Data.Unique as Exports +import Data.Version as Exports +import Data.Void as Exports +import Data.Word as Exports +import Debug.Trace as Exports +import Foreign.ForeignPtr as Exports +import Foreign.Ptr as Exports +import Foreign.StablePtr as Exports +import Foreign.Storable as Exports +import GHC.Clock as Exports (getMonotonicTimeNSec) +import GHC.Conc as Exports hiding (orElse, threadWaitRead, threadWaitReadSTM, threadWaitWrite, threadWaitWriteSTM, withMVar) +import GHC.Exts as Exports (IsList (..), groupWith, inline, lazy, sortWith) +import GHC.Generics as Exports (Generic) +import GHC.IO.Exception as Exports +import Numeric as Exports +import 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 diff --git a/src/library/Hasql/Pool/SessionErrorDestructors.hs b/src/library/Hasql/Pool/SessionErrorDestructors.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/SessionErrorDestructors.hs @@ -0,0 +1,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" diff --git a/test/Main.hs b/test/Main.hs deleted file mode 100644 --- a/test/Main.hs +++ /dev/null @@ -1,29 +0,0 @@ -module Main where - -import BasePrelude -import Test.Hspec -import Hasql.Pool -import qualified Hasql.Connection as Connection -import qualified Hasql.Decoders as Decoders -import qualified Hasql.Encoders as Encoders -import qualified Hasql.Session as Session -import qualified Hasql.Statement as Statement - - -main = hspec $ do - describe "Hasql.Pool.use" $ do - it "releases a spot in the pool when there is an error" $ do - pool <- acquire (1, 1, "host=localhost port=5432 user=postgres dbname=postgres") - let - statement = Statement.Statement "" Encoders.noParams Decoders.noResult True - session = Session.statement () statement - in do - use pool session `shouldNotReturn` (Right ()) - let - session = let - statement = let - decoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)) - in Statement.Statement "SELECT 1" Encoders.noParams decoder True - in Session.statement () statement - in do - use pool session `shouldReturn` (Right 1)