hasql 1.10.3.6 → 1.10.3.7

raw patch · 4 files changed

+242/−2 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

hasql.cabal +2/−1
src/comms-tests/Hasql/Comms/Session/CleanUpAfterInterruptionSpec.hs +103/−0
src/library-tests/Sharing/ByBug/PipelineAbortedInterruptionCleanupSpec.hs +134/−0
src/library/Hasql/Comms/Session.hs +3/−1

hasql.cabal view

@@ -1,6 +1,6 @@ cabal-version: 3.0 name: hasql-version: 1.10.3.6+version: 1.10.3.7 category: Hasql, Database, PostgreSQL synopsis: Fast PostgreSQL driver with a flexible mapping API description:@@ -326,6 +326,7 @@     Isolated.ByUnit.Connection.AcquireSpec     Pure.ByUnit.ErrorsSpec     Sharing.ByBug.ExceptionConnectionResetRaceSpec+    Sharing.ByBug.PipelineAbortedInterruptionCleanupSpec     Sharing.ByFeature.ConcurrencySpec     Sharing.ByFeature.DecoderCompatibilityCacheSpec     Sharing.ByFeature.DecoderCompatibilityCheckSpec

src/comms-tests/Hasql/Comms/Session/CleanUpAfterInterruptionSpec.hs view

@@ -27,6 +27,91 @@         status <- Pq.pipelineStatus connection         status `shouldBe` Pq.PipelineOff +    it "cleans up when the pipeline is aborted with no pending transaction or sync" \config -> do+      withConnection config \connection -> do+        success <- Pq.enterPipelineMode connection+        success `shouldBe` True++        sent <- Pq.sendQueryParams connection "SELECT * FROM nonexistent_table" [] Pq.Text+        sent `shouldBe` True+        flushRequestSent <- Pq.sendFlushRequest connection+        flushRequestSent `shouldBe` True+        flushStatus <- Pq.flush connection+        flushStatus `shouldBe` Pq.FlushOk++        let waitForResult attempts = do+              when (attempts <= 0) do+                expectationFailure "Timed out waiting for libpq result after flushing pipeline query"+              consumed <- Pq.consumeInput connection+              consumed `shouldBe` True+              busy <- Pq.isBusy connection+              if busy+                then threadDelay 1000 >> waitForResult (attempts - 1)+                else Pq.getResult connection++        result <- waitForResult (10000 :: Int)+        resultStatus <- traverse Pq.resultStatus result+        resultStatus `shouldBe` Just Pq.FatalError+        pipelineStatus <- Pq.pipelineStatus connection+        pipelineStatus `shouldBe` Pq.PipelineAborted++        cleanupResult <- Session.toHandler Session.cleanUpAfterInterruption connection+        cleanupResult `shouldBe` Right ()+        finalPipelineStatus <- Pq.pipelineStatus connection+        finalPipelineStatus `shouldBe` Pq.PipelineOff++    -- Deterministic counterpart to the timing-based reproduction in+    -- "Sharing.ByBug.PipelineAbortedInterruptionCleanupSpec": instead of+    -- racing an async exception into the narrow window during which libpq's+    -- pipeline is in the aborted state, we put the connection into that state+    -- directly and invoke the cleanup that an interruption would have+    -- invoked. This exercises the `leavePipeline` branch that used to only+    -- handle `PipelineOn`, leaving an aborted pipeline to fall through to+    -- `bringTransactionStatusToIdle`, which then tried to send "ABORT" as a+    -- serial command while still in pipeline mode.+    it "cleans up when the pipeline is aborted" \config -> do+      withConnection config \connection -> do+        -- Open a transaction, so that the failing statement below leaves the+        -- connection in the error transaction status -- the status that makes+        -- `bringTransactionStatusToIdle` want to send "ABORT" as a serial+        -- command, which libpq refuses while pipeline mode is still on+        _ <- Pq.exec connection "BEGIN"++        -- Enter pipeline mode+        success <- Pq.enterPipelineMode connection+        success `shouldBe` True++        -- Queue a statement that is guaranteed to fail on the server,+        -- followed by another one. The trailing statement matters: it makes+        -- the abort survive the drains that `cleanUpAfterInterruption`+        -- performs before reaching `leavePipeline`, since `drainResults`+        -- stops at each command boundary+        success <- Pq.sendQueryParams connection "select 1/0" [] Pq.Text+        success `shouldBe` True++        success <- Pq.sendQueryParams connection "select 1" [] Pq.Text+        success `shouldBe` True++        -- Flush the queued statements to the server+        success <- Pq.pipelineSync connection+        success `shouldBe` True++        -- Consume results until libpq registers the error and flips the+        -- pipeline into the aborted state, leaving the trailing sync marker+        -- undrained -- exactly the state an interruption would leave behind+        consumeUntilPipelineAborted connection++        -- Run cleanup+        result <- Session.toHandler Session.cleanUpAfterInterruption connection+        result `shouldBe` Right ()++        -- Verify we're out of pipeline mode and usable again+        status <- Pq.pipelineStatus connection+        status `shouldBe` Pq.PipelineOff++        transStatus <- Pq.transactionStatus connection+        transStatus `shouldBe` Pq.TransIdle+     it "cleans up when in an open transaction" \config -> do       withConnection config \connection -> do         -- Start a transaction@@ -106,6 +191,24 @@           Nothing -> expectationFailure "Expected error result from exec prepared after deallocation"  -- * Helpers++-- | Pull results from a pipelined connection until libpq reports the pipeline+-- as aborted, failing the example if that never happens.+consumeUntilPipelineAborted :: Pq.Connection -> IO ()+consumeUntilPipelineAborted connection =+  go (10 :: Int)+  where+    go attemptsLeft = do+      status <- Pq.pipelineStatus connection+      case status of+        Pq.PipelineAborted -> pure ()+        _+          | attemptsLeft <= 0 ->+              expectationFailure+                ("Pipeline did not reach the aborted state. Last status: " <> show status)+          | otherwise -> do+              _ <- Pq.getResult connection+              go (pred attemptsLeft)  withConnection :: (Text, Word16) -> (Pq.Connection -> IO a) -> IO a withConnection (host, port) action =

