diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,8 @@ +0.10.0 +----------------------------------------------------------------------------- +- Add `MonadClient` type-class. +- Add retry settings. + 0.9.7 ----------------------------------------------------------------------------- - Bugfix release. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -14,3 +14,6 @@ **Support for connection streams**. Requests can be multiplexed over a few connections. + +**Customisable retry settings**. Support for default retry settings as well +as local overrides per query. 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: 0.9.7 +version: 0.10.1 synopsis: Cassandra CQL client. stability: experimental license: OtherLicense @@ -21,15 +21,18 @@ which implements Cassandra's CQL protocol and complements it with the neccessary I/O operations. The feature-set includes: . - * __Node discovery__. The driver discovers nodes automatically from a small + * /Node discovery/. The driver discovers nodes automatically from a small set of bootstrap nodes. . - * __Customisable load-balancing policies__. In addition to pre-built LB + * /Customisable load-balancing policies/. In addition to pre-built LB policies such as round-robin, users of this library can provide their own policies if desired. . - * __Support for connection streams__. Requests can be multiplexed over a + * /Support for connection streams/. Requests can be multiplexed over a few connections. + . + * /Customisable retry settings/. Support for default retry settings as well + as local overrides per query. source-repository head type: git @@ -45,6 +48,7 @@ Database.CQL.IO other-modules: + Control.Retry Database.CQL.IO.Client Database.CQL.IO.Cluster.Discovery Database.CQL.IO.Cluster.Host @@ -62,25 +66,25 @@ Database.CQL.IO.Types build-depends: - async == 2.0.* - , auto-update == 0.1.* - , base >= 4.7 && < 5.0 - , bytestring >= 0.10 && < 1.0 - , containers >= 0.5 && < 1.0 - , cql == 3.0.* - , exceptions >= 0.4 && < 1.0 - , hashable >= 1.2 && < 2.0 - , iproute >= 1.3 && < 1.4 - , lens >= 4.4 && < 4.5 - , mtl >= 2.1 && < 2.3 - , mwc-random >= 0.13 && < 0.14 - , network >= 2.4 && < 3.0 - , retry >= 0.5 && < 0.6 - , semigroups >= 0.15 && < 0.16 - , stm >= 2.4 && < 3.0 - , text >= 0.11 && < 2.0 - , tinylog >= 0.8 && < 0.11 - , time >= 1.4 && < 2.0 - , transformers >= 0.3 && < 0.5 - , uuid >= 1.2.6 && < 2.0 - , vector >= 0.10 && < 1.0 + async == 2.0.* + , auto-update == 0.1.* + , base >= 4.7 && < 5.0 + , bytestring >= 0.10 && < 1.0 + , containers >= 0.5 && < 1.0 + , cql == 3.0.* + , data-default-class + , exceptions >= 0.4 && < 1.0 + , hashable >= 1.2 && < 2.0 + , iproute >= 1.3 && < 1.4 + , lens >= 4.4 && < 4.5 + , mtl >= 2.1 && < 2.3 + , mwc-random >= 0.13 && < 0.14 + , network >= 2.4 && < 3.0 + , semigroups >= 0.15 && < 0.16 + , stm >= 2.4 && < 3.0 + , text >= 0.11 && < 2.0 + , tinylog >= 0.8 && < 0.11 + , time >= 1.4 && < 2.0 + , transformers >= 0.3 && < 0.5 + , uuid >= 1.2.6 && < 2.0 + , vector >= 0.10 && < 1.0 diff --git a/src/Control/Retry.hs b/src/Control/Retry.hs new file mode 100644 --- /dev/null +++ b/src/Control/Retry.hs @@ -0,0 +1,287 @@ +-- This is a copy of Ozgun Ataman's retry module version 0.5.1 +-- available from http://hackage.haskell.org/package/retry +-- +-- It includes changes introduced by pull request 17 in +-- https://github.com/Soostone/retry/pull/17 +-- +-- If PR 17 is merged into upstream it can be removed again. + +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Retry +-- Copyright : Ozgun Ataman +-- License : BSD3 +-- +-- Maintainer : Ozgun Ataman +-- Stability : provisional +-- +-- This module exposes combinators that can wrap arbitrary monadic +-- actions. They run the action and potentially retry running it with +-- some configurable delay for a configurable number of times. +-- +-- The express purpose of this library is to make it easier to work +-- with IO and especially network IO actions that often experience +-- temporary failure that warrant retrying of the original action. For +-- example, a database query may time out for a while, in which case +-- we should delay a bit and retry the query. +---------------------------------------------------------------------------- + + +module Control.Retry + ( + -- * High Level Operation + RetryPolicy (..) + + , retrying + , recovering + , recoverAll + + -- * Retry Policies + , constantDelay + , exponentialBackoff + , fibonacciBackoff + , limitRetries + , capDelay + + -- * Re-export from Data.Monoid + + , (<>) + + ) where + +------------------------------------------------------------------------------- +import Control.Concurrent +import Control.Monad.Catch +import Control.Monad.IO.Class +import Data.Default.Class +import Data.Monoid +------------------------------------------------------------------------------- + + +------------------------------------------------------------------------------- +-- | A 'RetryPolicy' is a function that takes an iteration number and +-- possibly returns a delay in miliseconds. *Nothing* implies we have +-- reached the retry limit. +-- +-- Please note that 'RetryPolicy' is a 'Monoid'. You can collapse +-- multiple strategies into one using 'mappend' or '<>'. The semantics +-- of this combination are as follows: +-- +-- 1. If either policy returns 'Nothing', the combined policy returns +-- 'Nothing'. This can be used to @inhibit@ after a number of retries, +-- for example. +-- +-- 2. If both policies return a delay, the larger delay will be used. +-- This is quite natural when combining multiple policies to achieve a +-- certain effect. +-- +-- Example: +-- +-- One can easily define an exponential backoff policy with a limited +-- number of retries: +-- +-- >> limitedBackoff = exponentialBackoff 50 <> limitedRetries 5 +-- +-- Naturally, 'mempty' will retry immediately (delay 0) for an +-- unlimited number of retries, forming the identity for the 'Monoid'. +-- +-- The default under 'def' implements a constant 50ms delay, up to 5 times: +-- +-- >> def = constantDelay 50000 <> limitRetries 5 +-- +-- For anything more complex, just define your own 'RetryPolicy': +-- +-- >> myPolicy = RetryPolicy $ \ n -> if n > 10 then Just 1000 else Just 10000 +newtype RetryPolicy = RetryPolicy { getRetryPolicy :: Int -> Maybe Int } + + +instance Default RetryPolicy where + def = constantDelay 50000 <> limitRetries 5 + +instance Monoid RetryPolicy where + mempty = RetryPolicy $ (const (Just 0)) + (RetryPolicy a) `mappend` (RetryPolicy b) = RetryPolicy $ \ n -> do + a' <- a n + b' <- b n + return $! max a' b' + + +------------------------------------------------------------------------------- +-- | Retry immediately, but only up to @n@ times. +limitRetries + :: Int + -- ^ Maximum number of retries. + -> RetryPolicy +limitRetries i = RetryPolicy $ \ n -> if n >= i then Nothing else (Just 0) + + +------------------------------------------------------------------------------- +-- | Implement a constant delay with unlimited retries. +constantDelay + :: Int + -- ^ Base delay in microseconds + -> RetryPolicy +constantDelay delay = RetryPolicy (const (Just delay)) + + +------------------------------------------------------------------------------- +-- | Grow delay exponentially each iteration. +exponentialBackoff + :: Int + -- ^ Base delay in microseconds + -> RetryPolicy +exponentialBackoff base = RetryPolicy $ \ n -> Just (2^n * base) + + +------------------------------------------------------------------------------- +-- | Implement Fibonacci backoff. +fibonacciBackoff + :: Int + -- ^ Base delay in microseconds + -> RetryPolicy +fibonacciBackoff base = RetryPolicy $ \ n -> Just $ fib (n + 1) (0, base) + where + fib 0 (a, _) = a + fib !m (!a, !b) = fib (m-1) (b, a + b) + + +------------------------------------------------------------------------------- +-- | Set an upperbound for any delays that may be directed by the +-- given policy. +capDelay + :: Int + -- ^ A maximum delay in microseconds + -> RetryPolicy + -> RetryPolicy +capDelay limit p = RetryPolicy $ \ n -> min limit `fmap` (getRetryPolicy p) n + + +------------------------------------------------------------------------------- +-- | Retry combinator for actions that don't raise exceptions, but +-- signal in their type the outcome has failed. Examples are the +-- 'Maybe', 'Either' and 'EitherT' monads. +-- +-- Let's write a function that always fails and watch this combinator +-- retry it 5 additional times following the initial run: +-- +-- >>> import Data.Maybe +-- >>> let f = putStrLn "Running action" >> return Nothing +-- >>> retrying def isNothing f +-- Running action +-- Running action +-- Running action +-- Running action +-- Running action +-- Running action +-- Nothing +-- +-- Note how the latest failing result is returned after all retries +-- have been exhausted. +retrying :: MonadIO m + => RetryPolicy + -> (Int -> b -> m Bool) + -- ^ An action to check whether the result should be retried. + -- If True, we delay and retry the operation. + -> (Int -> m b) + -- ^ Action to run + -> m b +retrying (RetryPolicy policy) chk f = go 0 + where + go n = do + res <- f n + chk' <- chk n res + case chk' of + True -> + case (policy n) of + Just delay -> do + liftIO (threadDelay delay) + go $! n+1 + Nothing -> return res + False -> return res + + + +------------------------------------------------------------------------------- +-- | Retry ALL exceptions that may be raised. To be used with caution; +-- this matches the exception on 'SomeException'. +-- +-- See how the action below is run once and retried 5 more times +-- before finally failing for good: +-- +-- >>> let f = putStrLn "Running action" >> error "this is an error" +-- >>> recoverAll def f +-- Running action +-- Running action +-- Running action +-- Running action +-- Running action +-- Running action +-- *** Exception: this is an error +recoverAll :: (MonadIO m, MonadCatch m) + => RetryPolicy + -> (Int -> m a) + -> m a +recoverAll set f = recovering set [h] f + where + h _ = Handler $ \ (_ :: SomeException) -> return True + + +------------------------------------------------------------------------------- +-- | Run an action and recover from a raised exception by potentially +-- retrying the action a number of times. +recovering :: forall m a. (MonadIO m, MonadCatch m) + => RetryPolicy + -- ^ Just use 'def' faor default settings + -> [(Int -> Handler m Bool)] + -- ^ Should a given exception be retried? Action will be + -- retried if this returns True. + -> (Int -> m a) + -- ^ Action to perform + -> m a +recovering (RetryPolicy policy) hs f = go 0 + where + + -- | Convert a (e -> m Bool) handler into (e -> m a) so it can + -- be wired into the 'catches' combinator. + transHandler :: Int -> Handler m Bool -> Handler m a + transHandler n (Handler h) = Handler $ \ e -> do + chk <- h e + case chk of + True -> + case policy n of + Just delay -> do + liftIO (threadDelay delay) + go $! n+1 + Nothing -> throwM e + False -> throwM e + + go n = f n `catches` map (transHandler n . ($ n)) hs + + + + + ------------------ + -- Simple Tests -- + ------------------ + + + +-- data TestException = TestException deriving (Show, Typeable) +-- data AnotherException = AnotherException deriving (Show, Typeable) + +-- instance Exception TestException +-- instance Exception AnotherException + + +-- test = retrying def [h1,h2] f +-- where +-- f = putStrLn "Running action" >> throwM AnotherException +-- h1 = Handler $ \ (e :: TestException) -> return False +-- h2 = Handler $ \ (e :: AnotherException) -> return True 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 @@ -47,17 +47,33 @@ , setProtocolVersion , setResponseTimeout , setSendTimeout + , setRetrySettings + -- ** Retry Settings + , RetrySettings + , noRetry + , retryForever + , maxRetries + , adjustConsistency + , constDelay + , expBackoff + , fibBackoff + , adjustSendTimeout + , adjustResponseTimeout + -- * Client monad , Client + , MonadClient (..) , ClientState - , DebugInfo (..) + , DebugInfo (..) , init , runClient + , retry , shutdown , debugInfo , query + , query1 , write , schema , batch @@ -95,7 +111,7 @@ import Control.Applicative import Control.Monad.Catch import Control.Monad (void) -import Data.Maybe (isJust) +import Data.Maybe (isJust, listToMaybe) import Database.CQL.Protocol import Database.CQL.IO.Client import Database.CQL.IO.Cluster.Host @@ -104,7 +120,7 @@ import Database.CQL.IO.Types import Prelude hiding (init) -runQuery :: (Tuple a, Tuple b) => QueryString k a b -> QueryParams a -> Client (Response k a b) +runQuery :: (MonadClient m, Tuple a, Tuple b) => QueryString k a b -> QueryParams a -> m (Response k a b) runQuery q p = do r <- request (RqQuery (Query q p)) case r of @@ -112,19 +128,23 @@ _ -> return r -- | Run a CQL read-only query against a Cassandra node. -query :: (Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> Client [b] +query :: (MonadClient m, Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> m [b] query q p = do r <- runQuery q p case r of RsResult _ (RowsResult _ b) -> return b _ -> throwM UnexpectedResponse +-- | Run a CQL read-only query against a Cassandra node. +query1 :: (MonadClient m, Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> m (Maybe b) +query1 q p = listToMaybe <$> query q p + -- | Run a CQL insert/update query against a Cassandra node. -write :: Tuple a => QueryString W a () -> QueryParams a -> Client () +write :: (MonadClient m, Tuple a) => QueryString W a () -> QueryParams a -> m () write q p = void $ runQuery q p -- | Run a CQL schema query against a Cassandra node. -schema :: Tuple a => QueryString S a () -> QueryParams a -> Client (Maybe SchemaChange) +schema :: (MonadClient m, Tuple a) => QueryString S a () -> QueryParams a -> m (Maybe SchemaChange) schema x y = do r <- runQuery x y case r of @@ -133,7 +153,7 @@ _ -> throwM UnexpectedResponse -- | Run a batch query against a Cassandra node. -batch :: Batch -> Client () +batch :: MonadClient m => Batch -> m () batch b = command (RqBatch b) -- | Return value of 'paginate'. Contains the actual result values as well @@ -158,7 +178,7 @@ -- Please note that -- as of Cassandra 2.1.0 -- if your requested page size -- is equal to the result size, 'hasMore' might be true and a subsequent -- 'nextPage' will return an empty list in 'result'. -paginate :: (Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> Client (Page b) +paginate :: (MonadClient m, Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> m (Page b) paginate q p = do let p' = p { pageSize = pageSize p <|> Just 10000 } r <- runQuery q p' 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 @@ -9,13 +9,15 @@ module Database.CQL.IO.Client ( Client + , MonadClient (..) , ClientState - , DebugInfo (..) + , DebugInfo (..) , runClient , Database.CQL.IO.Client.init , shutdown , request , command + , retry , debugInfo ) where @@ -52,6 +54,7 @@ import Network.Socket (SockAddr (..), PortNumber) import System.Logger.Class hiding (Settings, new, settings, create) +import qualified Control.Monad.Reader as Reader import qualified Data.List.NonEmpty as NE import qualified Data.Map.Strict as Map import qualified Database.CQL.IO.Connection as C @@ -71,10 +74,10 @@ } data Context = Context - { _settings :: Settings - , _logger :: Logger - , _timeouts :: TimeoutManager - , _sigMonit :: Signal HostEvent + { _settings :: Settings + , _logger :: Logger + , _timeouts :: TimeoutManager + , _sigMonit :: Signal HostEvent } -- | Opaque client state/environment. @@ -115,6 +118,18 @@ instance MonadLogger Client where log l m = view (context.logger) >>= \g -> Logger.log g l m +-- | Monads in which 'Client' actions may be embedded. +class (Functor m, Applicative m, Monad m, MonadIO m, MonadCatch m) => MonadClient m + where + -- | Lift a computation to the 'Client' monad. + liftClient :: Client a -> m a + -- | Execute an action with a modified 'ClientState'. + localState :: (ClientState -> ClientState) -> m a -> m a + +instance MonadClient Client where + liftClient = id + localState = Reader.local + ----------------------------------------------------------------------------- -- API @@ -122,6 +137,10 @@ runClient :: MonadIO m => ClientState -> Client a -> m a runClient p a = liftIO $ runReaderT (client a) p +-- | Use given 'RetrySettings' during execution of some client action. +retry :: MonadClient m => RetrySettings -> m a -> m a +retry r = localState (set (context.settings.retrySettings) r) + -- | Send a CQL 'Request' to the server and return a 'Response'. -- -- This function will first ask the clients load-balancing 'Policy' for @@ -131,49 +150,81 @@ -- If all available hosts are busy (i.e. their connection pools are fully -- utilised), the function will block until a connection becomes available -- or the maximum wait-queue length has been reached. -request :: (Tuple a, Tuple b) => Request k a b -> Client (Response k a b) -request a = do +request :: (MonadClient m, Tuple a, Tuple b) => Request k a b -> m (Response k a b) +request a = liftClient $ do s <- ask n <- liftIO $ hostCount (s^.policy) - start s n + recovering (s^.context.settings.retrySettings.retryPolicy) recoverFrom $ \i -> + if i == 0 then + getResponse a s n + else let s' = adjust s in + getResponse (newRequest s) s' n where - start s n = do - h <- pickHost (s^.policy) - p <- Map.lookup h <$> readTVarIO' (s^.hostmap) - case p of - Just x -> action s n x `catches` handlers h - Nothing -> do - err $ msg (val "no pool for host " +++ h) - p' <- mkPool (s^.context) (h^.hostAddr) - atomically' $ modifyTVar' (s^.hostmap) (Map.alter (maybe (Just p') Just) h) - start s n + adjust s = + let x = s^.context.settings.retrySettings.sendTimeoutChange + y = s^.context.settings.retrySettings.recvTimeoutChange + in over (context.settings.connSettings.sendTimeout) (+ x) + . over (context.settings.connSettings.responseTimeout) (+ y) + $ s - action s n p = do - res <- tryWith p (go s) + newRequest s = + case s^.context.settings.retrySettings.reducedConsistency of + Nothing -> a + Just c -> + case a of + RqQuery (Query q p) -> RqQuery (Query q p { consistency = c }) + RqBatch b -> RqBatch b { batchConsistency = c } + _ -> a + + recoverFrom = + [ const $ Handler $ \e -> case e of + ReadTimeout {} -> return True + WriteTimeout {} -> return True + Overloaded {} -> return True + Unavailable {} -> return True + ServerError {} -> return True + _ -> return False + , const $ Handler $ \(_ :: ConnectionError) -> return True + ] + +getResponse :: (Tuple a, Tuple b) => Request k a b -> ClientState -> Word -> Client (Response k a b) +getResponse a s n = do + h <- pickHost (s^.policy) + p <- Map.lookup h <$> readTVarIO' (s^.hostmap) + case p of + Just x -> action x `catches` handlers h + Nothing -> do + err $ msg (val "no pool for host " +++ h) + p' <- mkPool (s^.context) (h^.hostAddr) + atomically' $ modifyTVar' (s^.hostmap) (Map.alter (maybe (Just p') Just) h) + getResponse a s n + where + action p = do + res <- tryWith p go case res of Just r -> return r Nothing -> if n > 0 then - start s (n - 1) + getResponse a s (n - 1) else case s^.context.settings.maxWaitQueue of - Nothing -> with p (transaction s) - Just q -> retry s q p + Nothing -> with p transaction + Just q -> again q p - go s h = do - atomically $ modifyTVar' (s^.failures) $ \n -> if n > 0 then n - 1 else 0 - transaction s h + go h = do + atomically $ modifyTVar' (s^.failures) $ \x -> if x > 0 then x - 1 else 0 + transaction h - retry s q p = do - n <- atomically' $ do - n <- readTVar (s^.failures) - unless (n == q) $ - writeTVar (s^.failures) (n + 1) - return n - unless (n < q) $ + again q p = do + k <- atomically' $ do + k <- readTVar (s^.failures) + unless (k == q) $ + writeTVar (s^.failures) (k + 1) + return k + unless (k < q) $ throwM HostsBusy - with p (go s) + with p go - transaction s c = do + transaction c = do let x = s^.context.settings.connSettings.compression let v = s^.context.settings.protoVersion r <- parse x <$> C.request c (serialise v x a) @@ -187,7 +238,7 @@ pickHost p = maybe (throwM NoHostAvailable) return =<< liftIO (select p) -- | Like 'request' but not returning any result. -command :: Request k () () -> Client () +command :: MonadClient m => Request k () () -> m () command = void . request data DebugInfo = DebugInfo @@ -205,8 +256,8 @@ . shows (policyInfo dbg) $ "" -debugInfo :: Client DebugInfo -debugInfo = do +debugInfo :: MonadClient m => m DebugInfo +debugInfo = liftClient $ do hosts <- Map.keys <$> (readTVarIO' =<< view hostmap) pols <- liftIO . display =<< view policy jbs <- Jobs.showJobs =<< view jobs @@ -353,7 +404,7 @@ else return Nothing maybe (liftIO . ignore . onEvent (e^.policy) $ HostDown (h^.hostAddr)) - (liftIO . void . async . recovering reconnectPolicy reconnectHandlers . continue e) + (liftIO . void . async . recovering reconnectPolicy reconnectHandlers . const . continue e) c Jobs.add (e^.jobs) (h^.hostAddr) $ monitor (e^.context) 0 30000000 h diff --git a/src/Database/CQL/IO/Settings.hs b/src/Database/CQL/IO/Settings.hs --- a/src/Database/CQL/IO/Settings.hs +++ b/src/Database/CQL/IO/Settings.hs @@ -8,7 +8,9 @@ module Database.CQL.IO.Settings where import Control.Lens hiding ((<|)) +import Control.Retry import Data.List.NonEmpty (NonEmpty (..), (<|)) +import Data.Monoid import Data.Time import Data.Word import Database.CQL.Protocol @@ -19,9 +21,17 @@ import Database.CQL.IO.Types (Milliseconds (..)) import Network.Socket (PortNumber (..)) +data RetrySettings = RetrySettings + { _retryPolicy :: RetryPolicy + , _reducedConsistency :: Maybe Consistency + , _sendTimeoutChange :: Milliseconds + , _recvTimeoutChange :: Milliseconds + } + data Settings = Settings { _poolSettings :: PoolSettings , _connSettings :: ConnectionSettings + , _retrySettings :: RetrySettings , _protoVersion :: Version , _portnumber :: PortNumber , _contacts :: NonEmpty String @@ -29,6 +39,7 @@ , _policyMaker :: IO Policy } +makeLenses ''RetrySettings makeLenses ''Settings -- | Default settings: @@ -51,10 +62,13 @@ -- * no compression is applied to frame bodies -- -- * no default keyspace is used. +-- +-- * no retries are done defSettings :: Settings defSettings = Settings P.defSettings C.defSettings + noRetry V3 (fromInteger 9042) ("localhost" :| []) @@ -108,7 +122,7 @@ -- number of CPU cores this codes is running on. setPoolStripes :: Int -> Settings -> Settings setPoolStripes v s - | v < 1 = error "Database.CQL.IO.Settings: stripes must be greater than 0" + | v < 1 = error "cql-io settings: stripes must be greater than 0" | otherwise = set (poolSettings.poolStripes) v s -- | When receiving a response times out, we can no longer use the stream of the @@ -131,8 +145,8 @@ -- to 32768 streams. setMaxStreams :: Int -> Settings -> Settings setMaxStreams v s = case s^.protoVersion of - V2 | v < 1 || v > 128 -> error "Database.CQL.IO.Settings: max. streams must be within [1, 128]" - V3 | v < 1 || v > 32768 -> error "Database.CQL.IO.Settings: max. streams must be within [1, 32768]" + V2 | v < 1 || v > 128 -> error "cql-io settings: max. streams must be within [1, 128]" + V3 | v < 1 || v > 32768 -> error "cql-io settings: max. streams must be within [1, 32768]" _ -> set (connSettings.maxStreams) v s -- | Set the connect timeout of a connection. @@ -154,3 +168,64 @@ -- initialised to use this keyspace. setKeyspace :: Keyspace -> Settings -> Settings setKeyspace v = set (connSettings.defKeyspace) (Just v) + +-- | Set default retry settings to use. +setRetrySettings :: RetrySettings -> Settings -> Settings +setRetrySettings v = set retrySettings v + +----------------------------------------------------------------------------- +-- Retry Settings + +-- | Never retry. +noRetry :: RetrySettings +noRetry = RetrySettings (RetryPolicy $ const Nothing) Nothing 0 0 + +-- | Forever retry immediately. +retryForever :: RetrySettings +retryForever = RetrySettings mempty Nothing 0 0 + +-- | Limit number of retries. +maxRetries :: Word -> RetrySettings -> RetrySettings +maxRetries v = over retryPolicy (mappend (limitRetries $ fromIntegral v)) + +-- | When retrying a (batch-) query, change consistency to the given value. +adjustConsistency :: Consistency -> RetrySettings -> RetrySettings +adjustConsistency v = set reducedConsistency (Just v) + +-- | Wait a constant time between retries. +constDelay :: NominalDiffTime -> RetrySettings -> RetrySettings +constDelay v = setDelayFn constantDelay v v + +-- | Delay retries with exponential backoff. +expBackoff :: NominalDiffTime + -- ^ Initial delay. + -> NominalDiffTime + -- ^ Maximum delay. + -> RetrySettings + -> RetrySettings +expBackoff = setDelayFn exponentialBackoff + +-- | Delay retries using Fibonacci sequence as backoff. +fibBackoff :: NominalDiffTime + -- ^ Initial delay. + -> NominalDiffTime + -- ^ Maximum delay. + -> RetrySettings + -> RetrySettings +fibBackoff = setDelayFn fibonacciBackoff + +-- | On retry adjust the send timeout. +adjustSendTimeout :: NominalDiffTime -> RetrySettings -> RetrySettings +adjustSendTimeout v = set sendTimeoutChange (Ms $ round (1000 * v)) + +-- | On retry adjust the response timeout. +adjustResponseTimeout :: NominalDiffTime -> RetrySettings -> RetrySettings +adjustResponseTimeout v = set recvTimeoutChange (Ms $ round (1000 * v)) + +setDelayFn :: (Int -> RetryPolicy) + -> NominalDiffTime + -> NominalDiffTime + -> RetrySettings + -> RetrySettings +setDelayFn d v w = over retryPolicy + (mappend $ capDelay (round (1000000 * w)) $ d (round (1000000 * v)))