pqi-conformance 1.0.12.0 → 1.0.13.0

raw patch · 10 files changed

+669/−3 lines, 10 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +6/−0
• pqi-conformance.cabal +6/−1
• src/library/Pqi/Conformance/Fixture/ConnectionLoss.hs +238/−0
• src/library/Pqi/Conformance/Operation/Exec.hs +2/−0
• src/library/Pqi/Conformance/Operation/Exec/LostConnection.hs +37/−0
• src/library/Pqi/Conformance/Operation/ExecParams.hs +4/−1
• src/library/Pqi/Conformance/Operation/ExecParams/LostConnection.hs +35/−0
• src/library/Pqi/Conformance/Operation/GetResult.hs +4/−1
• src/library/Pqi/Conformance/Operation/GetResult/LostConnection.hs +66/−0
• src/library/Pqi/Conformance/Scenario/LostConnection.hs +271/−0

@@ -1,3 +1,9 @@+# v1.0.13.0++## Non-breaking++- Added `getResult`, `exec` and `execParams` specs covering a connection that is lost while it is idle and is then used: `sendQuery` (a multi-statement text), `sendQueryParams`, `sendPrepare`, `sendQueryPrepared` (a statement prepared before the loss) and a two-command pipeline run through `getResult` until the sync, plus `exec` and `execParams`, each against three ways of dying - a silent close (FIN with no protocol message), a server-announced termination (`pg_terminate_backend` from a second connection) and a reset (RST). The handle is then used once more and finished. Every call of the victim runs under `try` and a 5 s `timeout`, so that a throw or a hang is recorded in the compared transcript as `THREW …` or `TIMED OUT` instead of aborting the run, and the transcript is compared line by line. The connection goes through an in-process TCP proxy (`Pqi.Conformance.Fixture.ConnectionLoss`) that keeps the client side open and drained once the database side is gone, and the scenario waits for the victim's own socket to have noticed the death before the victim's next command. A direct connection cannot be killed deterministically: the dead server answers the victim's write with a reset that races with reading the `FATAL`/EOF (38 EOF to 2 `Connection reset by peer` in 40 attempts of one cell), and a victim that writes before an RST has been processed by its kernel takes a different path than one that writes after. Found `pqi-native` 1.0.1.12 diverging from `libpq` in every cell. Where the connection is closed or terminated, `getResult` throws `pqi-native: connection closed by server: end of file` where `libpq` returns a `FatalError` result; `status` stays `ConnectionOk` where `libpq` reports `ConnectionBad` (for a termination `libpq` is `ConnectionOk` after the server's `FATAL` and only goes bad on the second `getResult`, which `pqi-native` never reaches); `errorMessage` is empty where `libpq` carries its "server closed the connection unexpectedly" text (after a termination, appended to the `FATAL` text); and the handle accepts a further `sendQuery` (`True`) where `libpq` answers `False` with "no connection to the server", then throws again on `getResult`. After a reset the send already fails (`False`, and for a pipeline the sends after the first) where `libpq` defers a write failure and reports the loss from the following `getResult`. `exec` and `execParams` no longer throw, but their `FatalError` result carries a `DiagMessagePrimary` error field where `libpq`'s made-up result has no error fields at all, after a termination `errorMessage` lacks the preceding `FATAL` text, and after the loss `errorMessage` keeps the "server closed the connection" text where `libpq` reports "no connection to the server". Found via `hasql` issue #341.+ # v1.0.12.0  ## Non-breaking

@@ -1,6 +1,6 @@ cabal-version: 3.0 name: pqi-conformance-version: 1.0.12.0+version: 1.0.13.0 category: Database, PostgreSQL, Testing synopsis: Differential conformance tests for pqi adapters description:@@ -84,6 +84,7 @@     Pqi.Conformance    other-modules:+    Pqi.Conformance.Fixture.ConnectionLoss     Pqi.Conformance.Harness     Pqi.Conformance.Observation     Pqi.Conformance.Operation.BackendPID@@ -116,7 +117,9 @@     Pqi.Conformance.Operation.EscapeStringConn     Pqi.Conformance.Operation.Exec     Pqi.Conformance.Operation.Exec.ConnectionLostMidQuery+    Pqi.Conformance.Operation.Exec.LostConnection     Pqi.Conformance.Operation.ExecParams+    Pqi.Conformance.Operation.ExecParams.LostConnection     Pqi.Conformance.Operation.ExecPrepared     Pqi.Conformance.Operation.ExitPipelineMode     Pqi.Conformance.Operation.Fformat@@ -136,6 +139,7 @@     Pqi.Conformance.Operation.Getlength     Pqi.Conformance.Operation.GetNotice     Pqi.Conformance.Operation.GetResult+    Pqi.Conformance.Operation.GetResult.LostConnection     Pqi.Conformance.Operation.Getvalue     Pqi.Conformance.Operation.GetvalueCopy     Pqi.Conformance.Operation.Host@@ -205,6 +209,7 @@     Pqi.Conformance.Prelude     Pqi.Conformance.Reference     Pqi.Conformance.Scenario+    Pqi.Conformance.Scenario.LostConnection    build-depends:     base >=4.11 && <5,

