pqi-native 1.0.1.13 → 1.0.1.14
raw patch · 7 files changed
+317/−147 lines, 7 filesdep ~pqi-conformancePVP ok
version bump matches the API change (PVP)
Dependency ranges changed: pqi-conformance
API changes (from Hackage documentation)
Files
- CHANGELOG.md +8/−0
- pqi-native.cabal +2/−2
- src/library/Pqi/Native.hs +27/−21
- src/library/Pqi/Native/Connection.hs +162/−34
- src/library/Pqi/Native/Query.hs +78/−80
- src/library/Pqi/Native/Types.hs +25/−3
- src/transport/Pqi/Native/Transport.hs +15/−7
CHANGELOG.md view
@@ -1,3 +1,11 @@+# v1.0.1.14++## Fixes++- Fixed the `hostaddr` conninfo parameter being ignored ([#11](https://github.com/nikita-volkov/pqi-native/issues/11)). With a `hostaddr`, libpq connects to that numeric address and never resolves `host`, which it keeps only for authentication and for naming the server in messages. `pqi-native` had no notion of it: it resolved `host` through DNS regardless, so a conninfo whose `host` does not resolve failed to connect, and the `hostaddr` key was forwarded to the server in the startup packet as if it were a run-time setting, which libpq never does. `hostaddr` is now parsed from both the `key=value` form and the URI's query string, kept out of the startup parameters, and dialled over TCP in preference to `host`, also when `host` is a Unix-domain socket directory, and also by a later `cancel` and `reset`. `host` is still what `Pqi.host` reports. Caught by the `pqi-conformance` spec `Pqi.Conformance.Operation.Connectdb.Hostaddr`.++- Fixed a lost connection being reported by throwing an `IOException` instead of the way libpq reports it. When the server closed the connection, or the socket failed, while a flow was waiting for results, `Pqi.getResult` (and, for a send that failed, the send itself) used to throw `pqi-native: connection closed by server: end of file` or a socket error such as `Connection reset by peer`; `status` stayed `ConnectionOk`; `errorMessage` stayed empty or lost the server's own `FATAL` text; and a further send was accepted and threw again. libpq never throws here ([hasql#341](https://github.com/nikita-volkov/hasql/issues/341)). Now, as in `pqReadData`, `PQgetResult` and `pqDropConnection`, the loss returns a `FatalError` result with no error fields (so no SQLSTATE) whose `resultErrorMessage` is libpq's "server closed the connection unexpectedly" text, appended to the connection's `errorMessage` after whatever was recorded before it. The connection becomes `ConnectionBad`, its socket is dropped (`socket` is `Nothing`, `finish` stays safe) and the next `getResult` returns `Nothing`. When the server announced the termination with a `FATAL`, that result is still returned first and unchanged, with `status` `ConnectionOk`, and the loss is reported by the read that follows. A pipeline that is lost ends the same way, without `PipelineAbort` results made up for the outstanding commands. A failed write (`Broken pipe`, `Connection reset by peer`) no longer throws either: as in libpq's `pqsecure_raw_write` it is remembered, the send reports success, and the loss is reported by the following read. A send, or a synchronous `exec` family call, on a connection that is not usable returns `False` (`Nothing`) and sets `errorMessage` to "no connection to the server", and `transactionStatus` is `TransUnknown`, as libpq's are. `putCopyData` and `putCopyEnd` on such a connection return `CopyInError` instead of attempting the write. Also fixed in a pipeline: `errorMessage` was empty after a server error where libpq has the server's text, and was not cleared when the next command's results begin. Only `IOException` from the transport is caught, so asynchronous exceptions such as `ThreadKilled` and `Timeout` still propagate untouched. Caught by the `pqi-conformance` specs `Pqi.Conformance.Operation.GetResult.LostConnection`, `Pqi.Conformance.Operation.Exec.LostConnection` and `Pqi.Conformance.Operation.ExecParams.LostConnection`, whose scenario (`pqi-conformance` 1.0.14.0) also pins `transactionStatus` and `socket` after the loss.+ # v1.0.1.13 ## Non-breaking
pqi-native.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: pqi-native-version: 1.0.1.13+version: 1.0.1.14 category: Database, PostgreSQL synopsis: Native (pure-Haskell) adapter for pqi description:@@ -160,5 +160,5 @@ build-depends: base >=4.11 && <5, hspec >=2.11 && <2.12,- pqi-conformance ^>=1.0.12,+ pqi-conformance ^>=1.0.15, pqi-native,
src/library/Pqi/Native.hs view
@@ -9,7 +9,6 @@ ) where -import Control.Exception (catch) import qualified Data.ByteString as ByteString import qualified Data.ByteString.Char8 as ByteString.Char8 import qualified Data.Map.Strict as Map@@ -68,6 +67,7 @@ { Pqi.connectPoll = pure Pqi.PollingOk, Pqi.isNullConnection = (Connection.isNull connection), Pqi.finish = do+ -- Safe after a lost connection, which has closed the socket already. transport <- readIORef (Connection.transport connection) Transport.close transport -- Mark the connection unusable so that any further protocol@@ -90,7 +90,14 @@ Pqi.port = pure (Just (ByteString.Char8.pack (show (Connection.port (Connection.info connection))))), Pqi.options = pure (Just ""), Pqi.status = readIORef (Connection.connStatus connection),- Pqi.transactionStatus = transactionStatusOf <$> readIORef (Connection.txStatus connection),+ Pqi.transactionStatus = do+ -- As libpq's @PQtransactionStatus@: nothing is known of the+ -- transaction on a connection that is not usable, whatever the+ -- last @ReadyForQuery@ said.+ status <- readIORef (Connection.connStatus connection)+ case status of+ Pqi.ConnectionOk -> transactionStatusOf <$> readIORef (Connection.txStatus connection)+ _ -> pure Pqi.TransUnknown, Pqi.parameterStatus = \name -> Map.lookup name <$> readIORef (Connection.parameters connection), Pqi.protocolVersion = pure 3, Pqi.serverVersion =@@ -99,7 +106,8 @@ Pqi.socket = do transport <- readIORef (Connection.transport connection) fd <- Transport.socketFd transport- pure (Just (fromIntegral fd :: Fd)),+ -- A dropped connection has no socket, as libpq's @PQsocket@ reports.+ pure (if fd < 0 then Nothing else Just (fromIntegral fd :: Fd)), Pqi.backendPID = maybe 0 fst <$> readIORef (Connection.backendKey connection), Pqi.connectionNeedsPassword = do needed <- readIORef (Connection.passwordNeeded connection)@@ -148,22 +156,16 @@ then pure False else writeIORef (Connection.pipelineStatus connection) Pqi.PipelineOff $> True, -- Both of these send directly, bypassing 'Query.sendAsync' (there is no- -- result-bearing command to track), so they need their own escape- -- hatch for the same class of failure 'Query.sendAsync' guards- -- against: a socket death mid-send throwing instead of the @False@- -- @PQpipelineSync@\/@PQsendFlushRequest@ return on a fatal send.- Pqi.pipelineSync =- catch- do- Connection.sendMessage connection syncMessage- modifyIORef' (Connection.pendingSyncs connection) (+ 1)- writeIORef (Connection.asyncPending connection) True- pure True- (\err -> False <$ Connection.markConnectionLost connection err),- Pqi.sendFlushRequest =- catch- (Connection.sendMessage connection flushMessage $> True)- (\err -> False <$ Connection.markConnectionLost connection err),+ -- result-bearing command to track), so they refuse a connection that+ -- is not usable themselves, the way it does.+ Pqi.pipelineSync = Connection.whenUsable connection False do+ Connection.sendMessage connection syncMessage+ modifyIORef' (Connection.pendingSyncs connection) (+ 1)+ writeIORef (Connection.asyncPending connection) True+ pure True,+ Pqi.sendFlushRequest = Connection.whenUsable connection False do+ Connection.sendMessage connection flushMessage+ pure True, Pqi.getCancel = do key <- readIORef (Connection.backendKey connection) pure@@ -172,6 +174,7 @@ mkCancel NativeCancel { host = Connection.host (Connection.info connection),+ hostaddr = Connection.hostaddr (Connection.info connection), port = Connection.port (Connection.info connection), pid, secret,@@ -185,8 +188,11 @@ Pqi.disableNoticeReporting = writeIORef (Connection.noticeReporting connection) False, Pqi.enableNoticeReporting = writeIORef (Connection.noticeReporting connection) True, Pqi.getNotice = popFirst (Connection.notices connection),- Pqi.putCopyData = \payload -> Connection.sendMessage connection (copyDataMessage payload) $> Pqi.CopyInOk,- Pqi.putCopyEnd = \reason -> do+ Pqi.putCopyData = \payload ->+ Connection.whenUsable connection Pqi.CopyInError+ $ Connection.sendMessage connection (copyDataMessage payload)+ $> Pqi.CopyInOk,+ Pqi.putCopyEnd = \reason -> Connection.whenUsable connection Pqi.CopyInError do Connection.sendMessage connection (maybe copyDoneMessage copyFailMessage reason) writeIORef (Connection.asyncPending connection) True pure Pqi.CopyInOk,
src/library/Pqi/Native/Connection.hs view
@@ -11,22 +11,25 @@ nullConnection, reconnect, nextMessage,+ receiveMessage,+ ConnectionLost (..), sendMessage, fieldValue, setError,- connectionLostMessage,- markConnectionLost,+ loseConnection,+ whenUsable, ) where -import Control.Exception (IOException, SomeException, catch, try)+import Control.Exception (Exception, IOException, SomeException, catch, throwIO, try) import qualified Data.ByteString as ByteString import qualified Data.ByteString.Char8 as ByteString.Char8 import Data.List (isInfixOf) import qualified Data.Map.Strict as Map import qualified Data.Sequence as Seq import qualified Data.Set as Set-import GHC.IO.Exception (IOErrorType (ResourceVanished), ioe_description, ioe_location, ioe_type)+import Foreign.C.Error (Errno (..), errnoToIOError)+import GHC.IO.Exception (IOErrorType (ResourceVanished), ioe_description, ioe_errno, ioe_location, ioe_type) import Pqi (ConnStatus (..), Notify (..), PipelineStatus (..), Verbosity (..)) import qualified Pqi.Native.Auth as Auth import Pqi.Native.Prelude@@ -46,6 +49,10 @@ -- | Parsed connection parameters (the @key=value@ subset we support). data ConnInfo = ConnInfo { host :: ByteString,+ -- | The numeric address to connect to, or empty when none was given.+ -- When present it is the one dialled, over TCP, and 'host' is left+ -- unresolved: see 'Transport.connect'.+ hostaddr :: ByteString, port :: Int, user :: ByteString, database :: ByteString,@@ -155,6 +162,7 @@ parseKeyValue dfltUser dfltHost raw = ConnInfo { host = defaultIfEmpty dfltHost (get "host" ""),+ hostaddr = get "hostaddr" "", port = maybe 5432 fst (ByteString.Char8.readInt (get "port" "5432")), user = theUser, database = get "dbname" theUser,@@ -179,14 +187,14 @@ -- | Conninfo keys already surfaced via their own 'ConnInfo' fields, so they're -- excluded from 'extraParams' rather than duplicated there. reservedKeys :: Set.Set ByteString-reservedKeys = Set.fromList ["host", "port", "user", "dbname", "password"]+reservedKeys = Set.fromList ["host", "hostaddr", "port", "user", "dbname", "password"] -- | Parse the authority+path portion of a @postgresql://@ URI (scheme already -- stripped). Handles @[user[:password]@][host[:port]][/dbname]@; ignores query -- parameters other than what appears in those components. parseUri :: ByteString -> ByteString -> ByteString -> ConnInfo parseUri dfltUser dfltHost withoutScheme =- ConnInfo {host, port, user, database, password, extraParams}+ ConnInfo {host, hostaddr, port, user, database, password, extraParams} where -- Split off optional "userinfo@" prefix. The '@' is unambiguous in this -- position: hosts do not contain '@' in practice.@@ -217,11 +225,12 @@ -- (e.g. ?application_name=foo&sslmode=disable). rawQuery = ByteString.drop 1 (ByteString.dropWhile (/= 0x3f) pathAndQuery) - extraParams =- Map.withoutKeys- (Map.fromList (mapMaybe toQueryPair (ByteString.split 0x26 rawQuery)))- reservedKeys+ queryParams = Map.fromList (mapMaybe toQueryPair (ByteString.split 0x26 rawQuery)) + extraParams = Map.withoutKeys queryParams reservedKeys++ hostaddr = Map.findWithDefault "" "hostaddr" queryParams+ toQueryPair token = case ByteString.elemIndex 0x3d token of Just i -> let (k, v) = ByteString.splitAt i token@@ -320,39 +329,92 @@ -- mirroring libpq's @password_needed@. Trust\/peer\/GSS authentication -- never sets this, regardless of whether a password was supplied in the -- conninfo.- passwordNeeded :: IORef Bool+ passwordNeeded :: IORef Bool,+ -- | Set once a socket-level write has failed, mirroring libpq's+ -- @write_failed@. From then on nothing more is written, and the failure+ -- is reported by the read that follows. See 'sendMessage'.+ writeFailed :: IORef Bool } -- | Send a serialized frontend message.+--+-- A failed write does not surface here, the way it does not in libpq+-- (@pqsecure_raw_write@ in @fe-secure.c@): the failure is remembered in+-- 'writeFailed', the data is discarded, and the caller is told the send+-- succeeded. The reason is that the server may have sent an error that+-- explains the closed socket - a @FATAL@ announcing its termination, say -+-- and reporting the write failure first would hide it. The loss is detected+-- by the receive that follows ('receiveMessage'), which is also where+-- libpq reports it. Everything after a failed write is discarded for the+-- same reason libpq gives: the stream has lost its message boundaries.+--+-- Only 'IOException' is caught, so an asynchronous exception interrupting a+-- send propagates untouched. sendMessage :: Connection -> Poker.Write -> IO () sendMessage connection write = do- transport <- readIORef (transport connection)- Transport.send transport write+ failed <- readIORef (writeFailed connection)+ unless failed do+ transport <- readIORef (transport connection)+ Transport.send transport write+ `catch` \(_ :: IOException) -> writeIORef (writeFailed connection) True -- | Receive the next /protocol-relevant/ backend message, transparently -- consuming and recording the asynchronous messages the backend may interleave -- at any time: @ParameterStatus@ (updates the parameter map), @NoticeResponse@ -- (collected when notice reporting is on), and @NotificationResponse@ (queued).+--+-- A transport failure propagates as the 'IOException' it is. The handshake+-- classifies it itself; every other flow uses 'receiveMessage'. nextMessage :: Connection -> IO BackendMessage-nextMessage connection = do+nextMessage = nextMessageVia Transport.receiveFrame++-- | 'nextMessage' for the flows that report a lost connection the way libpq+-- does.+--+-- A transport failure - the peer closed the connection (EOF), or the+-- socket failed - is turned into the loss libpq's @pqReadData@ records:+-- 'loseConnection' with libpq's text for the cause, and then the+-- 'ConnectionLost' failure, which the flow boundaries catch and turn into+-- libpq's made-up result. That is the only 'IOException' caught: a protocol+-- decode error, and any asynchronous exception interrupting the wait (which+-- hasql relies on to clean up afterwards), propagate untouched.+receiveMessage :: Connection -> IO BackendMessage+receiveMessage connection = nextMessageVia receive connection+ where+ receive transport =+ Transport.receiveFrame transport `catch` \err -> do+ let message = receiveFailureMessage err+ loseConnection connection message+ throwIO (ConnectionLost message)++-- | The connection was lost while a flow was reading from it: raised by+-- 'receiveMessage' once 'loseConnection' has recorded the loss, carrying the+-- text that was recorded. Not exported from the package.+newtype ConnectionLost = ConnectionLost ByteString+ deriving stock (Show)++instance Exception ConnectionLost++nextMessageVia :: (Transport -> IO (Word8, ByteString)) -> Connection -> IO BackendMessage+nextMessageVia receiveFrame connection = do transport <- readIORef (transport connection)- (typeByte, body) <- Transport.receiveFrame transport+ (typeByte, body) <- receiveFrame transport case decodeBackendMessage typeByte body of Left err -> ioError (userError ("pqi-native: protocol decode error: " <> show err)) Right message -> case message of ParameterStatus key value -> do modifyIORef' (parameters connection) (Map.insert key value)- nextMessage connection+ nextMessageVia receiveFrame connection NoticeResponse fields -> do reporting <- readIORef (noticeReporting connection) when reporting $ do let noticeText = formatErrorFields (Map.fromList fields) unless (ByteString.null noticeText) $ modifyIORef' (notices connection) (noticeText :)- nextMessage connection+ nextMessageVia receiveFrame connection NotificationResponse pid channel payload -> do modifyIORef' (pendingNotifications connection) (Notify channel pid payload :)- nextMessage connection+ nextMessageVia receiveFrame connection other -> pure other -- | Look up an error\/notice field by its single-byte code.@@ -365,12 +427,19 @@ writeIORef (lastError connection) (Just message) writeIORef (connStatus connection) ConnectionBad +-- | Whether the connection goes through a Unix-domain socket. A @hostaddr@+-- always makes it a TCP one, whatever the @host@ looks like: see+-- 'Transport.connect'.+viaUnixSocket :: ConnInfo -> Bool+viaUnixSocket connInfo =+ ByteString.null (hostaddr connInfo) && Transport.isUnixSocketHost (host connInfo)+ -- | Format an initial-connect failure (the socket couldn't even be opened), -- matching libpq's distinct phrasing for a Unix-domain socket, a DNS -- resolution failure, and any other TCP connect failure. connectFailureMessage :: ConnInfo -> IOException -> ByteString connectFailureMessage connInfo err- | Transport.isUnixSocketHost (host connInfo) =+ | viaUnixSocket connInfo = unixSocketFailureMessage connInfo ( ByteString.Char8.pack (ioe_description err)@@ -383,11 +452,17 @@ -- rather than the generic "could not connect to server". | "getAddrInfo" `isInfixOf` ioe_location err = "could not translate host name \""- <> host connInfo+ <> dialled <> "\" to address: " <> ByteString.Char8.pack (ioe_description err) <> "\n" | otherwise = "could not connect to server: " <> ByteString.Char8.pack (show err)+ where+ -- libpq names the address it resolved, which is the @hostaddr@ when there+ -- is one.+ dialled+ | ByteString.null (hostaddr connInfo) = host connInfo+ | otherwise = hostaddr connInfo -- | Format a handshake-time 'IOException' - e.g. the server closing the -- socket mid-rejection, as when shedding load with \"sorry, too many clients@@ -404,7 +479,7 @@ let fmtFields | isConnectionLost err = connectionLostMessage err <> "\n" | otherwise = connectionLostMessage err- if Transport.isUnixSocketHost (host connInfo)+ if viaUnixSocket connInfo then pure (unixSocketFailureMessage connInfo fmtFields) else tcpFailureMessage connection connInfo fmtFields @@ -428,17 +503,68 @@ "server closed the connection unexpectedly\n\tThis probably means the server terminated abnormally\n\tbefore or while processing the request." | otherwise = ByteString.Char8.pack (show err) --- | Record an escaped 'IOException' from the transport - a send or a read,--- at any point in a flow - the way libpq's own internals do: never throw,--- just mark the connection bad with the classified message ('connectionLostMessage'),--- so every subsequent call - starting with the one already in flight - sees--- it via 'Pqi.status'\/'Pqi.errorMessage'.-markConnectionLost :: Connection -> IOException -> IO ByteString-markConnectionLost connection err = do- let message = connectionLostMessage err- setError connection (message <> "\n")- pure message+-- | libpq's text for a failed read, from @pqReadData@ and+-- @pqsecure_raw_read@: the connection lost outright (EOF, or a reset) gets+-- the \"server closed the connection unexpectedly\" sentence, and any other+-- socket error \"could not receive data from server\" followed by the+-- @strerror@ of its errno. Ends with a newline, as every message libpq+-- appends to the connection's error message does.+receiveFailureMessage :: IOException -> ByteString+receiveFailureMessage err+ | isConnectionLost err = connectionLostMessage err <> "\n"+ | otherwise = "could not receive data from server: " <> strerror err <> "\n" +-- | The @strerror@ text of an 'IOException' that came from a failed system+-- call, derived from its errno rather than from its 'show'n form.+strerror :: IOException -> ByteString+strerror err =+ ByteString.Char8.pack case ioe_errno err of+ Just errno -> ioe_description (errnoToIOError "" (Errno errno) Nothing Nothing)+ Nothing -> ioe_description err++-- | The connection has been lost: do what libpq does when its read finds+-- the connection closed or failed (@pqReadData@ and @pqDropConnection@ in+-- @fe-misc.c@\/@fe-connect.c@, then the @IDLE@ that @PQgetResult@ sets).+-- This is the one place that knows what a lost connection looks like.+--+-- * The text is /appended/ to the connection's error message, after+-- whatever was recorded before it (the @FATAL@ the server announced the+-- termination with, say).+-- * The status becomes 'ConnectionBad', and the socket is dropped. 'finish'+-- closes it again, which is harmless.+-- * Nothing is in flight any more, and the commands that were queued in+-- pipeline mode are forgotten, without any 'PipelineAbort' results being+-- made up for them: those are for errors the /server/ reported.+--+-- The pipeline status itself is left as it is, as it is in libpq.+loseConnection :: Connection -> ByteString -> IO ()+loseConnection connection message = do+ modifyIORef' (lastError connection) (Just . (<> message) . fromMaybe "")+ writeIORef (connStatus connection) ConnectionBad+ transport <- readIORef (transport connection)+ Transport.close transport+ writeIORef (asyncPending connection) False+ writeIORef (pendingCommands connection) 0+ writeIORef (pendingSyncs connection) 0+ writeIORef (pendingParseOrigins connection) Seq.empty+ writeIORef (pipelineSeparatorPending connection) False++-- | Run an operation on a usable connection. On any other it is refused as+-- libpq's @PQsendQueryStart@ refuses a send: the given refusal is returned+-- in its place, and the error message becomes \"no connection to the+-- server\", replacing whatever was there unless commands are still queued in+-- a pipeline, in which case it belongs to them and is appended to.+whenUsable :: Connection -> a -> IO a -> IO a+whenUsable connection refusal operation = do+ status <- readIORef (connStatus connection)+ case status of+ ConnectionOk -> operation+ _ -> do+ queued <- readIORef (pendingCommands connection)+ when (queued == 0) $ writeIORef (lastError connection) (Just "")+ modifyIORef' (lastError connection) (Just . (<> "no connection to the server\n") . fromMaybe "")+ pure refusal+ -- | The handshake-failure message ('failWith', inside 'handshake') for a -- Unix-domain socket connection: names the socket path rather than a -- host\/port pair, matching libpq's phrasing.@@ -499,7 +625,7 @@ pure connection Right dfltUser -> do let info = parseConnInfo conninfo dfltUser dfltHost- transportResult <- try @IOException (Transport.connect (host info) (port info))+ transportResult <- try @IOException (Transport.connect (host info) (hostaddr info) (port info)) case transportResult of Left err -> do transport <- Transport.unconnected@@ -531,7 +657,7 @@ reconnect connection = do oldTransport <- readIORef (transport connection) Transport.close oldTransport- newTransport <- Transport.connect (host (info connection)) (port (info connection))+ newTransport <- Transport.connect (host (info connection)) (hostaddr (info connection)) (port (info connection)) writeIORef (transport connection) newTransport writeIORef (parameters connection) Map.empty writeIORef (backendKey connection) Nothing@@ -539,6 +665,7 @@ writeIORef (connStatus connection) ConnectionBad writeIORef (lastError connection) (Just "") writeIORef (passwordNeeded connection) False+ writeIORef (writeFailed connection) False sendMessage connection (startupMessage (startupParams (info connection))) handshake connection @@ -572,6 +699,7 @@ <*> newIORef ErrorsDefault <*> newIORef "" <*> newIORef False+ <*> newIORef False -- | The startup\/authentication state machine, ending at the first -- @ReadyForQuery@ (success) or @ErrorResponse@ (failure).@@ -613,7 +741,7 @@ let fmtFields = formatErrorFields (Map.fromList fields) connInfo = info connection message <-- if Transport.isUnixSocketHost (host connInfo)+ if viaUnixSocket connInfo then pure (unixSocketFailureMessage connInfo fmtFields) else tcpFailureMessage connection connInfo fmtFields setError connection message
src/library/Pqi/Native/Query.hs view
@@ -17,7 +17,7 @@ ) where -import Control.Exception (IOException, catch, mask_)+import Control.Exception (catch, mask_) import qualified Data.Map.Strict as Map import qualified Data.Sequence as Seq import qualified Data.Vector.Mutable as MVector@@ -25,7 +25,7 @@ import Pqi.Native.Connection import Pqi.Native.Prelude import Pqi.Native.Transport.Message-import Pqi.Native.Types (NativeResult (..), formatResultError)+import Pqi.Native.Types (NativeResult (..), formatResultError, lostConnectionResult) import qualified PtrPoker.Write as Poker -- * Message construction@@ -77,66 +77,50 @@ -- | Simple query. Returns the last result, mirroring @PQexec@. exec :: Connection -> ByteString -> IO (Maybe NativeResult)-exec connection sql = withReady connection do- catch- (sendMessage connection (queryMessage sql) >> (lastMaybe <$> collectSimple connection sql))- (fmap Just . connectionLostResult connection sql)+exec connection sql = withReady connection $ guardLoss do+ sendMessage connection (queryMessage sql)+ lastMaybe <$> collectSimple connection sql -- | Parameterized query via the extended protocol. execParams :: Connection -> ByteString -> [Maybe (Word32, ByteString, Format)] -> Format -> IO (Maybe NativeResult) execParams connection sql params resultFormat | tooManyParams params = pure Nothing- | otherwise = withReady connection do- catch- (sendMessage connection (paramsWrite sql params resultFormat) >> (Just <$> collectExtended connection sql))- (fmap Just . connectionLostResult connection sql)+ | otherwise = withReady connection $ guardLoss do+ sendMessage connection (paramsWrite sql params resultFormat)+ Just <$> collectExtended connection sql -- | Prepare a named statement. prepare :: Connection -> ByteString -> ByteString -> Maybe [Word32] -> IO (Maybe NativeResult) prepare connection name sql parameterTypes | maybe False tooManyParams parameterTypes = pure Nothing- | otherwise = withReady connection do- catch- (sendMessage connection (prepareWrite name sql parameterTypes) >> (Just <$> collectExtended connection sql))- (fmap Just . connectionLostResult connection sql)+ | otherwise = withReady connection $ guardLoss do+ sendMessage connection (prepareWrite name sql parameterTypes)+ Just <$> collectExtended connection sql -- | Execute a previously prepared statement. execPrepared :: Connection -> ByteString -> [Maybe (ByteString, Format)] -> Format -> IO (Maybe NativeResult) execPrepared connection name params resultFormat | tooManyParams params = pure Nothing- | otherwise = withReady connection do- catch- (sendMessage connection (preparedWrite name params resultFormat) >> (Just <$> collectExtended connection ""))- (fmap Just . connectionLostResult connection "")+ | otherwise = withReady connection $ guardLoss do+ sendMessage connection (preparedWrite name params resultFormat)+ Just <$> collectExtended connection "" -- * Asynchronous flows -- | Send a write in async mode, tracking pending commands for pipeline abort. ----- A socket death mid-send (e.g. @EPIPE@\/@ECONNRESET@) surfaces here as an--- escaped 'IOException' from 'Transport.send', which 'sendMessage' does not--- catch. Left uncaught, it would blow straight through this function - and--- every caller layered on it ('Pqi.Native.sendQuery' etc., 'Hasql.Comms.Send')--- - as a raw exception instead of the @False@ that @PQsendQuery@ always--- returns for a fatal send. Catching it here and marking the connection bad--- keeps the contract: the caller sees a normal failed send, discoverable via--- 'Pqi.status', exactly as libpq's own internals never throw and always--- record the failure on the connection instead.+-- Sending never throws. A failed write is not reported by the send but by+-- the receive that follows, as in libpq (see 'sendMessage'), so a send on a+-- usable connection returns 'True'. A send on a connection that is not+-- usable returns 'False' and records why (see 'whenUsable'). sendAsync :: Connection -> ByteString -> Poker.Write -> IO Bool-sendAsync connection sql write = do- status <- readIORef (connStatus connection)- case status of- ConnectionOk ->- catch- do- sendMessage connection write- writeIORef (currentQuery connection) sql- writeIORef (asyncPending connection) True- pipeStatus <- readIORef (pipelineStatus connection)- when (pipeStatus /= PipelineOff) $ modifyIORef' (pendingCommands connection) (+ 1)- pure True- (\err -> False <$ markConnectionLost connection err)- _ -> pure False+sendAsync connection sql write = whenUsable connection False do+ sendMessage connection write+ writeIORef (currentQuery connection) sql+ writeIORef (asyncPending connection) True+ pipeStatus <- readIORef (pipelineStatus connection)+ when (pipeStatus /= PipelineOff) $ modifyIORef' (pendingCommands connection) (+ 1)+ pure True -- | Whether the connection is in pipeline mode. inPipeline :: Connection -> IO Bool@@ -232,8 +216,12 @@ -- stall with no timer behind it. Masking keeps each transition atomic. The -- blocking read inside stays interruptible (see 'Transport.receiveFrame'), so -- this costs no abandonability.+--+-- A connection lost while waiting is reported as libpq reports it: this+-- returns the result libpq makes up for it (see 'guardLoss'), and the call+-- after that returns 'Nothing'. getNextResult :: Connection -> IO (Maybe NativeResult)-getNextResult connection = mask_ do+getNextResult connection = mask_ $ guardLoss do pending <- readIORef (asyncPending connection) if not pending then pure Nothing@@ -242,6 +230,7 @@ if sepPending then do writeIORef (pipelineSeparatorPending connection) False+ startNextCommand pure Nothing else do singleRow <- readIORef (singleRowMode connection)@@ -260,6 +249,15 @@ modifyIORef' (pendingCommands connection) (subtract 1) writeIORef (pipelineSeparatorPending connection) True + -- Where libpq moves on to the next queued command (@pqPipelineProcessQueue@)+ -- it clears the error message, which until then belongs to the command+ -- that has just been reported. Nothing is cleared when that was the last+ -- command.+ startNextCommand = do+ commands <- readIORef (pendingCommands connection)+ syncs <- readIORef (pendingSyncs connection)+ when (commands > 0 || syncs > 0) $ writeIORef (lastError connection) (Just "")+ -- Pop and discard this command's 'pendingParseOrigins' entry if it hasn't -- already been consumed by its own 'ParseComplete' (see the field's -- Haddock on 'Connection'). Called at every terminal message besides a@@ -284,11 +282,11 @@ popOriginIfPending pipeStatus builder modifyIORef' (pendingCommands connection) (subtract 1) writeIORef (pipelineSeparatorPending connection) True- pure (Just (NativeResult PipelineAbort [] [] Nothing Map.empty [] ""))+ pure (Just (NativeResult PipelineAbort [] [] Nothing Map.empty [] "" Nothing)) else readAndProcess singleRow builder pipeStatus readAndProcess singleRow builder pipeStatus = do- message <- nextMessage connection+ message <- receiveMessage connection case message of RowDescription fs -> go singleRow builder {accFields = fs, accSawRowDescription = True, accHadResponse = True}@@ -300,7 +298,7 @@ if singleRow then do writeIORef (singleRowFields connection) (accFields builder)- pure (Just (NativeResult SingleTuple (accFields builder) [values] Nothing Map.empty [] ""))+ pure (Just (NativeResult SingleTuple (accFields builder) [values] Nothing Map.empty [] "" Nothing)) else do pushRow (accRows builder) values go singleRow builder@@ -318,7 +316,7 @@ case origin of Just (Just True) -> do finishCommand pipeStatus- pure (Just (NativeResult CommandOk [] [] Nothing Map.empty [] ""))+ pure (Just (NativeResult CommandOk [] [] Nothing Map.empty [] "" Nothing)) Just _ -> go singleRow builder {accHadResponse = True, accOriginPopped = True} Nothing -> go singleRow builder {accHadResponse = True} BindComplete ->@@ -331,7 +329,7 @@ writeIORef (singleRowMode connection) False writeIORef (singleRowFields connection) [] writeIORef (lastError connection) (Just "")- pure (Just (NativeResult TuplesOk (accFields builder) [] (Just tag) Map.empty [] ""))+ pure (Just (NativeResult TuplesOk (accFields builder) [] (Just tag) Map.empty [] "" Nothing)) else do writeIORef (lastError connection) (Just "") popOriginIfPending pipeStatus builder@@ -341,7 +339,7 @@ writeIORef (lastError connection) (Just "") popOriginIfPending pipeStatus builder finishCommand pipeStatus- pure (Just (NativeResult EmptyQuery [] [] Nothing Map.empty [] ""))+ pure (Just (NativeResult EmptyQuery [] [] Nothing Map.empty [] "" Nothing)) ErrorResponse fs -> do let errMap = Map.fromList fs case pipeStatus of@@ -350,16 +348,17 @@ -- mode) but handle defensively. popOriginIfPending pipeStatus builder finishCommand pipeStatus- pure (Just (NativeResult PipelineAbort [] [] Nothing Map.empty [] ""))+ pure (Just (NativeResult PipelineAbort [] [] Nothing Map.empty [] "" Nothing)) PipelineOn -> do writeIORef (pipelineStatus connection) PipelineAborted+ writeIORef (lastError connection) (Just (formatResultError "" errMap)) popOriginIfPending pipeStatus builder finishCommand PipelineOn- pure (Just (NativeResult FatalError [] [] Nothing errMap [] ""))+ pure (Just (NativeResult FatalError [] [] Nothing errMap [] "" Nothing)) PipelineOff -> do sql <- readIORef (currentQuery connection) writeIORef (lastError connection) (Just (formatResultError sql errMap))- pure (Just (NativeResult FatalError [] [] Nothing errMap [] sql))+ pure (Just (NativeResult FatalError [] [] Nothing errMap [] sql Nothing)) PortalSuspended -> Just <$> commandResult builder Nothing ReadyForQuery txState -> do@@ -380,7 +379,8 @@ -- so the next 'getNextResult' reads straight on to the next sync. remaining <- atomicModifyIORef' (pendingSyncs connection) (\n -> (n - 1, n - 1)) when (remaining == 0) $ writeIORef (asyncPending connection) False- pure (Just (NativeResult PipelineSync [] [] Nothing Map.empty [] ""))+ startNextCommand+ pure (Just (NativeResult PipelineSync [] [] Nothing Map.empty [] "" Nothing)) _ -> go singleRow builder -- | Pop the next in-flight pipelined command's origin: @Just (Just True)@ if@@ -400,15 +400,15 @@ -- | Describe a prepared statement. describePrepared :: Connection -> ByteString -> IO (Maybe NativeResult)-describePrepared connection name = withReady connection do+describePrepared connection name = withReady connection $ guardLoss do sendMessage connection (describeStatementMessage name <> syncMessage)- catch (Just <$> collectExtended connection "") (fmap Just . connectionLostResult connection "")+ Just <$> collectExtended connection "" -- | Describe a portal. describePortal :: Connection -> ByteString -> IO (Maybe NativeResult)-describePortal connection name = withReady connection do+describePortal connection name = withReady connection $ guardLoss do sendMessage connection (describePortalMessage name <> syncMessage)- catch (Just <$> collectExtended connection "") (fmap Just . connectionLostResult connection "")+ Just <$> collectExtended connection "" -- * Parameter projections @@ -437,24 +437,20 @@ -- * Result collection -- | Only run a flow on a ready connection; mirror libpq returning no result--- when the connection is not usable.+-- when the connection is not usable, with \"no connection to the server\" as+-- the error message (see 'whenUsable'). withReady :: Connection -> IO (Maybe a) -> IO (Maybe a)-withReady connection action = do- status <- readIORef (connStatus connection)- case status of- ConnectionOk -> action- _ -> pure Nothing+withReady connection = whenUsable connection Nothing --- | Turn a read loop's escaped 'IOException' - e.g. a connection reset while--- a result is still in flight - into a classified 'FatalError' result,--- matching @PQexec@: libpq never throws here, it reports the same--- "server closed the connection unexpectedly" wording it uses for a--- handshake-time loss (see 'connectionLostMessage'), and marks the--- connection bad so the caller's next call sees it too.-connectionLostResult :: Connection -> ByteString -> IOException -> IO NativeResult-connectionLostResult connection sql err = do- message <- markConnectionLost connection err- pure (NativeResult FatalError [] [] Nothing (Map.singleton 0x4d message) [] sql)+-- | Run a flow that reads results off the connection, and if the connection+-- is lost while it does ('receiveMessage' has recorded the loss by then),+-- return the result libpq makes up for it. This is also what @PQexec@ returns+-- when the loss comes after results that were already received, as it is the+-- last result that counts. Everything else, an asynchronous exception+-- included, propagates.+guardLoss :: IO (Maybe NativeResult) -> IO (Maybe NativeResult)+guardLoss flow =+ flow `catch` \(ConnectionLost message) -> pure (Just (lostConnectionResult message)) -- | A growable row buffer, replacing a cons-list-plus-reverse accumulation -- with amortized-O(1) appends into a doubling mutable vector. A statement's@@ -531,7 +527,7 @@ go builder0 [] where go builder acc = do- message <- nextMessage connection+ message <- receiveMessage connection case message of RowDescription fs -> go builder {accFields = fs, accSawRowDescription = True} acc DataRow values -> do@@ -545,18 +541,18 @@ EmptyQueryResponse -> do writeIORef (lastError connection) (Just "") builder' <- newBuilder- go builder' (NativeResult EmptyQuery [] [] Nothing Map.empty [] "" : acc)+ go builder' (NativeResult EmptyQuery [] [] Nothing Map.empty [] "" Nothing : acc) ErrorResponse fs -> do let errMap = Map.fromList fs writeIORef (lastError connection) (Just (formatResultError sql errMap)) builder' <- newBuilder- go builder' (NativeResult FatalError [] [] Nothing errMap [] sql : acc)+ go builder' (NativeResult FatalError [] [] Nothing errMap [] sql Nothing : acc) CopyInResponse _ formats -> let fields = map copyField formats- in pure (reverse (NativeResult CopyIn fields [] Nothing Map.empty [] "" : acc))+ in pure (reverse (NativeResult CopyIn fields [] Nothing Map.empty [] "" Nothing : acc)) CopyOutResponse _ formats -> let fields = map copyField formats- in pure (reverse (NativeResult CopyOut fields [] Nothing Map.empty [] "" : acc))+ in pure (reverse (NativeResult CopyOut fields [] Nothing Map.empty [] "" Nothing : acc)) ReadyForQuery txState -> do writeIORef (txStatus connection) txState pure (reverse acc)@@ -569,7 +565,7 @@ go builder0 Nothing where go builder finished = do- message <- nextMessage connection+ message <- receiveMessage connection case message of RowDescription fs -> go builder {accFields = fs, accSawRowDescription = True} finished ParameterDescription oids -> go builder {accParamOids = oids} finished@@ -594,12 +590,12 @@ EmptyQueryResponse -> do writeIORef (lastError connection) (Just "") builder' <- newBuilder- go builder' (Just (NativeResult EmptyQuery [] [] Nothing Map.empty [] ""))+ go builder' (Just (NativeResult EmptyQuery [] [] Nothing Map.empty [] "" Nothing)) ErrorResponse fs -> do let errMap = Map.fromList fs writeIORef (lastError connection) (Just (formatResultError sql errMap)) builder' <- newBuilder- go builder' (Just (NativeResult FatalError [] [] Nothing errMap [] sql))+ go builder' (Just (NativeResult FatalError [] [] Nothing errMap [] sql Nothing)) ReadyForQuery txState -> do writeIORef (txStatus connection) txState case finished of@@ -621,6 +617,7 @@ Map.empty (accParamOids builder) ""+ Nothing -- | A result with no command completion (a @Describe@\/@Parse@-only flow): -- 'CommandOk', carrying any column descriptions and parameter OIDs.@@ -636,6 +633,7 @@ Map.empty (accParamOids builder) ""+ Nothing lastMaybe :: [a] -> Maybe a lastMaybe = foldl (\_ x -> Just x) Nothing
src/library/Pqi/Native/Types.hs view
@@ -4,6 +4,7 @@ module Pqi.Native.Types ( NativeResult (..), NativeCancel (..),+ lostConnectionResult, mkResult, mkCancel, formatErrorFields,@@ -43,10 +44,24 @@ -- position context in 'resultErrorMessage', matching libpq's behaviour). -- Empty for non-error results and for async results where the query is -- unavailable.- queryText :: ByteString+ queryText :: ByteString,+ -- | A message that is not derived from wire error fields, returned+ -- verbatim by 'resultErrorMessage' in place of the formatted fields. Only+ -- the result libpq makes up for a lost connection has one (see+ -- 'lostConnectionResult'); every result built from server messages has+ -- 'Nothing'.+ literalErrorMessage :: Maybe ByteString } deriving stock (Eq, Show) +-- | The result libpq makes up when the connection is lost while a flow is+-- waiting for the server (@pqPrepareAsyncResult@ in @fe-exec.c@): a+-- 'FatalError' with no error fields at all, hence no SQLSTATE, whose+-- message is the given text verbatim.+lostConnectionResult :: ByteString -> NativeResult+lostConnectionResult message =+ NativeResult FatalError [] [] Nothing Map.empty [] "" (Just message)+ -- | A standalone cancellation handle for the native adapter. -- -- Carries references to connection state so the cancel implementation can@@ -63,6 +78,8 @@ -- completions are in, @pendingCommands@ is 0 and the server is idle. data NativeCancel = NativeCancel { host :: ByteString,+ -- | Empty when the connection was not given a @hostaddr@.+ hostaddr :: ByteString, port :: Int, pid :: Int32, secret :: Int32,@@ -146,7 +163,12 @@ mkResult result = Result { resultStatus = pure (status result),- resultErrorMessage = pure (Just (formatResultError (queryText result) (errorFields result))),+ resultErrorMessage =+ pure+ $ Just+ $ fromMaybe+ (formatResultError (queryText result) (errorFields result))+ (literalErrorMessage result), resultErrorField = \field -> pure (Map.lookup (fieldCodeByte field) (errorFields result)), unsafeFreeResult = pure (), ntuples = pure (fromIntegral (length (rows result))),@@ -185,7 +207,7 @@ if not pending then pure (Right ()) else do- transport <- Transport.connect (host nc) (port nc)+ transport <- Transport.connect (host nc) (hostaddr nc) (port nc) Transport.send transport (cancelRequest (pid nc) (secret nc)) -- Read until EOF to ensure the server has processed the cancel request -- before we close the connection. This matches libpq's PQcancel behavior
src/transport/Pqi/Native/Transport.hs view
@@ -37,13 +37,21 @@ readBuffer :: IORef ByteString } --- | Open a connection to the given host and port. Mirroring libpq: if the--- host looks like an absolute path (starts with @\/@ - see 'isUnixSocketHost'),--- it names a Unix-domain socket /directory/ rather than a TCP host, and the--- connection is made to the socket file @'unixSocketPath' host port@ within--- it. Otherwise a TCP connection is made, resolving the host via DNS.-connect :: ByteString -> Int -> IO Transport-connect host port+-- | Open a connection to the given host, numeric host address and port, in+-- the order libpq considers them:+--+-- 1. A non-empty @hostaddr@ is connected to over TCP, and the @host@ is not+-- consulted at all: it is not resolved, and a path-like one does not+-- select a Unix-domain socket. libpq keeps it only for authentication and+-- for naming the server in messages.+-- 2. Otherwise, if the host looks like an absolute path (starts with @\/@ -+-- see 'isUnixSocketHost'), it names a Unix-domain socket /directory/+-- rather than a TCP host, and the connection is made to the socket file+-- @'unixSocketPath' host port@ within it.+-- 3. Otherwise a TCP connection is made, resolving the host via DNS.+connect :: ByteString -> ByteString -> Int -> IO Transport+connect host hostaddr port+ | not (ByteString.null hostaddr) = connectTcp hostaddr port | isUnixSocketHost host = connectUnix (unixSocketPath host port) | otherwise = connectTcp host port