pqi-conformance 1.0.2.1 → 1.0.3.0

raw patch · 4 files changed

+131/−1 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +6/−0
• pqi-conformance.cabal +2/−1
• src/library/Pqi/Conformance/Operation/PipelineSync.hs +2/−0
• src/library/Pqi/Conformance/Operation/PipelineSync/Interruption.hs +121/−0

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

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

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

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