@@ -0,0 +1,238 @@+-- | Fixtures for killing a live connection in a deterministic way: an+-- in-process TCP proxy between the victim and the container, which can end+-- the connection in each of the ways a connection can really die.+--+-- A direct connection cannot be killed deterministically. A dead server+-- answers the victim's next write with a reset, and that reset races with the+-- victim reading whatever the server said before it went away (the @FATAL@+-- of a terminated backend, or just the EOF): the same scenario run against+-- the same server surfaces as a clean EOF in most attempts and as a+-- @Connection reset by peer@ in the rest. The proxy removes the race by+-- being the one party that decides what the victim sees:+--+-- * Once the database side is gone, the client side of the proxy /stays+--   open and keeps being drained/. The victim's next write is swallowed+--   instead of reset, so \"the send still succeeds, the read finds the EOF\"+--   holds on every run.+-- * A reset is delivered exactly when asked for, instead of whenever the+--   kernel gets around to answering a write.+--+-- The proxy rewrites nothing at the protocol level, so it works for any+-- adapter.+module Pqi.Conformance.Fixture.ConnectionLoss+  ( -- * Proxy+    withProxy,+    Proxy (..),++    -- * Server-side termination+    terminateBackend,+  )+where++import Control.Concurrent (MVar, forkIO, newEmptyMVar, putMVar, readMVar, tryPutMVar, tryReadMVar)+import Control.Exception (IOException, SomeException, bracket, bracketOnError, catch, finally, onException, 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+import qualified Pqi+import Pqi.Conformance.Prelude+import qualified Pqi.Conformance.Reference as Reference++-- | The means of ending the proxied connection. All of them need the victim+-- to have connected first.+data Proxy = Proxy+  { -- | Silent close: shut down the database-facing socket, so that the+    -- victim receives a graceful close (FIN) without any protocol message+    -- before it. The client-facing socket stays open and keeps being+    -- drained. Returns once the FIN has been delivered to the victim.+    sever :: IO (),+    -- | Reset: close the client-facing socket with @SO_LINGER@ @{on, 0}@, so+    -- that the victim receives an @RST@, and close the database side. Returns+    -- once both forwarding threads have stopped.+    reset :: IO (),+    -- | Block until the /database/ side has closed on its own (after+    -- 'terminateBackend', for instance) and its last bytes, together with+    -- the FIN, have been delivered to the victim. From then on the victim's+    -- writes are swallowed, never forwarded to the dead server (which would+    -- answer them with a reset that races with the victim's read).+    serverClosed :: IO ()+  }++-- | Hand the action a conninfo that routes through a proxy to the same+-- server as the given one, together with the means of killing the connection.+--+-- The proxy accepts one client. The listening socket and every thread+-- started here are gone when the action returns.+withProxy :: ByteString -> (ByteString -> Proxy -> IO a) -> IO a+withProxy conninfo action =+  bracket listenOnLoopback Socket.close \listener -> do+    port <- Socket.socketPort listener+    sessionVar <- newEmptyMVar+    let serve = try @SomeException (establish listener host upstreamPort) >>= void . tryPutMVar sessionVar+        established = readMVar sessionVar >>= either (ioError . userError . show) pure+        proxy =+          Proxy+            { sever = do+                Session {..} <- established+                quietly (Socket.shutdown upstream Socket.ShutdownBoth)+                readMVar upstreamGone,+              reset = do+                Session {..} <- established+                quietly (Socket.setSockOptValue client Socket.Linger (Socket.SockOptValue (Socket.StructLinger 1 0)))+                quietly (Socket.close client)+                quietly (Socket.shutdown upstream Socket.ShutdownBoth)+                readMVar clientToUpstreamStopped+                readMVar upstreamToClientStopped,+              serverClosed = do+                Session {..} <- established+                readMVar upstreamGone+            }+        proxiedConninfo = conninfo <> " host=127.0.0.1 port=" <> ByteString.Char8.pack (show port)+    bracket (forkIO serve) (\_ -> teardown listener sessionVar) \_ -> action proxiedConninfo proxy+  where+    (host, upstreamPort) = upstreamAddress conninfo+    teardown listener sessionVar = do+      -- Closing the listener wakes an acceptor still waiting for a client,+      -- and the acceptor always fills the session variable, so this is+      -- bounded.+      quietly (Socket.close listener)+      readMVar sessionVar >>= \case+        Left _ -> pure ()+        Right Session {..} -> do+          quietly (Socket.shutdown client Socket.ShutdownBoth)+          quietly (Socket.shutdown upstream Socket.ShutdownBoth)+          readMVar clientToUpstreamStopped+          readMVar upstreamToClientStopped+          quietly (Socket.close client)+          quietly (Socket.close upstream)++-- | One proxied connection: the two sockets, and the state of the two+-- forwarding threads.+data Session = Session+  { client :: Socket.Socket,+    upstream :: Socket.Socket,+    -- | Filled when the database side has reached EOF (or failed) and the+    -- FIN has been forwarded to the client. Writes of the client are+    -- swallowed from then on.+    upstreamGone :: MVar (),+    clientToUpstreamStopped :: MVar (),+    upstreamToClientStopped :: MVar ()+  }++-- | Accept the client, dial the database and start the forwarding threads.+establish :: Socket.Socket -> String -> String -> IO Session+establish listener host port = do+  (client, _) <- Socket.accept listener+  upstream <- connectUpstream host port `onException` Socket.close client+  quietly (Socket.setSocketOption client Socket.NoDelay 1)+  quietly (Socket.setSocketOption upstream Socket.NoDelay 1)+  upstreamGone <- newEmptyMVar+  clientToUpstreamStopped <- newEmptyMVar+  upstreamToClientStopped <- newEmptyMVar+  _ <- forkIO (quietly (clientToUpstream client upstream upstreamGone) `finally` putMVar clientToUpstreamStopped ())+  _ <- forkIO (quietly (upstreamToClient client upstream upstreamGone) `finally` putMVar upstreamToClientStopped ())+  pure Session {..}++-- | Forward what the victim writes to the database, until the victim closes+-- its side. After the database side is gone, keep reading and discard.+clientToUpstream :: Socket.Socket -> Socket.Socket -> MVar () -> IO ()+clientToUpstream client upstream upstreamGone =+  pump `finally` quietly (Socket.shutdown upstream Socket.ShutdownSend)+  where+    pump = do+      chunk <- Socket.ByteString.recv client chunkSize+      unless (ByteString.null chunk) do+        gone <- isJust <$> tryReadMVar upstreamGone+        unless gone (quietly (Socket.ByteString.sendAll upstream chunk))+        pump++-- | Forward what the database writes to the victim, until the database side+-- ends, and pass the end on to the victim as a FIN.+upstreamToClient :: Socket.Socket -> Socket.Socket -> MVar () -> IO ()+upstreamToClient client upstream upstreamGone =+  pump `finally` do+    quietly (Socket.shutdown client Socket.ShutdownSend)+    void (tryPutMVar upstreamGone ())+  where+    pump = do+      chunk <- Socket.ByteString.recv upstream chunkSize+      unless (ByteString.null chunk) do+        Socket.ByteString.sendAll client chunk+        pump++chunkSize :: Int+chunkSize = 16384++-- | Terminate a backend from a second connection, opened directly to the+-- container (the conninfo is the harness's, not a proxied one), and wait+-- until the backend is gone from @pg_stat_activity@.+--+-- Uses the one-argument @pg_terminate_backend@. The pid is the victim's+-- 'Pqi.backendPID'.+terminateBackend :: ByteString -> Int32 -> IO ()+terminateBackend conninfo pid =+  bracket (Pqi.connectdb Reference.adapter conninfo) Pqi.finish \connection -> do+    _ <- Pqi.exec connection ("select pg_terminate_backend(" <> pidText <> ")")+    awaitGone connection (500 :: Int)+  where+    pidText = ByteString.Char8.pack (show pid)+    awaitGone connection attemptsLeft = do+      remaining <-+        Pqi.exec connection ("select count(*) from pg_stat_activity where pid = " <> pidText) >>= \case+          Nothing -> ioError (userError "terminateBackend: pg_stat_activity query returned no result")+          Just result -> Pqi.getvalue result 0 0+      case remaining of+        Just "0" -> pure ()+        _+          | attemptsLeft <= 0 -> ioError (userError ("terminateBackend: backend " <> show pid <> " is still alive"))+          | otherwise -> threadDelay 10_000 >> awaitGone connection (attemptsLeft - 1)++-- | The host and port of the database in a @keyword=value@ conninfo. The+-- last occurrence of a keyword wins, as in libpq.+upstreamAddress :: ByteString -> (String, String)+upstreamAddress conninfo =+  (keyword "host" "localhost", keyword "port" "5432")+  where+    assignments =+      [ (key, drop 1 value)+      | token <- words (ByteString.Char8.unpack conninfo),+        let (key, value) = break (== '=') token+      ]+    keyword key fallback = fromMaybe fallback (lookup key (reverse assignments))++listenOnLoopback :: IO Socket.Socket+listenOnLoopback = do+  address : _ <-+    Socket.getAddrInfo+      (Just Socket.defaultHints {Socket.addrSocketType = Socket.Stream})+      (Just "127.0.0.1")+      (Just "0")+  bracketOnError (Socket.socket (Socket.addrFamily address) (Socket.addrSocketType address) (Socket.addrProtocol address)) Socket.close \sock -> do+    Socket.bind sock (Socket.addrAddress address)+    Socket.listen sock 1+    pure sock++-- | Connect to the first of the host's addresses that accepts.+connectUpstream :: String -> String -> IO Socket.Socket+connectUpstream host port = do+  addresses :: [Socket.AddrInfo] <-+    Socket.getAddrInfo+      (Just Socket.defaultHints {Socket.addrSocketType = Socket.Stream})+      (Just host)+      (Just port)+  foldr+    (\address next -> attempt address `catch` \(_ :: IOException) -> next)+    (ioError (userError ("could not connect to " <> host <> ":" <> port)))+    addresses+  where+    attempt address =+      bracketOnError (Socket.socket (Socket.addrFamily address) (Socket.addrSocketType address) (Socket.addrProtocol address)) Socket.close \sock -> do+        Socket.connect sock (Socket.addrAddress address)+        pure sock++-- | Run an action for its effect, ignoring any exception: the sockets here+-- are being shut down from several sides at once, and a socket that is+-- already gone is the expected outcome rather than an error.+quietly :: IO () -> IO ()+quietly action = void (try @SomeException action)

