diff --git a/CHANGELOG b/CHANGELOG --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,10 @@ +1.0.1 +------ +- Address an issue whereby Cassandra 'Error' responses might be mistakenly + thrown by the (internal) withRetries function, possibly leading to + problems with the automatic (re-)preparation of prepared queries upon + 'Unprepared' server responses. + 1.0.0 ------ - Add support for CQL V4 binary protocol. diff --git a/cql-io.cabal b/cql-io.cabal --- a/cql-io.cabal +++ b/cql-io.cabal @@ -1,5 +1,5 @@ name: cql-io -version: 1.0.0 +version: 1.0.1 synopsis: Cassandra CQL client. license: MPL-2.0 license-file: LICENSE @@ -79,7 +79,7 @@ build-depends: async >= 2.0 , auto-update >= 0.1 - , base >= 4.7 && < 5.0 + , base >= 4.9 && < 5.0 , bytestring >= 0.10 , containers >= 0.5 , cql >= 4.0 diff --git a/src/Database/CQL/IO.hs b/src/Database/CQL/IO.hs --- a/src/Database/CQL/IO.hs +++ b/src/Database/CQL/IO.hs @@ -205,7 +205,10 @@ -- | A type which can be run as a query. class RunQ q where - runQ :: (MonadClient m, Tuple a, Tuple b) => q k a b -> QueryParams a -> m (Response k a b) + runQ :: (MonadClient m, Tuple a, Tuple b) + => q k a b + -> QueryParams a + -> m (Response k a b) instance RunQ QueryString where runQ q p = request (RqQuery (Query q p)) @@ -259,7 +262,7 @@ r <- runQ q p getResult r >>= \case RowsResult _ b -> return b - _ -> throwM $ UnexpectedResponse' r + _ -> throwM $ UnexpectedResponse r -- | Run a CQL schema query, returning 'SchemaChange' information, if any. schema :: (MonadClient m, Tuple a, RunQ q) => q S a () -> QueryParams a -> m (Maybe SchemaChange) diff --git a/src/Database/CQL/IO/Batch.hs b/src/Database/CQL/IO/Batch.hs --- a/src/Database/CQL/IO/Batch.hs +++ b/src/Database/CQL/IO/Batch.hs @@ -23,6 +23,7 @@ import Control.Monad.Trans import Control.Monad.Trans.State.Strict import Database.CQL.IO.Client +import Database.CQL.IO.Cluster.Host import Database.CQL.IO.PrepQuery import Database.CQL.IO.Types import Database.CQL.Protocol @@ -38,9 +39,9 @@ batch m = do b <- execStateT (unBatchM m) (Batch BatchLogged [] Quorum Nothing) r <- executeWithPrepare Nothing (RqBatch b :: Raw Request) - getResult r >>= \case + getResult (hrResponse r) >>= \case VoidResult -> return () - _ -> throwM $ UnexpectedResponse' r + _ -> throwM $ UnexpectedResponse (hrResponse r) -- | Add a query to this batch. addQuery :: (Show a, Tuple a, Tuple b) => QueryString W a b -> a -> BatchM () diff --git a/src/Database/CQL/IO/Client.hs b/src/Database/CQL/IO/Client.hs --- a/src/Database/CQL/IO/Client.hs +++ b/src/Database/CQL/IO/Client.hs @@ -54,7 +54,7 @@ #if MIN_VERSION_transformers(0,4,0) import Control.Monad.Trans.Except #endif -import Control.Retry (capDelay, exponentialBackoff, rsIterNumber, applyPolicy) +import Control.Retry (capDelay, exponentialBackoff, rsIterNumber) import Control.Retry (recovering) import Data.Foldable (for_, foldrM) import Data.List (find) @@ -235,12 +235,12 @@ request :: (MonadClient m, Tuple a, Tuple b) => Request k a b -> m (Response k a b) request a = liftClient $ do n <- liftIO . hostCount =<< view policy - snd <$> withRetries (requestN n) a + hrResponse <$> withRetries (requestN n) a -- | Invoke 'request1' up to @n@ times with different hosts if no -- connection is available. May return 'Nothing' if no connection -- is available on any of the tried hosts. -requestN :: (Tuple b, Tuple a) => Word -> Request k a b -> ClientState -> Client (Maybe (Host, Response k a b)) +requestN :: (Tuple b, Tuple a) => Word -> Request k a b -> ClientState -> Client (Maybe (HostResponse k a b)) requestN !n a s = do hst <- pickHost (s^.policy) res <- request1 hst a s @@ -252,13 +252,13 @@ -- | Get 'Response' from a single 'Host'. -- May return 'Nothing' if no connection is available. -request1 :: (Tuple a, Tuple b) => Host -> Request k a b -> ClientState -> Client (Maybe (Host, Response k a b)) +request1 :: (Tuple a, Tuple b) => Host -> Request k a b -> ClientState -> Client (Maybe (HostResponse k a b)) request1 h a s = do p <- Map.lookup h <$> readTVarIO' (s^.hostmap) case p of Just x -> do result <- with x transaction `catches` handlers - for_ result $ \(_, r) -> + for_ result $ \(HostResponse _ r) -> for_ (Cql.warnings r) $ \w -> warn $ msg (val "server warning: " +++ w) return result @@ -272,7 +272,7 @@ let x = s^.context.settings.connSettings.compression let v = s^.context.settings.protoVersion r <- parse x <$> C.request c (serialise v x a) - r `seq` return (h, r) + r `seq` return (HostResponse h r) handlers = [ Handler $ \(e :: ConnectionError) -> onConnectionError h e >> throwM e @@ -283,11 +283,11 @@ -- | Execute the given request. If an 'Unprepared' error is returned, this -- function will automatically try to re-prepare the query and re-execute -- the original request using the same host which was used for re-preparation. -executeWithPrepare :: (Tuple b, Tuple a) => Maybe Host -> Request k a b -> Client (Response k a b) +executeWithPrepare :: (Tuple b, Tuple a) => Maybe Host -> Request k a b -> Client (HostResponse k a b) executeWithPrepare h q = do f <- selectAction h r <- withRetries f q - case snd r of + case hrResponse r of RsError _ _ (Unprepared _ i) -> do pq <- preparedQueries qs <- atomically' (PQ.lookupQueryString (QueryId i) pq) @@ -296,8 +296,7 @@ Just s -> do (g, _) <- prepare (Just LazyPrepare) (s :: Raw QueryString) executeWithPrepare (Just g) q - RsError _ _ e -> throwM e - x -> return x + _ -> return r where selectAction Nothing = view policy >>= liftIO . hostCount >>= return . requestN selectAction (Just x) = return (request1 x) @@ -309,8 +308,8 @@ prepare (Just LazyPrepare) qs = do s <- ask n <- liftIO $ hostCount (s^.policy) - (h, r) <- withRetries (requestN n) (RqPrepare (Prepare qs)) - (h,) <$> getPreparedQueryId r + r <- withRetries (requestN n) (RqPrepare (Prepare qs)) + getPreparedQueryId r prepare (Just EagerPrepare) qs = view policy >>= liftIO . current @@ -318,8 +317,8 @@ >>= first where action rq h = do - r <- snd <$> withRetries (request1 h) rq - (h,) <$> getPreparedQueryId r + r <- withRetries (request1 h) rq + getPreparedQueryId r first (x:_) = return x first [] = throwM NoHostAvailable @@ -334,7 +333,7 @@ pq <- view prepQueries maybe (new pq) (run Nothing) =<< atomically' (PQ.lookupQueryId q pq) where - run h i = executeWithPrepare h (RqExecute (Execute i p)) + run h i = hrResponse <$> executeWithPrepare h (RqExecute (Execute i p)) new pq = do (h, i) <- prepare (Just LazyPrepare) (PQ.queryString q) atomically' (PQ.insert q i pq) @@ -500,24 +499,31 @@ ----------------------------------------------------------------------------- -- Exception handling +-- [Note: Error responses] +-- Cassandra error responses are locally thrown as 'ResponseError's to achieve +-- a unified handling of retries in the context of a single retry policy, +-- together with other recoverable (i.e. retryable) exceptions. However, this +-- is just an internal technicality for handling retries - generally error +-- responses must not escape this function as exceptions. Deciding if and when +-- to actually throw a 'ResponseError' upon inspection of the 'HostResponse' +-- must be left to the caller. withRetries :: (Tuple a, Tuple b) - => (Request k a b -> ClientState -> Client (Maybe (Host, Response k a b))) + => (Request k a b -> ClientState -> Client (Maybe (HostResponse k a b))) -> Request k a b - -> Client (Host, Response k a b) + -> Client (HostResponse k a b) withRetries fn a = do s <- ask let p = s^.context.settings.retrySettings.retryPolicy - recovering p recoverFrom $ \i -> do + r <- try $ recovering p canRetry $ \i -> do r <- if rsIterNumber i == 0 then fn a s else fn (newRequest s) (adjust s) case r of Nothing -> throwM HostsBusy - Just hr -> case snd hr of - RsError _ _ e -> applyPolicy p i - >>= maybe (return hr) (const (throwM e)) - _ -> return hr + -- [Note: Error responses] + Just hr -> maybe (return hr) throwM (toResponseError hr) + return $ either fromResponseError id r where adjust s = let x = s^.context.settings.retrySettings.sendTimeoutChange @@ -536,14 +542,14 @@ RqBatch b -> RqBatch b { batchConsistency = c } _ -> a - recoverFrom = - [ const $ Handler $ \e -> return $ case e of - ReadTimeout {} -> True - WriteTimeout {} -> True - Overloaded {} -> True - Unavailable {} -> True - ServerError {} -> True - _ -> False + canRetry = + [ const $ Handler $ \(e :: ResponseError) -> case reCause e of + ReadTimeout {} -> return True + WriteTimeout {} -> return True + Overloaded {} -> return True + Unavailable {} -> return True + ServerError {} -> return True + _ -> return False , const $ Handler $ \(_ :: ConnectionError) -> return True , const $ Handler $ \(_ :: IOException) -> return True , const $ Handler $ \(_ :: HostError) -> return True @@ -664,15 +670,15 @@ getResult :: MonadThrow m => Response k a b -> m (Result k a b) getResult (RsResult _ _ r) = return r getResult (RsError _ _ e) = throwM e -getResult r = throwM $ UnexpectedResponse r +getResult hr = throwM (UnexpectedResponse hr) {-# INLINE getResult #-} -getPreparedQueryId :: MonadThrow m => Response k a b -> m (QueryId k a b) -getPreparedQueryId r = do - rs <- getResult r - case rs of - PreparedResult i _ _ -> return i - _ -> throwM $ UnexpectedResponse r +getPreparedQueryId :: MonadThrow m => HostResponse k a b -> m (Host, QueryId k a b) +getPreparedQueryId hr = do + r <- getResult (hrResponse hr) + case r of + PreparedResult i _ _ -> return (hrHost hr, i) + _ -> throwM $ UnexpectedResponse (hrResponse hr) {-# INLINE getPreparedQueryId #-} peer2Host :: PortNumber -> Peer -> Host diff --git a/src/Database/CQL/IO/Cluster/Host.hs b/src/Database/CQL/IO/Cluster/Host.hs --- a/src/Database/CQL/IO/Cluster/Host.hs +++ b/src/Database/CQL/IO/Cluster/Host.hs @@ -2,16 +2,33 @@ -- License, v. 2.0. If a copy of the MPL was not distributed with this -- file, You can obtain one at http://mozilla.org/MPL/2.0/. +{-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} module Database.CQL.IO.Cluster.Host where import Control.Lens ((^.), Lens') import Data.ByteString.Lazy.Char8 (unpack) +import Database.CQL.Protocol (Response (..)) +import Data.IP import Data.Text (Text) -import Database.CQL.IO.Types (InetAddr) +import Network.Socket (SockAddr (..), PortNumber) import System.Logger.Message +-- | Host representation. +data Host = Host + { _hostAddr :: !InetAddr + , _dataCentre :: !Text + , _rack :: !Text + } deriving (Eq, Ord) + +-- | A response that is known to originate from a specific +-- host of a cluster. +data HostResponse k a b = HostResponse + { hrHost :: !Host + , hrResponse :: !(Response k a b) + } deriving (Show) + -- | This event will be passed to a 'Policy' to inform it about -- cluster changes. data HostEvent @@ -20,13 +37,6 @@ | HostUp !InetAddr -- ^ a host has been started | HostDown !InetAddr -- ^ a host has been stopped --- | Host representation. -data Host = Host - { _hostAddr :: !InetAddr - , _dataCentre :: !Text - , _rack :: !Text - } deriving (Eq, Ord) - -- | The IP address and port number of this host. hostAddr :: Lens' Host InetAddr hostAddr f ~(Host a c r) = fmap (\x -> Host x c r) (f a) @@ -47,3 +57,43 @@ instance ToBytes Host where bytes h = h^.dataCentre +++ val ":" +++ h^.rack +++ val ":" +++ h^.hostAddr + +----------------------------------------------------------------------------- +-- InetAddr + +newtype InetAddr = InetAddr { sockAddr :: SockAddr } deriving (Eq, Ord) + +instance Show InetAddr where + show (InetAddr (SockAddrInet p a)) = + let i = fromIntegral p :: Int in + shows (fromHostAddress a) . showString ":" . shows i $ "" + show (InetAddr (SockAddrInet6 p _ a _)) = + let i = fromIntegral p :: Int in + shows (fromHostAddress6 a) . showString ":" . shows i $ "" + show (InetAddr (SockAddrUnix unix)) = unix +#if MIN_VERSION_network(2,6,1) && !MIN_VERSION_network(3,0,0) + show (InetAddr (SockAddrCan int32)) = show int32 +#endif + +instance ToBytes InetAddr where + bytes (InetAddr (SockAddrInet p a)) = + let i = fromIntegral p :: Int in + show (fromHostAddress a) +++ val ":" +++ i + bytes (InetAddr (SockAddrInet6 p _ a _)) = + let i = fromIntegral p :: Int in + show (fromHostAddress6 a) +++ val ":" +++ i + bytes (InetAddr (SockAddrUnix unix)) = bytes unix +#if MIN_VERSION_network(2,6,1) && !MIN_VERSION_network(3,0,0) + bytes (InetAddr (SockAddrCan int32)) = bytes int32 +#endif + +ip2inet :: PortNumber -> IP -> InetAddr +ip2inet p (IPv4 a) = InetAddr $ SockAddrInet p (toHostAddress a) +ip2inet p (IPv6 a) = InetAddr $ SockAddrInet6 p 0 (toHostAddress6 a) 0 + +inet2ip :: InetAddr -> IP +inet2ip (InetAddr (SockAddrInet _ a)) = IPv4 (fromHostAddress a) +inet2ip (InetAddr (SockAddrInet6 _ _ a _)) = IPv6 (fromHostAddress6 a) +inet2ip _ = error "inet2Ip: not IP4/IP6 address" + + diff --git a/src/Database/CQL/IO/Cluster/Policies.hs b/src/Database/CQL/IO/Cluster/Policies.hs --- a/src/Database/CQL/IO/Cluster/Policies.hs +++ b/src/Database/CQL/IO/Cluster/Policies.hs @@ -17,7 +17,6 @@ import Data.Map.Strict (Map) import Data.Word import Database.CQL.IO.Cluster.Host -import Database.CQL.IO.Types (InetAddr) import System.Random.MWC import Prelude diff --git a/src/Database/CQL/IO/Connection.hs b/src/Database/CQL/IO/Connection.hs --- a/src/Database/CQL/IO/Connection.hs +++ b/src/Database/CQL/IO/Connection.hs @@ -43,6 +43,7 @@ import Data.Unique import Data.Vector (Vector, (!)) import Database.CQL.Protocol +import Database.CQL.IO.Cluster.Host import Database.CQL.IO.Connection.Socket (Socket) import Database.CQL.IO.Connection.Settings import Database.CQL.IO.Hexdump @@ -272,7 +273,7 @@ res <- request c enc case parse (c^.settings.compression) res :: Raw Response of RsReady _ _ Ready -> c^.eventSig |-> f - other -> throwM (UnexpectedResponse' other) + other -> throwM (UnexpectedResponse' other) validateSettings :: MonadIO m => Connection -> m () validateSettings c = liftIO $ do @@ -287,7 +288,7 @@ res <- request c (serialise (c^.protocol) noCompression options) case parse noCompression res :: Raw Response of RsSupported _ _ x -> return x - other -> throwM (UnexpectedResponse' other) + other -> throwM (UnexpectedResponse' other) useKeyspace :: MonadIO m => Connection -> Keyspace -> m () useKeyspace c ks = liftIO $ do @@ -298,7 +299,7 @@ res <- request c (serialise (c^.protocol) cmp req) case parse cmp res :: Raw Response of RsResult _ _ (SetKeyspaceResult _) -> return () - other -> throwM (UnexpectedResponse' other) + other -> throwM (UnexpectedResponse' other) query :: forall k a b m. (Tuple a, Tuple b, Show b, MonadIO m) => Connection @@ -312,7 +313,7 @@ res <- request c enc case parse (c^.settings.compression) res :: Response k a b of RsResult _ _ (RowsResult _ b) -> return b - other -> throwM (UnexpectedResponse' other) + other -> throwM (UnexpectedResponse' other) where params = QueryParams cons False p Nothing Nothing Nothing Nothing diff --git a/src/Database/CQL/IO/Connection/Settings.hs b/src/Database/CQL/IO/Connection/Settings.hs --- a/src/Database/CQL/IO/Connection/Settings.hs +++ b/src/Database/CQL/IO/Connection/Settings.hs @@ -37,6 +37,7 @@ import Data.HashMap.Strict (HashMap) import Data.Int import Database.CQL.Protocol +import Database.CQL.IO.Cluster.Host import Database.CQL.IO.Types import OpenSSL.Session (SSLContext) import Prelude diff --git a/src/Database/CQL/IO/Connection/Socket.hs b/src/Database/CQL/IO/Connection/Socket.hs --- a/src/Database/CQL/IO/Connection/Socket.hs +++ b/src/Database/CQL/IO/Connection/Socket.hs @@ -23,6 +23,7 @@ import Data.ByteString.Builder import Data.Maybe (isJust) import Data.Monoid +import Database.CQL.IO.Cluster.Host import Database.CQL.IO.Types import Foreign.C.Types (CInt (..)) import Network.Socket hiding (Stream, Socket, connect, close, recv, send, shutdown) diff --git a/src/Database/CQL/IO/Types.hs b/src/Database/CQL/IO/Types.hs --- a/src/Database/CQL/IO/Types.hs +++ b/src/Database/CQL/IO/Types.hs @@ -2,7 +2,6 @@ -- License, v. 2.0. If a copy of the MPL was not distributed with this -- file, You can obtain one at http://mozilla.org/MPL/2.0/. -{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} @@ -13,14 +12,13 @@ import Control.Monad.Catch import Data.Hashable -import Data.IP import Data.String import Data.Text (Text) import Data.Typeable import Data.Unique +import Data.UUID +import Database.CQL.IO.Cluster.Host import Database.CQL.Protocol -import Network.Socket (SockAddr (..), PortNumber) -import System.Logger.Message import qualified Data.Text.Lazy as Lazy @@ -39,44 +37,6 @@ hashWithSalt _ (ConnId u) = hashUnique u ----------------------------------------------------------------------------- --- InetAddr - -newtype InetAddr = InetAddr { sockAddr :: SockAddr } deriving (Eq, Ord) - -instance Show InetAddr where - show (InetAddr (SockAddrInet p a)) = - let i = fromIntegral p :: Int in - shows (fromHostAddress a) . showString ":" . shows i $ "" - show (InetAddr (SockAddrInet6 p _ a _)) = - let i = fromIntegral p :: Int in - shows (fromHostAddress6 a) . showString ":" . shows i $ "" - show (InetAddr (SockAddrUnix unix)) = unix -#if MIN_VERSION_network(2,6,1) && !MIN_VERSION_network(3,0,0) - show (InetAddr (SockAddrCan int32)) = show int32 -#endif - -instance ToBytes InetAddr where - bytes (InetAddr (SockAddrInet p a)) = - let i = fromIntegral p :: Int in - show (fromHostAddress a) +++ val ":" +++ i - bytes (InetAddr (SockAddrInet6 p _ a _)) = - let i = fromIntegral p :: Int in - show (fromHostAddress6 a) +++ val ":" +++ i - bytes (InetAddr (SockAddrUnix unix)) = bytes unix -#if MIN_VERSION_network(2,6,1) && !MIN_VERSION_network(3,0,0) - bytes (InetAddr (SockAddrCan int32)) = bytes int32 -#endif - -ip2inet :: PortNumber -> IP -> InetAddr -ip2inet p (IPv4 a) = InetAddr $ SockAddrInet p (toHostAddress a) -ip2inet p (IPv6 a) = InetAddr $ SockAddrInet6 p 0 (toHostAddress6 a) 0 - -inet2ip :: InetAddr -> IP -inet2ip (InetAddr (SockAddrInet _ a)) = IPv4 (fromHostAddress a) -inet2ip (InetAddr (SockAddrInet6 _ _ a _)) = IPv6 (fromHostAddress6 a) -inet2ip _ = error "inet2Ip: not IP4/IP6 address" - ------------------------------------------------------------------------------ -- InvalidSettings data InvalidSettings @@ -102,6 +62,25 @@ show (InternalError e) = "cql-io: internal error: " ++ show e ----------------------------------------------------------------------------- +-- ResponseError + +data ResponseError = ResponseError + { reHost :: !Host + , reTrace :: !(Maybe UUID) + , reWarn :: ![Text] + , reCause :: !Error + } deriving (Show, Typeable) + +instance Exception ResponseError + +toResponseError :: HostResponse k a b -> Maybe ResponseError +toResponseError (HostResponse h (RsError t w c)) = Just (ResponseError h t w c) +toResponseError _ = Nothing + +fromResponseError :: ResponseError -> HostResponse k a b +fromResponseError (ResponseError h t w c) = HostResponse h (RsError t w c) + +----------------------------------------------------------------------------- -- HostError data HostError @@ -147,8 +126,8 @@ data NoShow = NoShow deriving Show data UnexpectedResponse where - UnexpectedResponse :: !(Response k a b) -> UnexpectedResponse - UnexpectedResponse' :: Show b => !(Response k a b) -> UnexpectedResponse + UnexpectedResponse :: !(Response k a b) -> UnexpectedResponse + UnexpectedResponse' :: Show b => !(Response k a b) -> UnexpectedResponse deriving instance Typeable UnexpectedResponse instance Exception UnexpectedResponse @@ -156,8 +135,8 @@ instance Show UnexpectedResponse where show x = showString "cql-io: unexpected response: " . case x of - UnexpectedResponse r -> shows (f r) - UnexpectedResponse' r -> shows r + UnexpectedResponse r -> shows (f r) + UnexpectedResponse' r -> shows r $ "" where f :: Response k a b -> Response k a NoShow