pqi-conformance 1.0.11.0 → 1.0.12.0

raw patch · 7 files changed

+193/−5 lines, 7 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/Connectdb/HandshakeReset.hs +0/−1
• src/library/Pqi/Conformance/Operation/PipelineSync/Interruption.hs +0/−1
• src/library/Pqi/Conformance/Operation/SendQuery/ConnectionLostBeforeSend.hs +0/−1
• src/library/Pqi/Conformance/Operation/SendQueryParams.hs +3/−1
• src/library/Pqi/Conformance/Operation/SendQueryParams/PipelineFlowControl.hs +182/−0

@@ -1,3 +1,9 @@+# v1.0.12.0++## Non-breaking++- Added a `sendQueryParams` spec covering pipeline flow control: a burst of pipelined commands larger than the socket buffers along the path can absorb, submitted before a single result is read, with each command producing output of its own. The sender must keep draining the socket while it still has bytes to write, like libpq's `pqSendSome` (which absorbs incoming data via `pqReadData` and waits for read-or-write readiness on an incomplete send). Found `pqi-native`'s `Transport.send` was a bare blocking `sendAll` that never reads while sending, so once the unread results fill the server's output path the server stops reading the client's commands and both sides block forever. Surfaced as hasql's `manyLargeResultsViaPipeline` benchmark hanging from `pqi-native` 1.0.1.4 onward, exposed by the Unix-socket default host (whose 8KB buffers saturate on a ~10KB burst that TCP's 128KB ones absorbed); the spec scales the burst to force the same saturation over the suite's TCP container link. Because the failure is a hang rather than a wrong value, the scenario runs under a deadline and the bounded outcome is compared, so the deadlocked candidate fails as a deadline miss instead of stalling the suite.+ # v1.0.11.0  ## Non-breaking

@@ -1,6 +1,6 @@ cabal-version: 3.0 name: pqi-conformance-version: 1.0.11.0+version: 1.0.12.0 category: Database, PostgreSQL, Testing synopsis: Differential conformance tests for pqi adapters description:@@ -189,6 +189,7 @@     Pqi.Conformance.Operation.SendQuery     Pqi.Conformance.Operation.SendQuery.ConnectionLostBeforeSend     Pqi.Conformance.Operation.SendQueryParams+    Pqi.Conformance.Operation.SendQueryParams.PipelineFlowControl     Pqi.Conformance.Operation.SendQueryPrepared     Pqi.Conformance.Operation.ServerVersion     Pqi.Conformance.Operation.SetClientEncoding

@@ -31,7 +31,6 @@  import Control.Concurrent (forkIO) import Control.Exception (SomeException, bracket, try)-import qualified Data.ByteString as ByteString import qualified Data.ByteString.Char8 as ByteString.Char8 import qualified Network.Socket as Socket import qualified Network.Socket.ByteString as Socket.ByteString

@@ -41,7 +41,6 @@ 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)

@@ -76,7 +76,6 @@ where  import Control.Exception (SomeException, try)-import Control.Monad (void, when) import Foreign.C.Error (throwErrnoIfMinus1_) import Foreign.C.Types (CInt (..)) import qualified Pqi

@@ -8,12 +8,14 @@ import qualified Pqi import qualified Pqi as Lq import Pqi.Conformance.Harness+import qualified Pqi.Conformance.Operation.SendQueryParams.PipelineFlowControl as PipelineFlowControl import Pqi.Conformance.Prelude import Pqi.Conformance.Scenario (drainResults, execScenario, int4Oid, takeCommandResults, takeResult) import Test.Hspec  spec :: Pqi.Adapter -> SpecWith ByteString-spec adapter =+spec adapter = do+  PipelineFlowControl.spec adapter   describe "sendQueryParams" do     it "sends a parameterized query and collects its result" \conninfo ->       differential adapter conninfo \connection -> do