@@ -8,6 +8,7 @@ import qualified Pqi import Pqi.Conformance.Harness import qualified Pqi.Conformance.Operation.Exec.ConnectionLostMidQuery as ConnectionLostMidQuery+import qualified Pqi.Conformance.Operation.Exec.LostConnection as LostConnection import Pqi.Conformance.Prelude import Pqi.Conformance.Scenario import Test.Hspec@@ -15,6 +16,7 @@ spec :: Pqi.Adapter -> SpecWith ByteString spec adapter = do   ConnectionLostMidQuery.spec adapter+  LostConnection.spec adapter    describe "exec" do     let forCase title sql =

@@ -0,0 +1,37 @@+-- | Coverage for 'Pqi.exec' when the connection was lost before the call,+-- as it is for the next user of a pooled connection that died while idle,+-- against each way of dying: a silent close, a server-announced termination+-- (@pg_terminate_backend@), and a reset.+--+-- The contract is @libpq@'s: 'Pqi.exec' never throws on a lost connection.+-- It returns a 'Pqi.FatalError' result (or 'Nothing', where @libpq@ does),+-- and 'Pqi.status' becomes 'Pqi.ConnectionBad' with the message in+-- 'Pqi.errorMessage'. Distinct from+-- "Pqi.Conformance.Operation.Exec.ConnectionLostMidQuery", where a+-- hand-rolled server drops the connection /after/ reading the query; here+-- the server is the real one, and it is gone before the query is sent. See+-- <https://github.com/nikita-volkov/hasql/issues/341>.+--+-- See "Pqi.Conformance.Scenario.LostConnection" for what is observed and how+-- a throw or a hang is folded into the observation, and+-- "Pqi.Conformance.Fixture.ConnectionLoss" for why the connection goes+-- through a proxy instead of being killed directly.+module Pqi.Conformance.Operation.Exec.LostConnection+  ( spec,+  )+where++import qualified Pqi+import Pqi.Conformance.Prelude+import qualified Pqi.Conformance.Scenario.LostConnection as LostConnection+import Test.Hspec++spec :: Pqi.Adapter -> SpecWith ByteString+spec adapter =+  describe "exec" do+    LostConnection.spec+      adapter+      [ ( "a single statement",+          LostConnection.synchronous (`Pqi.exec` "select 1")+        )+      ]

