diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,54 @@ +# 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 diff --git a/hasql-pool.cabal b/hasql-pool.cabal --- a/hasql-pool.cabal +++ b/hasql-pool.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: hasql-pool -version: 1.4.2.3 +version: 1.5.1.1 category: Hasql, Database, PostgreSQL synopsis: Pool of connections for Hasql homepage: https://github.com/nikita-volkov/hasql-pool @@ -78,6 +78,7 @@ other-modules: Hasql.Pool.Config.Config + Hasql.Pool.Config.ReuseOrder Hasql.Pool.Config.Setting Hasql.Pool.Prelude Hasql.Pool.SessionErrorDestructors @@ -85,7 +86,8 @@ build-depends: base >=4.11 && <5, bytestring >=0.10 && <0.14, - hasql >=1.10 && <1.11, + hasql ^>=2.0, + pqi >=1.0 && <1.2, stm >=2.5 && <3, text >=1.2 && <3, time >=1.9 && <2, @@ -97,19 +99,21 @@ hs-source-dirs: src/integration-tests main-is: Main.hs other-modules: + Helpers.Adapters Helpers.Hooks Helpers.Scripts Helpers.Sessions - Specs.BySubject.Config.AgingTimeoutSpec - Specs.BySubject.Config.IdlenessTimeoutSpec - Specs.BySubject.Config.InitSessionSpec - Specs.BySubject.Helpers.Sessions.CountConnectionsSpec - Specs.BySubject.Helpers.Sessions.GetSettingSpec - Specs.BySubject.ReleaseSpec - Specs.BySubject.SpecHook - Specs.BySubject.UsageError.AcquisitionTimeoutSpec - Specs.BySubject.UsageError.SessionSpec - Specs.BySubject.UseSpec + Specs.Config.AgingTimeoutSpec + Specs.Config.IdlenessTimeoutSpec + Specs.Config.InitSessionSpec + Specs.Helpers.Sessions.CountConnectionsSpec + Specs.Helpers.Sessions.GetSettingSpec + Specs.InterruptionSpec + Specs.ReleaseSpec + Specs.SpecHook + Specs.UsageError.AcquisitionTimeoutSpec + Specs.UsageError.SessionSpec + Specs.UseSpec ghc-options: -threaded build-tool-depends: @@ -120,7 +124,9 @@ hasql, hasql-pool, hspec >=2.6 && <3, - postgresql-libpq >=0.10 && <0.12, + 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, 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/Scripts.hs b/src/integration-tests/Helpers/Scripts.hs --- a/src/integration-tests/Helpers/Scripts.hs +++ b/src/integration-tests/Helpers/Scripts.hs @@ -3,20 +3,23 @@ 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. --- Host and port of a running isolated postgres server. -type ScopeParams = (Text, Word16) +-- Adapter, host and port of a running isolated postgres server. +type ScopeParams = (Pqi.Adapter, Text, Word16) onTaggedPool :: Int -> DiffTime -> DiffTime -> DiffTime -> Text -> ScopeParams -> (Pool.Pool -> IO ()) -> IO () -onTaggedPool poolSize acqTimeout maxLifetime maxIdletime appName (host, port) = +onTaggedPool poolSize acqTimeout maxLifetime maxIdletime appName (adapter, host, port) = bracket ( Pool.acquire + adapter ( Config.settings [ Config.size poolSize, Config.acquisitionTimeout acqTimeout, @@ -37,15 +40,46 @@ Pool.release onAutotaggedPool :: Int -> DiffTime -> DiffTime -> DiffTime -> ScopeParams -> (Text -> Pool.Pool -> IO ()) -> IO () -onAutotaggedPool poolSize acqTimeout maxLifetime maxIdletime (host, port) cont = do +onAutotaggedPool poolSize acqTimeout maxLifetime maxIdletime scopeParams cont = do -- Generate app name appName <- generateName "hasql-pool-test-" - onTaggedPool poolSize acqTimeout maxLifetime maxIdletime appName (host, port) (cont appName) + onTaggedPool poolSize acqTimeout maxLifetime maxIdletime appName scopeParams (cont appName) +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 (host, port) = +onTaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession appName (adapter, host, port) = bracket ( Pool.acquire + adapter ( Config.settings [ Config.size poolSize, Config.acquisitionTimeout acqTimeout, @@ -67,9 +101,39 @@ Pool.release onAutotaggedPoolWithInitSession :: Int -> DiffTime -> DiffTime -> DiffTime -> Session.Session () -> ScopeParams -> (Text -> Pool.Pool -> IO ()) -> IO () -onAutotaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession (host, port) cont = do +onAutotaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession (adapter, host, port) cont = do appName <- generateName "hasql-pool-test-" - onTaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession appName (host, port) (cont appName) + onTaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession appName (adapter, host, port) (cont appName) + +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 = diff --git a/src/integration-tests/Helpers/Sessions.hs b/src/integration-tests/Helpers/Sessions.hs --- a/src/integration-tests/Helpers/Sessions.hs +++ b/src/integration-tests/Helpers/Sessions.hs @@ -6,15 +6,16 @@ getSetting, countConnections, sleep, + backendPid, ) where import Data.Tuple.All -import Database.PostgreSQL.LibPQ qualified as Pq import Hasql.Decoders qualified as Decoders import Hasql.Encoders qualified as Encoders import Hasql.Session qualified as Session import Hasql.Statement qualified as Statement +import Pqi qualified import Prelude selectOne :: Session.Session Int64 @@ -34,7 +35,7 @@ closeConn :: Session.Session () closeConn = Session.onLibpqConnection \conn -> do - Pq.finish conn + Pqi.finish conn pure (Right (), conn) setSetting :: Text -> Text -> Session.Session () @@ -71,3 +72,10 @@ 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/Specs/BySubject/Config/AgingTimeoutSpec.hs b/src/integration-tests/Specs/BySubject/Config/AgingTimeoutSpec.hs deleted file mode 100644 --- a/src/integration-tests/Specs/BySubject/Config/AgingTimeoutSpec.hs +++ /dev/null @@ -1,32 +0,0 @@ -module Specs.BySubject.Config.AgingTimeoutSpec where - -import Hasql.Pool -import Helpers.Scripts qualified as Scripts -import Helpers.Sessions qualified as Sessions -import Prelude -import Test.Hspec - -spec :: SpecWith Scripts.ScopeParams -spec = do - it "Actively times out old connections" \scopeParams -> do - Scripts.onAutotaggedPool 3 10 1_800 1_800 scopeParams \_appName1 pool1 -> do - Scripts.onAutotaggedPool 3 10 0.5 1_800 scopeParams \appName2 pool2 -> do - res <- use pool2 $ Sessions.selectOne - res `shouldBe` Right 1 - res2 <- use pool1 $ Sessions.countConnections appName2 - res2 `shouldBe` Right 1 - threadDelay 1_000_000 -- 1s - res3 <- use pool1 $ Sessions.countConnections appName2 - res3 `shouldBe` Right 0 - - it "Passively times out old connections" \scopeParams -> do - -- 0.5s connection lifetime - Scripts.onAutotaggedPool 1 10 0.5 1_800 scopeParams \_ pool -> do - varName <- Scripts.generateVarname - res <- use pool $ Sessions.setSetting varName "hello world" - res `shouldBe` Right () - res2 <- use pool $ Sessions.getSetting varName - res2 `shouldBe` Right (Just "hello world") - threadDelay 1_000_000 -- 1s - res3 <- use pool $ Sessions.getSetting varName - res3 `shouldBe` Right Nothing diff --git a/src/integration-tests/Specs/BySubject/Config/IdlenessTimeoutSpec.hs b/src/integration-tests/Specs/BySubject/Config/IdlenessTimeoutSpec.hs deleted file mode 100644 --- a/src/integration-tests/Specs/BySubject/Config/IdlenessTimeoutSpec.hs +++ /dev/null @@ -1,44 +0,0 @@ -module Specs.BySubject.Config.IdlenessTimeoutSpec where - -import Hasql.Pool -import Helpers.Scripts qualified as Scripts -import Helpers.Sessions qualified as Sessions -import Prelude -import Test.Hspec - -spec :: SpecWith Scripts.ScopeParams -spec = do - it "Times out old connections (maxIdletime)" \scopeParams -> do - -- 0.5s connection idle time - Scripts.onAutotaggedPool 1 10 1_800 0.5 scopeParams \_ pool -> do - varName <- Scripts.generateVarname - res <- use pool $ Sessions.setSetting varName "hello world" - res `shouldBe` Right () - res2 <- use pool $ Sessions.getSetting varName - res2 `shouldBe` Right (Just "hello world") - -- busy sleep, to keep connection alive - forM_ [1 :: Int .. 10] $ \_ -> do - r <- use pool $ Sessions.selectOne - r `shouldBe` Right 1 - threadDelay 100_000 -- 0.1s - res3 <- use pool $ Sessions.getSetting varName - res3 `shouldBe` Right (Just "hello world") - -- idle sleep, connection times out - threadDelay 1_000_000 -- 1s - res4 <- use pool $ Sessions.getSetting varName - res4 `shouldBe` Right Nothing - - it "Passively times out idle connections" \scopeParams -> do - -- 0.5s connection idle time, large lifetime, so only idleness can explain a passive close. - Scripts.onAutotaggedPool 3 10 1_800 1_800 scopeParams \_appName1 pool1 -> do - Scripts.onAutotaggedPool 3 10 1_800 0.5 scopeParams \appName2 pool2 -> do - res <- use pool2 $ Sessions.selectOne - res `shouldBe` Right 1 - res2 <- use pool1 $ Sessions.countConnections appName2 - res2 `shouldBe` Right 1 - -- Give the background reaper (1s tick) a chance to passively evict - -- the now-idle connection, without ever calling `use pool2` again - -- (which would trigger the separate active idleness check). - threadDelay 1_500_000 -- 1.5s - res3 <- use pool1 $ Sessions.countConnections appName2 - res3 `shouldBe` Right 0 diff --git a/src/integration-tests/Specs/BySubject/Config/InitSessionSpec.hs b/src/integration-tests/Specs/BySubject/Config/InitSessionSpec.hs deleted file mode 100644 --- a/src/integration-tests/Specs/BySubject/Config/InitSessionSpec.hs +++ /dev/null @@ -1,59 +0,0 @@ -module Specs.BySubject.Config.InitSessionSpec where - -import Hasql.Pool -import Helpers.Scripts qualified as Scripts -import Helpers.Sessions qualified as Sessions -import Prelude -import Test.Hspec - -spec :: SpecWith Scripts.ScopeParams -spec = do - it "Persists after exceptions thrown in session" \scopeParams -> do - Scripts.onAutotaggedPool 1 10 60 60 scopeParams \_ pool -> do - varName <- Scripts.generateVarname - - res <- use pool do - Sessions.setSetting varName "1" - Sessions.getSetting varName - shouldBe res (Right (Just "1")) - - try @SomeException do - use pool do - liftIO do - throwIO (userError "Intentional error for testing") - - res <- use pool do - Sessions.getSetting varName - shouldBe res (Right (Just "1")) - - it "Persists after bad query" \scopeParams -> do - Scripts.onAutotaggedPool 1 10 60 60 scopeParams \_ pool -> do - varName <- Scripts.generateVarname - - res <- use pool do - Sessions.setSetting varName "1" - Sessions.getSetting varName - shouldBe res (Right (Just "1")) - - use pool do - Sessions.badQuery - - res <- use pool do - Sessions.getSetting varName - shouldBe res (Right (Just "1")) - - -- https://github.com/nikita-volkov/hasql-pool/issues/56 - it "Does not exhaust the pool capacity when it fails" \scopeParams -> do - -- Pool of size 1 whose init session always fails, with a short - -- acquisition timeout so that a leaked capacity slot shows up as - -- an AcquisitionTimeoutUsageError instead of hanging the test. - Scripts.onAutotaggedPoolWithInitSession 1 1 60 60 Sessions.badQuery scopeParams \_ pool -> do - res1 <- use pool Sessions.selectOne - res1 `shouldSatisfy` \case - Left (SessionUsageError _) -> True - _ -> False - - res2 <- use pool Sessions.selectOne - res2 `shouldSatisfy` \case - Left (SessionUsageError _) -> True - _ -> False diff --git a/src/integration-tests/Specs/BySubject/Helpers/Sessions/CountConnectionsSpec.hs b/src/integration-tests/Specs/BySubject/Helpers/Sessions/CountConnectionsSpec.hs deleted file mode 100644 --- a/src/integration-tests/Specs/BySubject/Helpers/Sessions/CountConnectionsSpec.hs +++ /dev/null @@ -1,14 +0,0 @@ -module Specs.BySubject.Helpers.Sessions.CountConnectionsSpec where - -import Hasql.Pool -import Helpers.Scripts qualified as Scripts -import Helpers.Sessions qualified as Sessions -import Prelude -import Test.Hspec - -spec :: SpecWith Scripts.ScopeParams -spec = do - it "Counts active connections" \scopeParams -> do - Scripts.onAutotaggedPool 3 10 1_800 1_800 scopeParams \appName pool -> do - res <- use pool $ Sessions.countConnections appName - res `shouldBe` Right 1 diff --git a/src/integration-tests/Specs/BySubject/Helpers/Sessions/GetSettingSpec.hs b/src/integration-tests/Specs/BySubject/Helpers/Sessions/GetSettingSpec.hs deleted file mode 100644 --- a/src/integration-tests/Specs/BySubject/Helpers/Sessions/GetSettingSpec.hs +++ /dev/null @@ -1,36 +0,0 @@ -module Specs.BySubject.Helpers.Sessions.GetSettingSpec where - -import Hasql.Pool -import Helpers.Scripts qualified as Scripts -import Helpers.Sessions qualified as Sessions -import Prelude -import Test.Hspec - -spec :: SpecWith Scripts.ScopeParams -spec = do - it "Getting and setting session variables works" \scopeParams -> - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - varName <- Scripts.generateVarname - res <- use pool $ Sessions.getSetting varName - res `shouldBe` Right Nothing - res <- use pool $ do - Sessions.setSetting varName "hello world" - Sessions.getSetting varName - res `shouldBe` Right (Just "hello world") - - it "Session variables stay set when a connection gets reused" \scopeParams -> - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - varName <- Scripts.generateVarname - res <- use pool $ Sessions.setSetting varName "hello world" - res `shouldBe` Right () - res2 <- use pool $ Sessions.getSetting varName - res2 `shouldBe` Right (Just "hello world") - - it "Releasing the pool resets session variables" \scopeParams -> do - varName <- Scripts.generateVarname - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - res <- use pool $ Sessions.setSetting varName "hello world" - res `shouldBe` Right () - release pool - res <- use pool $ Sessions.getSetting varName - res `shouldBe` Right Nothing diff --git a/src/integration-tests/Specs/BySubject/ReleaseSpec.hs b/src/integration-tests/Specs/BySubject/ReleaseSpec.hs deleted file mode 100644 --- a/src/integration-tests/Specs/BySubject/ReleaseSpec.hs +++ /dev/null @@ -1,16 +0,0 @@ -module Specs.BySubject.ReleaseSpec where - -import Hasql.Pool -import Helpers.Scripts qualified as Scripts -import Helpers.Sessions qualified as Sessions -import Prelude -import Test.Hspec - -spec :: SpecWith Scripts.ScopeParams -spec = do - it "The pool remains usable after release" \scopeParams -> - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - _ <- use pool $ Sessions.selectOne - release pool - res <- use pool $ Sessions.selectOne - shouldSatisfy res $ isRight diff --git a/src/integration-tests/Specs/BySubject/SpecHook.hs b/src/integration-tests/Specs/BySubject/SpecHook.hs deleted file mode 100644 --- a/src/integration-tests/Specs/BySubject/SpecHook.hs +++ /dev/null @@ -1,11 +0,0 @@ --- Docs: https://hspec.github.io/hspec-discover.html -module Specs.BySubject.SpecHook where - -import Helpers.Hooks qualified as Hooks -import Helpers.Scripts qualified as Scripts -import Prelude -import Test.Hspec - -hook :: SpecWith Scripts.ScopeParams -> Spec -hook = - aroundAll Hooks.postgres17 . parallel diff --git a/src/integration-tests/Specs/BySubject/UsageError/AcquisitionTimeoutSpec.hs b/src/integration-tests/Specs/BySubject/UsageError/AcquisitionTimeoutSpec.hs deleted file mode 100644 --- a/src/integration-tests/Specs/BySubject/UsageError/AcquisitionTimeoutSpec.hs +++ /dev/null @@ -1,33 +0,0 @@ -module Specs.BySubject.UsageError.AcquisitionTimeoutSpec where - -import Control.Concurrent.Async (race) -import Hasql.Pool -import Helpers.Scripts qualified as Scripts -import Helpers.Sessions qualified as Sessions -import Prelude -import Test.Hspec - -spec :: SpecWith Scripts.ScopeParams -spec = do - it "Gets produced on timeout" \scopeParams -> - -- 1ms timeout - Scripts.onAutotaggedPool 1 0.001 1_800 1_800 scopeParams \_ pool -> do - sleeping <- newEmptyMVar - t0 <- getCurrentTime - res <- - race - ( use pool - $ liftIO - $ do - putMVar sleeping () - -- 1s - threadDelay 1_000_000 - ) - ( do - takeMVar sleeping - use pool $ Sessions.selectOne - ) - t1 <- getCurrentTime - res `shouldBe` Right (Left AcquisitionTimeoutUsageError) - -- 0.5s - diffUTCTime t1 t0 `shouldSatisfy` (< 0.5) diff --git a/src/integration-tests/Specs/BySubject/UsageError/SessionSpec.hs b/src/integration-tests/Specs/BySubject/UsageError/SessionSpec.hs deleted file mode 100644 --- a/src/integration-tests/Specs/BySubject/UsageError/SessionSpec.hs +++ /dev/null @@ -1,16 +0,0 @@ -module Specs.BySubject.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/BySubject/UseSpec.hs b/src/integration-tests/Specs/BySubject/UseSpec.hs deleted file mode 100644 --- a/src/integration-tests/Specs/BySubject/UseSpec.hs +++ /dev/null @@ -1,133 +0,0 @@ -module Specs.BySubject.UseSpec where - -import Control.Concurrent.Async (race) -import Data.Text qualified as Text -import Hasql.Decoders qualified as Decoders -import Hasql.Encoders qualified as Encoders -import Hasql.Errors qualified as Errors -import Hasql.Pool -import Hasql.Session qualified as Session -import Hasql.Statement qualified as Statement -import Helpers.Scripts qualified as Scripts -import Helpers.Sessions qualified as Sessions -import Prelude -import Test.Hspec - -spec :: SpecWith Scripts.ScopeParams -spec = do - it "Releases a spot in the pool when there is a query error" \scopeParams -> - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - use pool Sessions.badQuery `shouldNotReturn` (Right ()) - use pool Sessions.selectOne `shouldReturn` (Right 1) - - it "Connection errors cause eviction of connection" \scopeParams -> - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - _ <- use pool $ Sessions.closeConn >> Sessions.selectOne - _ <- use pool $ Sessions.closeConn >> Sessions.selectOne - _ <- use pool $ Sessions.closeConn >> Sessions.selectOne - res <- use pool $ Sessions.selectOne - shouldSatisfy res $ isRight - - it "Driver errors cause eviction of connection" \scopeParams -> do - settingName <- Scripts.generateVarname - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - use pool (Sessions.setSetting settingName "present") `shouldReturn` Right () - result <- use pool driverError - result `shouldSatisfy` \case - Left (SessionUsageError (Errors.DriverSessionError _)) -> True - _ -> False - use pool (Sessions.getSetting settingName) `shouldReturn` Right Nothing - - it "Connection gets returned to the pool after normal use" \scopeParams -> - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - _ <- use pool $ Sessions.selectOne - _ <- use pool $ Sessions.selectOne - _ <- use pool $ Sessions.selectOne - _ <- use pool $ Sessions.selectOne - res <- use pool $ Sessions.selectOne - shouldSatisfy res $ isRight - - it "Connection gets returned to the pool after non-connection error" \scopeParams -> - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - _ <- use pool $ Sessions.badQuery - _ <- use pool $ Sessions.badQuery - _ <- use pool $ Sessions.badQuery - _ <- use pool $ Sessions.badQuery - res <- use pool $ Sessions.selectOne - shouldSatisfy res $ isRight - - -- https://github.com/nikita-volkov/hasql-pool/issues/38 - -- - -- When a session is interrupted by an asynchronous exception (e.g., a - -- caller-side timeout racing the query, as simulated here via `race`) - -- while it is genuinely blocked waiting on the server's response, the - -- underlying libpq connection is left mid-command: the query was sent, - -- but its result was never read. `onLiveConn` in Hasql.Pool.use still - -- unconditionally returns such a connection to the pool (the `Left exc` - -- branch calls `returnConn` for all exceptions, not just synchronous - -- ones), so the next `use` call hands out a connection whose protocol - -- state is desynced from libpq's expectations. This is a plausible root - -- cause of the "connection pointer is NULL" reports: two independent - -- consumers of hasql-pool end up driving the same libpq connection state - -- machine without coordination. - it "Does not return a connection to the pool when the session is interrupted by an asynchronous exception" \scopeParams -> - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - started <- newEmptyMVar - _ <- - race - ( use pool do - liftIO $ putMVar started () - Sessions.sleep 2 - ) - ( do - takeMVar started - -- Give the query time to actually reach the server and for - -- the client to start blocking on the socket read, as - -- opposed to being cancelled while still sending. - threadDelay 200_000 - ) - res <- use pool Sessions.selectOne - res `shouldSatisfy` isRight - - it "Cached type errors cause eviction of connection" \scopeParams -> do - typeName <- Text.replace "-" "_" <$> Scripts.generateName "cached_type_" - Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do - use pool (Session.script (createTypeSql typeName)) `shouldReturn` Right () - use pool (roundtripEnum typeName "ok") `shouldReturn` Right "ok" - use pool (Session.script (recreateTypeSql typeName)) `shouldReturn` Right () - res <- use pool (roundtripEnum typeName "ok") - shouldSatisfy res \case - Left (SessionUsageError _) -> True - _ -> False - use pool (roundtripEnum typeName "ok") `shouldReturn` Right "ok" - -quoteIdentifier :: Text -> Text -quoteIdentifier identifier = - "\"" <> Text.replace "\"" "\"\"" identifier <> "\"" - -createTypeSql :: Text -> Text -createTypeSql typeName = - "create type " <> quotedTypeName <> " as enum ('sad', 'ok', 'happy')" - where - quotedTypeName = quoteIdentifier typeName - -recreateTypeSql :: Text -> Text -recreateTypeSql typeName = - "drop type " <> quotedTypeName <> "; create type " <> quotedTypeName <> " as enum ('sad', 'ok', 'happy')" - where - quotedTypeName = quoteIdentifier typeName - -roundtripEnum :: Text -> Text -> Session.Session Text -roundtripEnum typeName value = - Session.statement value statement - where - statement = - Statement.preparable - ("select $1 :: " <> quoteIdentifier typeName) - (Encoders.param (Encoders.nonNullable (Encoders.enum Nothing typeName id))) - (Decoders.singleRow (Decoders.column (Decoders.nonNullable (Decoders.enum Nothing typeName Just)))) - -driverError :: Session.Session () -driverError = - Session.onLibpqConnection \connection -> - pure (Left (Errors.DriverSessionError "synthetic driver error"), connection) 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 --- a/src/library/Hasql/Pool.hs +++ b/src/library/Hasql/Pool.hs @@ -16,10 +16,12 @@ 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 @@ -37,28 +39,63 @@ 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 microseconds. - poolAcquisitionTimeout :: Int, + -- | Acquisition timeout, in nanoseconds. + poolAcquisitionTimeout :: Word64, -- | Maximal connection lifetime, in nanoseconds. poolMaxLifetime :: Word64, -- | Maximal connection idle time, in nanoseconds. poolMaxIdletime :: Word64, - -- | Avail connections. + -- | 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 thread via garbage collection. + -- | To stop the manager and expirer threads via garbage collection. poolReaperRef :: IORef (), -- | Action for reporting the observations. poolObserver :: Observation -> IO (), @@ -66,18 +103,33 @@ 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. -acquire :: Config.Config -> IO Pool -acquire config = do +-- +-- 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 @@ -88,23 +140,22 @@ (idleEntries, liveEntries) = partition (entryIsIdle maxIdletimeNanos now) unagedEntries traverse_ (writeTQueue connectionQueue) liveEntries return $ do - forM_ agedEntries $ \entry -> do - Connection.release (entryConnection entry) - atomically $ modifyTVar' capVar succ - (Config.observationHandler config) (ConnectionObservation (entryId entry) (TerminatedConnectionStatus AgingConnectionTerminationReason)) - forM_ idleEntries $ \entry -> do - Connection.release (entryConnection entry) - atomically $ modifyTVar' capVar succ - (Config.observationHandler config) (ConnectionObservation (entryId entry) (TerminatedConnectionStatus IdlenessConnectionTerminationReason)) + 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, stop the manager. + -- 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) (Config.connectionSettingsProvider config) acqTimeoutMicros agingTimeoutNanos maxIdletimeNanos connectionQueue capVar reuseVar reaperRef (Config.observationHandler config) (Config.initSession config) + 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 - acqTimeoutMicros = - div (fromIntegral (diffTimeToPicoseconds (Config.acquisitionTimeout config))) 1_000_000 + acqTimeoutNanos = + div (fromIntegral (diffTimeToPicoseconds (Config.acquisitionTimeout config))) 1_000 agingTimeoutNanos = div (fromIntegral (diffTimeToPicoseconds (Config.agingTimeout config))) 1_000 maxIdletimeNanos = @@ -125,127 +176,327 @@ newReuse <- newTVar True writeTVar poolReuseVar newReuse entries <- flushTQueue poolConnectionQueue - return $ forM_ entries $ \entry -> do - Connection.release (entryConnection entry) - atomically $ modifyTVar' poolCapacity succ - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason)) + 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 'Session.ClientError' gets interpreted as a loss of +-- 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. -use :: Pool -> Session.Session a -> IO (Either UsageError a) +-- +-- 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 - timeout <- do - delay <- registerDelay poolAcquisitionTimeout - return $ readTVar delay - join . atomically $ do - reuseVar <- readTVar poolReuseVar - asum - [ readTQueue poolConnectionQueue <&> onConn reuseVar, - do - capVal <- readTVar poolCapacity - if capVal > 0 - then do - writeTVar poolCapacity $! pred capVal - return $ onNewConn reuseVar - else retry, - do - timedOut <- timeout - if timedOut - then return . return . Left $ AcquisitionTimeoutUsageError - else retry - ] + 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 - onNewConn reuseVar = do - settings <- poolFetchConnectionSettings + 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 - poolObserver (ConnectionObservation id ConnectingConnectionStatus) - Connection.acquire settings >>= \case + 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 - poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason connErrText))) - atomically $ modifyTVar' poolCapacity succ + unmask $ poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason connErrText))) + atomically returnSlot return $ Left $ ConnectionUsageError connErr - Right connection -> do - Connection.use connection poolInitSession >>= \case + Right connection -> + unmask (Connection.use connection poolInitSession) >>= \case Left err -> do Connection.release connection - atomically $ modifyTVar' poolCapacity succ - ErrorsDestruction.reset - ( \details -> do - poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (Just details)))) - ) - (poolObserver (ConnectionObservation id (TerminatedConnectionStatus (InitializationErrorTerminationReason err)))) - err + 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 - poolObserver (ConnectionObservation id (ReadyForUseConnectionStatus EstablishedConnectionReadyForUseReason)) - onLiveConn reuseVar (Entry connection now now id) - - onConn reuseVar entry = do - now <- getMonotonicTimeNSec - if entryIsAged poolMaxLifetime now entry - then do - Connection.release (entryConnection entry) - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus AgingConnectionTerminationReason)) - onNewConn reuseVar - else - if entryIsIdle poolMaxIdletime now entry - then do - Connection.release (entryConnection entry) - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus IdlenessConnectionTerminationReason)) - onNewConn reuseVar - else do - onLiveConn reuseVar entry {entryUseTimeNSec = now} + unmask $ poolObserver (ConnectionObservation id (ReadyForUseConnectionStatus EstablishedConnectionReadyForUseReason)) + return $ Right (Entry connection now now id) - onLiveConn reuseVar entry = do - poolObserver (ConnectionObservation (entryId entry) InUseConnectionStatus) - sessRes <- try @SomeException (Connection.use (entryConnection entry) sess) + -- \| 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 + returnConn reuseVar entry throwIO exc - Right (Left err) -> - if ErrorsDestruction.requiresConnectionDiscard err + 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 $ modifyTVar' poolCapacity succ - poolObserver - ( ConnectionObservation - (entryId entry) - (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (ErrorsDestruction.discardDetails err))) - ) + atomically returnSlot + unmask + $ poolObserver + ( ConnectionObservation + (entryId entry) + (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (Just (Errors.toMessage err)))) + ) return $ Left $ SessionUsageError err else do - returnConn - poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus (SessionFailedConnectionReadyForUseReason err))) + returnConn reuseVar entry + unmask $ poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus (SessionFailedConnectionReadyForUseReason err))) return $ Left $ SessionUsageError err Right (Right res) -> do - returnConn - poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus SessionSucceededConnectionReadyForUseReason)) + returnConn reuseVar entry + unmask $ poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus SessionSucceededConnectionReadyForUseReason)) return $ Right res - where - returnConn = - join . atomically $ do - reuse <- readTVar reuseVar - if reuse - then writeTQueue poolConnectionQueue entry $> return () - else return $ do - Connection.release (entryConnection entry) - atomically $ modifyTVar' poolCapacity succ - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason)) + + -- 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 diff --git a/src/library/Hasql/Pool/Config.hs b/src/library/Hasql/Pool/Config.hs --- a/src/library/Hasql/Pool/Config.hs +++ b/src/library/Hasql/Pool/Config.hs @@ -11,10 +11,13 @@ 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 diff --git a/src/library/Hasql/Pool/Config/Config.hs b/src/library/Hasql/Pool/Config/Config.hs --- a/src/library/Hasql/Pool/Config/Config.hs +++ b/src/library/Hasql/Pool/Config/Config.hs @@ -2,6 +2,7 @@ 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 @@ -14,7 +15,8 @@ idlenessTimeout :: DiffTime, connectionSettingsProvider :: IO Connection.Settings.Settings, observationHandler :: Observation -> IO (), - initSession :: Session.Session () + initSession :: Session.Session (), + reuseOrder :: ReuseOrder } -- | Reasonable defaults, which can be built upon. @@ -27,5 +29,6 @@ idlenessTimeout = Defaults.idlenessTimeout, connectionSettingsProvider = Defaults.dynamicConnectionSettings, observationHandler = Defaults.observationHandler, - initSession = Defaults.initSession + initSession = Defaults.initSession, + reuseOrder = Defaults.reuseOrder } diff --git a/src/library/Hasql/Pool/Config/Defaults.hs b/src/library/Hasql/Pool/Config/Defaults.hs --- a/src/library/Hasql/Pool/Config/Defaults.hs +++ b/src/library/Hasql/Pool/Config/Defaults.hs @@ -1,6 +1,7 @@ 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 @@ -45,3 +46,8 @@ -- > 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 --- a/src/library/Hasql/Pool/Config/Setting.hs +++ b/src/library/Hasql/Pool/Config/Setting.hs @@ -3,6 +3,7 @@ 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 @@ -74,6 +75,8 @@ -- -- 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. -- @@ -95,3 +98,14 @@ 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/SessionErrorDestructors.hs b/src/library/Hasql/Pool/SessionErrorDestructors.hs --- a/src/library/Hasql/Pool/SessionErrorDestructors.hs +++ b/src/library/Hasql/Pool/SessionErrorDestructors.hs @@ -19,12 +19,6 @@ -- cleanup after an interruption fails, so it must not be reused by the pool. Errors.DriverSessionError {} -> True -discardDetails :: Errors.SessionError -> Maybe Text -discardDetails err = - if requiresConnectionDiscard err - then Just $ Errors.toMessage err - else Nothing - statementRequiresConnectionDiscard :: Errors.StatementError -> Bool statementRequiresConnectionDiscard = \case Errors.ServerStatementError serverError -> isStaleServerError serverError