@@ -0,0 +1,182 @@+-- | Coverage for pipeline flow control: a burst of pipelined commands larger+-- than the socket buffers along the path can absorb, submitted back-to-back+-- before a single result is read, where every command produces output of its+-- own. Submitting such a burst must not deadlock: a client that still has+-- bytes to write while the server streams unread results back has to keep+-- /reading/ the socket while sending, so the server's output path drains and+-- the server keeps consuming the client's commands.+--+-- The burst is submitted through 'Pqi.sendQueryParams' because @libpq@+-- rejects @PQsendQuery@ itself in pipeline mode (\"PQsendQuery not allowed+-- in pipeline mode\") - only the extended protocol is accepted there. The+-- flow-control property being exercised is not specific to it, though: every+-- send operation goes through the same driver-level flush, which is where+-- the divergence lived.+--+-- The divergence: @pqi-native@'s 'Pqi.Native.Transport.send' was a bare+-- blocking @sendAll@, which never reads while sending. Once the server has+-- filled its output path with results, it blocks in @ClientWrite@ and stops+-- reading the client's commands; the client, with bytes still unsent,+-- blocks in @send@. Neither side can make progress again. The real @libpq@+-- survives the same scenario: @pqSendSome@ in @fe-misc.c@, on an incomplete+-- send, calls @pqReadData()@ to absorb incoming data and then+-- @pqWait(true, true, …)@ for read-or-write readiness, looping until its+-- output is fully sent - the comment above that loop describes this exact+-- deadlock.+--+-- Where it surfaced: hasql's @manyLargeResultsViaPipeline@ benchmark hung+-- (non-deterministically across runs, but always within the 100-command+-- pipeline) from @pqi-native@ 1.0.1.4 onward, after its default host+-- resolution moved from TCP @localhost@ to the Unix-domain socket (@\/tmp@).+-- The deadlock was always there; the Unix socket merely exposed it, because+-- its default 8KB+8KB buffers saturate on a ~10-16KB command burst, where+-- TCP's larger buffers absorb the same burst without the client's send ever+-- blocking.+--+-- Why the burst is as large as it is: this suite reaches its container+-- through Docker's published-port forwarding, and on a workstation that+-- path buffers far more than the raw socket buffers - measured on this+-- platform (macOS host, Docker Desktop), ~8.5MB of commands are accepted+-- with the server never reading (send-buffer autotuning plus the+-- forwarder's own buffering), and ~5.4MB of results are accepted with the+-- client never reading (receive-buffer autotuning capped at 4MB plus the+-- forwarder's buffering). A direct kernel route (a Linux CI host, a Unix+-- socket) exposes only the true socket buffers, an order of magnitude less.+-- The scenario therefore sends 1000 commands of ~32KB each - ~32MB of wire+-- traffic, more than double the sum of both directions' absorption even on+-- the forwarding path - with each command returning an ~32KB result, so+-- output expands to roughly the size of the input. By buffer arithmetic the+-- server's output path fills while it has consumed only a fraction of the+-- burst, and the client cannot finish sending unless the server consumes+-- several more MB, which it cannot do while the client reads nothing: both+-- sides are guaranteed stuck, whatever the platform's capacities are below+-- the burst's.+--+-- Why this observation: the failure mode is a hang, not a wrong value, and a+-- hung candidate must fail the spec rather than stall the whole suite. The+-- scenario therefore runs under a deadline and the /bounded outcome/ is what+-- gets compared: the reference completes within seconds and reports the+-- aggregate shape of the pipeline, while the deadlocked candidate never+-- returns and is reported as a deadline miss. When the deadline fires, the+-- worker thread is abandoned rather than killed - it is blocked inside a+-- foreign call no async exception can interrupt - and the harness's bracket+-- closes the connection, so the server sees EOF and exits. The results are+-- aggregated to counts rather than observed per-result: at this scale full+-- per-result observations would be enormous, per-statement agreement is+-- already covered by the per-operation specs, and what this scenario is+-- about is that all 1000 commands complete 'Pqi.TuplesOk' and the pipeline+-- reaches its sync.+module Pqi.Conformance.Operation.SendQueryParams.PipelineFlowControl+  ( spec,+  )+where++import qualified Control.Concurrent as Concurrent+import qualified Control.Exception as Exception+import qualified Data.ByteString.Char8 as ByteString.Char8+import qualified Pqi+import Pqi.Conformance.Harness+import Pqi.Conformance.Prelude+import Test.Hspec++-- | The payload each command carries: 32KB, so a command is ~32KB on the+-- wire (extended-protocol framing included) and returns an ~32KB result -+-- output expands to roughly the size of the input, so the server's output+-- path fills while it has consumed only a small fraction of the burst.+statementLiteral :: ByteString+statementLiteral = ByteString.Char8.replicate 32768 'x'++statement :: ByteString+statement = "select repeat('" <> statementLiteral <> "', 1)"++-- | 1000 commands, ~32MB of wire traffic in all - more than double what the+-- container's forwarding path can absorb in /both/ directions combined on a+-- workstation (measured ~8.5MB of unread input, ~5.4MB of unread output),+-- and far more than a direct kernel route can (~0.4MB). The client cannot+-- finish sending unless the server keeps reading, which it cannot do unless+-- the client keeps draining results.+commandCount :: Int+commandCount = 1000++-- | The reference collects the whole pipeline in seconds. Anything that+-- cannot within a minute is not slow but deadlocked.+deadlineMicros :: Int+deadlineMicros = 60_000_000++spec :: Pqi.Adapter -> SpecWith ByteString+spec adapter =+  describe "sendQueryParams" do+    describe "pipeline flow control" do+      it "sends a burst larger than the socket buffers while results stream back, without deadlocking" \conninfo ->+        differential adapter conninfo scenario++-- | Pipeline @statement@ @commandCount@ times, sync, then collect the whole+-- pipeline: every command's result and separator, the sync result, and the+-- idle end. The observation aggregates them (how many commands, how many+-- came back 'Pqi.TuplesOk', the total tuple count, any separator that wasn't+-- the expected empty one, the sync status, whether the stream really ended).+scenario :: Pqi.Connection -> IO (Either String (Bool, Int, Bool, Int, Int, Int32, Int, Maybe Pqi.ExecStatus, Bool, Bool))+scenario connection = withDeadline deadlineMicros do+  entered <- Pqi.enterPipelineMode connection+  sent <- traverse (\_ -> Pqi.sendQueryParams connection statement [] Pqi.Text) [1 .. commandCount]+  synced <- Pqi.pipelineSync connection+  (commands, okCount, tuples, separatorAnomalies) <- collectCommands 0 0 0 0+  syncResult <- Pqi.getResult connection+  syncStatus <- traverse Pqi.resultStatus syncResult+  idle <- Pqi.getResult connection+  exited <- Pqi.exitPipelineMode connection+  pure+    ( entered,+      length (filter id sent),+      synced,+      commands,+      okCount,+      tuples,+      separatorAnomalies,+      syncStatus,+      isNothing idle,+      exited+    )+  where+    collectCommands !commands !okCount !tuples !separatorAnomalies+      | commands >= commandCount = pure (commands, okCount, tuples, separatorAnomalies)+      | otherwise =+          Pqi.getResult connection >>= \case+            Nothing -> pure (commands, okCount, tuples, separatorAnomalies)+            Just result -> do+              status <- Pqi.resultStatus result+              rows <- Pqi.ntuples result+              separator <- Pqi.getResult connection+              collectCommands+                (commands + 1)+                (okCount + (if status == Pqi.TuplesOk then 1 else 0))+                (tuples + rows)+                (separatorAnomalies + maybe 0 (const 1) separator)++-- | Run an action under a deadline, reporting @Left@ if it does not finish+-- in time. Unlike 'System.Timeout.timeout', this /returns/ even when the+-- action is stuck inside a blocking foreign call (where @timeout@'s async+-- exception cannot be delivered and the wait itself would hang): the action+-- runs in a worker thread that is simply abandoned when the deadline passes.+-- Being abandoned is what makes a deadlocked candidate fail the spec instead+-- of freezing the whole suite.+withDeadline :: Int -> IO a -> IO (Either String a)+withDeadline micros action = do+  box <- Concurrent.newEmptyMVar+  let deliver = Concurrent.tryPutMVar box+  _ <-+    Concurrent.forkIO+      ( void+          ( Exception.try @Exception.SomeException action+              >>= deliver+              . either (Left . show) Right+          )+      )+  timer <-+    Concurrent.forkIO+      ( Concurrent.threadDelay micros+          >> void (deliver (Left ("deadline exceeded: no outcome within " <> show micros <> "us")))+      )+  outcome <- Concurrent.takeMVar box+  Concurrent.killThread timer+  pure outcome