diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +# v1.0.3.0 + +## Non-breaking + +- Added a `pipelineSync` spec covering resilience to an asynchronous interruption around a pipeline abort. Found the native adapter can leave a connection permanently unusable instead of failing outright, since none of its transport/connection code is exception-safe + # v1.0.2.1 ## Fixes diff --git a/pqi-conformance.cabal b/pqi-conformance.cabal --- a/pqi-conformance.cabal +++ b/pqi-conformance.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: pqi-conformance -version: 1.0.2.1 +version: 1.0.3.0 category: Database, PostgreSQL, Testing synopsis: Differential conformance tests for pqi adapters description: @@ -160,6 +160,7 @@ Pqi.Conformance.Operation.Pass Pqi.Conformance.Operation.PipelineStatus Pqi.Conformance.Operation.PipelineSync + Pqi.Conformance.Operation.PipelineSync.Interruption Pqi.Conformance.Operation.PipelineSync.Parity Pqi.Conformance.Operation.Port Pqi.Conformance.Operation.Prepare diff --git a/src/library/Pqi/Conformance/Operation/PipelineSync.hs b/src/library/Pqi/Conformance/Operation/PipelineSync.hs --- a/src/library/Pqi/Conformance/Operation/PipelineSync.hs +++ b/src/library/Pqi/Conformance/Operation/PipelineSync.hs @@ -8,6 +8,7 @@ import qualified Pqi import qualified Pqi as Lq import Pqi.Conformance.Harness +import qualified Pqi.Conformance.Operation.PipelineSync.Interruption as Interruption import qualified Pqi.Conformance.Operation.PipelineSync.Parity as Parity import Pqi.Conformance.Prelude import Pqi.Conformance.Scenario (takeCommandResults, takeResult) @@ -17,6 +18,7 @@ spec adapter = describe "pipelineSync" do Parity.spec adapter + Interruption.spec adapter it "collects pipelined queries per sync" \conninfo -> differential adapter conninfo \connection -> do entered <- Lq.enterPipelineMode connection diff --git a/src/library/Pqi/Conformance/Operation/PipelineSync/Interruption.hs b/src/library/Pqi/Conformance/Operation/PipelineSync/Interruption.hs new file mode 100644 --- /dev/null +++ b/src/library/Pqi/Conformance/Operation/PipelineSync/Interruption.hs @@ -0,0 +1,121 @@ +-- | Coverage for 'Pqi.pipelineSync' under asynchronous interruption: a +-- resilience gap found in @pqi-native@ while chasing a hang in @hasql@'s +-- @Integration.Sharing.Connection.Use.PipelineAbortedInterruptionCleanupSpec@: +-- an asynchronous exception (e.g. from 'System.Timeout.timeout') landing +-- while a pipelined command is being aborted, or while cleaning up +-- afterwards, can leave a connection permanently unusable instead of merely +-- failing outright. +-- +-- The suspected mechanism: none of @pqi-native@'s transport or connection +-- code (@Pqi.Native.Transport@, @Pqi.Native.Connection@, @Pqi.Native.Query@) +-- uses 'Control.Exception.mask', 'Control.Exception.bracket', or any other +-- exception-safety device - all of its state (the read buffer, pipeline +-- status, pending-command counters, ...) lives in plain 'Data.IORef.IORef's +-- updated as a sequence of ordinary, unprotected steps. Its reads go through +-- @network@'s non-blocking-socket + GHC-I\/O-manager path +-- ('Network.Socket.ByteString.recv'), which is /interruptible/ - it can +-- receive an async exception mid-call even under 'Control.Exception.mask' - +-- unlike @pqi-ffi@, whose blocking @libpq@ reads go through a @safe@ FFI +-- import that defers async exceptions until the call returns. An interrupt +-- landing mid-read can therefore desynchronize @pqi-native@'s message +-- framing (bytes already pulled off the socket are dropped instead of being +-- folded back into the read buffer) or abandon its pipeline-tracking state +-- mid-transition, in a way @pqi-ffi@ structurally cannot. +-- +-- This scenario mirrors the hasql-level reproduction as closely as possible +-- while staying entirely at the @Pqi@ level (no @hasql@ involved): pipeline +-- a statement that sleeps just long enough to widen the vulnerable window, +-- followed by one guaranteed to fail, sweep an async interrupt finely across +-- the moment the pipeline transitions to \"aborted\", then run the same +-- clean-up sequence @Hasql.Comms.Session.leavePipeline@ runs (sync, drain, +-- flush, drain, exit) - itself timed, since an adapter that can lose an +-- interrupt mid-read can just as easily lose one mid-clean-up - and finally +-- probe the connection with a trivial round-trip. A sound adapter always +-- either finishes promptly or leaves the connection in a state where a +-- fresh round-trip fails fast; it never goes quiet. +module Pqi.Conformance.Operation.PipelineSync.Interruption + ( spec, + ) +where + +import Control.Exception (bracket) +import qualified Data.ByteString.Char8 as ByteString.Char8 +import qualified Pqi +import Pqi.Conformance.Harness +import Pqi.Conformance.Prelude +import Pqi.Conformance.Scenario (drainResults, float8Oid) +import System.Timeout (timeout) +import Test.Hspec + +spec :: Pqi.Adapter -> SpecWith ByteString +spec adapter = + describe "resilience to an async interruption around a pipeline abort" do + it "the connection recovers or fails fast instead of going quiet" \conninfo -> do + reproduced <- go conninfo delays + reproduced `shouldBe` False + where + go _ [] = pure False + go conninfo (d : ds) = do + stalled <- attempt conninfo d + if stalled then pure True else go conninfo ds + -- Long enough to give an interrupt landing anywhere near the sleep + -- statement's completion (where the pipeline flips to "aborted") a + -- realistic chance to land mid-read, short enough to keep the sweep + -- affordable. + sleepMicros = 20000 :: Int + sleepSeconds = fromIntegral sleepMicros / 1000000 :: Double + + -- Sweep across the moment the sleep statement's result arrives and the + -- failing statement's error is processed - the vulnerable window is far + -- narrower than this timer's granularity. + delays = [sleepMicros + step | step <- [-3000, -2900 .. 6000]] :: [Int] + + attemptsPerDelay = 15 :: Int + + -- An adapter that isn't stuck settles a trivial round-trip in + -- milliseconds; anything still running after this long counts as a + -- reproduction of the stall. + cleanupTimeoutMicros = 1000000 :: Int + probeTimeoutMicros = 1000000 :: Int + + attempt conninfo delayMicros = + or <$> replicateM attemptsPerDelay (attemptOnce conninfo delayMicros) + + attemptOnce conninfo delayMicros = + bracket (Pqi.connectdb adapter conninfo) Pqi.finish \connection -> do + _ <- Pqi.enterPipelineMode connection + _ <- + Pqi.sendQueryParams + connection + "select pg_sleep($1)" + [Just (float8Oid, showBS sleepSeconds, Pqi.Text)] + Pqi.Text + _ <- Pqi.sendQueryParams connection "select 1/0" [] Pqi.Text + _ <- Pqi.pipelineSync connection + + -- Race the interrupt against draining the pipeline's results, tuned + -- to land around the PipelineOn -> PipelineAborted transition. + _ <- timeout delayMicros (drainResults connection) + + -- Mirror Hasql.Comms.Session.leavePipeline's clean-up sequence, + -- itself timed: a second interrupt landing here is just as capable + -- of desynchronizing the connection as the first one. + cleanedUp <- timeout cleanupTimeoutMicros do + _ <- drainResults connection + status <- Pqi.pipelineStatus connection + when (status /= Pqi.PipelineOff) do + _ <- Pqi.pipelineSync connection + _ <- drainResults connection + _ <- Pqi.sendFlushRequest connection + _ <- drainResults connection + _ <- Pqi.exitPipelineMode connection + pure () + + probe <- timeout probeTimeoutMicros do + _ <- Pqi.sendQuery connection "select 1" + drainResults connection + + pure (isNothing cleanedUp || isNothing probe) + + showBS :: Double -> ByteString + showBS = ByteString.Char8.pack . show