@@ -9,12 +9,15 @@ import qualified Pqi as Lq import Pqi.Conformance.Harness import Pqi.Conformance.Observation+import qualified Pqi.Conformance.Operation.ExecParams.LostConnection as LostConnection import Pqi.Conformance.Prelude import Pqi.Conformance.Scenario import Test.Hspec  spec :: Lq.Adapter -> SpecWith ByteString-spec adapter =+spec adapter = do+  LostConnection.spec adapter+   describe "execParams" do     it "text result format" \conninfo ->       differential adapter conninfo (paramsScenario Lq.Text)

@@ -0,0 +1,35 @@+-- | Coverage for 'Pqi.execParams' when the connection was lost before the+-- call, as it is for the next user of a pooled connection that died while+-- idle, against each way of dying: a silent close, a server-announced+-- termination (@pg_terminate_backend@), and a reset.+--+-- The contract is @libpq@'s: 'Pqi.execParams' never throws on a lost+-- connection. It returns a 'Pqi.FatalError' result (or 'Nothing', where+-- @libpq@ does), and 'Pqi.status' becomes 'Pqi.ConnectionBad' with the+-- message in 'Pqi.errorMessage'. See+-- <https://github.com/nikita-volkov/hasql/issues/341>.+--+-- See "Pqi.Conformance.Scenario.LostConnection" for what is observed and how+-- a throw or a hang is folded into the observation, and+-- "Pqi.Conformance.Fixture.ConnectionLoss" for why the connection goes+-- through a proxy instead of being killed directly.+module Pqi.Conformance.Operation.ExecParams.LostConnection+  ( spec,+  )+where++import qualified Pqi+import Pqi.Conformance.Prelude+import qualified Pqi.Conformance.Scenario.LostConnection as LostConnection+import Test.Hspec++spec :: Pqi.Adapter -> SpecWith ByteString+spec adapter =+  describe "execParams" do+    LostConnection.spec+      adapter+      [ ( "a single statement",+          LostConnection.synchronous \connection ->+            Pqi.execParams connection "select 1" [] Pqi.Text+        )+      ]

