diff --git a/hasql.cabal b/hasql.cabal --- a/hasql.cabal +++ b/hasql.cabal @@ -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 diff --git a/src/comms-tests/Hasql/Comms/Session/CleanUpAfterInterruptionSpec.hs b/src/comms-tests/Hasql/Comms/Session/CleanUpAfterInterruptionSpec.hs --- a/src/comms-tests/Hasql/Comms/Session/CleanUpAfterInterruptionSpec.hs +++ b/src/comms-tests/Hasql/Comms/Session/CleanUpAfterInterruptionSpec.hs @@ -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 = diff --git a/src/library-tests/Sharing/ByBug/PipelineAbortedInterruptionCleanupSpec.hs b/src/library-tests/Sharing/ByBug/PipelineAbortedInterruptionCleanupSpec.hs new file mode 100644 --- /dev/null +++ b/src/library-tests/Sharing/ByBug/PipelineAbortedInterruptionCleanupSpec.hs @@ -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` [] diff --git a/src/library/Hasql/Comms/Session.hs b/src/library/Hasql/Comms/Session.hs --- a/src/library/Hasql/Comms/Session.hs +++ b/src/library/Hasql/Comms/Session.hs @@ -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