mariadb-haskell-0.1.0.0: src/Database/MySQL/Connection.hs
{-# LANGUAGE ScopedTypeVariables #-}
{-|
Module : Database.MySQL.Connection
Description : Internal MariaDB transport and connection lifecycle
Copyright : (c) Winterland, 2016
License : BSD
Maintainer : Anton Antich <anton.antich@gmail.com>
Stability : experimental
Portability : macOS and Linux tested
Applications should use the abstract t'MySQLConn' exported by
"Database.MySQL.Base". This exposed internal module also supports deterministic
transport fixtures; modifying its records bypasses the public safety contract.
-}
module Database.MySQL.Connection where
import Control.Exception hiding (handle)
import Control.Monad
import qualified Crypto.ECC.Edwards25519 as Edwards
import qualified Crypto.PubKey.Ed25519 as Ed25519
import qualified Crypto.KDF.PBKDF2 as PBKDF2
import Crypto.Error (throwCryptoError)
import Crypto.Random (getRandomBytes)
import qualified Crypto.Hash as Crypto
import Crypto.Hash.Algorithms (SHA1(..))
import qualified Crypto.Number.Serialize as Serialize
import qualified Crypto.PubKey.RSA as RSA
import qualified Crypto.PubKey.RSA.OAEP as OAEP
import qualified Data.ASN1.BinaryEncoding as ASN1B
import qualified Data.ASN1.BitArray as ASN1BA
import qualified Data.ASN1.Encoding as ASN1E
import qualified Data.ASN1.Types as ASN1
import qualified Data.Binary as Binary
import qualified Data.Binary.Put as Binary
import Data.Bits
import qualified Data.ByteArray as BA
import Data.ByteString (ByteString)
import qualified Data.ByteString as B
import qualified Data.ByteString.Lazy as L
import Data.IORef (IORef, newIORef, readIORef, atomicModifyIORef',
writeIORef)
import qualified Data.PEM as PEM
import Data.Typeable
import Data.Word
import Database.MySQL.Protocol.Auth
import Database.MySQL.Authentication
import qualified Database.MySQL.Authentication.GSSAPI as GSSAPI
import Database.MySQL.Protocol.Command
import Database.MySQL.Protocol.Packet
import Network.Socket (HostName, PortNumber)
import qualified Network.Socket as N
import System.IO.Streams (InputStream)
import qualified System.IO.Streams as Stream
import qualified System.IO.Streams.TCP as TCP
import qualified Data.Connection as Conn
--------------------------------------------------------------------------------
import Control.Concurrent (ThreadId, myThreadId)
import System.Timeout (timeout)
import qualified Data.Binary.Parser as P
import qualified Data.ByteString.Char8 as BC
import qualified Data.Text.Encoding as TE
import qualified Data.Text as T
import qualified Data.Vector as V
import Data.Int (Int64)
import Data.Maybe (isJust)
import qualified Data.IntMap.Strict as IM
import Database.MySQL.Session
import Database.MySQL.Protocol.ColumnDef
import Database.MySQL.Protocol.MySQLValue
import Database.MySQL.Protocol.Progress
import Database.MySQL.Protocol.Compression
import Database.MySQL.Query
-- | Per-connection resource bounds. Limits are validated before opening a
-- transport; none can be zero. Message limits include continuation fragments.
data ConnectionOptions = ConnectionOptions
{ maximumMessageBytes :: !Word64 -- ^ Maximum inbound/outbound logical message.
, maximumMetadataBytes :: !Word64 -- ^ Per-metadata-block and OK tracking budget.
, maximumColumns :: !Int -- ^ Maximum fields in a result or statement metadata block.
, cleanupTimeoutMicros :: !Int -- ^ One deadline for draining and rollback.
, connectTimeoutMicros :: !Int -- ^ Whole establishment deadline: DNS, TLS, authentication and setup.
, maximumAuthIterations :: !Int -- ^ Maximum server-requested PARSEC PBKDF2 work.
, enableSessionTracking :: !Bool -- ^ Negotiate tracking and OK terminators together.
, enableMetadataCaching :: !Bool -- ^ Negotiate MariaDB prepared metadata omission.
, enableExtendedMetadata :: !Bool -- ^ Retain native type/format attributes.
, maximumPreparedStatements :: !Int -- ^ Maximum registered server statements.
, maximumBulkRows :: !Int -- ^ Rows per bulk command; batches are never silently split.
, maximumLongDataBytes :: !Word64 -- ^ Total streamed parameter bytes per execution.
, maximumCursorFetchRows :: !Word32 -- ^ Maximum requested rows in one server cursor fetch.
, enableProgressReporting :: !Bool -- ^ Negotiate reports; install callbacks separately.
, enableCompression :: !Bool -- ^ Negotiate zlib transport framing; off by default.
, compressionThresholdBytes :: !Int -- ^ Try compression for outbound frames at least this large.
} deriving (Eq, Show)
-- | 64 MiB messages, 16 MiB metadata, 65,535 columns, one-second cleanup,
-- ten-second establishment, at most 2^20 PARSEC iterations and session tracking.
defaultConnectionOptions :: ConnectionOptions
defaultConnectionOptions = ConnectionOptions
{ maximumMessageBytes = 64 * 1024 * 1024
, maximumMetadataBytes = 16 * 1024 * 1024
, maximumColumns = 65535
, cleanupTimeoutMicros = 1000000
, connectTimeoutMicros = 10000000
, maximumAuthIterations = 1048576
, enableSessionTracking = True
, enableMetadataCaching = True
, enableExtendedMetadata = True
, maximumPreparedStatements = 1024
, maximumBulkRows = 10000
, maximumLongDataBytes = 64 * 1024 * 1024
, maximumCursorFetchRows = 65536
, enableProgressReporting = False
, enableCompression = False
, compressionThresholdBytes = 1024
}
-- A cached fact must not pin a large result buffer or grow without bound.
-- Startup snapshots and subsequent tracker deltas share this byte limit.
maximumRenderingValueBytes :: Int
maximumRenderingValueBytes = 4096
validateOptions :: ConnectionOptions -> IO ()
validateOptions options = unless
(maximumMessageBytes options > 0 && maximumMessageBytes options <= fromIntegral (maxBound :: Word32)
&& maximumMessageBytes options <= fromIntegral (maxBound :: Int)
&& maximumMetadataBytes options > 0 && maximumColumns options > 0 && cleanupTimeoutMicros options > 0
&& connectTimeoutMicros options > 0 && maximumAuthIterations options >= 1024
&& maximumAuthIterations options <= 1073741824
&& maximumPreparedStatements options > 0 && maximumBulkRows options > 0
&& maximumLongDataBytes options > 0 && maximumCursorFetchRows options > 0
&& compressionThresholdBytes options >= 0 && compressionThresholdBytes options <= 65536)
(throwIO (ProtocolError "invalid connection resource options"))
-- | Observable lifecycle. Only 'Ready' admits a new command. 'Broken' and
-- 'Closed' are terminal; an unfinished stream retains its creating thread.
data ConnectionStatus = Ready | Executing | Streaming | Replicating | Broken | Closed deriving (Eq, Show)
-- Completion must be distinct from "bytes were sent": a concurrent close after
-- a terminal COMMIT reply does not make a known server error uncertain.
data ExchangeProgress = BeforeSend | WireInFlight | ReplyComplete deriving (Eq,Show)
data Runtime = Runtime
{ runtimeStatus :: !ConnectionStatus
, runtimeGeneration :: !Word64
, runtimeOwner :: !(Maybe ThreadId)
, transactionOwner :: !(Maybe ThreadId)
} deriving Show
-- | Bounded diagnostics, deliberately excluding SQL, parameters and credentials.
data ConnectionFailure = ConnectionFailure
{ failurePhase :: !ConnectionStatus -- ^ Phase in which the connection was retired.
, commitOutcomeUncertain :: !Bool -- ^ COMMIT may have reached MariaDB without a reply.
, secondaryCleanupFailure :: !Bool -- ^ Cleanup also failed after a primary exception.
} deriving (Eq, Show)
-- | Local ownership, resource, session-context and transaction failures.
-- 'CommitOutcomeUnknown' retains the original synchronous exception; inspect
-- connection diagnostics for uncertain commits interrupted by cancellation.
data ConnectionError = ConnectionUnavailable ConnectionStatus | ConcurrentConnectionUse | StaleResultStream
| NestedTransactionUnsupported | SessionContextUnavailable | ProtocolError String
| CommitOutcomeUnknown SomeException | CleanupDeadlineExceeded
deriving (Typeable, Show)
instance Exception ConnectionError
-- | A MariaDB session with one command/stream owner at a time. Obtain it through
-- the connection constructors and close it with 'close'. The public Base module
-- keeps this representation abstract; direct internal record access bypasses
-- the ownership, resource and session-state contracts.
data MySQLConn = MySQLConn
{ mysqlRead :: !(InputStream Packet)
, mysqlWrite :: !(Packet -> IO ())
, mysqlCloseSocket :: !(IO ())
, connectionOptions :: !ConnectionOptions
, connectionRuntime :: !(IORef Runtime)
, connectionFailureRef :: !(IORef (Maybe ConnectionFailure))
, connectionServerStatus :: !(IORef (Maybe Word16))
, connectionRenderContext :: !(IORef (Maybe RenderContext))
, connectionCapabilities :: !(IORef Word64)
, connectionSequence :: !(IORef FramingState)
, connectionExchangeProgress :: !(IORef ExchangeProgress)
, connectionReleased :: !(IORef Bool)
, connectionDrain :: !(IORef (Maybe (IO ())))
, connectionServerPacketLimit :: !(IORef Word64)
, connectionQueryEffect :: !(IORef (Maybe Bool))
, connectionOpaqueStatements :: !(IORef [Word32])
, connectionStatements :: !(IORef (Maybe StatementRegistry))
, connectionStatementSerial :: !(IORef Word64)
, connectionActiveStatement :: !(IORef (Maybe StmtID))
, connectionResultIndex :: !(IORef Int)
, connectionSession :: !(IORef SessionCache)
, connectionInitialSession :: !(IORef (Maybe InitialSession))
, connectionProgressCallback :: !(IORef (Maybe (Progress -> IO ())))
, connectionCallbackActive :: !(IORef Bool)
, connectionCompressionStart :: !(IO ())
, connectionCompressionReset :: !(IO ())
, connectionCompressionAllowReplicationResets :: !(IO ())
}
-- A decoder is compiled per metadata shape, not per row/execution. Metadata
-- cannot be evicted independently: MariaDB may omit it in the next response.
-- The aggregate byte charge shares the connection's metadata resource budget.
data CachedColumns = CachedColumns
{ cachedDetails :: ![ColumnMetadata]
, cachedFields :: ![ColumnDef]
, cachedColumnCount :: !Int
-- Intentionally lazy: ordinary text results must not construct binary
-- field parsers. The first binary read forces and shares this decoder.
, cachedBinaryDecoder :: Binary.Get [MySQLValue]
, cachedMetadataBytes :: !Word64
}
data StatementState = StatementState
{ statementSerial :: !Word64
, statementArity :: !Int
, statementResults :: !(IM.IntMap CachedColumns)
, statementMetadataBytes :: !Word64 -- ^ Sum of this statement's cached blocks.
}
-- | Keep byte charges beside the entries they describe, under one atomic update.
-- A metadata refresh must not scan unrelated statements: disabling omission or
-- changing a parameter type can put this operation on every execution's path.
data StatementRegistry = StatementRegistry
{ registryStatements :: !(IM.IntMap StatementState)
, registryMetadataBytes :: !Word64
}
-- | Read lifecycle diagnostics without claiming a command or performing I/O.
connectionStatus :: MySQLConn -> IO ConnectionStatus
connectionStatus = fmap runtimeStatus . readIORef . connectionRuntime
-- | Return retirement diagnostics, or 'Nothing' if no failure has been recorded.
connectionFailure :: MySQLConn -> IO (Maybe ConnectionFailure)
connectionFailure = readIORef . connectionFailureRef
-- Shared construction is also used by deterministic transport fixtures.
newConnection :: ConnectionOptions -> InputStream Packet -> (Packet -> IO ()) -> IO () -> IO MySQLConn
newConnection options input output abort = newConnectionUsing options makeReader output abort
where
makeReader framing = Stream.makeInputStream $ do
packet <- Stream.read input
case packet of
Nothing -> pure Nothing
Just value -> advanceFraming (maximumMessageBytes options) framing (pLen value) (pSeqN value) >> pure (Just value)
-- Live sockets validate framing in the byte decoder itself, without another
-- InputStream wrapper and another read/Maybe allocation for each physical row.
newByteConnection :: ConnectionOptions -> InputStream ByteString -> (Packet -> IO ()) -> IO () -> IO MySQLConn
newByteConnection options input output abort = newConnectionUsing options (\framing -> decodeFramedInput (maximumMessageBytes options) framing input) output abort
-- Raw transports share one optional byte layer; ordinary connections retain
-- their existing reader/writer path without compression checks per packet.
newTransportConnection :: ConnectionOptions -> InputStream ByteString -> (L.ByteString -> IO ()) -> IO () -> IO MySQLConn
newTransportConnection = buildTransportConnection (\conn _ -> pure conn)
-- | Construct two views of one transport. Authentication rejects oversized
-- physical headers before reading their bodies. After authentication, the
-- ordinary view uses the caller's full message budget without an extra mutable
-- phase check on every query packet. Both views share framing and unread bytes.
newAuthenticatingTransportConnection :: ConnectionOptions -> InputStream ByteString -> (L.ByteString -> IO ()) -> IO () -> IO (MySQLConn, MySQLConn)
newAuthenticatingTransportConnection = buildTransportConnection $ \conn input -> do
packets <- decodeFramedInput (authenticationWireLimit (connectionOptions conn))
(connectionSequence conn) input
pure (conn {mysqlRead=packets}, conn)
-- Select compression once; both connection views must read its same byte stream.
buildTransportConnection :: (MySQLConn -> InputStream ByteString -> IO a) -> ConnectionOptions -> InputStream ByteString -> (L.ByteString -> IO ()) -> IO () -> IO a
buildTransportConnection finish options input output abort
| not (enableCompression options) = do
conn <- newByteConnection options input (output . Binary.encode) abort
finish conn input
| otherwise = do
transport <- newCompressionTransport (maximumMessageBytes options + 4)
(compressionThresholdBytes options) input output
conn <- newByteConnection options (compressionInput transport)
(compressionSend transport . Binary.encode) abort
finish conn {connectionCompressionStart = compressionStart transport,
connectionCompressionReset = compressionReset transport,
connectionCompressionAllowReplicationResets = compressionAllowReplicationResets transport}
(compressionInput transport)
-- | One MiB of plugin data plus the largest accepted AuthSwitch envelope
-- (marker, 128-byte plugin name and NUL). Caller budgets may be smaller. This
-- phase limit also covers the greeting and terminal authentication messages.
authenticationWireLimit :: ConnectionOptions -> Word64
authenticationWireLimit options = min (maximumMessageBytes options) (1048576 + 130)
newConnectionUsing :: ConnectionOptions -> (IORef FramingState -> IO (InputStream Packet)) -> (Packet -> IO ()) -> IO () -> IO MySQLConn
newConnectionUsing options makeReader output abort = do
validateOptions options
state <- newIORef (Runtime Ready 0 Nothing Nothing)
failure <- newIORef Nothing
status <- newIORef Nothing
context <- newIORef Nothing
caps <- newIORef 0
framing <- newIORef (FramingState 0 0)
started <- newIORef BeforeSend
released <- newIORef False
drain <- newIORef Nothing
serverLimit <- newIORef (maximumMessageBytes options)
effect <- newIORef Nothing
opaque <- newIORef []
statements <- newIORef (Just (StatementRegistry IM.empty 0))
serial <- newIORef 0
active <- newIORef Nothing
resultIndex <- newIORef 0
observations <- newIORef (emptySessionCache 0)
initial <- newIORef Nothing
progressCallback <- newIORef Nothing
callbackActive <- newIORef False
input <- makeReader framing
let checkedOutput packet = advanceFraming (maximumMessageBytes options) framing (pLen packet) (pSeqN packet) >> output packet
pure MySQLConn
{ mysqlRead = input
, mysqlWrite = checkedOutput
, mysqlCloseSocket = abort
, connectionOptions = options
, connectionRuntime = state
, connectionFailureRef = failure
, connectionServerStatus = status
, connectionRenderContext = context
, connectionCapabilities = caps
, connectionSequence = framing
, connectionExchangeProgress = started
, connectionReleased = released
, connectionDrain = drain
, connectionServerPacketLimit = serverLimit
, connectionQueryEffect = effect
, connectionOpaqueStatements = opaque
, connectionStatements = statements
, connectionStatementSerial = serial
, connectionActiveStatement = active
, connectionResultIndex = resultIndex
, connectionSession = observations
, connectionInitialSession = initial
, connectionProgressCallback = progressCallback
, connectionCallbackActive = callbackActive
, connectionCompressionStart = pure ()
, connectionCompressionReset = pure ()
, connectionCompressionAllowReplicationResets = pure ()
}
-- Retain only restoration facts, never ConnectInfo or authentication material.
data InitialSession = InitialSession
{ initialSchema :: !ByteString
, initialCharset :: !ByteString
, initialCollation :: !ByteString
, initialVersion :: !Int
, initialRole :: !(Maybe ByteString)
}
data FramingState = FramingState {-# UNPACK #-} !Word16 {-# UNPACK #-} !Word64
-- The 257 possible empty states are constants, shared by every connection.
-- Ordinary short packets need no new framing-state allocation per row.
emptyFramingStates :: V.Vector FramingState
emptyFramingStates=V.generate 257 (\n -> FramingState (fromIntegral n) 0)
{-# NOINLINE emptyFramingStates #-}
-- 256 permits physical packet-number resets in replication and compressed
-- exchanges. Logical continuations still use the assembler's modulo-256 check.
setExchangeSequence :: MySQLConn -> Word16 -> IO ()
setExchangeSequence conn sequenceN = do
-- MariaDB net_flush synchronizes ordinary pkt_nr to compress_pkt_nr.
-- Thus ordinary packet numbers may jump at compressed flush boundaries;
-- Connector/C also validates only compressed-frame sequence in this mode.
-- Logical continuation ordering remains checked by readPacketWithLimit.
effective <- if sequenceN == 0 && enableCompression (connectionOptions conn)
then do
caps <- readIORef (connectionCapabilities conn)
pure (if testBit caps 5 then 256 else sequenceN)
else pure sequenceN
writeIORef (connectionSequence conn) (FramingState effective 0)
advanceFraming :: Word64 -> IORef FramingState -> Int64 -> Word8 -> IO ()
advanceFraming limit ref lengthBytes sequenceN = do
FramingState expected accumulated <- readIORef ref
unless (expected==256 || expected==fromIntegral sequenceN)
(throwIO (ProtocolError ("unexpected packet sequence: expected " ++ show expected ++ ", received " ++ show sequenceN)))
when (lengthBytes<0 || lengthBytes>16777215) (throwIO (ProtocolError "invalid physical packet length"))
let bytes=fromIntegral lengthBytes
when (accumulated>limit || bytes>limit-accumulated) (throwIO (ProtocolError "logical packet exceeds client limit"))
let next=if expected==256 then 256 else (expected+1) .&. 255
continued=if lengthBytes==16777215 then accumulated+bytes else 0
writeIORef' ref (if continued==0 then V.unsafeIndex emptyFramingStates (fromIntegral next) else FramingState next continued)
{-# INLINE advanceFraming #-}
-- Claim the finalizer once so close, cancellation and a failed stream cannot
-- double-close a transport. Transport finalizers must release resources even
-- if their graceful TLS/socket shutdown fails.
releaseTransport :: MySQLConn -> IO ()
releaseTransport conn = do
already <- atomicModifyIORef' (connectionReleased conn) (\x -> (True,x))
unless already (mysqlCloseSocket conn)
abortConnection :: MySQLConn -> Bool -> IO ()
abortConnection conn uncertain = mask_ $ do
clearStatements conn
writeIORef (connectionProgressCallback conn) Nothing
invalidateSession conn
phase <- atomicModifyIORef' (connectionRuntime conn) $ \rt ->
(rt { runtimeStatus=if runtimeStatus rt == Closed then Closed else Broken, runtimeOwner=Nothing },runtimeStatus rt)
atomicModifyIORef' (connectionFailureRef conn) $ \prior ->
let failure=case prior of
Nothing -> ConnectionFailure phase uncertain False
Just previous -> previous {commitOutcomeUncertain=commitOutcomeUncertain previous || uncertain}
in (Just failure,())
writeIORef (connectionRenderContext conn) Nothing
writeIORef (connectionDrain conn) Nothing
-- An ordinary close error must not replace the original protocol failure.
-- A new cancellation still belongs to the caller and must propagate.
releaseTransport conn `catch` (\(err :: SomeException) -> when (isCancellation err) (throwIO err))
-- The atomic claim rejects misuse instead of queueing commands. The surrounding
-- mask in withCommand closes the cancellation gap between claiming and setup.
claimCommand :: MySQLConn -> IO ()
claimCommand conn = do
tid <- myThreadId
result <- atomicModifyIORef' (connectionRuntime conn) $ \rt ->
let unavailable err = (rt,Left err) in
if maybe False (/=tid) (transactionOwner rt) then unavailable ConcurrentConnectionUse
else case runtimeStatus rt of
Ready -> (rt {runtimeStatus=Executing,runtimeGeneration=runtimeGeneration rt+1,runtimeOwner=Just tid},Right ())
Streaming -> unavailable (ConnectionUnavailable Streaming)
Executing -> unavailable ConcurrentConnectionUse
other -> unavailable (ConnectionUnavailable other)
case result of
Left (ConnectionUnavailable Streaming) -> throwIO UnconsumedResultSet
Left err -> throwIO err
Right () -> do
writeIORef (connectionExchangeProgress conn) BeforeSend
writeIORef (connectionQueryEffect conn) Nothing
-- Finalize every exchange-local reference before publishing Ready. Masking
-- protects this thread from cancellation; it does not exclude another thread
-- from claiming Ready immediately after the atomic runtime update.
finishResponse :: MySQLConn -> IO ()
finishResponse conn = mask_ $ do
writeIORef (connectionDrain conn) Nothing
caps <- readIORef (connectionCapabilities conn)
started <- readIORef (connectionExchangeProgress conn)
writeIORef (connectionExchangeProgress conn) ReplyComplete
effect <- readIORef (connectionQueryEffect conn)
when (started/=BeforeSend && (caps .&. 0x800000 == 0 || effect==Just True)) $ do
writeIORef (connectionRenderContext conn) Nothing
-- Preserve the latest actual envelope for observers, while declining
-- to present cached facts as current after an opaque effect.
invalidateSessionFacts conn
atomicModifyIORef' (connectionRuntime conn) $ \rt ->
(if runtimeStatus rt `elem` [Executing,Streaming] then rt {runtimeStatus=Ready,runtimeOwner=Nothing} else rt,())
-- A pre-send validation failure releases the claim. Once writing may have
-- started, any incomplete reply retires the connection rather than guessing
-- whether the next packet belongs to this command or the next one.
withCommand :: MySQLConn -> IO a -> IO a
withCommand conn action = mask $ \restore -> do
claimCommand conn
generation <- runtimeGeneration <$> readIORef (connectionRuntime conn)
result <- restore action `catch` \(err :: SomeException) -> do
runtime <- readIORef (connectionRuntime conn)
started <- readIORef (connectionExchangeProgress conn)
-- The action may have completed its reply before throwing. A successor
-- can already own Ready, so finalizers may touch only their generation.
when (runtimeGeneration runtime == generation && runtimeStatus runtime `elem` [Executing,Streaming,Replicating]) $
if started==WireInFlight then abortConnection conn False else finishResponse conn
throwIO err
runtime <- readIORef (connectionRuntime conn)
when (runtimeGeneration runtime == generation && runtimeStatus runtime == Executing) (finishResponse conn)
pure result
-- A completed stream must never read from a later exchange. Check generation
-- before ownership so stale handles fail without consuming a single packet.
streamGuard :: MySQLConn -> Word64 -> IO a -> IO a
streamGuard conn generation action = do
when (enableProgressReporting (connectionOptions conn)) $ do
callbackActive <- readIORef (connectionCallbackActive conn)
when callbackActive (throwIO ConcurrentConnectionUse)
rt <- readIORef (connectionRuntime conn)
tid <- myThreadId
unless (runtimeGeneration rt == generation) (throwIO StaleResultStream)
unless (runtimeStatus rt `elem` [Streaming,Replicating]) (throwIO (ConnectionUnavailable (runtimeStatus rt)))
unless (runtimeOwner rt == Just tid) (throwIO ConcurrentConnectionUse)
action `catch` \(err :: SomeException) -> do
current <- readIORef (connectionRuntime conn)
when (runtimeGeneration current == generation && runtimeStatus current `elem` [Streaming,Replicating])
(abortConnection conn False)
throwIO err
markStreaming :: MySQLConn -> IO Word64
markStreaming conn = markResponseStream conn Streaming
-- A concurrent close is terminal, including the gap between metadata receipt
-- and stream construction. Never overwrite Closed/Broken with a stream phase.
markResponseStream :: MySQLConn -> ConnectionStatus -> IO Word64
markResponseStream conn phase = do
result <- atomicModifyIORef' (connectionRuntime conn) $ \rt ->
if runtimeStatus rt == Executing
then (rt {runtimeStatus=phase}, Right (runtimeGeneration rt))
else (rt, Left (ConnectionUnavailable (runtimeStatus rt)))
either throwIO pure result
beginTransactionLease :: MySQLConn -> IO ()
beginTransactionLease conn = do
tid <- myThreadId
result <- atomicModifyIORef' (connectionRuntime conn) $ \rt ->
if isJust (transactionOwner rt) then (rt,False) else (rt {transactionOwner=Just tid},True)
unless result (throwIO NestedTransactionUnsupported)
endTransactionLease :: MySQLConn -> IO ()
endTransactionLease conn = atomicModifyIORef' (connectionRuntime conn) (\rt -> (rt {transactionOwner=Nothing},()))
-- | Endpoint, credentials and initial one-byte protocol collation ID. The
-- 'Show' instance always redacts the password, including an empty password.
data ConnectInfo = ConnectInfo
{ ciHost :: HostName
-- ^ TCP address; TLS certificate identity is passed separately.
, ciPort :: PortNumber
-- ^ TCP port (default 3306); ignored by Unix-socket constructors.
, ciDatabase :: ByteString
-- ^ Initial schema bytes; choose an explicit schema for pooled reset.
, ciUser :: ByteString
-- ^ MariaDB account name, interpreted according to server authentication.
, ciPassword :: ByteString
-- ^ Password bytes, excluded from the diagnostic 'Show' representation.
, ciCharset :: Word8
-- ^ Handshake collation ID; defaults to utf8mb4_unicode_ci (224).
}
-- Preserve ordinary record-style diagnostics without ever evaluating or
-- rendering credential bytes. Do not implement this by replacing a substring
-- in derived output: quoting/escaping or overlapping fields could leak secrets.
instance Show ConnectInfo where
showsPrec precedence ci = showParen (precedence > 10) $
showString "ConnectInfo {ciHost = " . shows (ciHost ci)
. showString ", ciPort = " . shows (ciPort ci)
. showString ", ciDatabase = " . shows (ciDatabase ci)
. showString ", ciUser = " . shows (ciUser ci)
. showString ", ciPassword = <redacted>, ciCharset = "
. shows (ciCharset ci) . showChar '}'
-- | Local TCP defaults using full Unicode (utf8mb4). MB4 is a compatibility alias.
defaultConnectInfo, defaultConnectInfoMB4 :: ConnectInfo
defaultConnectInfo = ConnectInfo "127.0.0.1" 3306 "" "root" "" utf8mb4_unicode_ci
defaultConnectInfoMB4 = defaultConnectInfo
utf8_general_ci, utf8mb4_unicode_ci :: Word8
utf8_general_ci=33
utf8mb4_unicode_ci=224
bUFSIZE :: Int
bUFSIZE=16384
-- | Open TCP with 'defaultConnectionOptions' and establish a rendering snapshot.
connect :: ConnectInfo -> IO MySQLConn
connect = connectWithOptions defaultConnectionOptions
-- | Open TCP with explicit bounds. Constructor failure closes the transport.
connectWithOptions :: ConnectionOptions -> ConnectInfo -> IO MySQLConn
connectWithOptions opts ci = snd <$> connectDetailWithOptions opts ci
-- | Open TCP and also return the server greeting.
connectDetail :: ConnectInfo -> IO (Greeting,MySQLConn)
connectDetail = connectDetailWithOptions defaultConnectionOptions
-- | Explicit-options variant of 'connectDetail'.
connectDetailWithOptions :: ConnectionOptions -> ConnectInfo -> IO (Greeting,MySQLConn)
connectDetailWithOptions opts = connectDetailWithAuthentication opts defaultAuthenticationOptions
-- | Open TCP with explicit external authentication. PAM still requires TLS or
-- a Unix socket; TCP permits principal-pinned mutually authenticated GSSAPI.
-- The authentication settings are discarded after establishment.
connectWithAuthentication :: ConnectionOptions -> AuthenticationOptions -> ConnectInfo -> IO MySQLConn
connectWithAuthentication opts auth ci = snd <$> connectDetailWithAuthentication opts auth ci
-- | Greeting-returning variant of 'connectWithAuthentication'.
connectDetailWithAuthentication :: ConnectionOptions -> AuthenticationOptions -> ConnectInfo -> IO (Greeting,MySQLConn)
connectDetailWithAuthentication opts auth ci = withConnectDeadline opts $ do
validateOptions opts
bracketOnError (TCP.connectSocketTimeout (connectTimeoutMicros opts) (ciHost ci) (ciPort ci) >>= TCP.socketToConnectionAdaptive bUFSIZE) Conn.close $ \raw -> do
(authConn,conn) <- newAuthenticatingTransportConnection opts (Conn.source raw) (Conn.send raw) (Conn.close raw)
greet <- authenticationDiagnostic (readGreeting authConn)
let caps=negotiatedCapabilities opts greet
writeIORef (connectionCapabilities conn) caps
mysqlWrite conn (encodeToPacket 1 (mkAuthForAuthentication auth opts caps ci greet))
q <- authenticationDiagnostic (readConnPacket authConn)
completeAuthWithOptions auth authConn greet False (ciPassword ci) q
when (testBit caps 5) (connectionCompressionStart conn)
role <- initializeSession conn
rememberInitialSession conn ci role
pure (greet,conn)
-- | Open a Unix-domain socket with the same session and ownership contracts as TCP.
connectUnixSocket :: FilePath -> ConnectInfo -> IO MySQLConn
connectUnixSocket = connectUnixSocketWithOptions defaultConnectionOptions
-- | Open a Unix-domain socket with explicit resource bounds.
connectUnixSocketWithOptions :: ConnectionOptions -> FilePath -> ConnectInfo -> IO MySQLConn
connectUnixSocketWithOptions opts path ci = snd <$> connectUnixSocketDetailWithOptions opts path ci
-- | Open a Unix-domain socket and also return the server greeting.
connectUnixSocketDetail :: FilePath -> ConnectInfo -> IO (Greeting,MySQLConn)
connectUnixSocketDetail = connectUnixSocketDetailWithOptions defaultConnectionOptions
-- | Explicit-options variant of 'connectUnixSocketDetail'.
connectUnixSocketDetailWithOptions :: ConnectionOptions -> FilePath -> ConnectInfo -> IO (Greeting,MySQLConn)
connectUnixSocketDetailWithOptions opts = connectUnixSocketDetailWithAuthentication opts defaultAuthenticationOptions
-- | Open a Unix socket with explicit PAM/GSSAPI settings. The local socket is
-- a permitted cleartext authentication transport; its filesystem access policy
-- remains the application's responsibility.
connectUnixSocketWithAuthentication :: ConnectionOptions -> AuthenticationOptions -> FilePath -> ConnectInfo -> IO MySQLConn
connectUnixSocketWithAuthentication opts auth path ci = snd <$> connectUnixSocketDetailWithAuthentication opts auth path ci
-- | Greeting-returning variant of 'connectUnixSocketWithAuthentication'.
connectUnixSocketDetailWithAuthentication :: ConnectionOptions -> AuthenticationOptions -> FilePath -> ConnectInfo -> IO (Greeting,MySQLConn)
connectUnixSocketDetailWithAuthentication opts auth path ci = withConnectDeadline opts $ do
validateOptions opts
let open=bracketOnError (N.socket N.AF_UNIX N.Stream 0) N.close $ \sock -> do
N.connect sock (N.SockAddrUnix path)
TCP.socketToConnectionAdaptive bUFSIZE (sock,N.SockAddrUnix path)
bracketOnError open Conn.close $ \raw -> do
(authConn,conn) <- newAuthenticatingTransportConnection opts (Conn.source raw) (Conn.send raw) (Conn.close raw)
greet <- authenticationDiagnostic (readGreeting authConn)
let caps=negotiatedCapabilities opts greet
writeIORef (connectionCapabilities conn) caps
mysqlWrite conn (encodeToPacket 1 (mkAuthForAuthentication auth opts caps ci greet))
q <- authenticationDiagnostic (readConnPacket authConn)
completeAuthWithOptions auth authConn greet True (ciPassword ci) q
when (testBit caps 5) (connectionCompressionStart conn)
role <- initializeSession conn
rememberInitialSession conn ci role
pure (greet,conn)
-- Greeting parsing consumes the whole envelope; an initial ERR is a server
-- failure, rather than an attempted protocol-10 decode of error text.
readGreeting :: MySQLConn -> IO Greeting
readGreeting conn = do
packet <- readConnPacket conn
when (isERR packet) (decodeFromPacket packet >>= throwIO . ERRException)
getFromPacketExact getGreeting packet
negotiatedCapabilities :: ConnectionOptions -> Greeting -> Word64
-- Advertise tracking and EOF deprecation only as a pair: both OK-envelope and
-- result-terminator parsing must be available before either is used on the wire.
negotiatedCapabilities options greeting = (implemented .&. greetingCaps greeting) .|.
(if enableSessionTracking options && greetingCaps greeting .&. 0x01800000 == 0x01800000 then 0x01800000 else 0)
where
implemented = clientCap .|. bit 34 .|. bit 37
.|. (if enableProgressReporting options then bit 32 else 0)
.|. (if enableCompression options then bit 5 else 0)
.|. (if enableMetadataCaching options then bit 36 else 0)
.|. (if enableExtendedMetadata options then bit 35 else 0)
-- | A required GSSAPI principal also forbids sending a password proof before
-- that method has been negotiated. MariaDB normally greets with native-password
-- even for GSSAPI accounts; send an empty initial response and require its switch.
mkAuthForAuthentication :: AuthenticationOptions -> ConnectionOptions -> Word64 -> ConnectInfo -> Greeting -> Auth
mkAuthForAuthentication auth options caps ci greeting
| isJust (authenticationGSSAPIPrincipal auth) = Auth caps
(fromIntegral (maximumMessageBytes options)) (ciCharset ci) (ciUser ci)
B.empty (ciDatabase ci) (greetingAuthPlugin greeting)
| otherwise = mkAuthWithOptions options caps ci greeting
mkAuthWithOptions :: ConnectionOptions -> Word64 -> ConnectInfo -> Greeting -> Auth
mkAuthWithOptions options caps ci greeting =
let Auth _ _ charset user response database plugin = mkAuth (ciDatabase ci) (ciUser ci) (ciPassword ci) (ciCharset ci) greeting
in Auth caps (fromIntegral (maximumMessageBytes options)) charset user response database plugin
mkAuth :: ByteString -> ByteString -> ByteString -> Word8 -> Greeting -> Auth
mkAuth db user pass charset greet =
let salt = greetingSalt1 greet `B.append` greetingSalt2 greet
plugin = greetingAuthPlugin greet
scambleBuf = scrambleForPlugin plugin salt pass
in Auth clientCap clientMaxPacketSize charset user scambleBuf db plugin
-- | Dispatch scramble based on the authentication plugin name.
scrambleForPlugin :: ByteString -> ByteString -> ByteString -> ByteString
scrambleForPlugin plugin salt pass
| plugin == "caching_sha2_password" = checkedNonce 20 (scrambleSHA256 salt pass)
| plugin == "mysql_native_password" = checkedNonce 20 (scrambleSHA1 salt pass)
| plugin == "client_ed25519" = checkedNonce 32 (scrambleEd25519 salt pass)
| plugin == "parsec" = checkedNonce 32 B.empty
-- External plugins begin their conversation after the client header. No
-- password/token may be emitted before transport and opt-in checks.
| plugin `elem` ["dialog", "mysql_clear_password", "auth_gssapi_client"] = B.empty
| otherwise = throw (AuthException ("unsupported MariaDB authentication plugin: " ++ show plugin))
where
checkedNonce count result
| B.length salt == count = result
| otherwise = throw (AuthException "invalid authentication challenge length")
-- | SHA1-based scramble for @mysql_native_password@.
scrambleSHA1 :: ByteString -> ByteString -> ByteString
scrambleSHA1 salt pass
| B.null pass = B.empty
| otherwise = B.pack (B.zipWith xor sha1pass withSalt)
where sha1pass = sha1 pass
withSalt = sha1 (salt `B.append` sha1 sha1pass)
sha1 :: ByteString -> ByteString
sha1 = BA.convert . (Crypto.hash :: ByteString -> Crypto.Digest Crypto.SHA1)
-- | SHA256-based scramble for @caching_sha2_password@.
-- XOR(SHA256(password), SHA256(SHA256(SHA256(password)) + nonce))
scrambleSHA256 :: ByteString -> ByteString -> ByteString
scrambleSHA256 salt pass
| B.null pass = B.empty
| otherwise = B.pack (B.zipWith xor sha256pass withSalt)
where sha256pass = sha256 pass
withSalt = sha256 (sha256 sha256pass `B.append` salt)
sha256 :: ByteString -> ByteString
sha256 = BA.convert . (Crypto.hash :: ByteString -> Crypto.Digest Crypto.SHA256)
-- | MariaDB's password-based ed25519 variant. The password itself is hashed
-- once to an expanded scalar/prefix, unlike standard Ed25519's 32-byte seed.
-- Arithmetic uses crypton's constant-time Edwards primitives; no server point
-- or client-implemented curve arithmetic is involved.
scrambleEd25519 :: ByteString -> ByteString -> ByteString
scrambleEd25519 nonce password = encodedR <> Edwards.scalarEncode signed
where
digest = BA.convert (Crypto.hash password :: Crypto.Digest Crypto.SHA512) :: ByteString
scalarBytes = B.pack ([B.head digest .&. 248] ++ B.unpack (B.take 30 (B.drop 1 digest))
++ [(B.index digest 31 .&. 63) .|. 64])
scalar = throwCryptoError (Edwards.scalarDecodeLong scalarBytes)
hashScalar bytes = throwCryptoError (Edwards.scalarDecodeLong
(BA.convert (Crypto.hash bytes :: Crypto.Digest Crypto.SHA512) :: BA.ScrubbedBytes))
public = Edwards.pointEncode (Edwards.toPoint scalar) :: ByteString
nonceScalar = hashScalar (B.drop 32 digest <> nonce)
encodedR = Edwards.pointEncode (Edwards.toPoint nonceScalar)
challenge = hashScalar (encodedR <> public <> nonce)
signed = Edwards.scalarAdd nonceScalar (Edwards.scalarMul challenge scalar)
-- | Bound the entire connection establishment, including resolver, TLS,
-- authentication and initial session queries. The enclosing brackets release
-- any acquired transports before this timeout becomes a public exception.
withConnectDeadline :: ConnectionOptions -> IO a -> IO a
withConnectDeadline options action = do
validateOptions options
timeout (connectTimeoutMicros options) action >>= maybe
(throwIO (AuthException "MariaDB connection establishment timed out")) pure
-- | Packet decoders retain failing buffers for ordinary query diagnostics.
-- Authentication parsing must not disclose them. Keep this wrapper around the
-- decoder itself so exceptions deliberately raised by application callbacks
-- retain their original type and value.
authenticationDiagnostic :: IO a -> IO a
authenticationDiagnostic action = action `catch` \(_ :: DecodePacketException) ->
throwIO (AuthException "malformed authentication packet")
-- | Complete a bounded plugin conversation on this connection. Plugin and nonce
-- always travel together, including auth switches. Cleartext full-auth responses
-- are allowed only over TLS or a Unix-domain socket. Malformed messages and
-- unsupported plugins fail closed; the constructor owns transport cleanup.
completeAuth :: MySQLConn -> Greeting -> Bool -> ByteString -> Packet -> IO ()
completeAuth = completeAuthWithOptions defaultAuthenticationOptions
-- | External-authentication variant. Switches embed raw plugin data; later
-- tokens use AuthMoreData. Decode that envelope before invoking a plugin, and
-- accept terminal OK/ERR only in their original SQL-protocol envelope.
completeAuthWithOptions :: AuthenticationOptions -> MySQLConn -> Greeting -> Bool -> ByteString -> Packet -> IO ()
completeAuthWithOptions auth conn greeting secure password first = do
checkPacket first
when (isERR first) (authError first)
if greetingAuthPlugin greeting `elem` externalPlugins
then external 0 (greetingAuthPlugin greeting) (unwrapPlugin first)
else loop 0 (greetingAuthPlugin greeting) initialNonce first
where
initialNonce = greetingSalt1 greeting <> greetingSalt2 greeting
input = mysqlRead conn
output = mysqlWrite conn
send packet body = output (Packet (fromIntegral (B.length body)) (pSeqN packet + 1) (L.fromStrict body))
next = readPacketWithLimit (authenticationWireLimit (connectionOptions conn)) input
authError packet = authenticationDiagnostic (decodeFromPacket packet) >>= throwIO . ERRException
authOK packet = void (authenticationDiagnostic (consumeOK conn packet))
checkPacket packet = when (pLen packet < 0 || fromIntegral (pLen packet) > authenticationWireLimit (connectionOptions conn))
(throwIO (AuthException "authentication packet exceeds message limit"))
loop :: Int -> ByteString -> ByteString -> Packet -> IO ()
loop rounds plugin nonce packet
| rounds >= 16 = throwIO (AuthException "too many authentication exchanges")
| isERR packet = authError packet
-- Requiring a principal is an authentication-method requirement, not
-- merely permission to use the credential cache if a server asks.
-- Reject OK and password continuations before any alternate response.
| isJust (authenticationGSSAPIPrincipal auth) && not (isAuthSwitch packet) =
throwIO (AuthException "MariaDB did not select required GSSAPI authentication")
| isOK packet = authOK packet
| isAuthSwitch packet = do
checkPacket packet
let (name, rest) = B.break (==0) (B.drop 1 (L.toStrict (pBody packet)))
when (B.null name || B.length name > 128 || B.null rest)
(throwIO (AuthException "malformed authentication switch"))
when (isJust (authenticationGSSAPIPrincipal auth) && name /= "auth_gssapi_client")
(throwIO (AuthException "MariaDB requested another method instead of required GSSAPI"))
let challenge = B.drop 1 rest
if name `elem` externalPlugins
then external (rounds+1) name
(Packet (fromIntegral (B.length challenge)) (pSeqN packet) (L.fromStrict challenge))
else passwordSwitch rounds name challenge packet
| isAuthMoreData packet && plugin == "parsec" = do
let parameters = B.drop 1 (L.toStrict (pBody packet))
unless (B.length parameters == 20 && B.head parameters == 80)
(throwIO (AuthException "invalid PARSEC salt parameters"))
let factor = fromIntegral (B.index parameters 1) :: Int
when (factor > 20 || (1024 :: Integer) `shiftL` factor > fromIntegral (maximumAuthIterations (connectionOptions conn)))
(throwIO (AuthException "PARSEC work exceeds configured iteration limit"))
let keyBytes = PBKDF2.fastPBKDF2_SHA512 (PBKDF2.Parameters (1024 `shiftL` factor) 32)
password (B.drop 2 parameters) :: BA.ScrubbedBytes
key = throwCryptoError (Ed25519.secretKey keyBytes)
clientNonce <- getRandomBytes 32
let signature = BA.convert (Ed25519.sign key (Ed25519.toPublic key) (nonce <> clientNonce))
send packet (clientNonce <> signature)
next >>= loop (rounds+1) "parsec-finished" nonce
| isAuthMoreData packet && plugin == "caching_sha2_password" = do
let dataBytes = B.drop 1 (L.toStrict (pBody packet))
case B.unpack dataBytes of
[3] -> next >>= terminal
[4] -> do
if secure then send packet (password <> "\0")
else do
send packet (B.singleton 2)
keyPacket <- next
when (isERR keyPacket) (authError keyPacket)
let body = L.toStrict (pBody keyPacket)
unless (isAuthMoreData keyPacket && B.length body > 1 && B.length body <= 16384)
(throwIO (AuthException "invalid RSA authentication key envelope"))
doRSAAuth output (pSeqN keyPacket+1) password nonce (B.drop 1 body)
next >>= terminal
_ -> throwIO (AuthException "invalid SHA-2 authentication continuation")
| otherwise = throwIO (AuthException "unexpected authentication response")
passwordSwitch rounds name challenge packet = do
let expected = if name `elem` ["client_ed25519", "parsec"] then 32 else 20
exact = if B.length challenge == expected+1 && B.last challenge == 0
then B.init challenge else challenge
response <- evaluate (scrambleForPlugin name exact password)
send packet response
next >>= loop (rounds+1) name exact
-- MariaDB wraps later plugin data with AuthMoreData (0x01), while data
-- embedded in AuthSwitchRequest is already raw. Strip the wrapper exactly
-- once at the wire boundary; never feed it to PAM or the GSS mechanism.
unwrapPlugin packet
| isAuthMoreData packet = packet {pLen=pLen packet-1,pBody=L.tail (pBody packet)}
| otherwise = packet
nextPlugin = do
packet <- next
when (isERR packet) (authError packet)
unless (isAuthMoreData packet) (throwIO (AuthException "unexpected authentication response"))
pure (unwrapPlugin packet)
externalPlugins = ["dialog", "mysql_clear_password", "auth_gssapi_client"]
bounded packet = do
when (pLen packet > 1048576) (throwIO (AuthException "authentication token exceeds 1 MiB"))
pure (L.toStrict (pBody packet))
external :: Int -> ByteString -> Packet -> IO ()
external rounds plugin packet
| isJust (authenticationGSSAPIPrincipal auth) && plugin /= "auth_gssapi_client" =
throwIO (AuthException "MariaDB did not select required GSSAPI authentication")
| rounds >= 16 = throwIO (AuthException "too many authentication exchanges")
| plugin == "auth_gssapi_client" = do
announcement <- bounded packet
let (principal,rest) = B.break (==0) announcement
mechanism = B.drop 1 rest
expected <- maybe (throwIO (AuthException "GSSAPI requires an expected server principal")) pure
(authenticationGSSAPIPrincipal auth)
unless (not (B.null expected) && B.length expected <= 1024 && not (B.elem 0 expected)
&& principal == expected && not (B.null rest)
&& mechanism `elem` ["", "\0", "Kerberos\0", "Negotiate\0"])
(throwIO (AuthException "GSSAPI server principal or mechanism does not match configuration"))
(GSSAPI.withGSSAPI expected (mechanism == "Negotiate\0") $ \step ->
gssLoop step rounds packet B.empty)
`catch` \(GSSAPI.GSSAPIError message) -> throwIO (AuthException message)
| otherwise = do
unless secure (throwIO (AuthException "PAM cleartext authentication requires TLS or a Unix socket"))
callback <- maybe (throwIO (AuthException "PAM requires an explicit dialog callback")) pure
(authenticationDialog auth)
if plugin == "mysql_clear_password"
then do
void (bounded packet)
-- A clear-password switch usually carries no plugin data.
-- Its sequence still belongs to the switch packet.
reply <- callback (DialogPrompt B.empty False True 0)
sendReply reply
next >>= terminal
else do
-- PAM can request a client-plugin switch before its
-- helper has produced a prompt. A zero-byte reply
-- acknowledges that switch; it contains no password.
question <- if pLen packet == 0
then send packet B.empty >> nextPlugin
else pure packet
dialogLoop callback rounds 0 question
sendReply reply = do
when (B.length reply > 65535 || B.elem 0 reply)
(throwIO (AuthException "PAM reply exceeds 65535 bytes or contains NUL"))
-- The checked writer owns the next sequence, including an empty-data
-- auth switch; using it avoids a second read before the required reply.
FramingState sequenceN _ <- readIORef (connectionSequence conn)
let seqN = fromIntegral sequenceN
output (Packet (fromIntegral (B.length reply+1)) seqN (L.fromStrict (reply <> "\0")))
dialogLoop callback rounds question packet = do
when (rounds >= 16) (throwIO (AuthException "too many authentication exchanges"))
-- This is raw plugin data, not an SQL envelope. In particular, an
-- ERR-shaped token must not become a server error containing its bytes.
body <- bounded packet
unless (not (B.null body) && B.head body `elem` [2,3,4,5])
(throwIO (AuthException "malformed PAM dialog question"))
let kind = B.head body
prompt = B.tail body
-- auth_pam sends packet-length-delimited text without a NUL;
-- dialog plugins may also supply one terminal NUL.
text = if not (B.null prompt) && B.last prompt == 0 then B.init prompt else prompt
when (B.length text > 65535 || B.elem 0 text)
(throwIO (AuthException "PAM prompt exceeds 65535 bytes or contains NUL"))
reply <- callback (DialogPrompt text (kind < 4) (odd kind) question)
sendReply reply
packet' <- next
if odd kind || isOK packet' || isERR packet'
then terminal packet'
else do
unless (isAuthMoreData packet') (throwIO (AuthException "unexpected authentication continuation"))
dialogLoop callback (rounds+1) (question+1) (unwrapPlugin packet')
gssLoop step rounds packet token = do
when (rounds >= 16) (throwIO (AuthException "too many authentication exchanges"))
(complete,response) <- step token
unless (B.null response) (send packet response)
reply <- next
if complete then terminal reply else do
when (isERR reply) (authError reply)
-- An early SQL OK is not evidence of a completed mutual context.
unless (isAuthMoreData reply)
(throwIO (AuthException "GSSAPI context ended before mutual authentication"))
inputToken <- bounded (unwrapPlugin reply)
gssLoop step (rounds+1) reply inputToken
terminal packet
| isOK packet = authOK packet
| isERR packet = authError packet
| otherwise = throwIO (AuthException "unexpected authentication response")
-- | Full auth handler for plain TCP connections using RSA encryption.
--
-- The nonce is captured via partial application at the call site:
-- @completeAuth is writePacket pass q (plainFullAuth salt)@
plainFullAuth :: ByteString -> Word8 -> ByteString -> (Packet -> IO ()) -> InputStream Packet -> IO ()
plainFullAuth nonce seqN pass writePacket is = do
-- Request server's RSA public key
writePacket (putToPacket seqN (Binary.putWord8 0x02))
keyPkt <- readPacket is
-- Key packet body: 0x01 <PEM bytes>
let pemKey = L.toStrict (L.drop 1 (pBody keyPkt))
doRSAAuth writePacket (pSeqN keyPkt + 1) pass nonce pemKey
q <- readPacket is
if isOK q then pure ()
else decodeFromPacket q >>= throwIO . ERRException
-- | Parse a PEM-encoded SubjectPublicKeyInfo RSA public key (PKCS#8 / RFC 5480).
parseRSAPublicKey :: ByteString -> Either String RSA.PublicKey
parseRSAPublicKey pemBytes =
either (Left . show) Right (PEM.pemParseBS pemBytes)
>>= firstPEM
>>= decodeDER
>>= spkiBitString
>>= decodeDER
>>= rsaKey
where
firstPEM [] = Left "empty PEM"
firstPEM (x:_) = Right (PEM.pemContent x)
decodeDER bs = either (Left . show) Right (ASN1E.decodeASN1' ASN1B.DER bs)
spkiBitString (ASN1.Start ASN1.Sequence
: ASN1.Start ASN1.Sequence
: ASN1.OID [1,2,840,113549,1,1,1]
: ASN1.Null
: ASN1.End ASN1.Sequence
: ASN1.BitString bs
: ASN1.End ASN1.Sequence
: [])
= Right (ASN1BA.bitArrayGetData bs)
spkiBitString xs = Left ("unexpected SPKI ASN1: " ++ show xs)
rsaKey (ASN1.Start ASN1.Sequence
: ASN1.IntVal n
: ASN1.IntVal e
: ASN1.End ASN1.Sequence
: [])
-- Check signed ASN.1 integers before positive-integer serialization.
-- The SPKI algorithm above must be rsaEncryption, the OAEP key type.
| n <= 0 || e < 3 || even e || e > 4294967295 = Left "invalid RSA public integers"
| otherwise = Right RSA.PublicKey
{ RSA.public_size = B.length (Serialize.i2osp n :: ByteString)
, RSA.public_n = n
, RSA.public_e = e
}
rsaKey xs = Left ("unexpected RSA key ASN1: " ++ show xs)
-- | XOR @(password ++ NUL)@ with a cyclically-extended nonce.
xorPasswordNonce :: ByteString -> ByteString -> ByteString
xorPasswordNonce pass nonce
| B.length nonce /= 20 = throw (AuthException "invalid RSA authentication nonce length")
| otherwise =
let msg = pass `B.snoc` 0x00
nonceLen = B.length nonce
cycled = B.pack [nonce `B.index` (i `mod` nonceLen) | i <- [0 .. B.length msg - 1]]
in B.pack (B.zipWith xor msg cycled)
-- | Encrypt password with the server's RSA public key and send it.
doRSAAuth :: (Packet -> IO ()) -> Word8 -> ByteString -> ByteString -> ByteString -> IO ()
doRSAAuth writePacket seqN pass nonce pemKey = do
pubKey <- either (\e -> throwIO (AuthException ("RSA key parse: " ++ e))) pure
(parseRSAPublicKey pemKey)
unless (RSA.public_size pubKey >= 256 && RSA.public_size pubKey <= 1024)
(throwIO (AuthException "RSA authentication key is outside supported bounds"))
result <- OAEP.encrypt (OAEP.defaultOAEPParams SHA1) pubKey
(xorPasswordNonce pass nonce)
ct <- either (\e -> throwIO (AuthException ("RSA encrypt: " ++ show e))) pure result
writePacket (putToPacket seqN (Binary.putByteString ct))
-- | Authentication/plugin validation or establishment-deadline failure.
-- Server ERR replies retain ERRException; cancellation keeps its original
-- exception identity. Constructors release their transport before propagating.
data AuthException = AuthException String deriving (Typeable, Show)
instance Exception AuthException
-- Physical framing reads exactly one packet. Limits are checked before its body.
decodeInputStream :: InputStream ByteString -> IO (InputStream Packet)
decodeInputStream = decodeInputStreamWithLimit (maximumMessageBytes defaultConnectionOptions)
decodeInputStreamWithLimit :: Word64 -> InputStream ByteString -> IO (InputStream Packet)
decodeInputStreamWithLimit limit input = do
framing <- newIORef (FramingState 256 0)
decodeFramedInput limit framing input
decodeFramedInput :: Word64 -> IORef FramingState -> InputStream ByteString -> IO (InputStream Packet)
decodeFramedInput limit framing input = Stream.makeInputStream $ do
header <- Stream.readExactly 4 input
let len=fromIntegral (B.index header 0) .|. (fromIntegral (B.index header 1) `shiftL` 8) .|. (fromIntegral (B.index header 2) `shiftL` 16)
advanceFraming limit framing len (B.index header 3)
body <- loop [] len
pure (Just (Packet len (B.index header 3) body))
where
loop acc 0 = pure (L.fromChunks (reverse acc))
loop acc remaining = do
chunk <- Stream.read input >>= maybe (throwIO NetworkException) pure
if B.null chunk then loop acc remaining else do
let count=fromIntegral (B.length chunk)
if count >= remaining then do
let (part,rest)=B.splitAt (fromIntegral remaining) chunk
unless (B.null rest) (Stream.unRead rest input)
pure (L.fromChunks (reverse (part:acc)))
else loop (chunk:acc) (remaining-count)
readPacket :: InputStream Packet -> IO Packet
readPacket = readPacketWithLimit (maximumMessageBytes defaultConnectionOptions)
readPacketWithLimit :: Word64 -> InputStream Packet -> IO Packet
readPacketWithLimit limit input = do
first <- Stream.read input >>= maybe (throwIO NetworkException) pure
validateLength (pLen first)
if pLen first < 16777215 then pure first else go (pLen first) (pSeqN first) [pBody first]
where
validateLength len = when (len < 0 || fromIntegral len > limit) (throwIO (ProtocolError "logical packet exceeds message limit"))
go total sequenceN chunks = do
packet <- Stream.read input >>= maybe (throwIO NetworkException) pure
unless (pSeqN packet == sequenceN+1) (throwIO (ProtocolError "invalid continuation sequence"))
when (pLen packet < 0 || pLen packet > 16777215) (throwIO (ProtocolError "invalid fragment length"))
let size=total+pLen packet
validateLength size
if pLen packet < 16777215 then pure (Packet size (pSeqN packet) (L.concat (reverse (pBody packet:chunks))))
else go size (pSeqN packet) (pBody packet:chunks)
readConnPacket :: MySQLConn -> IO Packet
readConnPacket conn
| not (enableProgressReporting (connectionOptions conn)) = readOne
| otherwise = reports
where
readOne = readPacketWithLimit (maximumMessageBytes (connectionOptions conn)) (mysqlRead conn)
reports = do
packet <- readOne
if not (isProgressPacket packet) then pure packet else do
caps <- readIORef (connectionCapabilities conn)
unless (testBit caps 32) (throwIO (ProtocolError "unnegotiated progress packet"))
when (pLen packet > 16384 || fromIntegral (pLen packet) > maximumMetadataBytes (connectionOptions conn))
(throwIO (ProtocolError "progress description exceeds client limit"))
progress <- getFromPacketExact getProgress packet
callback <- readIORef (connectionProgressCallback conn)
forM_ callback $ \notify -> bracket_
(writeIORef (connectionCallbackActive conn) True)
(writeIORef (connectionCallbackActive conn) False) (notify progress)
reports
-- | Replace the progress callback while idle. Negotiate
-- 'enableProgressReporting' at connect time; without a callback reports are
-- consumed and discarded. Callbacks run synchronously on the command owner:
-- they must not reenter this connection. Exceptions/cancellation retire an
-- incomplete exchange. No background worker or unbounded queue exists.
setProgressCallback :: MySQLConn -> Maybe (Progress -> IO ()) -> IO ()
setProgressCallback conn callback = withCommand conn $ do
caps <- readIORef (connectionCapabilities conn)
unless (testBit caps 32) (throwIO (ProtocolError "progress reporting was not negotiated"))
writeIORef (connectionProgressCallback conn) callback
writeCommand :: Command -> (Packet -> IO ()) -> IO ()
writeCommand cmd writer = let body=Binary.runPut (putCommand cmd) in writeEncoded (L.length body) body writer
writeEncoded :: Int64 -> L.ByteString -> (Packet -> IO ()) -> IO ()
writeEncoded count body writer = send count 0 body
where
send len sequenceN payload
| len < 16777215 = writer (Packet len sequenceN payload)
| otherwise = do
let (part,rest)=L.splitAt 16777215 payload
writer (Packet 16777215 sequenceN part)
send (len-16777215) (sequenceN+1) rest
-- Validate/encode the complete outgoing command before marking WireInFlight.
-- Traversing bounded lazy chunks keeps oversized builders from being flattened
-- into a second contiguous allocation merely to discover their size.
sendCommand :: MySQLConn -> Command -> IO ()
sendCommand conn cmd = do
policy <- readIORef (connectionQueryEffect conn)
when (policy==Nothing) $ case cmd of
COM_QUERY bytes -> classifyQueryEffect conn (Query bytes)
COM_STMT_PREPARE bytes -> classifyQueryEffect conn (Query bytes)
COM_STMT_EXECUTE ident _ _ -> do
opaque <- readIORef (connectionOpaqueStatements conn)
writeIORef (connectionQueryEffect conn) (Just (ident `elem` opaque))
COM_STMT_EXECUTE_EXT ident _ _ _ -> do
opaque <- readIORef (connectionOpaqueStatements conn)
writeIORef (connectionQueryEffect conn) (Just (ident `elem` opaque))
COM_STMT_BULK_EXECUTE ident _ _ _ -> do
opaque <- readIORef (connectionOpaqueStatements conn)
writeIORef (connectionQueryEffect conn) (Just (ident `elem` opaque))
_ -> pure ()
encoder <- case cmd of
COM_STMT_EXECUTE ident values nullMap -> do
let needsEncoding (MySQLText _)=True; needsEncoding _=False
charset <- if any needsEncoding values then renderClientCharset <$> getRenderContext conn else pure "utf8mb4"
mapM_ (either (throwIO . UnencodableParameter) pure . validateValue) values
encoded <- forM values $ \value -> case value of
MySQLText text -> either (throwIO . UnencodableParameter) (pure . putLenEncBytes) (encodeCharsetText charset text)
_ -> pure (putBinaryField value)
pure $ do
Binary.putWord8 0x17
Binary.putWord32le ident
Binary.putWord8 0
Binary.putWord32le 1
unless (null values) $ do
Binary.putByteString (fromBitMap nullMap)
Binary.putWord8 1
mapM_ putParamMySQLType values
sequence_ encoded
COM_STMT_EXECUTE_EXT ident flags values nullMap -> do
unless (flags `elem` [0,1]) (throwIO (ProtocolError "unsupported cursor mode"))
let ordinary = [value | ExecuteValue value <- values]
charset <- if any (\value -> case value of MySQLText _ -> True; _ -> False) ordinary
then renderClientCharset <$> getRenderContext conn else pure "utf8mb4"
mapM_ (either (throwIO . UnencodableParameter) pure . validateValue) ordinary
encoded <- forM values $ \value -> case value of
ExecuteValue (MySQLText text) -> either (throwIO . UnencodableParameter)
(pure . putLenEncBytes) (encodeCharsetText charset text)
ExecuteValue cell -> pure (putBinaryField cell)
_ -> pure (pure ())
pure $ do
Binary.putWord8 0x17
Binary.putWord32le ident
Binary.putWord8 flags
Binary.putWord32le 1
unless (null values) $ do
Binary.putByteString (fromBitMap nullMap)
Binary.putWord8 1
mapM_ (putParamMySQLType . executeParameterValue) values
sequence_ encoded
COM_STMT_BULK_EXECUTE ident flags types rows -> do
caps <- readIORef (connectionCapabilities conn)
unless (testBit caps 34 && (flags .&. 64 == 0 || testBit caps 37))
(throwIO (ProtocolError "server did not negotiate requested bulk capability"))
unless (flags .&. 0xff3f == 0 && flags .&. 128 /= 0)
(throwIO (ProtocolError "invalid bulk flags"))
let dimensionsError = ProtocolError "bulk dimensions exceed client limits or are empty"
width <- maybe (throwIO dimensionsError) pure
(lengthWithin (maximumColumns (connectionOptions conn)) types)
rowCount <- maybe (throwIO dimensionsError) pure
(lengthWithin (maximumBulkRows (connectionOptions conn)) rows)
unless (width > 0 && rowCount > 0)
(throwIO (ProtocolError "bulk dimensions exceed client limits or are empty"))
ensureBulkMinimumFits conn width rowCount
forM_ rows $ \row -> do
unless (lengthWithin width row == Just width)
(throwIO (ProtocolError "bulk parameter count mismatch"))
forM_ (zip types row) $ \(typ, cell) -> case cell of
BulkValue value | value /= MySQLNull -> do
unless (parameterTypeSignature typ == parameterTypeSignature value)
(throwIO (ProtocolError "bulk column has inconsistent wire types"))
either (throwIO . UnencodableParameter) pure (validateValue value)
_ -> pure ()
let hasText = any (any (\cell -> case cell of BulkValue (MySQLText _) -> True; _ -> False)) rows
charset <- if hasText then renderClientCharset <$> getRenderContext conn else pure "utf8mb4"
serverLimit <- readIORef (connectionServerPacketLimit conn)
let limit = min serverLimit (maximumMessageBytes (connectionOptions conn))
header = 7 + 2 * fromIntegral width
when (header > limit) (throwIO (ProtocolError "bulk header exceeds message limit"))
remaining <- newIORef (limit - header)
let normalize cell = do
left <- readIORef remaining
when (left == 0) (throwIO (ProtocolError "bulk exceeds message limit"))
normalized <- case cell of
BulkValue (MySQLText value) -> do
when (fromIntegral (T.length value) > left)
(throwIO (ProtocolError "bulk text exceeds message limit"))
either (throwIO . UnencodableParameter) (pure . BulkValue . MySQLBytes)
(encodeCharsetText charset value)
_ -> pure cell
let bytes = 1 + case normalized of
BulkValue value -> bulkValueSize value
_ -> 0
when (bytes > left) (throwIO (ProtocolError "bulk exceeds message limit"))
writeIORef remaining (left - bytes)
pure normalized
encodedRows <- mapM (mapM normalize) rows
pure (putCommand (COM_STMT_BULK_EXECUTE ident flags types encodedRows))
_ -> pure (putCommand cmd)
let body=Binary.runPut encoder
serverLimit <- readIORef (connectionServerPacketLimit conn)
size <- encodedLength (min serverLimit (maximumMessageBytes (connectionOptions conn))) body
writeIORef (connectionActiveStatement conn) $ case cmd of
COM_STMT_EXECUTE ident _ _ -> Just ident
COM_STMT_EXECUTE_EXT ident _ _ _ -> Just ident
COM_STMT_FETCH ident _ -> Just ident
COM_STMT_BULK_EXECUTE ident _ _ _ -> Just ident
_ -> Nothing
writeIORef (connectionResultIndex conn) 0
case cmd of
COM_RESET_CONNECTION -> do
-- Invalidate before writing: an interrupted reset must never leave
-- a usable handle to a statement MariaDB may already have removed.
atomicModifyIORef' (connectionStatements conn) (\registry ->
(fmap (const (StatementRegistry IM.empty 0)) registry, ()))
writeIORef (connectionOpaqueStatements conn) []
writeIORef (connectionRenderContext conn) Nothing
writeIORef (connectionServerStatus conn) Nothing
invalidateSession conn
_ -> pure ()
setExchangeSequence conn 0
connectionCompressionReset conn
writeIORef (connectionExchangeProgress conn) WireInFlight
writeEncoded size body (mysqlWrite conn)
-- Byte accounting happens while normalizing cells, before retaining another
-- batch worth of transcoded text. Fixed scalars avoid per-cell builders; only
-- bounded decimal/temporal values need their small existing encoder for sizing.
bulkValueSize :: MySQLValue -> Word64
bulkValueSize value = case value of
MySQLNull -> 0
MySQLBytes bytes -> stringSize bytes
MySQLGeometry bytes -> stringSize bytes
_ -> case parameterTypeSignature value .&. 255 of
1 -> 1
2 -> 2
3 -> 4
4 -> 4
5 -> 8
8 -> 8
_ -> fromIntegral (L.length (Binary.runPut (putBinaryField value)))
where
stringSize bytes = let size = fromIntegral (B.length bytes)
prefix | size < 251 = 1
| size < 65536 = 3
| size < 16777216 = 4
| otherwise = 9
in size + prefix
-- | Every cell needs an indicator even when all values are NULL/default. Reject
-- impossible batches before traversing their cells or choosing column types.
ensureBulkMinimumFits :: MySQLConn -> Int -> Int -> IO ()
ensureBulkMinimumFits conn width rows = do
serverLimit <- readIORef (connectionServerPacketLimit conn)
let minimumBytes = 7 + 2 * fromIntegral width + fromIntegral width * fromIntegral rows
when (minimumBytes > min serverLimit (maximumMessageBytes (connectionOptions conn)))
(throwIO (ProtocolError "bulk indicators exceed message limit"))
-- | Count a finite list only when it fits the bound. Inspect at most one excess
-- cons without evaluating its element. Unlike @take (limit + 1)@ this also works
-- when a caller uses 'maxBound' as a permissive resource limit.
lengthWithin :: Int -> [a] -> Maybe Int
lengthWithin limit = go 0
where
go !count [] = if count <= limit then Just count else Nothing
go !count (_ : rest)
| count < limit = go (count + 1) rest
| otherwise = Nothing
{-# INLINE lengthWithin #-}
-- | Idempotently close the transport. Ready connections attempt bounded COM_QUIT;
-- active ones are interrupted immediately without trying to drain their reply.
close :: MySQLConn -> IO ()
close conn = mask_ $ do
clearStatements conn
writeIORef (connectionProgressCallback conn) Nothing
invalidateSession conn
prior <- atomicModifyIORef' (connectionRuntime conn) (\rt -> (rt {runtimeStatus=Closed,runtimeOwner=Nothing},runtimeStatus rt))
unless (prior == Closed) $ do
when (prior `elem` [Executing,Streaming,Replicating]) $
atomicModifyIORef' (connectionFailureRef conn) (\failure ->
(case failure of Nothing -> Just (ConnectionFailure prior False False); Just _ -> failure,()))
writeIORef (connectionDrain conn) Nothing
let goodbye=when (prior==Ready) $ void $ timeout (cleanupTimeoutMicros (connectionOptions conn)) $ do
setExchangeSequence conn 0
connectionCompressionReset conn
writeCommand COM_QUIT (mysqlWrite conn)
(goodbye `catch` (\(err :: SomeException) -> when (isCancellation err) (throwIO err))) `finally` releaseTransport conn
-- The cheap path recognizes flat ASCII SQL with only known stable calls.
-- Everything else uses the same lexer as parameter templates; a prefix alone
-- never establishes that an arbitrary statement cannot alter session state.
classifyQueryEffect :: MySQLConn -> Query -> IO ()
classifyQueryEffect conn qry@(Query input) = do
context <- readIORef (connectionRenderContext conn)
let uncertain=if simpleStableQuery input then False else
case context >>= either (const Nothing) Just . (`compileQuery` qry) of
Just plan -> planOpaque plan || planMultiple plan
Nothing -> True
writeIORef (connectionQueryEffect conn) (Just uncertain)
-- | Check server responsiveness through the ordinary command ownership guard.
ping :: MySQLConn -> IO OK
ping conn = command conn COM_PING
-- | Low-level command expecting one OK/ERR reply. Commands producing rows or no
-- reply require their dedicated APIs instead (for example 'Database.MySQL.Base.closeStmt' in Base).
command :: MySQLConn -> Command -> IO OK
command conn COM_RESET_CONNECTION = resetConnection conn
command conn cmd = withCommand conn $ sendCommand conn cmd >> readConnPacket conn >>= consumeCommandOK conn
consumeCommandOK :: MySQLConn -> Packet -> IO OK
consumeCommandOK conn packet
| isERR packet = serverError conn packet
| isOK packet = consumeOK conn packet
| otherwise = throwIO (UnexpectedPacket packet)
serverError :: MySQLConn -> Packet -> IO a
serverError conn packet = do
err <- decodeFromPacket packet
writeIORef (connectionServerStatus conn) Nothing
writeIORef (connectionRenderContext conn) Nothing
invalidateSession conn
finishResponse conn
throwIO (ERRException err)
updateStatus :: MySQLConn -> Word16 -> IO ()
updateStatus conn flags = do
previous <- readIORef (connectionServerStatus conn)
unless (previous==Just flags) $ do
writeIORef (connectionServerStatus conn) (Just flags)
atomicModifyIORef' (connectionSession conn) (\cache -> (applySessionStatus flags cache, ()))
when (maybe True (\old -> (old .&. 0x8200) /= (flags .&. 0x8200)) previous) $
atomicModifyIORef' (connectionRenderContext conn) $ \ctx ->
let change c = c {renderSQLModes=toggle "NO_BACKSLASH_ESCAPES" 0x200 (toggle "ANSI_QUOTES" 0x8000 (renderSQLModes c))}
toggle name flag modes = if flags .&. flag /= 0 then name:filter (/=name) modes else filter (/=name) modes
in (fmap change ctx,())
consumeOK :: MySQLConn -> Packet -> IO OK
consumeOK conn packet = case P.parseDetailLazy getOK (pBody packet) of
Left (buffer,offset,message) -> throwIO (DecodePacketFailed buffer offset message)
Right (_,offset,ok) -> do
updateStatus conn (okStatus ok)
caps <- readIORef (connectionCapabilities conn)
when (caps .&. 0x800000 /= 0) $ do
when (pLen packet<offset || fromIntegral (pLen packet-offset)>maximumMetadataBytes (connectionOptions conn))
(throwIO (ProtocolError "OK/session envelope exceeds metadata budget"))
let trailer=L.toStrict (L.drop offset (pBody packet))
changed=okStatus ok .&. 0x4000 /= 0
if B.null trailer then when changed (throwIO (ProtocolError "missing session tracking payload")) else do
(_,rest) <- either (throwIO . ProtocolError) pure (lenString trailer)
if changed then do
(tracking,leftover) <- either (throwIO . ProtocolError) pure (lenString rest)
unless (B.null leftover) (throwIO (ProtocolError "trailing session payload"))
when (fromIntegral (B.length tracking)>maximumMetadataBytes (connectionOptions conn)) (throwIO (ProtocolError "session tracking payload exceeds metadata budget"))
changes <- either (throwIO . ProtocolError) pure (decodeSessionChanges tracking)
recordSessionChanges conn changes
mapM_ (updateTracked conn) [(name,value) | SystemVariableChanged name value <- changes]
-- MariaDB aliases these settings by storage offset. A
-- collation-only tracker cannot establish the new charset
-- required by the renderer; keep that context explicit.
when (hasVariableChange "collation_connection" changes &&
not (hasVariableChange "character_set_connection" changes)) $
writeIORef (connectionRenderContext conn) Nothing
else unless (B.null rest) (throwIO (ProtocolError "unexpected OK trailer"))
pure ok
-- Kept for internal wire fixtures and the rendering parser's legacy helper.
lenString :: ByteString -> Either String (ByteString,ByteString)
lenString = lengthString
trackingPairs :: ByteString -> Either String [(ByteString,ByteString)]
trackingPairs payload = do
changes <- decodeSessionChanges payload
pure [(name,value) | SystemVariableChanged name value <- changes]
-- | Read observed session state without claiming the wire or issuing SQL.
-- ERR, terminal failure, untracked/opaque effects and reset invalidate facts;
-- unknown values stay explicit. Last changes are a bounded latest envelope,
-- not a lossless change log. Do not infer that an absent variable is a default.
getSessionState :: MySQLConn -> IO SessionState
getSessionState conn = do
cache <- readIORef (connectionSession conn)
phase <- connectionStatus conn
pure (sessionObservation (if phase `elem` [Broken,Closed]
then emptySessionCache (sessionChangeNumber (sessionObservation cache)) else cache))
invalidateSession :: MySQLConn -> IO ()
invalidateSession conn = do
flags <- readIORef (connectionServerStatus conn)
atomicModifyIORef' (connectionSession conn) $ \cache ->
let empty = emptySessionCache (sessionChangeNumber (sessionObservation cache) + 1)
in (maybe empty (`applySessionStatus` empty) flags, ())
-- Opaque SQL may disable its own tracking. Keep the last real wire envelope
-- observable, but invalidate final-state facts rather than invent completeness.
invalidateSessionFacts :: MySQLConn -> IO ()
invalidateSessionFacts conn = do
flags <- readIORef (connectionServerStatus conn)
atomicModifyIORef' (connectionSession conn) $ \cache ->
let prior = sessionObservation cache
empty = emptySessionCache (sessionChangeNumber prior + 1)
current = maybe empty (`applySessionStatus` empty) flags
in (current {sessionObservation = (sessionObservation current)
{sessionLastChanges = sessionLastChanges prior}}, ())
recordSessionChanges :: MySQLConn -> [SessionChange] -> IO ()
recordSessionChanges conn = updateSessionObservations conn TrackingEnvelope
-- Seed facts read by explicit helper SQL without fabricating tracker events.
recordSessionFacts :: MySQLConn -> [SessionChange] -> IO ()
recordSessionFacts conn = updateSessionObservations conn SnapshotFacts
updateSessionObservations :: MySQLConn -> ObservationSource -> [SessionChange] -> IO ()
updateSessionObservations conn source changes = do
result <- atomicModifyIORef' (connectionSession conn) $ \cache ->
case applySessionObservations source (maximumMetadataBytes (connectionOptions conn)) changes cache of
Left message -> (cache, Left message)
Right updated -> (updated, Right ())
either (\message -> abortConnection conn False >> throwIO (ProtocolError message)) pure result
updateTracked :: MySQLConn -> (ByteString,ByteString) -> IO ()
updateTracked conn (name,value)
| name `elem` ["sql_mode","character_set_client","character_set_connection","collation_connection"] && B.length value>maximumRenderingValueBytes = throwIO (ProtocolError "tracked rendering value exceeds bound")
| name == "session_track_system_variables" = do
let selected=BC.split ',' value
unless (value=="*" || all (`elem` selected) ["sql_mode","character_set_client","character_set_connection"])
(writeIORef (connectionRenderContext conn) Nothing)
| otherwise = atomicModifyIORef' (connectionRenderContext conn) $ \ctx ->
let apply c = case name of
"sql_mode" -> c {renderSQLModes=BC.split ',' (B.copy value)}
"character_set_client" -> c {renderClientCharset=B.copy value}
"character_set_connection" -> c {renderConnectionCharset=B.copy value,renderConnectionCollation=Nothing}
"collation_connection" -> c {renderConnectionCollation=Just (B.copy value)}
_ -> c
in (fmap apply ctx,())
usesOKTerminator :: MySQLConn -> IO Bool
usesOKTerminator conn = (/=0) . (.&. 0x1000000) <$> readIORef (connectionCapabilities conn)
isResultEnd :: MySQLConn -> Packet -> IO Bool
isResultEnd conn packet = (\modern -> isResultEndFor modern packet) <$> usesOKTerminator conn
isResultEndFor :: Bool -> Packet -> Bool
isResultEndFor modern packet = if modern then pLen packet > 0 && pLen packet < 16777215 && not (L.null (pBody packet)) && L.head (pBody packet)==0xfe else isEOF packet
{-# INLINE isResultEndFor #-}
consumeEnd :: MySQLConn -> Packet -> IO Word16
consumeEnd conn packet = do
modern <- usesOKTerminator conn
if modern then okStatus <$> consumeOK conn packet else do
unless (pLen packet==5) (throwIO (ProtocolError "invalid legacy EOF length"))
EOF _ flags <- decodeFromPacket packet
updateStatus conn flags
pure flags
readMetadataEnd :: MySQLConn -> IO ()
readMetadataEnd conn = do
modern <- usesOKTerminator conn
unless modern $ do
packet <- readConnPacket conn
unless (isEOF packet) (throwIO (UnexpectedPacket packet))
void (consumeEnd conn packet)
-- Read a complete metadata block before publishing it into a cache. A partial
-- replacement never destroys the last fully validated decoder.
readColumnBlock :: MySQLConn -> Int -> IO CachedColumns
readColumnBlock conn count = do
unless (count >= 0 && count <= maximumColumns (connectionOptions conn))
(throwIO (ProtocolError "column count exceeds client limit"))
caps <- readIORef (connectionCapabilities conn)
let loop 0 total = pure ([], total)
loop n total = do
packet <- readConnPacket conn
when (isERR packet) (serverError conn packet)
let budget = maximumMetadataBytes (connectionOptions conn)
size = fromIntegral (pLen packet)
when (total > budget || size > budget - total)
(throwIO (ProtocolError "metadata exceeds client limit"))
field <- getFromPacketExact (getColumnMetadata (testBit caps 35)) packet
(rest, bytes) <- loop (n - 1) (total + size)
pure (field : rest, bytes)
(details, bytes) <- loop count 0
when (count > 0) (readMetadataEnd conn)
let fields = map metadataColumn details
pure (CachedColumns details fields count (prepareBinaryRow fields) bytes)
readColumns :: MySQLConn -> Int -> IO [ColumnDef]
readColumns conn count = cachedFields <$> readColumnBlock conn count
-- A header marker is present on text results too when caching is negotiated.
-- Omission is legal only for the exact registered statement/result ordinal.
-- Legacy metadata EOF still follows an omitted block when EOF deprecation is off.
readResultColumns :: MySQLConn -> Packet -> IO CachedColumns
readResultColumns conn packet = do
caps <- readIORef (connectionCapabilities conn)
(count, follows) <- getFromPacketExact (do
n <- getLenEncInt
flag <- if testBit caps 36 then P.getWord8 else pure 1
unless (flag <= 1) (fail "invalid metadata-follows flag")
pure (n, flag == 1)) packet
unless (count > 0 && count <= maximumColumns (connectionOptions conn))
(throwIO (ProtocolError "invalid result column count"))
active <- readIORef (connectionActiveStatement conn)
index <- atomicModifyIORef' (connectionResultIndex conn) (\n -> (n + 1, n))
if follows then do
columns <- readColumnBlock conn count
case active of
Just ident -> storeStatementColumns conn ident index columns
Nothing -> pure columns
else do
states <- statementMap conn
let found = active >>= (\ident -> IM.lookup (fromIntegral ident) states)
>>= IM.lookup index . statementResults
case found of
Just columns | cachedColumnCount columns == count -> do
readMetadataEnd conn
pure columns
_ -> throwIO (ProtocolError "omitted metadata has no matching statement/result cache")
-- The registry and its terminal marker share one atomic reference. A close
-- racing with metadata publication can never resurrect a handle or its buffers.
statementMap :: MySQLConn -> IO (IM.IntMap StatementState)
statementMap conn = maybe IM.empty registryStatements <$> readIORef (connectionStatements conn)
-- | Registration can replace a server ID. Remove that generation's entire charge
-- atomically, so its metadata cannot remain billed to the replacement handle.
registerStatement :: MySQLConn -> StmtID -> Word64 -> Int -> IO ()
registerStatement conn ident serial arity = do
inserted <- atomicModifyIORef' (connectionStatements conn) $ \current -> case current of
Nothing -> (Nothing, False)
Just registry ->
let key = fromIntegral ident
states = registryStatements registry
old = maybe 0 statementMetadataBytes (IM.lookup key states)
state = StatementState
{ statementSerial = serial
, statementArity = arity
, statementResults = IM.empty
, statementMetadataBytes = 0
}
updated = registry
{ registryStatements = IM.insert key state states
, registryMetadataBytes = registryMetadataBytes registry - old
}
in (Just updated, True)
unless inserted (throwIO (ConnectionUnavailable Closed))
-- Cached names must own their bytes. A tiny column name decoded from a shared
-- receive chunk must not pin that chunk (and neighbouring row data) indefinitely.
-- Copy only retained statement metadata; ordinary text results keep their path.
ownColumnBlock :: CachedColumns -> IO CachedColumns
ownColumnBlock columns = do
details <- forM (cachedDetails columns) $ \metadata -> do
let field = metadataColumn metadata
owned <- evaluate field
{ columnDB = B.copy (columnDB field)
, columnTable = B.copy (columnTable field)
, columnOrigTable = B.copy (columnOrigTable field)
, columnName = B.copy (columnName field)
, columnOrigName = B.copy (columnOrigName field)
}
attributes <- forM (metadataAttributes metadata) $ \(tag, bytes) -> do
value <- evaluate (B.copy bytes)
pure (tag, value)
pure metadata {metadataColumn = owned, metadataAttributes = attributes}
let fields = map metadataColumn details
pure columns {cachedDetails = details, cachedFields = fields,
cachedBinaryDecoder = prepareBinaryRow fields}
storeStatementColumns :: MySQLConn -> StmtID -> Int -> CachedColumns -> IO CachedColumns
storeStatementColumns conn ident index columns = do
owned <- ownColumnBlock columns
outcome <- atomicModifyIORef' (connectionStatements conn) $ \current -> case current of
Nothing -> (Nothing, Left (ConnectionUnavailable Closed))
Just registry -> case IM.lookup (fromIntegral ident) (registryStatements registry) of
Nothing -> (Just registry, Right ())
Just state ->
let old = maybe 0 cachedMetadataBytes (IM.lookup index (statementResults state))
remaining = registryMetadataBytes registry - old
replacement = cachedMetadataBytes columns
budget = maximumMetadataBytes (connectionOptions conn)
state' = state
{ statementResults = IM.insert index owned (statementResults state)
, statementMetadataBytes = statementMetadataBytes state - old + replacement
}
registry' = registry
{ registryStatements = IM.insert (fromIntegral ident) state' (registryStatements registry)
, registryMetadataBytes = remaining + replacement
}
in if remaining > budget || replacement > budget - remaining
then (Just registry, Left (ProtocolError "aggregate statement metadata exceeds client limit"))
else (Just registry', Right ())
either throwIO (const (pure owned)) outcome
forgetStatement :: MySQLConn -> StmtID -> IO ()
forgetStatement conn ident = atomicModifyIORef' (connectionStatements conn) $ \registry ->
let remove current =
let key = fromIntegral ident
states = registryStatements current
released = maybe 0 statementMetadataBytes (IM.lookup key states)
in current
{ registryStatements = IM.delete key states
, registryMetadataBytes = registryMetadataBytes current - released
}
in (fmap remove registry, ())
clearStatements :: MySQLConn -> IO ()
clearStatements conn = do
writeIORef (connectionStatements conn) Nothing
writeIORef (connectionOpaqueStatements conn) []
-- | Return known session facts without issuing SQL. A snapshot can become stale
-- after another operation; connected APIs obtain it under their command claim.
-- Unknown facts raise 'SessionContextUnavailable' rather than silently probing.
getRenderContext :: MySQLConn -> IO RenderContext
getRenderContext conn = readIORef (connectionRenderContext conn) >>= maybe (throwIO SessionContextUnavailable) pure
-- Startup and reset include CURRENT_ROLE in the existing snapshot exchange.
-- Ordinary public refresh keeps its six rendering facts and never redefines
-- the constructor's restoration target.
initializeSession :: MySQLConn -> IO (Maybe ByteString)
initializeSession = refreshSessionSnapshot RenderingAndRole
-- Capture the constructor's database and wire charset once. Later SET/USE
-- commands must not silently redefine what a pool's clean session means.
rememberInitialSession :: MySQLConn -> ConnectInfo -> Maybe ByteString -> IO ()
rememberInitialSession conn ci role = do
context <- getRenderContext conn
collation <- maybe (throwIO SessionContextUnavailable) pure (renderConnectionCollation context)
writeIORef (connectionInitialSession conn) (Just (InitialSession
(B.copy (ciDatabase ci)) (renderClientCharset context) collation (renderServerVersion context) role))
recordSessionFacts conn [SchemaChanged (B.copy (ciDatabase ci))]
-- | Roll back and reset a synchronized MariaDB session without reconnecting.
-- Restores the constructor's active role, nonempty database and charset, configures required
-- tracking, and refreshes rendering facts. Other settings return to server
-- defaults; use 'resetConnectionWith' for application-specific initialization.
-- All prepared handles/raw registrations and cached decoders are invalidated.
--
-- Pending streams, transaction helpers and other command owners are rejected
-- before sending. This cannot repair a broken transport. One cleanup deadline
-- covers reset and restoration; any failure after starting retires the socket.
-- The returned OK belongs to RESET; helper SQL changes statement diagnostics.
resetConnection :: MySQLConn -> IO OK
resetConnection conn = resetConnectionWith conn (const (pure ()))
-- | Reset and initialize under one exclusive lease and cleanup deadline.
-- The callback may issue ordinary commands, but must consume all results and
-- leave no active transaction. Failure or cancellation retires the connection;
-- SQL is never replayed. A pool should open connections with an explicit
-- database: MariaDB cannot reset a selected database back to "none". A
-- constructor with no database can reset only while that fact remains known.
-- This callback runs with the caller's masking state, including buffered work.
resetConnectionWith :: MySQLConn -> (MySQLConn -> IO ()) -> IO OK
resetConnectionWith conn initialize = mask $ \restore -> do
acquireResetLease conn
flip finally (endTransactionLease conn) $ do
initial <- readIORef (connectionInitialSession conn) >>= maybe
(throwIO (ProtocolError "initial session restoration facts unavailable")) pure
when (initialVersion initial < 100204)
(throwIO (ProtocolError "MariaDB reset requires server 10.2.4 or later"))
state <- getSessionState conn
when (B.null (initialSchema initial) && sessionSchema state /= Just B.empty)
(throwIO (ProtocolError "cannot reset a selected/unknown database to none; use an explicit constructor database"))
let run = do
ok <- withCommand conn $ do
sendCommand conn COM_RESET_CONNECTION
readConnPacket conn >>= consumeCommandOK conn
-- RESET preserves the active role. Snapshot it alongside the
-- rendering facts, then restore it before selecting a database:
-- access to that database may come exclusively from this role.
role <- initializeSession conn
when (role /= initialRole initial) $ do
context <- getRenderContext conn
sql <- case initialRole initial of
Nothing -> pure "SET ROLE NONE"
Just name -> do
text <- either (const (throwIO (ProtocolError "invalid initial role encoding"))) pure
(TE.decodeUtf8' name)
let renderRole current = renderParams current "SET ROLE ?" [MySQLText text]
utf8Role = do
-- A valid role can be outside the original
-- client charset. Change only this rare
-- restoration path, then let the ordinary
-- charset restoration below put it back.
void (command conn (COM_QUERY "SET NAMES utf8mb4"))
refreshSessionState conn
utf8 <- getRenderContext conn
either throwIO pure (renderRole utf8)
case renderRole context of
Left (UnencodableParameter _) -> utf8Role
Left (UnsupportedRenderingContext _) -> utf8Role
result -> either throwIO pure result
void (command conn (COM_QUERY (fromQuery sql)))
when (sessionSchema state /= Just (initialSchema initial)) $
void (command conn (COM_INIT_DB (initialSchema initial)))
recordSessionFacts conn [SchemaChanged (initialSchema initial)]
-- Without negotiated tracking, ROLE/INIT_DB invalidate the
-- rendering snapshot. Refresh only on that restoration path;
-- unchanged roles/databases keep the ordinary exchange count.
restorationContext <- readIORef (connectionRenderContext conn)
unless (isJust restorationContext) (refreshSessionState conn)
-- MariaDB restores the authentication charset on RESET. Avoid
-- a redundant SET NAMES unless the snapshot reports otherwise.
context <- getRenderContext conn
let identifier bytes = not (B.null bytes) && B.all
(\w -> w >= 97 && w <= 122 || w >= 65 && w <= 90 || w >= 48 && w <= 57 || w == 95) bytes
unless (identifier (initialCharset initial) && identifier (initialCollation initial))
(throwIO (ProtocolError "invalid initial charset/collation identifier"))
when (renderClientCharset context /= initialCharset initial ||
renderConnectionCollation context /= Just (initialCollation initial)) $ do
void (command conn (COM_QUERY (L.fromStrict ("SET NAMES " <> initialCharset initial
<> " COLLATE " <> initialCollation initial))))
refreshSessionState conn
restored <- getRenderContext conn
unless (renderClientCharset restored == initialCharset initial &&
renderConnectionCollation restored == Just (initialCollation initial))
(throwIO (ProtocolError "server did not restore the initial charset/collation"))
recordSessionFacts conn [SchemaChanged (initialSchema initial)]
initialize conn
guardUnconsumed conn
flags <- readIORef (connectionServerStatus conn)
unless (maybe False (\value -> value .&. 1 == 0) flags)
(throwIO (ProtocolError "reset initializer left an active or unknown transaction"))
known <- readIORef (connectionRenderContext conn)
when (not (isJust known)) (refreshSessionState conn)
pure ok
result <- timeout (cleanupTimeoutMicros (connectionOptions conn))
(restore run `onException` abortConnection conn False)
case result of
Just ok -> pure ok
Nothing -> abortConnection conn False >> throwIO CleanupDeadlineExceeded
-- A reset inside withTransaction/refresh would roll back its caller's work.
-- Unlike ordinary session refresh this operation never borrows an existing
-- lease, even on the same thread. The atomic claim also excludes stream owners.
acquireResetLease :: MySQLConn -> IO ()
acquireResetLease conn = do
tid <- myThreadId
result <- atomicModifyIORef' (connectionRuntime conn) $ \runtime ->
if runtimeStatus runtime == Ready && transactionOwner runtime == Nothing
then (runtime {transactionOwner = Just tid}, Right ())
else (runtime, Left (case runtimeStatus runtime of
Streaming -> toException UnconsumedResultSet
Broken -> toException (ConnectionUnavailable Broken)
Closed -> toException (ConnectionUnavailable Closed)
_ | transactionOwner runtime == Just tid -> toException NestedTransactionUnsupported
| otherwise -> toException ConcurrentConnectionUse))
either throwIO pure result
-- | Reconfigure tracking and explicitly snapshot rendering facts under one lease.
-- This executes SQL and therefore changes preceding-statement diagnostics such
-- as ROW_COUNT(). Existing tracker selections, including '*', are preserved.
refreshSessionState :: MySQLConn -> IO ()
refreshSessionState conn = void (refreshSessionSnapshot RenderingFacts conn)
-- Binary role bytes bypass character_set_results: lossy conversion could make
-- two distinct Unicode roles compare equal. The extra column costs no exchange.
-- Keep it out of ordinary refreshes, which do not establish a new clean-session role.
data SnapshotFields = RenderingFacts | RenderingAndRole deriving (Eq)
refreshSessionSnapshot :: SnapshotFields -> MySQLConn -> IO (Maybe ByteString)
refreshSessionSnapshot fieldsToRead conn = withSessionLease conn $ do
guardUnconsumed conn
caps <- readIORef (connectionCapabilities conn)
when (caps .&. 0x800000 /= 0) $ void $ withCommand conn $ do
-- This exact setup statement changes only tracker configuration. It
-- cannot execute a routine or invalidate an already observed schema.
writeIORef (connectionQueryEffect conn) (Just False)
sendCommand conn (COM_QUERY "SET SESSION session_track_schema=ON, session_track_system_variables=IF(@@session.session_track_system_variables='*','*',CONCAT_WS(',',NULLIF(@@session.session_track_system_variables,''),'sql_mode,character_set_client,character_set_connection,character_set_results,session_track_system_variables,session_track_schema,session_track_transaction_info'))")
readConnPacket conn >>= consumeCommandOK conn
(context, limit, role) <- withCommand conn $ do
writeIORef (connectionQueryEffect conn) (Just False)
let query = "SELECT @@sql_mode,@@character_set_client,@@character_set_connection,@@collation_connection,VERSION(),CAST(@@max_allowed_packet AS CHAR)"
<> if fieldsToRead == RenderingAndRole then ",CAST(CURRENT_ROLE() AS BINARY)" else ""
sendCommand conn (COM_QUERY query)
first <- readConnPacket conn
when (isERR first) (serverError conn first)
metadata <- readResultColumns conn first
let fields = cachedFields metadata
count = length fields
unless (count == if fieldsToRead == RenderingAndRole then 7 else 6)
(throwIO (ProtocolError "invalid session snapshot column count"))
rowPacket <- readConnPacket conn
when (fromIntegral (pLen rowPacket) > maximumMetadataBytes (connectionOptions conn))
(throwIO (ProtocolError "session snapshot exceeds metadata budget"))
rawValues <- getFromPacketExact (getTextRow fields) rowPacket
-- Force/validate the snapshot while this exchange is still owned. A
-- lazy UTF-8 exception after Ready would escape retirement, and a
-- permissive decoder could otherwise accept an unparsed row trailer.
let (renderingValues, roleValues) = splitAt 6 rawValues
role <- case roleValues of
[] -> pure Nothing
[MySQLNull] -> pure Nothing
[MySQLText value] -> Just <$> ownSnapshotValue (TE.encodeUtf8 value)
[MySQLBytes value] -> do
-- Binary results must still represent a valid role name when
-- later rendered in the constructor's client charset.
owned <- ownSnapshotValue value
either (const (throwIO (ProtocolError "invalid role snapshot encoding"))) (const (pure ()))
(TE.decodeUtf8' owned)
pure (Just owned)
_ -> throwIO (ProtocolError "invalid role snapshot value type")
values <- forM renderingValues $ \value -> do
bytes <- case value of
MySQLText text -> evaluate (TE.encodeUtf8 text)
MySQLBytes raw -> pure raw
_ -> throwIO (ProtocolError "invalid session snapshot value type")
-- These six cold-start facts outlive the receive buffers. Retain
-- explicitly owned, bounded storage regardless of codec internals;
-- ordinary result payloads are not copied by this policy.
ownSnapshotValue bytes
snapshot <- case values of
[modes, client, server, collation, version, packetLimit] -> do
versionID <- maybe (throwIO (ProtocolError "invalid server version snapshot")) pure (versionNumber version)
limit <- case BC.readInteger packetLimit of
Just (n, rest) | B.null rest && n > 0 && n <= fromIntegral (maxBound :: Word64) -> pure (fromIntegral n)
_ -> throwIO (ProtocolError "invalid server packet limit snapshot")
pure (RenderContext (BC.split ',' modes) client server (Just collation) versionID, limit)
_ -> throwIO (ProtocolError "invalid session snapshot")
end <- readConnPacket conn
isEnd <- isResultEnd conn end
unless isEnd (throwIO (UnexpectedPacket end))
flags <- consumeEnd conn end
when (flags .&. 8 /= 0) (throwIO (ProtocolError "unexpected additional session snapshot result"))
let (context, limit) = snapshot
pure (context, limit, role)
writeIORef (connectionServerPacketLimit conn) limit
writeIORef (connectionRenderContext conn) (Just context)
recordSessionFacts conn
[SystemVariableChanged "sql_mode" (BC.intercalate "," (renderSQLModes context)),
SystemVariableChanged "character_set_client" (renderClientCharset context),
SystemVariableChanged "character_set_connection" (renderConnectionCharset context),
SystemVariableChanged "collation_connection" (maybe "" id (renderConnectionCollation context))]
pure role
where
-- Failure must be detected under the command claim, not turned into version
-- zero after Ready: version gates determine which SQL bytes are executable.
versionNumber value = case traverse component (take 3 (BC.split '.' (BC.takeWhile (\w -> w == '.' || w >= '0' && w <= '9') value))) of
Just [a,b,c] | a > 0 -> Just (a*10000+b*100+c)
_ -> Nothing
component value = case BC.readInt value of
Just (n, rest) | n >= 0 && n < 100 && B.null rest -> Just n
_ -> Nothing
-- Force copies while the exchange is owned; neither a role nor a rendering
-- fact can retain a receive-buffer slice or defer work past Ready.
ownSnapshotValue bytes = boundedSnapshotValue bytes >>= evaluate . B.copy
boundedSnapshotValue bytes = do
when (B.length bytes > maximumRenderingValueBytes)
(throwIO (ProtocolError "session snapshot value exceeds bound"))
pure bytes
guardUnconsumed :: MySQLConn -> IO ()
guardUnconsumed conn = do
rt <- readIORef (connectionRuntime conn)
tid <- myThreadId
when (maybe False (/=tid) (transactionOwner rt)) (throwIO ConcurrentConnectionUse)
case runtimeStatus rt of
Ready -> pure ()
Streaming -> throwIO UnconsumedResultSet
Executing -> throwIO ConcurrentConnectionUse
other -> throwIO (ConnectionUnavailable other)
-- Legacy low-level helpers retain their behavior for independent packet streams.
waitCommandReply :: InputStream Packet -> IO OK
waitCommandReply input = do
packet <- readPacket input
if isERR packet then decodeFromPacket packet >>= throwIO . ERRException
else if isOK packet then decodeFromPacket packet else throwIO (UnexpectedPacket packet)
waitCommandReplys :: InputStream Packet -> IO [OK]
waitCommandReplys input = do
ok <- waitCommandReply input
if isThereMore ok then (ok:) <$> waitCommandReplys input else pure [ok]
writeIORef' :: IORef a -> a -> IO ()
writeIORef' ref value = value `seq` writeIORef ref value
-- | An expected packet/body could not be read from a closed packet source.
-- Other socket/TLS exceptions can propagate directly; inspect connection status
-- to decide whether a new connection is required.
data NetworkException = NetworkException deriving (Typeable,Show)
instance Exception NetworkException
-- | A command was attempted while the preceding result still owns the wire.
-- No bytes for the new command have been sent; consume or discard that result.
data UnconsumedResultSet = UnconsumedResultSet deriving (Typeable,Show)
instance Exception UnconsumedResultSet
-- | MariaDB returned an ERR reply. Ordinary commands consume the complete reply
-- before permitting reuse, but invalidate cached rendering facts. Replication
-- and constructor failures have their own terminal lifecycle rules.
data ERRException = ERRException ERR deriving (Typeable,Show)
instance Exception ERRException
-- | A packet marker/shape is unexpected for the current protocol phase.
data UnexpectedPacket = UnexpectedPacket Packet deriving (Typeable,Show)
instance Exception UnexpectedPacket
-- System.Timeout's private exception is not necessarily SomeAsyncException.
isCancellation :: SomeException -> Bool
isCancellation err@(SomeException inner) = isJust (fromException err :: Maybe SomeAsyncException)
|| tyConName (typeRepTyCon (typeOf inner)) == "Timeout"
-- A refresh's configuration and snapshot exchanges are one logical operation.
withSessionLease :: MySQLConn -> IO a -> IO a
withSessionLease conn action = mask $ \restore -> do
tid <- myThreadId
acquired <- atomicModifyIORef' (connectionRuntime conn) $ \rt -> case transactionOwner rt of
Nothing | runtimeStatus rt==Ready -> (rt {transactionOwner=Just tid},Right True)
Just owner | owner==tid && runtimeStatus rt==Ready -> (rt,Right False)
_ -> (rt,Left ConcurrentConnectionUse)
own <- either throwIO pure acquired
restore action `finally` when own (endTransactionLease conn)
-- Force bounded encoded chunks once, without first allocating an oversized message.
encodedLength :: Word64 -> L.ByteString -> IO Int64
encodedLength limit = loop 0 . L.toChunks
where
loop count []=pure (fromIntegral count)
loop count (chunk:rest)=do
let next=count+fromIntegral (B.length chunk)
when (next>limit) (throwIO (ProtocolError "outgoing message exceeds client/server limit"))
next `seq` loop next rest
{-# INLINE readConnPacket #-}
{-# INLINE streamGuard #-}