@@ -7,12 +7,15 @@  import qualified Pqi import Pqi.Conformance.Harness+import qualified Pqi.Conformance.Operation.GetResult.LostConnection as LostConnection import Pqi.Conformance.Prelude import Pqi.Conformance.Scenario (takeResult) import Test.Hspec  spec :: Pqi.Adapter -> SpecWith ByteString-spec adapter =+spec adapter = do+  LostConnection.spec adapter+   describe "getResult" do     it "yields each result then a Nothing terminator" \conninfo ->       differential adapter conninfo \connection -> do

@@ -0,0 +1,66 @@+-- | Coverage for 'Pqi.getResult' when the connection is lost between the+-- command being sent and its results being read - which is how a pool+-- connection that died while idle shows up to its next user. Every+-- asynchronous way of starting a command is covered ('Pqi.sendQuery',+-- 'Pqi.sendQueryParams', 'Pqi.sendPrepare', 'Pqi.sendQueryPrepared', and a+-- pipeline), against each way of dying: a silent close, a server-announced+-- termination (@pg_terminate_backend@), and a reset.+--+-- The contract is @libpq@'s: 'Pqi.getResult' never throws on a lost+-- connection. It returns a 'Pqi.FatalError' result - for a termination+-- first the server's own @FATAL@ (while 'Pqi.status' is still+-- 'Pqi.ConnectionOk'), then a made-up one without error fields whose message+-- is \"server closed the connection unexpectedly\" - and 'Pqi.status'+-- becomes 'Pqi.ConnectionBad'. A consumer written against that contract+-- (hasql's @Connection.use@ and pool, notably, which classify a dead+-- connection by its status) breaks on anything else: see+-- <https://github.com/nikita-volkov/hasql/issues/341>.+--+-- The scenario also reuses the handle afterwards and finishes it, because+-- the divergence that was found left the handle in a state in which+-- 'Pqi.finish' never returned. See+-- "Pqi.Conformance.Scenario.LostConnection" for what is observed and how a+-- throw or a hang is folded into the observation, and+-- "Pqi.Conformance.Fixture.ConnectionLoss" for why the connection goes+-- through a proxy instead of being killed directly.+module Pqi.Conformance.Operation.GetResult.LostConnection+  ( spec,+  )+where++import qualified Pqi+import Pqi.Conformance.Prelude+import qualified Pqi.Conformance.Scenario.LostConnection as LostConnection+import Test.Hspec++spec :: Pqi.Adapter -> SpecWith ByteString+spec adapter =+  describe "getResult" do+    LostConnection.spec+      adapter+      [ ( "sendQuery, a multi-statement text",+          LostConnection.asynchronous LostConnection.UntilNothing \connection ->+            pure <$> Pqi.sendQuery connection "select 1; select 2"+        ),+        ( "sendQueryParams",+          LostConnection.asynchronous LostConnection.UntilNothing \connection ->+            pure <$> Pqi.sendQueryParams connection "select 1" [] Pqi.Text+        ),+        ( "sendPrepare",+          LostConnection.asynchronous LostConnection.UntilNothing \connection ->+            pure <$> Pqi.sendPrepare connection "lost_connection" "select 1" Nothing+        ),+        ( "sendQueryPrepared, a statement prepared before the connection was lost",+          LostConnection.withPreparedStatement "lost_connection" "select 1" do+            LostConnection.asynchronous LostConnection.UntilNothing \connection ->+              pure <$> Pqi.sendQueryPrepared connection "lost_connection" [] Pqi.Text+        ),+        ( "a pipeline of two commands and a sync",+          LostConnection.asynchronous LostConnection.UntilPipelineSync \connection -> do+            entered <- Pqi.enterPipelineMode connection+            first <- Pqi.sendQueryParams connection "select 1" [] Pqi.Text+            second <- Pqi.sendQueryParams connection "select 2" [] Pqi.Text+            synced <- Pqi.pipelineSync connection+            pure [entered, first, second, synced]+        )+      ]

