diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# v0.1.0.0 + +## Breaking + +- Migrate to `pqi` 0.1's record-of-functions redesign and switch to exporting the `adapter` value instead of the `Connection` type. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -21,8 +21,8 @@ ## Status -All classes are implemented. Verified against the `postgresql-libpq` reference -via the conformance differential suite. +The full `Pqi.Connection` capability record is implemented. Verified against +the `postgresql-libpq` reference via the conformance differential suite. Authentication: **trust**, **MD5**, and **SCRAM-SHA-256** are implemented. SCRAM is verified against a password-auth PostgreSQL 17 container (which defaults to diff --git a/pqi-native.cabal b/pqi-native.cabal --- a/pqi-native.cabal +++ b/pqi-native.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: pqi-native -version: 0.0.1.1 +version: 0.1.0.0 category: Database, PostgreSQL synopsis: Native (pure-Haskell) adapter for pqi description: @@ -8,8 +8,8 @@ protocol directly, with no dependency on the C @libpq@ library. The transport layer (framing, message serialization\/deserialization, socket I\/O) is isolated in an internal @transport@ sub-library built on @ptr-poker@ and - @ptr-peeker@; the public library is an abstraction over it that implements the - @pqi@ capability classes. + @ptr-peeker@; the public library is an abstraction over it that builds the + @pqi@ @Adapter@. homepage: https://github.com/nikita-volkov/pqi-native bug-reports: https://github.com/nikita-volkov/pqi-native/issues @@ -131,6 +131,7 @@ Pqi.Native.Prelude Pqi.Native.Query Pqi.Native.Types + Pqi.Native.UnescapeBytea build-depends: base >=4.11 && <5, @@ -138,8 +139,9 @@ bytestring >=0.10 && <0.13, containers >=0.6 && <0.9, crypton >=0.34 && <2, - pqi ^>=0.0, + pqi ^>=0.1, pqi-native:transport, + ptr-peeker ^>=0.2, ptr-poker ^>=0.1.3, unix >=2.7 && <2.9, @@ -151,5 +153,5 @@ build-depends: base >=4.11 && <5, hspec >=2.11 && <2.12, - pqi-conformance ^>=0.0, + pqi-conformance ^>=0.1, pqi-native, diff --git a/src/comms/Pqi/Native/Comms.hs b/src/comms/Pqi/Native/Comms.hs --- a/src/comms/Pqi/Native/Comms.hs +++ b/src/comms/Pqi/Native/Comms.hs @@ -29,9 +29,9 @@ import Data.ByteString (ByteString) import Data.Int (Int16, Int32) import Data.Word (Word32, Word8) +import Prelude import qualified PtrPeeker as Peeker import qualified PtrPoker.Write as Poker -import Prelude -- | A semantic decoding failure (as opposed to a \"need more bytes\" framing -- shortfall, which the transport handles separately). diff --git a/src/library/Pqi/Native.hs b/src/library/Pqi/Native.hs --- a/src/library/Pqi/Native.hs +++ b/src/library/Pqi/Native.hs @@ -1,20 +1,19 @@ -{-# LANGUAGE UndecidableInstances #-} -{-# OPTIONS_GHC -Wno-orphans #-} - -- | The native (pure-Haskell) @pqi@ adapter. -- --- 'Connection' speaks the PostgreSQL wire protocol directly. --- Provides the 'IsConnection' instance. +-- 'adapter' bundles the three functions that produce a 'Pqi.Connection' +-- whose fields are closures over the underlying native 'Connection.Connection' +-- (which speaks the PostgreSQL wire protocol directly). 'Pqi.Result' and +-- 'Pqi.Cancel' values are constructed the same way, in "Pqi.Native.Types". module Pqi.Native - ( Connection, + ( adapter, ) where import qualified Data.ByteString as ByteString import qualified Data.ByteString.Char8 as ByteString.Char8 import qualified Data.Map.Strict as Map -import Pqi -import Pqi.Native.Connection (Connection (..)) +import qualified Pqi +import Pqi.Native.Connection (Connection) import qualified Pqi.Native.Connection as Connection import qualified Pqi.Native.LargeObject as LargeObject import Pqi.Native.Prelude @@ -28,158 +27,167 @@ flushMessage, syncMessage, ) -import Pqi.Native.Types (NativeCancel (..), NativeResult (..)) +import Pqi.Native.Types (NativeCancel (..), NativeResult (..), mkCancel, mkResult) +import qualified Pqi.Native.UnescapeBytea as UnescapeBytea import System.Posix.Types (Fd) -instance IsConnection Connection where - type ResultOf Connection = NativeResult - type CancelOf Connection = NativeCancel - - connectdb = Connection.establish - connectStart = Connection.establish - connectPoll _ = pure PollingOk - newNullConnection = Connection.nullConnection - isNullConnection connection = connection.isNull - finish connection = readIORef connection.transport >>= Transport.close - reset = Connection.reconnect - resetStart connection = Connection.reconnect connection $> True - resetPoll _ = pure PollingOk - db connection = pure (Just connection.info.database) - user connection = pure (Just connection.info.user) - pass connection = pure (Just connection.info.password) - host connection = pure (Just connection.info.host) - port connection = pure (Just (ByteString.Char8.pack (show connection.info.port))) - options _ = pure (Just "") - status connection = readIORef connection.connStatus - transactionStatus connection = transactionStatusOf <$> readIORef connection.txStatus - parameterStatus connection name = Map.lookup name <$> readIORef connection.parameters - protocolVersion _ = pure 3 - serverVersion connection = - maybe 0 parseServerVersion . Map.lookup "server_version" <$> readIORef connection.parameters - errorMessage connection = readIORef connection.lastError - socket connection = do - transport <- readIORef connection.transport - fd <- Transport.socketFd transport - pure (Just (fromIntegral fd :: Fd)) - backendPID connection = maybe 0 fst <$> readIORef connection.backendKey - connectionNeedsPassword _ = pure False - connectionUsedPassword connection = pure (not (ByteString.null connection.info.password)) - - exec connection sql = Query.exec connection sql - execParams connection sql params resultFormat = - Query.execParams connection sql params resultFormat - prepare connection name sql parameterTypes = - Query.prepare connection name sql parameterTypes - execPrepared connection name params resultFormat = - Query.execPrepared connection name params resultFormat - describePrepared connection name = Query.describePrepared connection name - describePortal connection name = Query.describePortal connection name - - escapeStringConn _ value - | isValidUtf8 value = - pure (Just (ByteString.intercalate "''" (ByteString.split 0x27 value))) - | otherwise = pure Nothing - escapeByteaConn _ value = - pure (Just ("\\x" <> hexEncode value)) - escapeIdentifier _ value - | isValidUtf8 value = - pure (Just ("\"" <> ByteString.intercalate "\"\"" (ByteString.split 0x22 value) <> "\"")) - | otherwise = pure Nothing - - sendQuery = Query.sendQuery - sendQueryParams = Query.sendQueryParams - sendPrepare = Query.sendPrepare - sendQueryPrepared = Query.sendQueryPrepared - sendDescribePrepared = Query.sendDescribePrepared - sendDescribePortal = Query.sendDescribePortal - getResult connection = Query.getNextResult connection - consumeInput _ = pure True - isBusy _ = pure False - setnonblocking connection flag = writeIORef connection.nonblocking flag $> True - isnonblocking connection = readIORef connection.nonblocking - setSingleRowMode connection = do - pending <- readIORef connection.asyncPending - if pending - then writeIORef connection.singleRowMode True $> True - else pure False - flush _ = pure FlushOk - - pipelineStatus connection = readIORef connection.pipelineStatus - enterPipelineMode connection = writeIORef connection.pipelineStatus PipelineOn $> True - exitPipelineMode connection = do - pending <- readIORef connection.asyncPending - if pending - then pure False - else writeIORef connection.pipelineStatus PipelineOff $> True - pipelineSync connection = do - Connection.sendMessage connection syncMessage - modifyIORef' connection.pendingSyncs (+ 1) - writeIORef connection.asyncPending True - pure True - sendFlushRequest connection = Connection.sendMessage connection flushMessage $> True - - getCancel connection = do - key <- readIORef connection.backendKey - pure - $ fmap - ( \(pid, secret) -> - NativeCancel - { host = connection.info.host, - port = connection.info.port, - pid, - secret, - asyncPendingRef = connection.asyncPending, - pipelineStatusRef = connection.pipelineStatus, - pendingCommandsRef = connection.pendingCommands - } - ) - key - - notifies connection = popFirst connection.pendingNotifications - disableNoticeReporting connection = writeIORef connection.noticeReporting False - enableNoticeReporting connection = writeIORef connection.noticeReporting True - getNotice connection = popFirst connection.notices +-- | The native adapter. +adapter :: Pqi.Adapter +adapter = + Pqi.Adapter + { Pqi.name = "pqi-native", + Pqi.connectdb = \conninfo -> mkConnection <$> Connection.establish conninfo, + Pqi.connectStart = \conninfo -> mkConnection <$> Connection.establish conninfo, + Pqi.newNullConnection = mkConnection <$> Connection.nullConnection, + Pqi.unescapeBytea = \input -> pure (Just (UnescapeBytea.unescapeBytea input)) + } - putCopyData connection payload = Connection.sendMessage connection (copyDataMessage payload) $> CopyInOk - putCopyEnd connection reason = do - Connection.sendMessage connection (maybe copyDoneMessage copyFailMessage reason) - writeIORef connection.asyncPending True - pure CopyInOk - getCopyData connection _nonBlocking = do - message <- Connection.nextMessage connection - case message of - CopyData payload -> pure (CopyOutRow payload) - CopyDone -> do +-- | Build a 'Pqi.Connection' whose fields close over the given native +-- connection. +mkConnection :: Connection -> Pqi.Connection +mkConnection connection = + Pqi.Connection + { Pqi.connectPoll = pure Pqi.PollingOk, + Pqi.isNullConnection = connection.isNull, + Pqi.finish = readIORef connection.transport >>= Transport.close, + Pqi.reset = Connection.reconnect connection, + Pqi.resetStart = Connection.reconnect connection $> True, + Pqi.resetPoll = pure Pqi.PollingOk, + Pqi.db = pure (Just connection.info.database), + Pqi.user = pure (Just connection.info.user), + Pqi.pass = pure (Just connection.info.password), + Pqi.host = pure (Just connection.info.host), + Pqi.port = pure (Just (ByteString.Char8.pack (show connection.info.port))), + Pqi.options = pure (Just ""), + Pqi.status = readIORef connection.connStatus, + Pqi.transactionStatus = transactionStatusOf <$> readIORef connection.txStatus, + Pqi.parameterStatus = \name -> Map.lookup name <$> readIORef connection.parameters, + Pqi.protocolVersion = pure 3, + Pqi.serverVersion = + maybe 0 parseServerVersion . Map.lookup "server_version" <$> readIORef connection.parameters, + Pqi.errorMessage = readIORef connection.lastError, + Pqi.socket = do + transport <- readIORef connection.transport + fd <- Transport.socketFd transport + pure (Just (fromIntegral fd :: Fd)), + Pqi.backendPID = maybe 0 fst <$> readIORef connection.backendKey, + Pqi.connectionNeedsPassword = pure False, + Pqi.connectionUsedPassword = pure (not (ByteString.null connection.info.password)), + Pqi.exec = \sql -> fmap mkResult <$> Query.exec connection sql, + Pqi.execParams = \sql params resultFormat -> + fmap mkResult <$> Query.execParams connection sql params resultFormat, + Pqi.prepare = \name sql parameterTypes -> + fmap mkResult <$> Query.prepare connection name sql parameterTypes, + Pqi.execPrepared = \name params resultFormat -> + fmap mkResult <$> Query.execPrepared connection name params resultFormat, + Pqi.describePrepared = \name -> fmap mkResult <$> Query.describePrepared connection name, + Pqi.describePortal = \name -> fmap mkResult <$> Query.describePortal connection name, + Pqi.escapeStringConn = \value -> + if isValidUtf8 value + then pure (Just (ByteString.intercalate "''" (ByteString.split 0x27 value))) + else pure Nothing, + Pqi.escapeByteaConn = \value -> pure (Just ("\\x" <> hexEncode value)), + Pqi.escapeIdentifier = \value -> + if isValidUtf8 value + then pure (Just ("\"" <> ByteString.intercalate "\"\"" (ByteString.split 0x22 value) <> "\"")) + else pure Nothing, + Pqi.sendQuery = Query.sendQuery connection, + Pqi.sendQueryParams = \sql params resultFormat -> Query.sendQueryParams connection sql params resultFormat, + Pqi.sendPrepare = \name sql parameterTypes -> Query.sendPrepare connection name sql parameterTypes, + Pqi.sendQueryPrepared = \name params resultFormat -> Query.sendQueryPrepared connection name params resultFormat, + Pqi.sendDescribePrepared = Query.sendDescribePrepared connection, + Pqi.sendDescribePortal = Query.sendDescribePortal connection, + Pqi.getResult = fmap mkResult <$> Query.getNextResult connection, + Pqi.consumeInput = pure True, + Pqi.isBusy = pure False, + Pqi.setnonblocking = \flag -> writeIORef connection.nonblocking flag $> True, + Pqi.isnonblocking = readIORef connection.nonblocking, + Pqi.setSingleRowMode = do + pending <- readIORef connection.asyncPending + if pending + then writeIORef connection.singleRowMode True $> True + else pure False, + Pqi.flush = pure Pqi.FlushOk, + Pqi.pipelineStatus = readIORef connection.pipelineStatus, + Pqi.enterPipelineMode = writeIORef connection.pipelineStatus Pqi.PipelineOn $> True, + Pqi.exitPipelineMode = do + pending <- readIORef connection.asyncPending + if pending + then pure False + else writeIORef connection.pipelineStatus Pqi.PipelineOff $> True, + Pqi.pipelineSync = do + Connection.sendMessage connection syncMessage + modifyIORef' connection.pendingSyncs (+ 1) writeIORef connection.asyncPending True - pure CopyOutDone - CommandComplete _ -> drainToReady connection $> CopyOutDone - ErrorResponse _ -> drainToReady connection $> CopyOutError - ReadyForQuery txState -> writeIORef connection.txStatus txState $> CopyOutDone - _ -> getCopyData connection _nonBlocking - - loCreat = LargeObject.loCreat - loCreate = LargeObject.loCreate - loImport = LargeObject.loImport - loImportWithOid = LargeObject.loImportWithOid - loExport = LargeObject.loExport - loOpen = LargeObject.loOpen - loWrite = LargeObject.loWrite - loRead = LargeObject.loRead - loSeek = LargeObject.loSeek - loTell = LargeObject.loTell - loTruncate = LargeObject.loTruncate - loClose = LargeObject.loClose - loUnlink = LargeObject.loUnlink + pure True, + Pqi.sendFlushRequest = Connection.sendMessage connection flushMessage $> True, + Pqi.getCancel = do + key <- readIORef connection.backendKey + pure + $ fmap + ( \(pid, secret) -> + mkCancel + NativeCancel + { host = connection.info.host, + port = connection.info.port, + pid, + secret, + asyncPendingRef = connection.asyncPending, + pipelineStatusRef = connection.pipelineStatus, + pendingCommandsRef = connection.pendingCommands + } + ) + key, + Pqi.notifies = popFirst connection.pendingNotifications, + Pqi.disableNoticeReporting = writeIORef connection.noticeReporting False, + Pqi.enableNoticeReporting = writeIORef connection.noticeReporting True, + Pqi.getNotice = popFirst connection.notices, + Pqi.putCopyData = \payload -> Connection.sendMessage connection (copyDataMessage payload) $> Pqi.CopyInOk, + Pqi.putCopyEnd = \reason -> do + Connection.sendMessage connection (maybe copyDoneMessage copyFailMessage reason) + writeIORef connection.asyncPending True + pure Pqi.CopyInOk, + Pqi.getCopyData = getCopyData connection, + Pqi.loCreat = LargeObject.loCreat connection, + Pqi.loCreate = LargeObject.loCreate connection, + Pqi.loImport = LargeObject.loImport connection, + Pqi.loImportWithOid = LargeObject.loImportWithOid connection, + Pqi.loExport = LargeObject.loExport connection, + Pqi.loOpen = LargeObject.loOpen connection, + Pqi.loWrite = LargeObject.loWrite connection, + Pqi.loRead = LargeObject.loRead connection, + Pqi.loSeek = LargeObject.loSeek connection, + Pqi.loTell = LargeObject.loTell connection, + Pqi.loTruncate = LargeObject.loTruncate connection, + Pqi.loClose = LargeObject.loClose connection, + Pqi.loUnlink = LargeObject.loUnlink connection, + Pqi.clientEncoding = + fromMaybe "SQL_ASCII" . Map.lookup "client_encoding" <$> readIORef connection.parameters, + Pqi.setClientEncoding = \encoding -> do + result <- Query.exec connection ("SET client_encoding TO '" <> encoding <> "'") + pure (maybe False (\value -> value.status /= Pqi.FatalError) result), + Pqi.setErrorVerbosity = \verbosity -> do + previous <- readIORef connection.errorVerbosity + writeIORef connection.errorVerbosity verbosity + pure previous + } - clientEncoding connection = - fromMaybe "SQL_ASCII" . Map.lookup "client_encoding" <$> readIORef connection.parameters - setClientEncoding connection encoding = do - result <- Query.exec connection ("SET client_encoding TO '" <> encoding <> "'") - pure (maybe False (\value -> value.status /= FatalError) result) - setErrorVerbosity connection verbosity = do - previous <- readIORef connection.errorVerbosity - writeIORef connection.errorVerbosity verbosity - pure previous +-- | Receive data on a @COPY TO STDOUT@ connection, as 'Pqi.getCopyData'. The +-- native adapter has no non-blocking transport, so the @Bool@ argument is +-- ignored; it always reads until a full chunk (or the end of the copy) is +-- available. +getCopyData :: Connection -> Bool -> IO Pqi.CopyOutResult +getCopyData connection nonBlocking = do + message <- Connection.nextMessage connection + case message of + CopyData payload -> pure (Pqi.CopyOutRow payload) + CopyDone -> do + writeIORef connection.asyncPending True + pure Pqi.CopyOutDone + CommandComplete _ -> drainToReady connection $> Pqi.CopyOutDone + ErrorResponse _ -> drainToReady connection $> Pqi.CopyOutError + ReadyForQuery txState -> writeIORef connection.txStatus txState $> Pqi.CopyOutDone + _ -> getCopyData connection nonBlocking -- | Read messages until @ReadyForQuery@, recording the transaction status. drainToReady :: Connection -> IO () @@ -196,12 +204,12 @@ [] -> ([], Nothing) oldest : rest -> (reverse rest, Just oldest) -transactionStatusOf :: Word8 -> TransactionStatus +transactionStatusOf :: Word8 -> Pqi.TransactionStatus transactionStatusOf = \case - 0x49 -> TransIdle -- 'I' - 0x54 -> TransInTrans -- 'T' - 0x45 -> TransInError -- 'E' - _ -> TransUnknown + 0x49 -> Pqi.TransIdle -- 'I' + 0x54 -> Pqi.TransInTrans -- 'T' + 0x45 -> Pqi.TransInError -- 'E' + _ -> Pqi.TransUnknown -- | Parse the @server_version@ parameter into libpq's @MMmmpp@ integer form -- (e.g. @\"17.2\"@ -> @170002@, @\"9.6.3\"@ -> @90603@). diff --git a/src/library/Pqi/Native/Types.hs b/src/library/Pqi/Native/Types.hs --- a/src/library/Pqi/Native/Types.hs +++ b/src/library/Pqi/Native/Types.hs @@ -1,8 +1,11 @@ -- | Internal types for the native adapter, separated from 'Connection.hs' to --- avoid orphan-instance warnings for 'IsResult' and 'IsCancel'. +-- keep 'mkResult'\/'mkCancel' (and the pure formatting helpers they build on) +-- next to the data they close over. module Pqi.Native.Types ( NativeResult (..), NativeCancel (..), + mkResult, + mkCancel, formatErrorFields, formatResultError, ) @@ -15,12 +18,12 @@ import Data.List (findIndex) import qualified Data.Map.Strict as Map import Pqi - ( ExecStatus (..), + ( Cancel (..), + ExecStatus (..), FieldCode (..), Format (..), - IsCancel (..), - IsResult (..), PipelineStatus (..), + Result (..), ) import Pqi.Native.Prelude import qualified Pqi.Native.Transport as Transport @@ -137,54 +140,62 @@ Just (n, _) | n > 0 -> Just n _ -> Nothing -instance IsResult NativeResult where - resultStatus result = pure result.status - resultErrorMessage result = pure (Just (formatResultError result.queryText result.errorFields)) - resultErrorField result field = pure (Map.lookup (fieldCodeByte field) result.errorFields) - unsafeFreeResult _ = pure () - ntuples result = pure (fromIntegral (length result.rows)) - nfields result = pure (fromIntegral (length result.fields)) - fname result column = pure $ do - fd <- atMay result.fields column - if ByteString.null fd.name then Nothing else Just fd.name - fnumber result name = - pure (fromIntegral <$> findIndex (\field -> field.name == folded) result.fields) - where - folded = foldIdentifier name - ftable result column = pure (maybe 0 (.tableOid) (atMay result.fields column)) - ftablecol result column = - pure (maybe 0 (\field -> fromIntegral (field.columnAttributeNumber :: Int16)) (atMay result.fields column)) - fformat result column = - pure (maybe Text (\field -> formatOf field.formatCode) (atMay result.fields column)) - ftype result column = pure (maybe 0 (.typeOid) (atMay result.fields column)) - fmod result column = pure (maybe 0 (\field -> fromIntegral (field.typeModifier :: Int32)) (atMay result.fields column)) - fsize result column = pure (maybe 0 (\field -> fromIntegral (field.typeSize :: Int16)) (atMay result.fields column)) - getvalue result row column = pure (join (cellAt result row column)) - getvalue' result row column = pure (join (cellAt result row column)) - getisnull result row column = pure (maybe True isNothing (cellAt result row column)) - getlength result row column = - pure (maybe 0 (maybe 0 ByteString.length) (cellAt result row column)) - nparams result = pure (fromIntegral (length result.paramOids)) - paramtype result index = pure (fromMaybe 0 (atMay result.paramOids index)) - cmdStatus result = pure (Just (fromMaybe "" result.commandTag)) - cmdTuples result = pure (Just (maybe "" affectedRows result.commandTag)) +-- | Build a 'Result' whose fields close over the given fully materialized +-- 'NativeResult'. +mkResult :: NativeResult -> Result +mkResult result = + Result + { resultStatus = pure result.status, + resultErrorMessage = pure (Just (formatResultError result.queryText result.errorFields)), + resultErrorField = \field -> pure (Map.lookup (fieldCodeByte field) result.errorFields), + unsafeFreeResult = pure (), + ntuples = pure (fromIntegral (length result.rows)), + nfields = pure (fromIntegral (length result.fields)), + fname = \column -> pure $ do + fd <- atMay result.fields column + if ByteString.null fd.name then Nothing else Just fd.name, + fnumber = \name -> + pure (fromIntegral <$> findIndex (\field -> field.name == foldIdentifier name) result.fields), + ftable = \column -> pure (maybe 0 (.tableOid) (atMay result.fields column)), + ftablecol = \column -> + pure (maybe 0 (\field -> fromIntegral (field.columnAttributeNumber :: Int16)) (atMay result.fields column)), + fformat = \column -> + pure (maybe Text (\field -> formatOf field.formatCode) (atMay result.fields column)), + ftype = \column -> pure (maybe 0 (.typeOid) (atMay result.fields column)), + fmod = \column -> + pure (maybe 0 (\field -> fromIntegral (field.typeModifier :: Int32)) (atMay result.fields column)), + fsize = \column -> + pure (maybe 0 (\field -> fromIntegral (field.typeSize :: Int16)) (atMay result.fields column)), + getvalue = \row column -> pure (join (cellAt result row column)), + getvalue' = \row column -> pure (join (cellAt result row column)), + getisnull = \row column -> pure (maybe True isNothing (cellAt result row column)), + getlength = \row column -> pure (maybe 0 (maybe 0 ByteString.length) (cellAt result row column)), + nparams = pure (fromIntegral (length result.paramOids)), + paramtype = \index -> pure (fromMaybe 0 (atMay result.paramOids index)), + cmdStatus = pure (Just (fromMaybe "" result.commandTag)), + cmdTuples = pure (Just (maybe "" affectedRows result.commandTag)) + } -instance IsCancel NativeCancel where - cancel nc = do - pending <- readIORef nc.asyncPendingRef - if not pending - then pure (Right ()) - else do - transport <- Transport.connect nc.host nc.port - Transport.send transport (cancelRequest nc.pid nc.secret) - -- Read until EOF to ensure the server has processed the cancel request - -- before we close the connection. This matches libpq's PQcancel behavior - -- and prevents the cancel signal from racing with the next query. - _ <- try @IOException (Transport.readUntilClosed transport) - Transport.close transport - pure (Right ()) +-- | Build a 'Cancel' whose field closes over the given 'NativeCancel'. +mkCancel :: NativeCancel -> Cancel +mkCancel nc = + Cancel + { cancel = do + pending <- readIORef nc.asyncPendingRef + if not pending + then pure (Right ()) + else do + transport <- Transport.connect nc.host nc.port + Transport.send transport (cancelRequest nc.pid nc.secret) + -- Read until EOF to ensure the server has processed the cancel request + -- before we close the connection. This matches libpq's PQcancel behavior + -- and prevents the cancel signal from racing with the next query. + _ <- try @IOException (Transport.readUntilClosed transport) + Transport.close transport + pure (Right ()) + } --- * Helpers for the 'IsResult' instance +-- * Helpers for 'mkResult' atMay :: [a] -> Int32 -> Maybe a atMay xs i diff --git a/src/library/Pqi/Native/UnescapeBytea.hs b/src/library/Pqi/Native/UnescapeBytea.hs new file mode 100644 --- /dev/null +++ b/src/library/Pqi/Native/UnescapeBytea.hs @@ -0,0 +1,158 @@ +-- | Pure implementation of @bytea@ unescaping. See 'unescapeBytea'. +module Pqi.Native.UnescapeBytea + ( unescapeBytea, + ) +where + +import Data.ByteString (ByteString) +import Data.Either (fromRight) +import Data.Word (Word8) +import Prelude +import PtrPeeker (Variable, fixed, hasMore, runVariableOnByteString, unsignedInt1) +import PtrPoker.Write (Write) +import qualified PtrPoker.Write as Write + +-- | Convert the textual representation of a @bytea@ value, as produced by +-- the server, back into raw bytes. Both the modern @\\x@ hex format (lowercase +-- @x@ only) and the legacy escape format are accepted. +-- +-- Malformed input is tolerated exactly the way @PQunescapeBytea@ tolerates it: +-- in hex format, characters that are not hex digits (including whitespace) are +-- silently skipped, and a hex digit whose pair character is invalid is +-- dropped; in escape format, an invalid escape simply drops the backslash, and +-- an octal escape must start with @0@..@3@. Input is treated as a C string: +-- the first NUL byte terminates processing. +unescapeBytea :: ByteString -> ByteString +unescapeBytea input = + Write.toByteString + $ fromRight mempty + $ runVariableOnByteString decoder input + +-- Inline NUL truncation and \x prefix detection so no intermediate ByteStrings +-- are allocated before dispatching to the format-specific decoder. +decoder :: Variable Write +decoder = do + more <- hasMore + if not more + then return mempty + else do + b0 <- fixed unsignedInt1 + case b0 of + 0x00 -> return mempty -- NUL: C-string terminator + 0x5c -> do + -- backslash: probe for the \x hex-format prefix + more2 <- hasMore + if not more2 + then return mempty -- single trailing backslash + else do + b1 <- fixed unsignedInt1 + if b1 == 0x78 -- lowercase 'x': enter hex mode + then hexDecoder + else afterBackslash b1 -- escape mode; b1 follows the consumed '\' + _ -> (Write.word8 b0 <>) <$> escapeDecoder + +-- | Hex-format decoder. Skips non-hex bytes (matching @PQunescapeBytea@), +-- pairs hex nibbles, and stops at a NUL byte (C-string terminator). +hexDecoder :: Variable Write +hexDecoder = do + more <- hasMore + if not more + then return mempty + else do + a <- fixed unsignedInt1 + if a == 0x00 + then return mempty -- NUL: stop + else case hexValue a of + Nothing -> hexDecoder -- skip non-hex byte + Just hi -> do + more2 <- hasMore + if not more2 + then return mempty -- drop unpaired nibble + else do + b <- fixed unsignedInt1 + if b == 0x00 + then return mempty -- NUL: stop, drop unpaired nibble + else case hexValue b of + Nothing -> hexDecoder -- skip b, look for next pair + Just lo -> (Write.word8 (hi * 16 + lo) <>) <$> hexDecoder + where + hexValue :: Word8 -> Maybe Word8 + hexValue w + | w >= 0x30 && w <= 0x39 = Just (w - 0x30) + | w >= 0x61 && w <= 0x66 = Just (w - 0x57) + | w >= 0x41 && w <= 0x46 = Just (w - 0x37) + | otherwise = Nothing + +-- | Escape-format decoder. Processes bytes as escape sequences and stops at +-- a NUL byte (C-string terminator). +escapeDecoder :: Variable Write +escapeDecoder = do + more <- hasMore + if not more + then return mempty + else do + b <- fixed unsignedInt1 + case b of + 0x00 -> return mempty + 0x5c -> handleEscapeBackslash + _ -> (Write.word8 b <>) <$> escapeDecoder + +-- | Handle the bytes that follow a consumed backslash in escape format. +-- Exported so the top-level dispatcher can reuse it after consuming the +-- @\\x@ prefix check. +afterBackslash :: Word8 -> Variable Write +afterBackslash next = case next of + 0x00 -> return mempty + 0x5c -> (Write.word8 0x5c <>) <$> escapeDecoder + _ -> octalOrLiteralDecoder next + +handleEscapeBackslash :: Variable Write +handleEscapeBackslash = do + more <- hasMore + if not more + then return mempty -- trailing backslash: drop it + else do + next <- fixed unsignedInt1 + afterBackslash next + +-- | Try to decode a 3-digit octal starting with @a@ (already consumed). +-- Falls back to emitting @a@ literally and re-routing the consumed lookahead +-- byte(s) through 'afterEscape', reproducing @PQunescapeBytea@'s backtracking. +octalOrLiteralDecoder :: Word8 -> Variable Write +octalOrLiteralDecoder a + | isFirstOctal a = do + more <- hasMore + if not more + then return (Write.word8 a) + else do + b <- fixed unsignedInt1 + if not (isOctal b) + then (Write.word8 a <>) <$> afterEscape b -- b isn't octal: emit a, re-route b + else do + more2 <- hasMore + if not more2 + then return (Write.word8 a <> Write.word8 b) -- only two digits: both literal + else do + c <- fixed unsignedInt1 + if isOctal c + then (Write.word8 (octal a b c) <>) <$> escapeDecoder + else (\x -> Write.word8 a <> Write.word8 b <> x) <$> afterEscape c + | otherwise = (Write.word8 a <>) <$> escapeDecoder + where + -- \| Route an already-consumed byte back through the escape-format main loop. + -- Used when a consumed lookahead byte must be re-processed after a failed + -- octal-triple attempt. + afterEscape :: Word8 -> Variable Write + afterEscape b = case b of + 0x00 -> return mempty + 0x5c -> handleEscapeBackslash + _ -> (Write.word8 b <>) <$> escapeDecoder + + isFirstOctal :: Word8 -> Bool + isFirstOctal w = w >= 0x30 && w <= 0x33 + + isOctal :: Word8 -> Bool + isOctal w = w >= 0x30 && w <= 0x37 + + octal :: Word8 -> Word8 -> Word8 -> Word8 + octal a b c = (a - 0x30) * 64 + (b - 0x30) * 8 + (c - 0x30) diff --git a/src/test/Spec.hs b/src/test/Spec.hs --- a/src/test/Spec.hs +++ b/src/test/Spec.hs @@ -4,11 +4,10 @@ -- SCRAM) against the FFI reference, and tears the container down again. module Main (main) where -import Data.Proxy (Proxy (..)) import Pqi.Conformance (specs) -import Pqi.Native (Connection) -import Test.Hspec +import qualified Pqi.Native import Prelude +import Test.Hspec main :: IO () -main = hspec (specs (Proxy @Connection)) +main = hspec (specs Pqi.Native.adapter)