@@ -0,0 +1,134 @@+module Sharing.ByBug.PipelineAbortedInterruptionCleanupSpec (spec) where++import Data.Text qualified as Text+import Hasql.Connection qualified as Connection+import Hasql.Decoders qualified as Decoders+import Hasql.Encoders qualified as Encoders+import Hasql.Pipeline qualified as Pipeline+import Hasql.Session qualified as Session+import Hasql.Statement qualified as Statement+import Helpers.Scripts qualified as Scripts+import Test.Hspec+import Prelude++-- | A statement that sleeps for the given number of seconds and succeeds.+--+-- Used to widen the wall-clock window during which the client is blocked+-- waiting on the network/socket for a pipelined result, so that an+-- asynchronous interruption has a realistic chance of landing right around+-- the moment the *next* statement in the same pipeline fails.+sleepStatement :: Statement.Statement Double ()+sleepStatement =+  Statement.preparable+    "select pg_sleep($1)"+    (Encoders.param (Encoders.nonNullable Encoders.float8))+    Decoders.noResult++-- | A statement that is guaranteed to fail on the server.+failingStatement :: Statement.Statement () ()+failingStatement =+  Statement.preparable+    "select 1/0"+    Encoders.noParams+    Decoders.noResult++-- | A pipeline of two statements: the first sleeps and succeeds, the+-- second fails. Executing this via 'Session.pipeline' drives libpq's+-- pipeline status through: Off -> On -> (once the sleep result has been+-- received and the divide error has been processed) Aborted -> (normally)+-- Off again, via the exit sequence inside 'toPipelineIO'.+--+-- The bug under test concerns what happens if an asynchronous exception+-- interrupts execution during the narrow "Aborted" window: right after+-- libpq has registered the error result for the second statement (which+-- flips its internal pipeline status to `PipelineAborted`) but before the+-- driver has drained the trailing pipeline-sync marker and called+-- `exitPipelineMode`. That window is only a couple of FFI calls wide, so+-- reliably landing an async exception inside it requires many attempts+-- across a fine-grained sweep of interrupt delays (see 'spec' below).+--+-- Note: an earlier version of this test tried to widen the window by+-- appending many trivial "filler" statements after the failing one (on+-- the theory that draining their results would take measurably longer).+-- That approach reproduced failures reliably, but for the wrong reason:+-- `Comms.Session.drainResults` only drains one queued command's worth of+-- results per call, so a large backlog of undrained filler results made+-- `exitPipelineMode` fail with "cannot exit pipeline mode with uncollected+-- results" regardless of whether the `PipelineOn`/`PipelineAborted` bug+-- under test was present or fixed. That's a real, separate limitation of+-- `drainResults`, but not the bug this test is about, so the pipeline here+-- is kept to exactly two statements and 'attempt' below specifically+-- checks for the "not allowed in pipeline mode" signature (the one the+-- one-line `leavePipeline` fix actually addresses) rather than any+-- "Failed to clean up after interruption" message.+racingPipelineSession :: Double -> Session.Session ()+racingPipelineSession sleepSeconds =+  Session.pipeline do+    Pipeline.statement sleepSeconds sleepStatement+      *> Pipeline.statement () failingStatement++-- | Try once to reproduce the bug: acquire a fresh connection, race a+-- `timeout` against the pipelined session (sleep-then-fail) tuned to fire+-- right around the moment the pipeline transitions to the aborted state,+-- and report whether `Connection.use` came back with the specific driver+-- error that signals the `leavePipeline` bug: it only checks for+-- `PipelineOn`, so when the connection is genuinely `PipelineAborted` at+-- interruption time, cleanup skips leaving the pipeline and falls through+-- to `bringTransactionStatusToIdle`, which tries to send "ABORT" as a+-- serial command while still in pipeline mode -- something libpq flatly+-- refuses ("PQsendQuery not allowed in pipeline mode").+--+-- Note on why checking `Connection.use`'s own return value is enough: when+-- `timeout` throws its internal exception into the thread running+-- `Connection.use`, that exception is caught by `use`'s own+-- @try \@SomeException@. If the bug is NOT triggered, `use` cleans up+-- successfully and rethrows the very same timeout exception, so `timeout`+-- observes it and returns 'Nothing'. If the bug IS triggered, `use`+-- reports the cleanup failure as an ordinary `Left (DriverSessionError _)`+-- return value instead of rethrowing, so `timeout` observes a normal+-- return and reports 'Just (Left _)'.+attempt :: (Text, Word16) -> Double -> Int -> IO (Maybe Text)+attempt config sleepSeconds delayMicros =+  Scripts.onPreparableConnection config \connection -> do+    result <- timeout delayMicros do+      Connection.use connection (racingPipelineSession sleepSeconds)+    pure case result of+      Just (Left err) ->+        let rendered = Text.pack (show err)+         in if "Failed to clean up after interruption"+              `Text.isInfixOf` rendered+              && "not allowed in pipeline mode"+              `Text.isInfixOf` rendered+              then Just rendered+              else Nothing+      Just (Right ()) -> Nothing+      Nothing -> Nothing++spec :: SpecWith (Text, Word16)+spec = do+  describe "Interruption of a pipeline while it is in the Aborted status" do+    it "Connection.use recovers cleanly instead of reporting a driver cleanup failure" \config -> do+      -- We sweep the timeout across a window that straddles the moment the+      -- sleep statement finishes and the failing statement's error result+      -- gets processed by libpq (which is when the pipeline status flips+      -- from `PipelineOn` to `PipelineAborted`). The genuinely vulnerable+      -- window is only a couple of FFI calls wide (nowhere near as wide as+      -- our timer granularity), so we compensate with a large number of+      -- attempts spread finely across the window and a fresh connection+      -- each time, rather than trying to widen the window itself.+      let sleepMicros = 20000 :: Int -- 20ms sleep statement duration+          sleepSeconds = fromIntegral sleepMicros / 1000000+          delays = [sleepMicros + step | step <- [(-3000), (-2900) .. 6000]]+          attemptsPerDelay = 15++      results <-+        sequence+          [ attempt config sleepSeconds d+          | d <- delays,+            _ <- [1 :: Int .. attemptsPerDelay]+          ]++      let reproductions = [msg | Just msg <- results]++      reproductions+        `shouldBe` []

src/library/Hasql/Comms/Session.hs view

@@ -75,7 +75,9 @@ leavePipeline :: Session () leavePipeline = do   pipelineStatus <- getPipelineStatus-  when (pipelineStatus == Pq.PipelineOn) do+  -- PipelineAborted is still pipeline mode. It must reach a sync point before+  -- libpq permits serial queries such as ABORT or DEALLOCATE ALL again.+  when (pipelineStatus /= Pq.PipelineOff) do     -- In pipeline mode, we need to ensure the pipeline is synchronized before exiting.     -- Send a pipeline sync marker to flush any pending operations.     syncSuccess <- sendPipelineSync