@@ -0,0 +1,271 @@+-- | The scenario shared by the @LostConnection@ specs of the operations that+-- run a command: the victim connects through a proxy+-- ("Pqi.Conformance.Fixture.ConnectionLoss"), the connection is killed in+-- one of three ways, and the victim runs its next command and then keeps+-- using the same handle.+--+-- The observation is the /transcript/ of everything the victim saw: the+-- return value of every call, the connection's @status@ and @errorMessage@+-- after it, and whether the call threw or hung. The scenario asserts nothing+-- about @libpq@'s values itself; the reference produces them and the+-- candidate is compared against them.+--+-- Every call of the victim is /bounded/: it runs under+-- 'System.Timeout.timeout' and under 'Control.Exception.try', so that a+-- candidate which throws or hangs fails the comparison with a transcript+-- that says so (@THREW …@, @TIMED OUT@), rather than killing the run - or+-- taking the whole suite down with it, as a @finish@ that never returns+-- does. The same applies to the reference, so that the comparison is of like+-- with like; a reference that throws would be a broken fixture.+module Pqi.Conformance.Scenario.LostConnection+  ( -- * Specs+    spec,+    Operation,+    asynchronous,+    synchronous,+    Collection (..),+    withPreparedStatement,+  )+where++import Control.Exception (SomeException, try)+import Data.List (intercalate)+import GHC.Conc (threadWaitRead)+import qualified Pqi+import Pqi.Conformance.Fixture.ConnectionLoss+import Pqi.Conformance.Harness+import Pqi.Conformance.Observation (ResultObservation (..), observeResult)+import Pqi.Conformance.Prelude+import System.Timeout (timeout)+import Test.Hspec++-- | The ways in which the connection is killed.+data Death+  = -- | A graceful close (FIN) with no protocol message.+    SilentClose+  | -- | @pg_terminate_backend@ from a second connection: the server announces+    -- it with a @FATAL@ and then closes.+    ServerAnnouncedTermination+  | -- | The peer answers with an @RST@.+    Reset+  deriving stock (Eq, Show, Enum, Bounded)++deathTitle :: Death -> String+deathTitle = \case+  SilentClose -> "after a silent close"+  ServerAnnouncedTermination -> "after a server-announced termination"+  Reset -> "after a reset"++-- | Kill the connection, given the harness conninfo, the proxy and the+-- victim's backend pid, and return once the victim's own socket has noticed.+--+-- The proxy only knows that it has closed or reset its side. Whether the+-- victim's kernel has /processed/ the resulting @FIN@, @RST@ or @FATAL@ is+-- another matter, and it matters: a victim that writes before the @RST@ has+-- arrived has its write accepted and fails on its next read, whereas after+-- it the write itself fails. @libpq@ reports the same either way, so the+-- reference does not show the difference, but an adapter that reports the+-- two paths differently flips between two outcomes from run to run. Waiting+-- for the victim's socket to turn readable - which a pending @FIN@, @RST@ or+-- unread bytes all make it, without consuming any of them - removes that+-- race.+kill :: Death -> ByteString -> Proxy -> Pqi.Connection -> Int32 -> IO ()+kill death conninfo proxy victim pid = do+  case death of+    SilentClose -> sever proxy+    ServerAnnouncedTermination -> terminateBackend conninfo pid >> serverClosed proxy+    Reset -> reset proxy+  void (try @SomeException (Pqi.socket victim >>= traverse_ awaitReadable))+  where+    awaitReadable fd = timeout callDeadlineMicros (threadWaitRead fd)++-- | What the victim does after the connection has been killed.+data Operation = Operation+  { -- | Runs before the connection is killed, so that the command that+    -- follows has something to rely on. Its outcome is part of the+    -- transcript, to show that the preparation really happened.+    setup :: Pqi.Connection -> IO [String],+    -- | The command, together with the observation of its outcome.+    command :: Pqi.Connection -> IO [String]+  }++-- | A command that is sent, and whose results are then collected with+-- 'Pqi.getResult'. The sends are all run and their returned flags+-- recorded, in order.+asynchronous :: Collection -> (Pqi.Connection -> IO [Bool]) -> Operation+asynchronous collection send =+  Operation+    { setup = \_ -> pure [],+      command = \connection -> do+        sent <- bounded (send connection)+        collected <- collect collection connection+        pure (("send: " <> renderCall show sent) : collected)+    }++-- | A command that returns its result synchronously: no 'Pqi.getResult'+-- loop.+synchronous :: (Pqi.Connection -> IO (Maybe Pqi.Result)) -> Operation+synchronous run =+  Operation+    { setup = \_ -> pure [],+      command = \connection -> do+        returned <- bounded (run connection >>= traverse observeResult)+        status <- bounded (Pqi.status connection)+        message <- bounded (Pqi.errorMessage connection)+        pure+          [ "returned: " <> renderCall renderMaybeResult returned,+            "status: " <> renderCall show status,+            "errorMessage: " <> renderCall show message+          ]+    }++-- | Prepare a statement before the connection is killed, so that the command+-- can use the cached statement, as a prepared-statement cache does.+withPreparedStatement :: ByteString -> ByteString -> Operation -> Operation+withPreparedStatement name sql operation =+  operation+    { setup = \connection -> do+        prepared <- Pqi.prepare connection name sql Nothing >>= traverse observeResult+        rest <- setup operation connection+        pure (("prepared: " <> renderMaybeResult prepared) : rest)+    }++-- | How the results of an asynchronous command are collected.+data Collection+  = -- | Until the first 'Nothing': the end of a command's results.+    UntilNothing+  | -- | Through the pipeline: until the sync result, or until the stream+    -- reports 'Nothing' twice in a row, since in pipeline mode a 'Nothing'+    -- separates every command's results from the next one's.+    UntilPipelineSync++-- | The most results that are collected. A connection that is lost yields+-- one or two; a candidate that fails to notice would otherwise loop.+resultCap :: Int+resultCap = 10++-- | How long any single call of the victim is allowed to take.+callDeadlineMicros :: Int+callDeadlineMicros = 5_000_000++-- | The outcome of one bounded call.+data Call a+  = Returned a+  | Threw String+  | TimedOut++bounded :: IO a -> IO (Call a)+bounded action =+  try @SomeException (timeout callDeadlineMicros action) <&> \case+    Left exception -> Threw (show exception)+    Right Nothing -> TimedOut+    Right (Just value) -> Returned value+  where+    (<&>) = flip fmap++renderCall :: (a -> String) -> Call a -> String+renderCall render = \case+  Returned value -> render value+  Threw exception -> "THREW " <> exception+  TimedOut -> "TIMED OUT after " <> show callDeadlineMicros <> "us"++renderMaybeResult :: Maybe ResultObservation -> String+renderMaybeResult = \case+  Nothing -> "Nothing"+  Just observation ->+    -- Only the error fields that are set: all of them are always observed,+    -- and nearly all are absent.+    "Just " <> show observation {errorFields = filter (isJust . snd) (errorFields observation)}++-- | Collect results and observe each of them, together with the+-- connection's @status@ and @errorMessage@ right after the call.+collect :: Collection -> Pqi.Connection -> IO [String]+collect collection connection = go 1 False+  where+    go :: Int -> Bool -> IO [String]+    go number previousWasNothing+      | number > resultCap = pure ["getResult: not finished after " <> show resultCap <> " calls"]+      | otherwise = do+          returned <- bounded (Pqi.getResult connection >>= traverse observeResult)+          status <- bounded (Pqi.status connection)+          message <- bounded (Pqi.errorMessage connection)+          let observation =+                [ "getResult #" <> show number <> ": " <> renderCall renderMaybeResult returned,+                  "  status: " <> renderCall show status,+                  "  errorMessage: " <> renderCall show message+                ]+              finished = case (collection, returned) of+                (_, Threw _) -> True+                (_, TimedOut) -> True+                (UntilNothing, Returned Nothing) -> True+                (UntilPipelineSync, Returned Nothing) -> previousWasNothing+                (UntilPipelineSync, Returned (Just ResultObservation {status = Pqi.PipelineSync})) -> True+                _ -> False+          if finished+            then pure observation+            else (observation <>) <$> go (number + 1) (isNothingResult returned)+    isNothingResult = \case+      Returned Nothing -> True+      _ -> False++-- | The same handle, used again: a new command, then what the connection+-- says about itself.+reuse :: Pqi.Connection -> IO [String]+reuse connection = do+  sent <- bounded (Pqi.sendQuery connection "select 1")+  message <- bounded (Pqi.errorMessage connection)+  result <- bounded (Pqi.getResult connection >>= traverse observeResult)+  status <- bounded (Pqi.status connection)+  pure+    [ "afterwards, sendQuery: " <> renderCall show sent,+      "afterwards, errorMessage: " <> renderCall show message,+      "afterwards, getResult: " <> renderCall renderMaybeResult result,+      "afterwards, status: " <> renderCall show status+    ]++-- | What the victim saw, one line per observation.+--+-- 'Show' renders the lines as they are, so that a failed comparison is+-- diffed line by line.+newtype Transcript = Transcript [String]+  deriving newtype (Eq)++instance Show Transcript where+  show (Transcript observations) = intercalate "\n" observations++scenario :: Death -> Operation -> Pqi.Adapter -> ByteString -> IO Transcript+scenario death operation adapter conninfo =+  withProxy conninfo \proxiedConninfo proxy -> do+    victim <- Pqi.connectdb adapter proxiedConninfo+    connected <- bounded (Pqi.status victim)+    prepared <- bounded (setup operation victim)+    pid <- Pqi.backendPID victim+    kill death conninfo proxy victim pid+    during <- command operation victim+    afterwards <- reuse victim+    finished <- bounded (Pqi.finish victim)+    pure+      ( Transcript+          ( concat+              [ ["connected: " <> renderCall show connected],+                case prepared of+                  Returned observations -> observations+                  Threw exception -> ["setup: THREW " <> exception]+                  TimedOut -> ["setup: TIMED OUT after " <> show callDeadlineMicros <> "us"],+                during,+                afterwards,+                ["finish: " <> renderCall show finished]+              ]+          )+      )++-- | The matrix: every way of dying, with every given operation as the+-- victim's next command.+spec :: Pqi.Adapter -> [(String, Operation)] -> SpecWith ByteString+spec adapter operations =+  describe "lost connection" do+    for_ [minBound .. maxBound] \death ->+      describe (deathTitle death) do+        for_ operations \(title, operation) ->+          it title \conninfo ->+            differentialConnect adapter conninfo (scenario death operation)