hasql-pool 1.5.0.1 → 1.5.1.0

raw patch · 14 files changed

+723/−102 lines, 14 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +25/−0
• hasql-pool.cabal +3/−1
• src/integration-tests/Helpers/Scripts.hs +61/−0
• src/integration-tests/Helpers/Sessions.hs +8/−0
• src/integration-tests/Specs/Config/IdlenessTimeoutSpec.hs +18/−0
• src/integration-tests/Specs/InterruptionSpec.hs +163/−0
• src/integration-tests/Specs/UsageError/AcquisitionTimeoutSpec.hs +13/−1
• src/integration-tests/Specs/UseSpec.hs +68/−0
• src/library/Hasql/Pool.hs +319/−98
• src/library/Hasql/Pool/Config.hs +3/−0
• src/library/Hasql/Pool/Config/Config.hs +5/−2
• src/library/Hasql/Pool/Config/Defaults.hs +6/−0
• src/library/Hasql/Pool/Config/ReuseOrder.hs +17/−0
• src/library/Hasql/Pool/Config/Setting.hs +14/−0

@@ -1,3 +1,28 @@+# 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.

@@ -1,6 +1,6 @@ cabal-version: 3.0 name: hasql-pool-version: 1.5.0.1+version: 1.5.1.0 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@@ -107,6 +108,7 @@     Specs.Config.InitSessionSpec     Specs.Helpers.Sessions.CountConnectionsSpec     Specs.Helpers.Sessions.GetSettingSpec+    Specs.InterruptionSpec     Specs.ReleaseSpec     Specs.SpecHook     Specs.UsageError.AcquisitionTimeoutSpec

@@ -3,6 +3,7 @@ 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@@ -44,6 +45,36 @@   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@@ -73,6 +104,36 @@ 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 =

@@ -6,6 +6,7 @@     getSetting,     countConnections,     sleep,+    backendPid,   ) where @@ -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))

@@ -1,6 +1,7 @@ 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@@ -42,3 +43,20 @@         threadDelay 1_500_000 -- 1.5s         res3 <- use pool1 $ Sessions.countConnections appName2         res3 `shouldBe` Right 0++  it "Times out the connections that aren't needed with LIFO" \scopeParams -> do+    -- 0.5s connection idle time+    Scripts.onAutotaggedPoolWithReuseOrder 2 10 1_800 0.5 Config.LIFO scopeParams \appName pool -> do+      held <- newEmptyMVar+      done <- newEmptyMVar+      -- Establish two connections.+      _ <- forkIO (void (use pool (liftIO (putMVar held () *> takeMVar done))))+      takeMVar held+      use pool Sessions.selectOne `shouldReturn` Right 1+      putMVar done ()+      -- Keep using the pool one pool request at a time, which needs only one+      -- connection, for longer than the idle time and the reaper's 1s tick.+      forM_ [1 :: Int .. 15] \_ -> do+        use pool Sessions.selectOne `shouldReturn` Right 1+        threadDelay 100_000 -- 0.1s+      use pool (Sessions.countConnections appName) `shouldReturn` Right 1

@@ -0,0 +1,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"

@@ -1,6 +1,6 @@ module Specs.UsageError.AcquisitionTimeoutSpec where -import Control.Concurrent.Async (race)+import Control.Concurrent.Async (forConcurrently, race) import Hasql.Pool import Helpers.Scripts qualified as Scripts import Helpers.Sessions qualified as Sessions@@ -31,3 +31,15 @@       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

@@ -6,6 +6,7 @@ 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@@ -89,6 +90,52 @@       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@@ -131,3 +178,24 @@ 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)

@@ -16,6 +16,7 @@ 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@@ -38,6 +39,33 @@ 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.@@ -46,22 +74,28 @@     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 (),@@ -69,6 +103,17 @@     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@@ -82,7 +127,9 @@   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@@ -93,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) adapter (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 =@@ -130,10 +176,19 @@     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.@@ -144,113 +199,279 @@ -- 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 poolAdapter 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             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 (ErrorsDestruction.discardDetails 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 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

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

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

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

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

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