diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,74 @@ +# Changelog for glue-core + +## 0.8.0.0 + +### Breaking changes + +- Options records for time-dependent combinators take the monad as a type parameter and carry a + `Clock m` defaulting to `systemClock`: `TimeoutOptions m`, `RetryOptions m a`, + `HedgingOptions m`, `BatchingOptions m`, `DogpileOptions m`, `ShutdownOptions m b` and + `PreloadedOptions` (new `preloadingClock` field). Code that updates fields of the defaults keeps + compiling; when an update replaces every field that mentions `m` (for example only the clock), + annotate the default, e.g. `(defaultTimeoutOptions :: TimeoutOptions IO) { timeoutClock = c }` (#13). +- `Glue.CircuitBreaker` counting is fixed and its semantics match the documentation (#5): + - Calls cancelled by an asynchronous exception (timeouts, losing hedges, killed threads) no + longer count as failures; the exception is still rethrown. + - `maxBreakerFailures` is replaced by `breakerFailureThreshold`, which opens on that failure + (the default 3 opens on the third; previously the breaker opened on failure `max + 1`). + - Failures are consecutive by default (a call that answers resets the count; previously the + count never reset), or counted within `breakerFailureWindowMs` when set. + - `resetTimeoutSecs` is replaced by `breakerOpenDurationMs`, how long the breaker stays open, + with millisecond resolution on the monotonic clock rather than wall-clock seconds. +- `CircuitBreakerOptions` and `CircuitBreakerState` take the monad (and options the failure type) + as type parameters. +- `RetryOptions m a failure`: `retryInitialWaitTimeMs` and `retryWaitTimeMultiplier` are replaced + by `retryBackoff` (`ConstantBackoff`, `LinearBackoff` or `ExponentialBackoff`) (#7). + +### Additions + +- Circuit breaker failure classification with `countsAsFailure`, a phase observer + `onBreakerTransition`, `circuitBreakerPhases`, and a pluggable `BreakerStore` (admit / settle / + read phase, with an atomic half-open trial) so state can be shared between instances, with + `newInMemoryBreakerStore` as the default (#9). +- `circuitBreakerEither` and `circuitBreakerEitherWithStore` for services returning failures as + values (#6). +- Retry decided by the failure (`retryOn`), with a delay ceiling (`retryMaxDelayMs`), full or equal + jitter from an injectable random source (`retryJitter`, `retryRandom`), a deadline after which no + retry starts (`retryRemainingMs`), an attempt observer (`onRetry`) and `retryingServiceEither` (#7). +- `failoverAcross` tries an ordered list of distinct backends, with `advanceOn` deciding which + failures move on and which stop the chain, a choice of reporting the first or last failure, an + indexed variant reporting which backend answered, and `Either` variants (#8). +- `Glue.Race`: a low-level race of attempts (`withRace`, `launch`, `awaitNext` with an optional + timeout, `abandon`, `abandonAll`) for callers driving their own launch decisions. Simultaneous + endings are reported earliest index first, and abandoning waits until the attempt has finished, + reporting how it really ended (#10). +- `hedgeAcross` hedges across distinct backends on top of `Glue.Race`: `admitLaunch` lets the + caller decline a backup without failing the primary, `onAttemptEnded` reports every attempt + (including cancelled losers), an early failure launches the next backend at once, and + `hedgeAcrossEither` supports failures as values. `hedgingService` is now built on it (#10). +- `weightedBy`: a deterministic weighted load balancer. Weights are basis points validated up + front (`Either WeightError`), and each request's backend is chosen from a caller-supplied stable + key by SipHash-2-4 under a `Salt`, so a key reaches the same backend on every instance and across + restarts and different salts split independently. `bucketOf` and `chooseIndex` expose the choice + without calling a backend (#11). +- Keyed concurrency limits: `limitConcurrencyBy` keeps one limit per key and refuses a full key + at once with `Saturated` as a value, releasing on every exit path (including cancellation) and + dropping idle keys. `withSlot` holds a slot for work that outlives the service call, and an + `AdmissionStore` (in-memory by default) is the extension point for limits shared across + instances. `limitConcurrencyEither` and `limitConcurrencyByEither` return refusals as the + caller's failure value (#12). +- `Glue.Periodic`: `periodically` runs a background pass every `periodMs` measured from the end of + the previous pass (optionally immediately and with jitter), reports failed passes without ending + the loop, and returns a handle with `runNow` (queued behind a running pass) and a graceful + `stopPeriodic` (#14). +- `addTimeoutEither` and `gracefulShutdownEither` return their refusal as a value for services + whose failures are values (#6). + +## 0.7.0.1 + +- PVP upper bounds on all dependencies. + +## 0.7.0 + +- Modernised for GHC 9.6 to 9.12, `MonadUnliftIO` throughout, new combinators (hedging, + concurrency limits, load balancing, graceful shutdown) and a Nix flake. diff --git a/glue-core.cabal b/glue-core.cabal --- a/glue-core.cabal +++ b/glue-core.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: glue-core -version: 0.7.0.1 +version: 0.8.0.0 synopsis: Make better services and clients. description: Combinator library to enhance the general functionality of services and clients. license: BSD-3-Clause @@ -9,6 +9,7 @@ maintainer: github@futurenotfound.com category: Network build-type: Simple +extra-doc-files: CHANGELOG.md tested-with: GHC == 9.6.7, GHC == 9.8.4, GHC == 9.10.3, GHC == 9.12.3 source-repository head @@ -39,16 +40,20 @@ Glue.Failover Glue.Hedging Glue.LoadBalancer + Glue.Periodic Glue.Preload + Glue.Race Glue.Retry Glue.Shutdown Glue.Switching Glue.Timeout build-depends: , base >= 4.14 && < 5 - , glue-common ^>= 0.7.0 + , bytestring >= 0.10 && < 0.13 + , containers >= 0.6 && < 0.8 + , glue-common ^>= 0.8.0 , hashable >= 1.3 && < 1.6 - , time >= 1.9 && < 1.15 + , random >= 1.2 && < 1.4 , unliftio >= 0.2 && < 0.3 , unordered-containers >= 0.2 && < 0.3 @@ -66,11 +71,13 @@ Glue.FailoverSpec Glue.HedgingSpec Glue.LoadBalancerSpec + Glue.PeriodicSpec Glue.PreloadSpec Glue.RetrySpec Glue.ShutdownSpec Glue.SwitchingSpec Glue.TimeoutSpec + Glue.WeightedLoadBalancerSpec Spec build-tool-depends: hspec-discover:hspec-discover ghc-options: @@ -79,13 +86,14 @@ -with-rtsopts=-N build-depends: , async >= 2.2 && < 2.3 + , bytestring >= 0.10 && < 0.13 + , containers >= 0.6 && < 0.8 , base >= 4.14 && < 5 - , glue-common ^>= 0.7.0 + , glue-common ^>= 0.8.0 , glue-core , hashable >= 1.3 && < 1.6 , hspec >= 2.7 && < 2.12 , QuickCheck >= 2.14 && < 2.16 , quickcheck-instances >= 0.3 && < 0.4 - , time >= 1.9 && < 1.15 , unliftio >= 0.2 && < 0.3 , unordered-containers >= 0.2 && < 0.3 diff --git a/src/Glue/Batcher.hs b/src/Glue/Batcher.hs --- a/src/Glue/Batcher.hs +++ b/src/Glue/Batcher.hs @@ -13,24 +13,27 @@ import Data.Hashable (Hashable) import qualified Data.HashMap.Strict as M import qualified Data.HashSet as S +import Glue.Clock import Glue.Types import UnliftIO (MonadIO, MonadUnliftIO, SomeException) -import UnliftIO.Concurrent (forkIO, threadDelay) +import UnliftIO.Concurrent (forkIO) import UnliftIO.IORef (IORef, atomicModifyIORef', newIORef) import UnliftIO.MVar (newEmptyMVar, putMVar) -- | Options for configuring the batching. -newtype BatchingOptions = BatchingOptions +data BatchingOptions m = BatchingOptions { batchWindowMs :: Int -- ^ Window in milliseconds over which to batch. + , batchClock :: Clock m + -- ^ Clock used to wait out the batching window. } - deriving (Eq, Show) -- | Default instance of t'BatchingOptions' with a batching window of 10ms. -defaultBatchingOptions :: BatchingOptions +defaultBatchingOptions :: MonadUnliftIO m => BatchingOptions m defaultBatchingOptions = BatchingOptions { batchWindowMs = 10 + , batchClock = systemClock } -- | Cumulative request batch of all the outstanding calls. @@ -55,14 +58,14 @@ result <- makeCall service requests traverse_ (applyToPending result) pendings -startBatch :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> m () +startBatch :: (Hashable a, MonadUnliftIO m) => BatchingOptions m -> MultiGetService m a b -> IORef (RequestBatch a b) -> m () startBatch options service ref = void $ forkIO $ do - threadDelay (1000 * batchWindowMs options) + clockSleep (batchClock options) (1000 * batchWindowMs options) pendings <- atomicModifyIORef' ref (\p -> (emptyBatch, p)) processCalls service pendings -addPending :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> PendingRequest a b -> m () +addPending :: (Hashable a, MonadUnliftIO m) => BatchingOptions m -> MultiGetService m a b -> IORef (RequestBatch a b) -> PendingRequest a b -> m () addPending options service ref p@(SingleRequest request _) = join $ atomicModifyIORef' @@ -74,14 +77,14 @@ ref (\(RequestBatch ps rs) -> (RequestBatch (p : ps) (S.union rs requests), if null ps then startBatch options service ref else pure ())) -singleService :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> BasicService m a (Maybe b) +singleService :: (Hashable a, MonadUnliftIO m) => BatchingOptions m -> MultiGetService m a b -> IORef (RequestBatch a b) -> BasicService m a (Maybe b) singleService options service ref request = do mvar <- newEmptyMVar let pending = SingleRequest request mvar addPending options service ref pending getResult mvar -multiService :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> MultiGetService m a b +multiService :: (Hashable a, MonadUnliftIO m) => BatchingOptions m -> MultiGetService m a b -> IORef (RequestBatch a b) -> MultiGetService m a b multiService options service ref requests = do mvar <- newEmptyMVar let pending = MultiRequest requests mvar @@ -91,7 +94,7 @@ -- | Function for constructing a batching service. batchingService :: (Hashable a, MonadUnliftIO m, MonadIO n) - => BatchingOptions + => BatchingOptions m -- ^ Options to configure the batched service. -> MultiGetService m a b -- ^ Service around which to batch requests. diff --git a/src/Glue/CircuitBreaker.hs b/src/Glue/CircuitBreaker.hs --- a/src/Glue/CircuitBreaker.hs +++ b/src/Glue/CircuitBreaker.hs @@ -6,53 +6,186 @@ -- to make unsuccessful calls. -- Often this is useful if the underlying service were to repeatedly time out, so as to -- reduce the number of calls inflight holding up upstream callers. +-- +-- The breaker is closed until 'breakerFailureThreshold' counted failures have occurred (consecutive +-- failures by default, or failures within 'breakerFailureWindowMs'), then open for +-- 'breakerOpenDurationMs', refusing calls. After that the next call is admitted as the single +-- half-open trial: if it answers the breaker closes, if it fails the breaker opens again. +-- +-- Only synchronous failures accepted by 'countsAsFailure' count. A call cancelled by an +-- asynchronous exception (a timeout, a losing hedge, a killed request thread) counts for nothing, +-- and the exception is rethrown. module Glue.CircuitBreaker ( CircuitBreakerOptions (..) + , defaultCircuitBreakerOptions + , BreakerPhase (..) + , BreakerTransition (..) + , Admission (..) + , Settlement (..) + , BreakerStore (..) + , newInMemoryBreakerStore , CircuitBreakerState , CircuitBreakerException (..) , combineCircuitBreakerStates + , circuitBreakerPhases , isCircuitBreakerOpen , isCircuitBreakerClosed - , defaultCircuitBreakerOptions , circuitBreaker + , circuitBreakerEither + , circuitBreakerWithStore + , circuitBreakerEitherWithStore ) where -import Data.Time.Clock.POSIX (getPOSIXTime) +import Data.Foldable (traverse_) +import Data.Int (Int64) +import Glue.Clock +import Glue.Failure import Glue.Types -import UnliftIO (Exception, MonadIO, MonadUnliftIO, - liftIO) -import UnliftIO.Exception (bracketOnError, throwIO) -import UnliftIO.IORef (IORef, atomicModifyIORef', newIORef, - readIORef, writeIORef) +import UnliftIO (Exception, MonadIO, MonadUnliftIO, + SomeException) +import UnliftIO.Exception (mask, onException, throwIO) +import UnliftIO.IORef (atomicModifyIORef', newIORef, readIORef) -- | Options for determining behaviour of circuit breaking services. -data CircuitBreakerOptions = CircuitBreakerOptions - { maxBreakerFailures :: Int - -- ^ How many times the underlying service must fail in the given window before the circuit opens. - , resetTimeoutSecs :: Int - -- ^ The window of time in which the underlying service must fail for the circuit to open. - , breakerDescription :: String +data CircuitBreakerOptions m failure = CircuitBreakerOptions + { breakerFailureThreshold :: Int + -- ^ The circuit opens on this many counted failures, so @3@ opens on the third (minimum 1). + , breakerFailureWindowMs :: Maybe Int + -- ^ 'Nothing' counts consecutive failures, a call that answers resetting the count. + -- @'Just' w@ counts the failures within the last @w@ milliseconds instead. + , breakerOpenDurationMs :: Int + -- ^ How long (in milliseconds) the circuit stays open before a single trial call is admitted. + , breakerDescription :: String -- ^ Description that is attached to the failure so as to identify the particular circuit. + , countsAsFailure :: failure -> Bool + -- ^ Whether a failure says something about the backend's health. Failures that don't count + -- (a caller's own bad request, say) are treated as the backend having answered. + , onBreakerTransition :: BreakerTransition -> m () + -- ^ Called after each change of phase, for metrics and alerts. + , breakerClock :: Clock m + -- ^ Clock used for the failure window and the open duration. } - deriving (Eq, Show) --- | Defaulted options for the circuit breaker with 3 failures over 60 seconds. -defaultCircuitBreakerOptions :: CircuitBreakerOptions +-- | Defaulted options for the circuit breaker: opens on the 3rd consecutive failure, for 60 seconds, +-- counting every failure. +defaultCircuitBreakerOptions :: MonadUnliftIO m => CircuitBreakerOptions m failure defaultCircuitBreakerOptions = CircuitBreakerOptions - { maxBreakerFailures = 3 - , resetTimeoutSecs = 60 - , breakerDescription = "Circuit breaker open." + { breakerFailureThreshold = 3 + , breakerFailureWindowMs = Nothing + , breakerOpenDurationMs = 60000 + , breakerDescription = "Circuit breaker open." + , countsAsFailure = const True + , onBreakerTransition = const (pure ()) + , breakerClock = systemClock } --- | Status indicating if the circuit is open. -data BreakerStatus - = BreakerClosed Int - | BreakerOpen Int +-- | The phase a circuit breaker is in. +data BreakerPhase + = BreakerClosed + -- ^ Calls pass through. + | BreakerOpen + -- ^ Calls are refused. + | BreakerHalfOpen + -- ^ A single trial call is in flight; other calls are refused. + deriving (Eq, Ord, Show, Enum, Bounded) + +-- | A change of phase. +data BreakerTransition = BreakerTransition + { transitionFrom :: BreakerPhase + , transitionTo :: BreakerPhase + } deriving (Eq, Show) +-- | The store's decision about a call. +data Admission + = AdmittedClosed + -- ^ The circuit is closed and the call may proceed. + | AdmittedTrial + -- ^ The call is the single half-open trial. + | Refused + -- ^ The circuit is open (or a trial is in flight) and the call must not be made. + deriving (Eq, Show) + +-- | How an admitted call ended, as far as the breaker is concerned. +data Settlement + = SettledAnswered + -- ^ The backend answered: a success, or a failure that 'countsAsFailure' rejected. + | SettledFailed + -- ^ A counted failure. + | SettledAbandoned + -- ^ The call was cancelled (or threw outside the failure mode) and says nothing about health. + deriving (Eq, Show) + +-- | The state behind a circuit breaker, abstracted so that it can be shared between instances of a +-- horizontally scaled service (a database row updated with one conditional statement using the +-- database clock, say). Implementations must make 'admitCall' atomic so only one caller is ever +-- given 'AdmittedTrial' for a given half-open period. +data BreakerStore m = BreakerStore + { admitCall :: m (Admission, Maybe BreakerTransition) + -- ^ Decide whether a call may proceed, reporting any phase change this caused. + , settleCall :: Admission -> Settlement -> m (Maybe BreakerTransition) + -- ^ Record how an admitted call ended, reporting any phase change this caused. + , readPhase :: m BreakerPhase + -- ^ The current phase, for display. + } + +data MemoryState + = MemoryClosed [Int64] + -- ^ Times of the counted failures still in play, newest first. + | MemoryOpen Int64 + -- ^ Open until the given time. + | MemoryHalfOpen + +-- | The in-memory t'BreakerStore' used by 'circuitBreaker', configured from the threshold, window, +-- open duration and clock in the options. +newInMemoryBreakerStore :: (MonadIO m, MonadIO n) => CircuitBreakerOptions m failure -> n (BreakerStore m) +newInMemoryBreakerStore options = do + ref <- newIORef (MemoryClosed []) + let clock = breakerClock options + threshold = max 1 (breakerFailureThreshold options) + openUs = fromIntegral (max 0 (breakerOpenDurationMs options)) * 1000 + windowUs = fmap (\w -> fromIntegral (max 0 w) * 1000) (breakerFailureWindowMs options) + inWindow now = maybe id (\w -> takeWhile (> now - w)) windowUs + transition from to = Just (BreakerTransition from to) + admit now state = case state of + MemoryClosed _ -> (state, (AdmittedClosed, Nothing)) + MemoryOpen until' + | now >= until' -> (MemoryHalfOpen, (AdmittedTrial, transition BreakerOpen BreakerHalfOpen)) + | otherwise -> (state, (Refused, Nothing)) + MemoryHalfOpen -> (state, (Refused, Nothing)) + settle now AdmittedClosed settlement (MemoryClosed failures) = case settlement of + SettledAnswered + | Just _ <- windowUs -> (MemoryClosed (inWindow now failures), Nothing) + | otherwise -> (MemoryClosed [], Nothing) + SettledFailed -> + let failures' = now : inWindow now failures + in if length failures' >= threshold + then (MemoryOpen (now + openUs), transition BreakerClosed BreakerOpen) + else (MemoryClosed failures', Nothing) + SettledAbandoned -> (MemoryClosed failures, Nothing) + settle now AdmittedTrial settlement MemoryHalfOpen = case settlement of + SettledAnswered -> (MemoryClosed [], transition BreakerHalfOpen BreakerClosed) + SettledFailed -> (MemoryOpen (now + openUs), transition BreakerHalfOpen BreakerOpen) + -- The trial told us nothing, so the next call may try again straight away. + SettledAbandoned -> (MemoryOpen now, transition BreakerHalfOpen BreakerOpen) + settle _ _ _ state = (state, Nothing) + phaseOf (MemoryClosed _) = BreakerClosed + phaseOf (MemoryOpen _) = BreakerOpen + phaseOf MemoryHalfOpen = BreakerHalfOpen + pure + BreakerStore + { admitCall = do + now <- clockNow clock + atomicModifyIORef' ref (admit now) + , settleCall = \admission settlement -> do + now <- clockNow clock + atomicModifyIORef' ref (settle now admission settlement) + , readPhase = phaseOf <$> readIORef ref + } + -- | Representation of the state the circuit breaker is currently in. -newtype CircuitBreakerState = CircuitBreakerState [IORef BreakerStatus] +newtype CircuitBreakerState m = CircuitBreakerState [m BreakerPhase] -- | Exception thrown when the circuit is open. newtype CircuitBreakerException = CircuitBreakerException String @@ -61,69 +194,97 @@ instance Exception CircuitBreakerException -- | Combines multiple states together. -combineCircuitBreakerStates :: Foldable t => t CircuitBreakerState -> CircuitBreakerState -combineCircuitBreakerStates = CircuitBreakerState . foldMap (\(CircuitBreakerState refs) -> refs) +combineCircuitBreakerStates :: Foldable t => t (CircuitBreakerState m) -> CircuitBreakerState m +combineCircuitBreakerStates = CircuitBreakerState . foldMap (\(CircuitBreakerState phases) -> phases) --- | Determines if a specific status is open. -isStatusOpen :: BreakerStatus -> Bool -isStatusOpen (BreakerOpen _) = True -isStatusOpen (BreakerClosed _) = False +-- | The phases of each breaker in the state. +circuitBreakerPhases :: Applicative m => CircuitBreakerState m -> m [BreakerPhase] +circuitBreakerPhases (CircuitBreakerState phases) = sequenceA phases --- | Determines if a specific status is closed. -isStatusClosed :: BreakerStatus -> Bool -isStatusClosed (BreakerOpen _) = False -isStatusClosed (BreakerClosed _) = True +-- | Determines if a circuit breaker is open (or half-open) in any of its breakers. +isCircuitBreakerOpen :: Applicative m => CircuitBreakerState m -> m Bool +isCircuitBreakerOpen state = any (/= BreakerClosed) <$> circuitBreakerPhases state --- | Determines if a circuit breaker is open. -isCircuitBreakerOpen :: MonadIO m => CircuitBreakerState -> m Bool -isCircuitBreakerOpen (CircuitBreakerState states) = - or <$> traverse (fmap isStatusOpen . readIORef) states +-- | Determines if a circuit breaker is closed in all of its breakers. +isCircuitBreakerClosed :: Applicative m => CircuitBreakerState m -> m Bool +isCircuitBreakerClosed state = all (== BreakerClosed) <$> circuitBreakerPhases state --- | Determines if a circuit breaker is closed. -isCircuitBreakerClosed :: MonadIO m => CircuitBreakerState -> m Bool -isCircuitBreakerClosed (CircuitBreakerState states) = - and <$> traverse (fmap isStatusClosed . readIORef) states +breakerVia + :: MonadUnliftIO m + => FailureMode m r failure + -> m r + -> CircuitBreakerOptions m failure + -> BreakerStore m + -> BasicService m a r + -> BasicService m a r +breakerVia mode refuse options store service request = + mask $ \restore -> do + let notify = traverse_ (onBreakerTransition options) + (admission, admitted) <- admitCall store + case admission of + Refused -> restore (notify admitted >> refuse) + _ -> do + outcome <- + restore (notify admitted >> observeCall mode (service request)) + `onException` (settleCall store admission SettledAbandoned >>= notify) + let settlement = either (\f -> if countsAsFailure options f then SettledFailed else SettledAnswered) (const SettledAnswered) outcome + settled <- settleCall store admission settlement + restore $ do + notify settled + either (reraise mode) pure outcome -- | Circuit breaking services can be constructed with this function. circuitBreaker :: (MonadUnliftIO m, MonadIO n) - => CircuitBreakerOptions + => CircuitBreakerOptions m SomeException -- ^ Options for specifying the circuit breaker behaviour. -> BasicService m a b -- ^ Service to protect with the circuit breaker. - -> n (CircuitBreakerState, BasicService m a b) + -> n (CircuitBreakerState m, BasicService m a b) circuitBreaker options service = do - let retrieveTime = liftIO $ round <$> getPOSIXTime - failureMax = maxBreakerFailures options - callIfClosed request ref = bracketOnError (pure ()) (\_ -> incErrors ref) (\_ -> service request) - canaryCall request ref = do - result <- callIfClosed request ref - writeIORef ref $ BreakerClosed 0 - pure result - incErrors ref = do - currentTime <- retrieveTime - atomicModifyIORef' ref $ \status -> case status of - BreakerClosed errorCount -> - ( if errorCount >= failureMax - then BreakerOpen (currentTime + resetTimeoutSecs options) - else BreakerClosed (errorCount + 1) - , () - ) - other -> (other, ()) - failingCall = throwIO $ CircuitBreakerException $ breakerDescription options - callIfOpen request ref = do - currentTime <- retrieveTime - canaryRequest <- atomicModifyIORef' ref $ \status -> case status of - BreakerClosed _ -> (status, False) - BreakerOpen time -> - if currentTime > time - then (BreakerOpen (currentTime + resetTimeoutSecs options), True) - else (status, False) - if canaryRequest then canaryCall request ref else failingCall - breakerService ref request = do - status <- readIORef ref - case status of - BreakerClosed _ -> callIfClosed request ref - BreakerOpen _ -> callIfOpen request ref - ref <- newIORef $ BreakerClosed 0 - pure (CircuitBreakerState [ref], breakerService ref) + store <- newInMemoryBreakerStore options + pure (CircuitBreakerState [readPhase store], circuitBreakerWithStore options store service) + +-- | Variant of 'circuitBreaker' using the given store, for state shared between instances. +-- The threshold, window and open duration in the options are the store's concern and unused here. +circuitBreakerWithStore + :: MonadUnliftIO m + => CircuitBreakerOptions m SomeException + -- ^ Options for specifying the circuit breaker behaviour. + -> BreakerStore m + -- ^ Store holding the breaker's state. + -> BasicService m a b + -- ^ Service to protect with the circuit breaker. + -> BasicService m a b +circuitBreakerWithStore options = + breakerVia exceptionFailures (throwIO $ CircuitBreakerException $ breakerDescription options) options + +-- | Variant of 'circuitBreaker' for services returning failures as values: @Left@ results are +-- classified by 'countsAsFailure', and a refused call returns the given failure instead of throwing. +circuitBreakerEither + :: (MonadUnliftIO m, MonadIO n) + => CircuitBreakerOptions m e + -- ^ Options for specifying the circuit breaker behaviour ('breakerDescription' is unused). + -> e + -- ^ Failure returned when the circuit is open. + -> EitherService m a e b + -- ^ Service to protect with the circuit breaker. + -> n (CircuitBreakerState m, EitherService m a e b) +circuitBreakerEither options onBreakerOpen service = do + store <- newInMemoryBreakerStore options + pure (CircuitBreakerState [readPhase store], circuitBreakerEitherWithStore options store onBreakerOpen service) + +-- | Variant of 'circuitBreakerEither' using the given store. +circuitBreakerEitherWithStore + :: MonadUnliftIO m + => CircuitBreakerOptions m e + -- ^ Options for specifying the circuit breaker behaviour. + -> BreakerStore m + -- ^ Store holding the breaker's state. + -> e + -- ^ Failure returned when the circuit is open. + -> EitherService m a e b + -- ^ Service to protect with the circuit breaker. + -> EitherService m a e b +circuitBreakerEitherWithStore options store onBreakerOpen = + breakerVia valueFailures (pure (Left onBreakerOpen)) options store diff --git a/src/Glue/ConcurrencyLimit.hs b/src/Glue/ConcurrencyLimit.hs --- a/src/Glue/ConcurrencyLimit.hs +++ b/src/Glue/ConcurrencyLimit.hs @@ -3,18 +3,39 @@ -- | Module supporting concurrent request constraints (bulkhead pattern), -- limiting the number of in-flight and optionally queued requests. +-- +-- 'limitConcurrency' caps one wrapped service. 'limitConcurrencyBy' keeps one limit per key (a +-- backend or target, say) and refuses a full key at once with v'Saturated' as a value, so a full +-- guard can feed a fallback without exception plumbing and is never mistaken for a backend failure. +-- A slot covers the wrapped call only; 'withSlot' holds one for work that outlives the call, such +-- as a streamed response consumed later. module Glue.ConcurrencyLimit ( ConcurrencyLimitOptions (..) , ConcurrencyLimitExceededException (..) , defaultConcurrencyLimitOptions , limitConcurrency + , limitConcurrencyEither + -- * Keyed limits + , Saturated (..) + , AdmissionLease (..) + , AdmissionStore (..) + , newInMemoryAdmissionStore + , withSlot + , limitConcurrencyBy + , limitConcurrencyByWithStore + , limitConcurrencyByEither ) where +import Data.Foldable (traverse_) +import Data.Hashable (Hashable) +import qualified Data.HashMap.Strict as M +import Glue.Failure import Glue.Types -import UnliftIO (Exception, MonadIO, MonadUnliftIO) -import UnliftIO.Exception (bracketOnError, finally, throwIO) -import UnliftIO.STM (atomically, modifyTVar', newTVarIO, - readTVar, retrySTM, writeTVar) +import UnliftIO (Exception, MonadIO, MonadUnliftIO) +import UnliftIO.Exception (bracket, bracketOnError, finally, + throwIO) +import UnliftIO.STM (atomically, modifyTVar', newTVarIO, + readTVar, retrySTM, writeTVar) -- | Options for configuring concurrent request limits. data ConcurrencyLimitOptions = ConcurrencyLimitOptions @@ -42,15 +63,13 @@ instance Exception ConcurrencyLimitExceededException --- | Wraps a service with a concurrency limit restricting in-flight and queued requests. -limitConcurrency +limitVia :: (MonadUnliftIO m, MonadIO n) - => ConcurrencyLimitOptions - -- ^ Options configuring the concurrency and queue bounds. - -> BasicService m a b - -- ^ Service to protect. - -> n (BasicService m a b) -limitConcurrency options service = do + => m r + -> ConcurrencyLimitOptions + -> BasicService m a r + -> n (BasicService m a r) +limitVia refuse options service = do let maxActive = max 1 (maxInFlight options) inFlightVar <- newTVarIO (0 :: Int) queuedVar <- newTVarIO (0 :: Int) @@ -77,11 +96,126 @@ writeTVar queuedVar (queued + 1) pure (Right False) case admission of - Left () -> - throwIO $ ConcurrencyLimitExceededException (concurrencyLimitDescription options) + Left () -> refuse Right True -> finally (service request) releaseInFlight Right False -> do bracketOnError (pure ()) (\_ -> releaseQueued) (\_ -> waitForSlot) finally (service request) releaseInFlight pure limitedService + +-- | Wraps a service with a concurrency limit restricting in-flight and queued requests. +limitConcurrency + :: (MonadUnliftIO m, MonadIO n) + => ConcurrencyLimitOptions + -- ^ Options configuring the concurrency and queue bounds. + -> BasicService m a b + -- ^ Service to protect. + -> n (BasicService m a b) +limitConcurrency options = + limitVia (throwIO $ ConcurrencyLimitExceededException (concurrencyLimitDescription options)) options + +-- | Variant of 'limitConcurrency' for services returning failures as values, returning the given +-- failure when both limits are reached instead of throwing. +limitConcurrencyEither + :: (MonadUnliftIO m, MonadIO n) + => ConcurrencyLimitOptions + -- ^ Options configuring the concurrency and queue bounds ('concurrencyLimitDescription' is unused). + -> e + -- ^ Failure returned when the limits are reached. + -> EitherService m a e b + -- ^ Service to protect. + -> n (EitherService m a e b) +limitConcurrencyEither options onSaturated = limitVia (pure (Left onSaturated)) options + +-- | Refusal by a keyed limit whose slots are all in use. +data Saturated = Saturated + deriving (Eq, Show) + +-- | A claimed slot, released exactly once by 'withSlot'. +newtype AdmissionLease m = AdmissionLease + { releaseLease :: m () + } + +-- | Where keyed slots are counted. The in-memory store is the default; an implementation backed by +-- shared storage (claim, renew and release of expiring leases, say) gives limits shared across +-- instances, renewing its own leases for as long as they are held. +data AdmissionStore m k = AdmissionStore + { claimSlot :: k -> m (Maybe (AdmissionLease m)) + -- ^ Claims a slot for the key, or 'Nothing' if its limit is reached. Must be atomic. + , slotsInUse :: k -> m Int + -- ^ Slots currently held for the key. + , trackedKeys :: m Int + -- ^ Number of keys with slots held. Keys with nothing in flight are not kept. + } + +-- | An in-memory store with the given limit per key (a limit of zero or less refuses every call). +newInMemoryAdmissionStore :: (Hashable k, MonadIO m, MonadIO n) => (k -> Int) -> n (AdmissionStore m k) +newInMemoryAdmissionStore limitFor = do + countsVar <- newTVarIO M.empty + let release key = atomically $ modifyTVar' countsVar (M.update (\n -> if n <= 1 then Nothing else Just (n - 1)) key) + claim key = atomically $ do + counts <- readTVar countsVar + let current = M.findWithDefault 0 key counts + if current < limitFor key + then do + writeTVar countsVar (M.insert key (current + 1) counts) + pure (Just (AdmissionLease (release key))) + else pure Nothing + pure + AdmissionStore + { claimSlot = claim + , slotsInUse = \key -> atomically $ M.findWithDefault 0 key <$> readTVar countsVar + , trackedKeys = atomically $ M.size <$> readTVar countsVar + } + +-- | Runs the work holding a slot for the key, or returns v'Saturated' at once if none is free. +-- The slot is released on every exit path, including asynchronous cancellation. +withSlot :: MonadUnliftIO m => AdmissionStore m k -> k -> m r -> m (Either Saturated r) +withSlot store key work = + bracket (claimSlot store key) (traverse_ releaseLease) $ + maybe (pure (Left Saturated)) (const (Right <$> work)) + +-- | Wraps a service with one concurrency limit per key, refusing a call whose key is full with +-- v'Saturated' instead of queueing it. +limitConcurrencyBy + :: (MonadUnliftIO m, MonadIO n, Hashable k) + => (a -> k) + -- ^ The key a request is limited under. + -> (k -> Int) + -- ^ The limit for each key. + -> BasicService m a b + -- ^ Service to protect. + -> n (BasicService m a (Either Saturated b)) +limitConcurrencyBy keyOf limitFor service = do + store <- newInMemoryAdmissionStore limitFor + pure (limitConcurrencyByWithStore store keyOf service) + +-- | Variant of 'limitConcurrencyBy' counting slots in the given store. +limitConcurrencyByWithStore + :: MonadUnliftIO m + => AdmissionStore m k + -- ^ Store counting the slots. + -> (a -> k) + -- ^ The key a request is limited under. + -> BasicService m a b + -- ^ Service to protect. + -> BasicService m a (Either Saturated b) +limitConcurrencyByWithStore store keyOf service request = withSlot store (keyOf request) (service request) + +-- | Variant of 'limitConcurrencyBy' for services returning failures as values, returning the given +-- failure when the key is full. +limitConcurrencyByEither + :: (MonadUnliftIO m, MonadIO n, Hashable k) + => (a -> k) + -- ^ The key a request is limited under. + -> (k -> Int) + -- ^ The limit for each key. + -> e + -- ^ Failure returned when the key is full. + -> EitherService m a e b + -- ^ Service to protect. + -> n (EitherService m a e b) +limitConcurrencyByEither keyOf limitFor onSaturated service = do + limited <- limitConcurrencyBy keyOf limitFor service + pure $ fmap (either (const (Left onSaturated)) id) . limited diff --git a/src/Glue/DogpileProtection.hs b/src/Glue/DogpileProtection.hs --- a/src/Glue/DogpileProtection.hs +++ b/src/Glue/DogpileProtection.hs @@ -13,27 +13,29 @@ import Control.Monad (void) import Data.Hashable (Hashable) import qualified Data.HashMap.Strict as M +import Glue.Clock import Glue.Types -import UnliftIO (MonadIO, MonadUnliftIO, SomeException, - liftIO) -import UnliftIO.Concurrent (forkIO, threadDelay) +import UnliftIO (MonadIO, MonadUnliftIO, SomeException) +import UnliftIO.Concurrent (forkIO) import UnliftIO.Exception (onException, throwIO, toException, tryAny) import UnliftIO.IORef (IORef, atomicModifyIORef', newIORef) import UnliftIO.MVar (MVar, newEmptyMVar, putMVar, readMVar, tryPutMVar) -- | Options for configuring dogpile protection. -newtype DogpileOptions = DogpileOptions +data DogpileOptions m = DogpileOptions { dogpileCacheTimeMs :: Int -- ^ How long (in milliseconds) to retain a completed result for subsequent callers. + , dogpileClock :: Clock m + -- ^ Clock used to expire retained results. } - deriving (Eq, Show) -- | Default options for dogpile protection, caching completed values for 5 seconds (5000ms). -defaultDogpileOptions :: DogpileOptions +defaultDogpileOptions :: MonadUnliftIO m => DogpileOptions m defaultDogpileOptions = DogpileOptions { dogpileCacheTimeMs = 5000 + , dogpileClock = systemClock } data DogpileResult b @@ -51,7 +53,7 @@ dogpileProtectWith :: forall m n a b. (MonadUnliftIO m, MonadIO n, Hashable a) - => DogpileOptions + => DogpileOptions m -- ^ Options configuring dogpile protection. -> BasicService m a b -- ^ The service to protect. @@ -76,8 +78,8 @@ if ttlMs > 0 then do atomicModifyIORef' mapRef (\m -> (M.insert request (CachedValue result) m, ())) - void $ liftIO $ forkIO $ do - threadDelay (ttlMs * 1000) + void $ forkIO $ do + clockSleep (dogpileClock options) (ttlMs * 1000) atomicModifyIORef' mapRef (\m -> (M.delete request m, ())) else atomicModifyIORef' mapRef (\m -> (M.delete request m, ())) diff --git a/src/Glue/Failover.hs b/src/Glue/Failover.hs --- a/src/Glue/Failover.hs +++ b/src/Glue/Failover.hs @@ -1,15 +1,27 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE ScopedTypeVariables #-} --- | Module supporting failover of a service by transforming the request each time a failure occurs up to a fixed number of attempts. +-- | Module supporting failover: either re-invoking one service with a transformed request each +-- time a failure occurs ('failover'), or trying an ordered list of distinct backends, a primary +-- then its backups, until one answers ('failoverAcross'). module Glue.Failover ( FailoverOptions (..) , defaultFailoverOptions , failover + , FailoverAcrossOptions (..) + , ReportedFailure (..) + , defaultFailoverAcrossOptions + , failoverAcross + , failoverAcrossIndexed + , failoverAcrossEither + , failoverAcrossIndexedEither ) where +import Data.List.NonEmpty (NonEmpty (..)) +import Data.Maybe (fromMaybe) +import Glue.Failure import Glue.Types -import UnliftIO (MonadUnliftIO) +import UnliftIO (MonadUnliftIO, SomeException) import UnliftIO.Exception (catchAny) -- | Options for determining behaviour of failover services. @@ -44,3 +56,93 @@ then invoke else catchAny invoke afterFail in invokeService 0 + +-- | Which failure is reported when every backend has failed. +data ReportedFailure + = ReportFirstFailure + -- ^ The primary's failure. + | ReportLastFailure + -- ^ The last backup's failure. + deriving (Eq, Show) + +-- | Options for failing over across distinct backends. +data FailoverAcrossOptions failure = FailoverAcrossOptions + { advanceOn :: failure -> Bool + -- ^ Whether a failure moves on to the next backend. A failure that doesn't (an invalid request, + -- say, which a backup would only repeat) stops the chain and is reported at once. + , reportedFailure :: ReportedFailure + -- ^ Which failure is reported when every backend has failed. + } + +-- | Defaults that advance on every failure and report the last one. +defaultFailoverAcrossOptions :: FailoverAcrossOptions failure +defaultFailoverAcrossOptions = + FailoverAcrossOptions + { advanceOn = const True + , reportedFailure = ReportLastFailure + } + +failoverAcrossVia + :: Monad m + => FailureMode m r failure + -> FailoverAcrossOptions failure + -> NonEmpty (BasicService m a r) + -> BasicService m a (Int, r) +failoverAcrossVia mode options (primary :| backups) request = go 0 Nothing primary backups + where + report index failure = (,) index <$> reraise mode failure + go index firstFailure service rest = do + outcome <- observeCall mode (service request) + case outcome of + Right result -> pure (index, result) + Left failure + | not (advanceOn options failure) -> report index failure + | otherwise -> case rest of + next : rest' -> go (index + 1) (Just (fromMaybe (index, failure) firstFailure)) next rest' + [] -> case (reportedFailure options, firstFailure) of + (ReportFirstFailure, Just (firstIndex, first)) -> report firstIndex first + _ -> report index failure + +-- | Creates a service that tries each backend in order and returns the first success. +-- Asynchronous exceptions always propagate and never advance to the next backend. +failoverAcross + :: MonadUnliftIO m + => FailoverAcrossOptions SomeException + -- ^ Options deciding which failures advance and which failure is reported. + -> NonEmpty (BasicService m a b) + -- ^ The backends, primary first. + -> BasicService m a b +failoverAcross options services = fmap snd . failoverAcrossIndexed options services + +-- | Variant of 'failoverAcross' that also reports the index of the backend that answered, +-- for tracing and billing. +failoverAcrossIndexed + :: MonadUnliftIO m + => FailoverAcrossOptions SomeException + -- ^ Options deciding which failures advance and which failure is reported. + -> NonEmpty (BasicService m a b) + -- ^ The backends, primary first. + -> BasicService m a (Int, b) +failoverAcrossIndexed = failoverAcrossVia exceptionFailures + +-- | Variant of 'failoverAcross' for services returning failures as values: @Left@ results are +-- judged by 'advanceOn'. Exceptions are not caught and do not advance. +failoverAcrossEither + :: Monad m + => FailoverAcrossOptions e + -- ^ Options deciding which failures advance and which failure is reported. + -> NonEmpty (EitherService m a e b) + -- ^ The backends, primary first. + -> EitherService m a e b +failoverAcrossEither options services = fmap (fmap snd) . failoverAcrossIndexedEither options services + +-- | Variant of 'failoverAcrossEither' that also reports the index of the backend that answered. +failoverAcrossIndexedEither + :: Monad m + => FailoverAcrossOptions e + -- ^ Options deciding which failures advance and which failure is reported. + -> NonEmpty (EitherService m a e b) + -- ^ The backends, primary first. + -> EitherService m a e (Int, b) +failoverAcrossIndexedEither options services = + fmap (\(index, result) -> (,) index <$> result) . failoverAcrossVia valueFailures options services diff --git a/src/Glue/Hedging.hs b/src/Glue/Hedging.hs --- a/src/Glue/Hedging.hs +++ b/src/Glue/Hedging.hs @@ -1,81 +1,176 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE ScopedTypeVariables #-} --- | Module supporting request hedging (duplication), dispatching multiple concurrent --- requests (optionally staggered) and completing with the first successful result. +-- | Module supporting request hedging: dispatching further attempts if earlier ones haven't +-- answered within a delay, and completing with the first successful result. +-- +-- 'hedgeAcross' races distinct backends, a primary then its backups, leaving the caller in control +-- of whether each backup may launch and informed of how every attempt ended. 'hedgingService' +-- duplicates the same call to one replicated backend. Both are built on "Glue.Race", which callers +-- with their own launch policy can drive directly. module Glue.Hedging - ( HedgingOptions (..) + ( HedgeOptions (..) + , defaultHedgeOptions + , hedgeAcross + , hedgeAcrossEither + , HedgingOptions (..) , defaultHedgingOptions , hedgingService + , AttemptOutcome (..) ) where -import Control.Monad (when) -import Data.Maybe (fromMaybe) +import Data.Foldable (traverse_) +import Data.List.NonEmpty (NonEmpty (..)) +import Data.Maybe (fromMaybe) +import Glue.Clock +import Glue.Failure +import Glue.Race import Glue.Types -import UnliftIO (Async, MonadUnliftIO, SomeException, - withAsync) -import UnliftIO.Concurrent (threadDelay) -import UnliftIO.Exception (throwIO, tryAny) -import UnliftIO.STM (atomically, newTVarIO, readTVar, retrySTM, - writeTVar) +import UnliftIO (MonadUnliftIO, SomeException) +import UnliftIO.Exception (throwIO) +-- | Options for hedging across distinct backends. +data HedgeOptions m failure r = HedgeOptions + { hedgeDelayMs :: Int + -- ^ How long (in milliseconds) to wait for an attempt to end before launching the next backend. + -- A failed attempt launches the next backend straight away. + , admitLaunch :: Int -> m Bool + -- ^ Asked before launching the backend with this index (from 1; the primary always launches). + -- Declining skips that backend without failing the attempts already running; the next backend + -- is considered after a further delay. + , onAttemptEnded :: Int -> AttemptOutcome failure r -> m () + -- ^ Told how each launched attempt ended, including losers that were cancelled, once each has + -- actually finished. + , onHedgeWon :: Int -> m () + -- ^ Told which attempt's answer was returned. + , hedgeClock :: Clock m + -- ^ Clock used for the delays. + } + +-- | Defaults launching each backend 100ms after the previous one with no further control. +defaultHedgeOptions :: MonadUnliftIO m => HedgeOptions m failure r +defaultHedgeOptions = + HedgeOptions + { hedgeDelayMs = 100 + , admitLaunch = const (pure True) + , onAttemptEnded = \_ _ -> pure () + , onHedgeWon = const (pure ()) + , hedgeClock = systemClock + } + +-- | Thrown if a hedge ends with no outcome to report, which the race guarantees cannot happen. +noOutcome :: MonadUnliftIO m => m a +noOutcome = throwIO (userError "Glue.Hedging: every attempt ended without an outcome") + +hedgeVia + :: forall m a r failure. + MonadUnliftIO m + => FailureMode m r failure + -> HedgeOptions m failure r + -> NonEmpty (BasicService m a r) + -> BasicService m a r +hedgeVia mode options (primary :| backups) request = + withRace mode (hedgeClock options) $ \race -> do + let finishLosers = abandonAll race >>= traverse_ (uncurry (onAttemptEnded options)) + + launchNext :: [(Int, BasicService m a r)] -> m ([(Int, BasicService m a r)], Bool) + launchNext [] = pure ([], False) + launchNext ((index, service) : rest) = do + admitted <- admitLaunch options index + if admitted + then launch race (service request) >> pure (rest, True) + else pure (rest, False) + + launchUntilRunning pending = do + (pending', launched) <- launchNext pending + if launched || null pending' then pure pending' else launchUntilRunning pending' + + loop :: [(Int, BasicService m a r)] -> Maybe failure -> m r + loop pending firstFailure = do + event <- awaitNext race (if null pending then Nothing else Just (hedgeDelayMs options)) + case event of + AttemptEnded index outcome@(AttemptAnswered result) -> do + onAttemptEnded options index outcome + finishLosers + onHedgeWon options index + pure result + AttemptEnded index outcome@(AttemptFailed failure) -> do + onAttemptEnded options index outcome + (pending', _) <- launchNext pending + loop pending' (Just (fromMaybe failure firstFailure)) + AttemptEnded index outcome@(AttemptCrashed e) -> do + onAttemptEnded options index outcome + finishLosers + throwIO e + AttemptEnded index outcome@AttemptCancelled -> do + onAttemptEnded options index outcome + loop pending firstFailure + RaceTimedOut -> do + (pending', _) <- launchNext pending + loop pending' firstFailure + NothingRunning + | null pending -> maybe noOutcome (reraise mode) firstFailure + | otherwise -> launchUntilRunning pending >>= \pending' -> loop pending' firstFailure + + _ <- launch race (primary request) + loop (zip [1 ..] backups) Nothing + +-- | Races distinct backends: the primary is launched at once and each backup after a further +-- 'hedgeDelayMs' if no attempt has ended (or straight away when an attempt fails). The first success is returned and the other attempts are cancelled, waiting until they +-- have finished. Attempts ending together are decided by their order in the list. If every attempt +-- fails, the first failure is rethrown. +hedgeAcross + :: MonadUnliftIO m + => HedgeOptions m SomeException b + -- ^ Options configuring the delay and the caller's control over launches and outcomes. + -> NonEmpty (BasicService m a b) + -- ^ The backends, primary first. + -> BasicService m a b +hedgeAcross = hedgeVia exceptionFailures + +-- | Variant of 'hedgeAcross' for services returning failures as values: a @Left@ result is a +-- failed attempt, and the first @Left@ is returned if every attempt fails. +hedgeAcrossEither + :: MonadUnliftIO m + => HedgeOptions m e (Either e b) + -- ^ Options configuring the delay and the caller's control over launches and outcomes. + -> NonEmpty (EitherService m a e b) + -- ^ The backends, primary first. + -> EitherService m a e b +hedgeAcrossEither = hedgeVia valueFailures + -- | Options for configuring hedged (duplicated) service calls. -data HedgingOptions = HedgingOptions +data HedgingOptions m = HedgingOptions { hedgingCount :: Int -- ^ Total number of concurrent attempts to make (including the primary call). , hedgingDelayMs :: Int -- ^ Stagger delay in milliseconds before each subsequent hedged call (`0` to fire all simultaneously). + , hedgingClock :: Clock m + -- ^ Clock used for the stagger delay. } - deriving (Eq, Show) -- | Default hedging options that fire 2 concurrent requests with no stagger delay. -defaultHedgingOptions :: HedgingOptions +defaultHedgingOptions :: MonadUnliftIO m => HedgingOptions m defaultHedgingOptions = HedgingOptions { hedgingCount = 2 , hedgingDelayMs = 0 + , hedgingClock = systemClock } -withAsyncs :: MonadUnliftIO m => [m a] -> ([Async a] -> m b) -> m b -withAsyncs [] k = k [] -withAsyncs (m:ms) k = withAsync m $ \a -> withAsyncs ms (\as -> k (a : as)) - -- | Creates a service that duplicates requests up to 'hedgingCount' times and returns -- the first successful response, automatically cancelling remaining in-flight attempts. -- If all attempts fail, the first encountered exception is rethrown. hedgingService - :: forall m a b. - MonadUnliftIO m - => HedgingOptions + :: MonadUnliftIO m + => HedgingOptions m -- ^ Options configuring the number of hedged calls and stagger delay. -> BasicService m a b -- ^ Underlying service to invoke. -> BasicService m a b -hedgingService options service request - | hedgingCount options <= 1 = service request - | otherwise = do - let total = hedgingCount options - delayUs = max 0 (hedgingDelayMs options) * 1000 - stateVar <- newTVarIO (Nothing :: Maybe b, total, Nothing :: Maybe SomeException) - let runAttempt :: Int -> m () - runAttempt idx = do - when (idx > 0 && delayUs > 0) $ - threadDelay (idx * delayUs) - res <- tryAny (service request) - atomically $ do - (mWin, remaining, mErr) <- readTVar stateVar - case res of - Right val -> writeTVar stateVar (Just (fromMaybe val mWin), remaining - 1, mErr) - Left err -> writeTVar stateVar (mWin, remaining - 1, Just (fromMaybe err mErr)) - withAsyncs [runAttempt i | i <- [0 .. total - 1]] $ \_ -> do - outcome <- atomically $ do - (mWin, remaining, mErr) <- readTVar stateVar - case mWin of - Just val -> pure (Right val) - Nothing - | remaining <= 0 -> pure (Left mErr) - | otherwise -> retrySTM - case outcome of - Right val -> pure val - Left (Just err) -> throwIO err - Left Nothing -> service request +hedgingService options service + | hedgingCount options <= 1 = service + | otherwise = + hedgeAcross + defaultHedgeOptions { hedgeDelayMs = max 0 (hedgingDelayMs options), hedgeClock = hedgingClock options } + (service :| replicate (hedgingCount options - 1) service) diff --git a/src/Glue/LoadBalancer.hs b/src/Glue/LoadBalancer.hs --- a/src/Glue/LoadBalancer.hs +++ b/src/Glue/LoadBalancer.hs @@ -1,16 +1,32 @@ +{-# LANGUAGE BangPatterns #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE ScopedTypeVariables #-} --- | Module providing round-robin load balancing across multiple services. +-- | Module providing load balancing across multiple services: round-robin, or a deterministic +-- weighted split keyed by a stable request key. module Glue.LoadBalancer ( NoServicesAvailableException (..) , roundRobinLoadBalancer , roundRobinLoadBalancerNE + -- * Weighted, keyed balancing + , Weight + , totalWeight + , WeightError (..) + , Salt (..) + , saltFromBytes + , validateWeights + , bucketOf + , chooseIndex + , weightedBy + , sipHash24 ) where +import Data.Bits (rotateL, shiftL, xor, (.|.)) +import qualified Data.ByteString as BS import Data.Foldable (toList) import Data.List.NonEmpty (NonEmpty (..)) import qualified Data.List.NonEmpty as NE +import Data.Word (Word64) import Glue.Types import UnliftIO (Exception, MonadIO) import UnliftIO.Exception (throwIO) @@ -49,3 +65,104 @@ case toList services of [] -> pure $ const $ throwIO NoServicesAvailableException (x:xs) -> roundRobinLoadBalancerNE (x :| xs) + +-- | A backend's share of traffic in basis points: the weights of a split sum to 'totalWeight'. +type Weight = Int + +-- | The sum every set of weights must have: 10000 basis points. +totalWeight :: Int +totalWeight = 10000 + +-- | Why a set of weights was rejected. +data WeightError + = NegativeWeight Int Weight + -- ^ The weight at this index is negative. + | WeightsDoNotSumToTotal Int + -- ^ The weights sum to this rather than 'totalWeight'. + deriving (Eq, Show) + +-- | A 128-bit SipHash key. Different salts give independent assignments, so two splits keyed on +-- the same request key never correlate. +data Salt = Salt Word64 Word64 + deriving (Eq, Show) + +-- | Derives a salt from arbitrary bytes (a split's name, say) by hashing them under two fixed keys. +saltFromBytes :: BS.ByteString -> Salt +saltFromBytes bytes = Salt (sipHash24 0 0 bytes) (sipHash24 0 1 bytes) + +-- | Checks that no weight is negative and that they sum to 'totalWeight'. +validateWeights :: NonEmpty Weight -> Either WeightError () +validateWeights weights = + case [(index, w) | (index, w) <- zip [0 ..] (toList weights), w < 0] of + (index, w) : _ -> Left (NegativeWeight index w) + [] + | sum weights /= totalWeight -> Left (WeightsDoNotSumToTotal (sum weights)) + | otherwise -> Right () + +-- | The bucket, from 0 to 'totalWeight' - 1, that a key falls in for a salt: the key's SipHash-2-4 +-- under the salt, modulo 'totalWeight'. Reducing a 64-bit hash this way favours the lowest +-- @2^64 mod 10000 = 1616@ buckets by one value in about @1.8 * 10^15@, a negligible bias. +-- The bucket never depends on the weights. +bucketOf :: Salt -> BS.ByteString -> Int +bucketOf (Salt k0 k1) key = fromIntegral (sipHash24 k0 k1 key `mod` fromIntegral totalWeight) + +-- | The index of the backend a key is assigned to: backends own consecutive ranges of buckets in +-- list order, sized by their weights. A key keeps its backend across instances and restarts, and +-- changing weights only moves the keys whose bucket changes owner. Weights should satisfy +-- 'validateWeights'; buckets beyond their sum go to the last backend. +chooseIndex :: Salt -> BS.ByteString -> NonEmpty Weight -> Int +chooseIndex salt key weights = go 0 0 (toList weights) + where + bucket = bucketOf salt key + go index _ [_] = index + go index start (w : ws) + | bucket < start + max 0 w = index + | otherwise = go (index + 1) (start + max 0 w) ws + go index _ [] = index + +-- | Creates a service splitting traffic between backends by weight, choosing each request's backend +-- from a stable key (a session id, else a request id) so the same key always reaches the same +-- backend. A failure of the chosen backend propagates; it is never redistributed (compose with +-- 'Glue.Failover.failoverAcross' for that). +weightedBy + :: (a -> BS.ByteString) + -- ^ The request's stable key. + -> Salt + -- ^ The split's salt. + -> NonEmpty (Weight, BasicService m a b) + -- ^ Backends with their weights in basis points. + -> Either WeightError (BasicService m a b) +weightedBy keyOf salt backends = do + let weights = fmap fst backends + services = fmap snd backends + validateWeights weights + pure $ \request -> (services NE.!! chooseIndex salt (keyOf request) weights) request + +-- | SipHash-2-4 of the bytes under the 128-bit key given as two little-endian words. +sipHash24 :: Word64 -> Word64 -> BS.ByteString -> Word64 +sipHash24 k0 k1 message = finish (compress (initial, message)) + where + initial = (k0 `xor` 0x736f6d6570736575, k1 `xor` 0x646f72616e646f6d, k0 `xor` 0x6c7967656e657261, k1 `xor` 0x7465646279746573) + len = BS.length message + compress (!v, bytes) + | BS.length bytes >= 8 = compress (absorb v (word (BS.take 8 bytes)), BS.drop 8 bytes) + | otherwise = absorb v (word bytes .|. (fromIntegral (len `mod` 256) `shiftL` 56)) + absorb (v0, v1, v2, v3) m = + let (v0', v1', v2', v3') = sipRound (sipRound (v0, v1, v2, v3 `xor` m)) + in (v0' `xor` m, v1', v2', v3') + finish (v0, v1, v2, v3) = + let (v0', v1', v2', v3') = iterate sipRound (v0, v1, v2 `xor` 0xff, v3) !! 4 + in v0' `xor` v1' `xor` v2' `xor` v3' + word = BS.foldr' (\b acc -> (acc `shiftL` 8) .|. fromIntegral b) 0 + sipRound (v0, v1, v2, v3) = + let a0 = v0 + v1 + a1 = rotateL v1 13 `xor` a0 + a0' = rotateL a0 32 + a2 = v2 + v3 + a3 = rotateL v3 16 `xor` a2 + b0 = a0' + a3 + b3 = rotateL a3 21 `xor` b0 + b2 = a2 + a1 + b1 = rotateL a1 17 `xor` b2 + b2' = rotateL b2 32 + in (b0, b1, b2', b3) diff --git a/src/Glue/Periodic.hs b/src/Glue/Periodic.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Periodic.hs @@ -0,0 +1,137 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | Module supporting background jobs run periodically (sweeping expired rows, flushing a queue, +-- polling for state) with a graceful stop. +-- +-- Passes never overlap: the period is measured from the end of the previous pass. A pass that fails +-- with a synchronous exception is reported and the loop carries on; an asynchronous exception ends +-- the loop. +module Glue.Periodic + ( PeriodicOptions (..) + , PeriodicHandle (..) + , defaultPeriodicOptions + , periodically + ) where + +import Control.Monad (unless, void, when) +import Data.Maybe (isJust) +import Glue.Clock +import System.Random (randomRIO) +import UnliftIO (MonadUnliftIO, SomeException, liftIO) +import UnliftIO.Async (async, cancel) +import UnliftIO.Exception (finally, tryAny) +import UnliftIO.STM (atomically, modifyTVar', newTVarIO, + readTVar, retrySTM, writeTVar) + +-- | Options for a periodic job. +data PeriodicOptions m = PeriodicOptions + { periodMs :: Int + -- ^ Time (in milliseconds) from the end of one pass to the start of the next. + , runImmediately :: Bool + -- ^ Whether the first pass runs at once rather than after the first period. + , periodJitterMs :: Int + -- ^ Up to this many milliseconds, chosen at random, are added to each period. + , periodicRandom :: Int -> m Int + -- ^ Source of randomness for jitter: a uniformly random value between zero and the argument inclusive. + , onPassFailed :: SomeException -> m () + -- ^ Told of each pass that failed with a synchronous exception. + , periodicGracePeriodMs :: Int + -- ^ How long (in milliseconds) 'stopPeriodic' lets a running pass finish before cancelling it. + , periodicClock :: Clock m + -- ^ Clock used for the period and grace period. + } + +-- | Defaults: every 60 seconds, first pass after one period, no jitter, failures ignored, and a +-- 5 second grace period. +defaultPeriodicOptions :: MonadUnliftIO m => PeriodicOptions m +defaultPeriodicOptions = + PeriodicOptions + { periodMs = 60000 + , runImmediately = False + , periodJitterMs = 0 + , periodicRandom = \n -> liftIO $ randomRIO (0, max 0 n) + , onPassFailed = const (pure ()) + , periodicGracePeriodMs = 5000 + , periodicClock = systemClock + } + +-- | Handle used to control a periodic job. +data PeriodicHandle m = PeriodicHandle + { stopPeriodic :: m Bool + -- ^ Stops the job: no new pass starts, and a running pass gets 'periodicGracePeriodMs' to finish + -- before it is cancelled. Returns 'True' if it stopped without cancelling a pass. + , runNow :: m Bool + -- ^ Triggers a pass, queued behind any pass that is running, and waits for it to complete. + -- Returns 'False' if the job stopped first. + , passesCompleted :: m Int + -- ^ Number of passes that have finished, whether they succeeded or failed. + } + +-- | Starts running the pass periodically in the background. +periodically :: forall m. MonadUnliftIO m => PeriodicOptions m -> m () -> m (PeriodicHandle m) +periodically options pass = do + stoppingVar <- newTVarIO False + requestedVar <- newTVarIO (0 :: Int) + runningVar <- newTVarIO False + completedVar <- newTVarIO (0 :: Int) + endedVar <- newTVarIO False + let clock = periodicClock options + + waitForPeriod :: m () + waitForPeriod = do + jitter <- if periodJitterMs options > 0 then periodicRandom options (periodJitterMs options) else pure 0 + let delayUs = (max 0 (periodMs options) + max 0 (min (periodJitterMs options) jitter)) * 1000 + void $ clockTimeout clock delayUs $ atomically $ do + stopping <- readTVar stoppingVar + requested <- readTVar requestedVar + unless (stopping || requested > 0) retrySTM + + runPass :: m () + runPass = do + result <- tryAny pass `finally` atomically (writeTVar runningVar False >> modifyTVar' completedVar (+ 1)) + either (void . tryAny . onPassFailed options) pure result + + loop :: Bool -> m () + loop first = do + unless (first && runImmediately options) waitForPeriod + started <- atomically $ do + stopping <- readTVar stoppingVar + if stopping + then pure False + else do + writeTVar runningVar True + writeTVar requestedVar 0 + pure True + when started $ runPass >> loop False + + worker <- async (loop True `finally` atomically (writeTVar endedVar True)) + pure + PeriodicHandle + { stopPeriodic = do + atomically $ writeTVar stoppingVar True + drained <- clockTimeout clock (max 0 (periodicGracePeriodMs options) * 1000) $ atomically $ do + running <- readTVar runningVar + when running retrySTM + cancel worker + pure (isJust drained) + , runNow = do + target <- atomically $ do + stopping <- readTVar stoppingVar + if stopping + then pure Nothing + else do + running <- readTVar runningVar + completed <- readTVar completedVar + modifyTVar' requestedVar (+ 1) + pure (Just (completed + if running then 2 else 1)) + case target of + Nothing -> pure False + Just wanted -> atomically $ do + completed <- readTVar completedVar + ended <- readTVar endedVar + if completed >= wanted + then pure True + else if ended then pure False else retrySTM + , passesCompleted = atomically $ readTVar completedVar + } diff --git a/src/Glue/Preload.hs b/src/Glue/Preload.hs --- a/src/Glue/Preload.hs +++ b/src/Glue/Preload.hs @@ -19,11 +19,12 @@ import qualified Data.HashMap.Strict as M import qualified Data.HashSet as S import Data.Maybe (fromJust) +import Glue.Clock import Glue.Types import UnliftIO (Exception, MonadIO, MonadUnliftIO, SomeException (..), liftIO) import UnliftIO.Async (race_) -import UnliftIO.Concurrent (forkIO, threadDelay) +import UnliftIO.Concurrent (forkIO) import UnliftIO.Exception (throwIO) import UnliftIO.IORef (IORef, atomicModifyIORef', newIORef, readIORef) @@ -37,15 +38,18 @@ -- ^ Amount of time between refreshes. , preloadingRun :: MToIO m -- ^ Get an 'IO' of the response for the caching. + , preloadingClock :: Clock m + -- ^ Clock used to wait between refreshes. } -- | Defaulted options for preloading a 'S.HashSet' of keys with a 30 second refresh time. -defaultPreloadedOptions :: S.HashSet a -> MToIO m -> PreloadedOptions m a b +defaultPreloadedOptions :: MonadUnliftIO m => S.HashSet a -> MToIO m -> PreloadedOptions m a b defaultPreloadedOptions toPreload pRun = PreloadedOptions { preloadedKeys = toPreload , preloadingRefreshTimeMs = 30 * 1000 , preloadingRun = pRun + , preloadingClock = systemClock } data PreloadedState a b @@ -105,7 +109,7 @@ continue <- liftIO $ preloadingRun runUpdate if continue then do - stopped <- liftIO $ raceWait stopVar (preloadingRefreshTimeMs * 1000) + stopped <- liftIO $ raceWait stopVar (preloadingRun (clockSleep preloadingClock (preloadingRefreshTimeMs * 1000))) if stopped then pure () else updatePreloaded else pure () let plService request = do @@ -123,9 +127,9 @@ void $ liftIO $ forkIO updatePreloaded pure (plService, stop) where - raceWait :: MVar () -> Int -> IO Bool - raceWait stopVar delayUs = do - race_ (readMVar stopVar) (threadDelay delayUs) + raceWait :: MVar () -> IO () -> IO Bool + raceWait stopVar sleep = do + race_ (readMVar stopVar) sleep not <$> tryPutMVar stopVar () -- | Preloads a single parameter-less call. @@ -141,7 +145,7 @@ -> n (m b, () -> n ()) preloadingCall toIO timeBetweenRefreshes call = do let multiGetService = basicToMultiGet (\_ -> call) :: MultiGetService m () b - let options = PreloadedOptions (S.singleton ()) timeBetweenRefreshes toIO + let options = (defaultPreloadedOptions (S.singleton ()) toIO) { preloadingRefreshTimeMs = timeBetweenRefreshes } (preloading, stop) <- preloadingService options multiGetService let preloadingBasicService = multiGetToBasic preloading let preloadingCallResult = fromJust <$> preloadingBasicService () diff --git a/src/Glue/Race.hs b/src/Glue/Race.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Race.hs @@ -0,0 +1,144 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | Module providing a low-level race of concurrent attempts, for callers that make their own +-- decisions about when to launch another attempt and what to do as each one ends (see +-- 'Glue.Hedging.hedgeAcross' for a ready-made policy built on it). +-- +-- Attempts are numbered from 0 in launch order. 'awaitNext' reports each attempt's ending once, +-- earliest index first when several have ended. 'abandon' cancels an attempt and waits until it +-- has actually finished, reporting how it really ended: an attempt that completed just before the +-- cancel reports its result. Leaving 'withRace' abandons every attempt still running. +module Glue.Race + ( Race + , AttemptOutcome (..) + , RaceEvent (..) + , withRace + , launch + , awaitNext + , abandon + , abandonAll + , finishedAttempts + ) where + +import Control.Monad (filterM) +import qualified Data.Map.Strict as M +import Data.Maybe (catMaybes, fromMaybe) +import Glue.Clock +import Glue.Failure +import UnliftIO (Async, AsyncCancelled (..), MonadUnliftIO, + SomeException, fromException) +import UnliftIO.Async (asyncWithUnmask, cancel, pollSTM, + waitCatch) +import UnliftIO.Exception (bracket, mask_) +import UnliftIO.STM (TVar, atomically, modifyTVar', newTVarIO, + orElse, readTVar, retrySTM, stateTVar) + +-- | How an attempt ended. +data AttemptOutcome failure r + = AttemptAnswered r + -- ^ The attempt returned a result. + | AttemptFailed failure + -- ^ The attempt failed, as seen by the race's failure mode. + | AttemptCancelled + -- ^ The attempt was abandoned before it ended. + | AttemptCrashed SomeException + -- ^ The attempt threw outside its failure mode: an asynchronous exception, or any exception + -- from a service whose failures are values. + deriving (Show) + +-- | What 'awaitNext' observed. +data RaceEvent failure r + = AttemptEnded Int (AttemptOutcome failure r) + -- ^ The attempt with this index ended. + | RaceTimedOut + -- ^ The limit passed with no attempt ending. + | NothingRunning + -- ^ Every launched attempt has already been reported or abandoned. + deriving (Show) + +-- | A race of attempts, each producing an @r@ whose failures are seen through a t'FailureMode'. +data Race m failure r = Race + { raceMode :: FailureMode m r failure + , raceClock :: Clock m + , raceRunning :: TVar (M.Map Int (Async (Either failure r))) + -- ^ Attempts not yet reported or abandoned. + , raceEnded :: TVar (M.Map Int (AttemptOutcome failure r)) + , raceNextId :: TVar Int + } + +toOutcome :: Either SomeException (Either failure r) -> AttemptOutcome failure r +toOutcome (Right (Right result)) = AttemptAnswered result +toOutcome (Right (Left failure)) = AttemptFailed failure +toOutcome (Left e) = case fromException e of + Just AsyncCancelled -> AttemptCancelled + Nothing -> AttemptCrashed e + +-- | Runs a race, abandoning any attempts still running when the body finishes or throws. +withRace :: MonadUnliftIO m => FailureMode m r failure -> Clock m -> (Race m failure r -> m x) -> m x +withRace mode clock = bracket create abandonAll + where + create = Race mode clock <$> newTVarIO M.empty <*> newTVarIO M.empty <*> newTVarIO 0 + +-- | Launches an attempt, returning its index. +launch :: MonadUnliftIO m => Race m failure r -> m r -> m Int +launch race action = mask_ $ do + attempt <- asyncWithUnmask $ \unmask -> unmask (observeCall (raceMode race) action) + atomically $ do + index <- stateTVar (raceNextId race) (\i -> (i, i + 1)) + modifyTVar' (raceRunning race) (M.insert index attempt) + pure index + +-- | Waits for the next attempt to end, for at most the given number of milliseconds if given +-- (zero or less only checks). When several have ended, the lowest index is reported first. +awaitNext :: MonadUnliftIO m => Race m failure r -> Maybe Int -> m (RaceEvent failure r) +awaitNext race limitMs = + case limitMs of + Nothing -> atomically next + Just ms + | ms <= 0 -> atomically (next `orElse` pure RaceTimedOut) + | otherwise -> fromMaybe RaceTimedOut <$> clockTimeout (raceClock race) (ms * 1000) (atomically next) + where + next = do + running <- readTVar (raceRunning race) + if M.null running + then pure NothingRunning + else do + polled <- traverse (\(index, attempt) -> fmap ((,) index) <$> pollSTM attempt) (M.toAscList running) + case catMaybes polled of + [] -> retrySTM + (index, result) : _ -> do + let outcome = toOutcome result + modifyTVar' (raceRunning race) (M.delete index) + modifyTVar' (raceEnded race) (M.insert index outcome) + pure (AttemptEnded index outcome) + +-- | Cancels an attempt and waits until it has finished, returning how it really ended. An +-- attempt already reported returns its reported outcome; an index never launched reports +-- 'AttemptCancelled'. +abandon :: MonadUnliftIO m => Race m failure r -> Int -> m (AttemptOutcome failure r) +abandon race index = do + (ended, running) <- atomically $ (,) <$> (M.lookup index <$> readTVar (raceEnded race)) + <*> (M.lookup index <$> readTVar (raceRunning race)) + case (ended, running) of + (Just outcome, _) -> pure outcome + (Nothing, Nothing) -> pure AttemptCancelled + (Nothing, Just attempt) -> do + cancel attempt + outcome <- toOutcome <$> waitCatch attempt + atomically $ do + modifyTVar' (raceRunning race) (M.delete index) + modifyTVar' (raceEnded race) (M.insert index outcome) + pure outcome + +-- | Abandons every attempt not yet reported, in index order, returning how each ended. +abandonAll :: MonadUnliftIO m => Race m failure r -> m [(Int, AttemptOutcome failure r)] +abandonAll race = do + running <- atomically $ M.keys <$> readTVar (raceRunning race) + traverse (\index -> (,) index <$> abandon race index) running + +-- | Indices of attempts that have ended but not yet been reported by 'awaitNext'. +finishedAttempts :: MonadUnliftIO m => Race m failure r -> m [Int] +finishedAttempts race = atomically $ do + running <- M.toAscList <$> readTVar (raceRunning race) + fmap fst <$> filterM (fmap (maybe False (const True)) . pollSTM . snd) running diff --git a/src/Glue/Retry.hs b/src/Glue/Retry.hs --- a/src/Glue/Retry.hs +++ b/src/Glue/Retry.hs @@ -2,71 +2,147 @@ {-# LANGUAGE ScopedTypeVariables #-} -- | Module containing retry functionality, allowing the construction of services that attempt multiple times in case of transient failure. +-- +-- Whether to retry is decided by the failure ('retryOn') as well as the request ('retryAllowed'). +-- Waits between attempts follow a 'Backoff', optionally capped and jittered, and a retry whose wait +-- would pass the caller's deadline ('retryRemainingMs') is skipped, returning the last failure. +-- Asynchronous exceptions are never retried. module Glue.Retry ( RetryOptions (..) + , Backoff (..) + , Jitter (..) , defaultRetryOptions + , backoffDelayMs , retryingService + , retryingServiceEither ) where -import Control.Monad (when) +import Glue.Clock +import Glue.Failure import Glue.Types -import UnliftIO (MonadUnliftIO, SomeAsyncException, - SomeException, fromException) -import UnliftIO.Concurrent (threadDelay) -import UnliftIO.Exception (catch, throwIO) +import System.Random (randomRIO) +import UnliftIO (MonadUnliftIO, SomeException, liftIO) +-- | How the wait before each retry grows. +data Backoff + = ConstantBackoff Int + -- ^ The same wait (in milliseconds) before every retry. + | LinearBackoff Int Int + -- ^ An initial wait, growing by the second value (both in milliseconds) for each retry. + | ExponentialBackoff Int Double + -- ^ An initial wait in milliseconds, multiplied by the second value for each retry. + deriving (Eq, Show) + +-- | Randomisation of the wait, so that many clients don't retry in lockstep. +-- See . +data Jitter + = NoJitter + -- ^ Wait exactly the backoff. + | FullJitter + -- ^ Wait a uniformly random time between zero and the backoff. + | EqualJitter + -- ^ Wait half the backoff plus a uniformly random time up to the other half. + deriving (Eq, Show) + -- | Options for determining behaviour of retrying services. -data RetryOptions a = RetryOptions - { retryAllowed :: a -> Bool +data RetryOptions m a failure = RetryOptions + { retryAllowed :: a -> Bool -- ^ Predicate for determining if we can retry a call, can be used to prevent retries on non-idempotent operations. - , retryInitialWaitTimeMs :: Int - -- ^ Amount of time (in milliseconds) to wait after the first failure. - , maximumRetries :: Int + , retryOn :: failure -> Bool + -- ^ Predicate on the failure: retry an unreachable backend or a timeout, say, but not a rejected request. + , maximumRetries :: Int -- ^ The upper bound on how many retry attempts to make when invoking the service. - , retryWaitTimeMultiplier :: Double - -- ^ How much to multiply 'retryInitialWaitTimeMs' by for each subsequent retry. + , retryBackoff :: Backoff + -- ^ How long to wait before each retry. + , retryMaxDelayMs :: Maybe Int + -- ^ Ceiling on the wait before any retry, applied before jitter. + , retryJitter :: Jitter + -- ^ Randomisation applied to the capped wait. + , retryRandom :: Int -> m Int + -- ^ Source of randomness for jitter: a uniformly random value between zero and the argument inclusive. + , retryRemainingMs :: m (Maybe Int) + -- ^ Time left (in milliseconds) before the caller's deadline, if there is one. A retry whose wait + -- would pass it is skipped. 'remainingUntil' builds this from a deadline on a t'Clock'. + , onRetry :: Int -> failure -> Int -> m () + -- ^ Called before each retry with the retry number (from 1), the failure and the chosen wait in milliseconds. + , retryClock :: Clock m + -- ^ Clock used to wait between attempts. } --- | Defaulted options for retrying 3 times with no wait time. -defaultRetryOptions :: RetryOptions a +-- | Defaulted options for retrying any failure 3 times with no wait time and no deadline. +defaultRetryOptions :: MonadUnliftIO m => RetryOptions m a failure defaultRetryOptions = RetryOptions - { retryAllowed = const True - , retryInitialWaitTimeMs = 0 - , maximumRetries = 3 - , retryWaitTimeMultiplier = 0 + { retryAllowed = const True + , retryOn = const True + , maximumRetries = 3 + , retryBackoff = ConstantBackoff 0 + , retryMaxDelayMs = Nothing + , retryJitter = NoJitter + , retryRandom = \n -> liftIO $ randomRIO (0, max 0 n) + , retryRemainingMs = pure Nothing + , onRetry = \_ _ _ -> pure () + , retryClock = systemClock } -possibleAsyncException :: SomeException -> Maybe SomeAsyncException -possibleAsyncException = fromException +-- | The un-jittered, uncapped wait in milliseconds before the retry with the given index (from 0). +backoffDelayMs :: Backoff -> Int -> Int +backoffDelayMs (ConstantBackoff delay) _ = max 0 delay +backoffDelayMs (LinearBackoff initial step) n = max 0 (initial + step * n) +backoffDelayMs (ExponentialBackoff initial multiplier) n = + let factor = if multiplier <= 0 then 1 else multiplier + delay = fromIntegral (max 0 initial) * factor ^ n :: Double + in if delay >= fromIntegral (maxBound :: Int) then maxBound else round delay +chooseDelay :: Monad m => RetryOptions m a failure -> Int -> m Int +chooseDelay options n = do + let capped = maybe id (min . max 0) (retryMaxDelayMs options) (backoffDelayMs (retryBackoff options) n) + case retryJitter options of + NoJitter -> pure capped + FullJitter -> clamp capped <$> retryRandom options capped + EqualJitter -> do + let half = capped `div` 2 + (+ half) . clamp (capped - half) <$> retryRandom options (capped - half) + where + clamp upper = max 0 . min upper + +retryVia :: Monad m => FailureMode m r failure -> RetryOptions m a failure -> BasicService m a r -> BasicService m a r +retryVia mode options service request = attempt 0 + where + attempt retryCount = do + outcome <- observeCall mode (service request) + case outcome of + Right result -> pure result + Left failure + | retryCount < maximumRetries options && retryAllowed options request && retryOn options failure -> do + delay <- chooseDelay options retryCount + remaining <- retryRemainingMs options + if maybe False (\left -> left <= 0 || delay > left) remaining + then reraise mode failure + else do + onRetry options (retryCount + 1) failure delay + clockSleep (retryClock options) (delay * 1000) + attempt (retryCount + 1) + | otherwise -> reraise mode failure + -- | Retries a call to a service multiple times, potentially backing off wait times between subsequent calls. -- Asynchronous exceptions don't result in a retry; they are immediately rethrown. retryingService - :: forall m a b. - MonadUnliftIO m - => RetryOptions a + :: MonadUnliftIO m + => RetryOptions m a SomeException -- ^ Instance of t'RetryOptions' to configure the retry functionality. -> BasicService m a b -- ^ The service to perform retries of. -> BasicService m a b -retryingService options service = - let catchHandler rc r e = - case possibleAsyncException e of - Just ae -> throwIO ae - Nothing -> wait rc >> attempt (rc + 1) r - attempt retryCount request = - if retryAllowed options request && maxRetries > retryCount - then catch (service request) (catchHandler retryCount request) - else service request - maxRetries = maximumRetries options - initialUs = retryInitialWaitTimeMs options * 1000 - multiplier = - if retryWaitTimeMultiplier options <= 0 - then 1.0 - else retryWaitTimeMultiplier options - wait :: Int -> m () - wait retryCount = - when (initialUs > 0) $ - threadDelay $ round $ fromIntegral initialUs * (multiplier ^ retryCount) - in attempt 0 +retryingService = retryVia exceptionFailures + +-- | Variant of 'retryingService' for services returning failures as values: @Left@ results are +-- judged by 'retryOn', and the last @Left@ is returned once retrying stops. Exceptions are not caught. +retryingServiceEither + :: Monad m + => RetryOptions m a e + -- ^ Instance of t'RetryOptions' to configure the retry functionality. + -> EitherService m a e b + -- ^ The service to perform retries of. + -> EitherService m a e b +retryingServiceEither = retryVia valueFailures diff --git a/src/Glue/Shutdown.hs b/src/Glue/Shutdown.hs --- a/src/Glue/Shutdown.hs +++ b/src/Glue/Shutdown.hs @@ -11,18 +11,21 @@ , ShutdownHandle (..) , defaultShutdownOptions , gracefulShutdown + , gracefulShutdownEither ) where +import Data.Maybe (fromMaybe) +import Glue.Clock +import Glue.Failure import Glue.Types import UnliftIO (Exception, MonadIO, MonadUnliftIO) import UnliftIO.Async (race) import UnliftIO.Exception (finally, throwIO) import UnliftIO.STM (atomically, modifyTVar', newTVarIO, readTVar, retrySTM, writeTVar) -import UnliftIO.Timeout (timeout) -- | Options for configuring graceful service shutdown. -data ShutdownOptions b = ShutdownOptions +data ShutdownOptions m b = ShutdownOptions { shutdownFallback :: Maybe b -- ^ Optional default value returned for rejected requests after shutdown is initiated. -- If 'Nothing', t'ServiceShuttingDownException' is thrown instead. @@ -30,16 +33,18 @@ -- ^ Maximum time in milliseconds to wait for existing in-flight requests to complete before cancelling them. , shutdownDescription :: String -- ^ Description attached to t'ServiceShuttingDownException'. + , shutdownClock :: Clock m + -- ^ Clock used to measure the grace period. } - deriving (Eq, Show) -- | Default shutdown options: throws t'ServiceShuttingDownException' on rejected calls and waits up to 5 seconds for in-flight calls. -defaultShutdownOptions :: ShutdownOptions b +defaultShutdownOptions :: MonadUnliftIO m => ShutdownOptions m b defaultShutdownOptions = ShutdownOptions { shutdownFallback = Nothing , shutdownGracePeriodMs = 5000 , shutdownDescription = "Service is shutting down." + , shutdownClock = systemClock } -- | Exception thrown when a service is shutting down and no 'shutdownFallback' is configured. @@ -62,25 +67,42 @@ -- | Wraps a service with graceful shutdown support. gracefulShutdown - :: forall m n a b. - (MonadUnliftIO m, MonadIO n) - => ShutdownOptions b + :: (MonadUnliftIO m, MonadIO n) + => ShutdownOptions m b -- ^ Options configuring fallback behaviour and grace period. -> BasicService m a b -- ^ Service to wrap. -> n (BasicService m a b, ShutdownHandle m) -gracefulShutdown options service = do +gracefulShutdown options = + shutdownVia (maybe (throwIO $ ServiceShuttingDownException (shutdownDescription options)) pure (shutdownFallback options)) options + +-- | Variant of 'gracefulShutdown' for services returning failures as values: rejected calls +-- return 'shutdownFallback' if set, otherwise the given failure, instead of throwing. +gracefulShutdownEither + :: (MonadUnliftIO m, MonadIO n) + => ShutdownOptions m (Either e b) + -- ^ Options configuring fallback behaviour and grace period ('shutdownDescription' is unused). + -> e + -- ^ Failure returned for rejected calls when no 'shutdownFallback' is set. + -> EitherService m a e b + -- ^ Service to wrap. + -> n (EitherService m a e b, ShutdownHandle m) +gracefulShutdownEither options onShuttingDown = + shutdownVia (pure (fromMaybe (Left onShuttingDown) (shutdownFallback options))) options + +shutdownVia + :: forall m n a b. + (MonadUnliftIO m, MonadIO n) + => m b + -> ShutdownOptions m b + -> BasicService m a b + -> n (BasicService m a b, ShutdownHandle m) +shutdownVia rejectRequest options service = do shuttingDownVar <- newTVarIO False forceCancelVar <- newTVarIO False inFlightVar <- newTVarIO (0 :: Int) - let rejectRequest :: m b - rejectRequest = - case shutdownFallback options of - Just fallback -> pure fallback - Nothing -> throwIO $ ServiceShuttingDownException (shutdownDescription options) - - waitForForceCancel :: m () + let waitForForceCancel :: m () waitForForceCancel = atomically $ do cancelled <- readTVar forceCancelVar if cancelled then pure () else retrySTM @@ -114,7 +136,7 @@ { initiateShutdown = do atomically $ writeTVar shuttingDownVar True let graceUs = max 0 (shutdownGracePeriodMs options) * 1000 - drained <- timeout graceUs waitDrained + drained <- clockTimeout (shutdownClock options) graceUs waitDrained case drained of Just () -> pure True Nothing -> do diff --git a/src/Glue/Timeout.hs b/src/Glue/Timeout.hs --- a/src/Glue/Timeout.hs +++ b/src/Glue/Timeout.hs @@ -7,28 +7,32 @@ , TimeoutException (..) , defaultTimeoutOptions , addTimeout + , addTimeoutEither ) where +import Glue.Clock +import Glue.Failure import Glue.Types import UnliftIO (Exception, MonadUnliftIO) import UnliftIO.Exception (throwIO) -import UnliftIO.Timeout (timeout) -- | Options for determining behaviour of services with a timeout. -data TimeoutOptions = TimeoutOptions +data TimeoutOptions m = TimeoutOptions { timeoutDescription :: String -- ^ Description added to the t'TimeoutException' thrown when the timeout is exceeded. , timeoutLimitMs :: Int -- ^ Timeout in milliseconds. + , timeoutClock :: Clock m + -- ^ Clock used to measure the timeout. } - deriving (Eq, Show) -- | Default instance of t'TimeoutOptions' with a timeout of 30 seconds. -defaultTimeoutOptions :: TimeoutOptions +defaultTimeoutOptions :: MonadUnliftIO m => TimeoutOptions m defaultTimeoutOptions = TimeoutOptions { timeoutDescription = "Service call timed out." , timeoutLimitMs = 30000 + , timeoutClock = systemClock } -- | Exception thrown when the timeout is exceeded. @@ -37,14 +41,30 @@ instance Exception TimeoutException +timeoutVia :: Monad m => m r -> TimeoutOptions m -> BasicService m a r -> BasicService m a r +timeoutVia onTimeout options service request = do + maybeResult <- clockTimeout (timeoutClock options) (1000 * timeoutLimitMs options) (service request) + maybe onTimeout pure maybeResult + -- | Function for producing services protected with a timeout. addTimeout :: MonadUnliftIO m - => TimeoutOptions + => TimeoutOptions m -- ^ Options to configure the timeout. -> BasicService m a b -- ^ Service to protect with a timeout. -> BasicService m a b -addTimeout options service request = do - maybeResult <- timeout (1000 * timeoutLimitMs options) (service request) - maybe (throwIO $ TimeoutException $ timeoutDescription options) pure maybeResult +addTimeout options = timeoutVia (throwIO $ TimeoutException $ timeoutDescription options) options + +-- | Variant of 'addTimeout' for services returning failures as values, returning the given +-- failure when the timeout is exceeded instead of throwing. +addTimeoutEither + :: MonadUnliftIO m + => TimeoutOptions m + -- ^ Options to configure the timeout ('timeoutDescription' is unused). + -> e + -- ^ Failure returned when the timeout is exceeded. + -> EitherService m a e b + -- ^ Service to protect with a timeout. + -> EitherService m a e b +addTimeoutEither options onTimeout = timeoutVia (pure (Left onTimeout)) options diff --git a/test/Glue/BatcherSpec.hs b/test/Glue/BatcherSpec.hs --- a/test/Glue/BatcherSpec.hs +++ b/test/Glue/BatcherSpec.hs @@ -18,7 +18,7 @@ let requestsAsSets = fmap S.fromList requests serviceFunctionality rs = M.fromList $ fmap (\r -> (r, r * 2)) $ S.toList rs service = (pure . serviceFunctionality) :: MultiGetService IO Int Int - (_, multiService) <- batchingService (BatchingOptions 2) service + (_, multiService) <- batchingService (defaultBatchingOptions { batchWindowMs = 2 }) service let expectedResults = fmap serviceFunctionality requestsAsSets mapConcurrently multiService requestsAsSets `shouldReturn` expectedResults @@ -26,6 +26,6 @@ property $ \(requests :: [Int]) -> do let serviceFunctionality rs = M.fromList $ fmap (\r -> (r, r * 2)) $ S.toList rs service = (pure . serviceFunctionality) :: MultiGetService IO Int Int - (singleService, _) <- batchingService (BatchingOptions 2) service + (singleService, _) <- batchingService (defaultBatchingOptions { batchWindowMs = 2 }) service let expectedResults = fmap (\r -> Just (r * 2)) requests mapConcurrently singleService requests `shouldReturn` expectedResults diff --git a/test/Glue/CircuitBreakerSpec.hs b/test/Glue/CircuitBreakerSpec.hs --- a/test/Glue/CircuitBreakerSpec.hs +++ b/test/Glue/CircuitBreakerSpec.hs @@ -2,52 +2,180 @@ module Glue.CircuitBreakerSpec (spec) where +import Control.Concurrent.Async (async, cancel, wait) +import Control.Monad (forM, replicateM_) import Data.IORef import Glue.CircuitBreaker +import Glue.Clock +import Glue.Clock.Virtual import Test.Hspec import Test.QuickCheck -import UnliftIO.Exception (Exception, throwIO, try) +import UnliftIO.Exception (Exception, fromException, throwIO, + try) +import UnliftIO.STM (atomically, newEmptyTMVarIO, + putTMVar, readTMVar) data CircuitBreakerTestException = CircuitBreakerTestException deriving (Eq, Show) instance Exception CircuitBreakerTestException -requests :: [Int] -requests = [1 .. 10] +newtype BadRequestException = BadRequestException Int + deriving (Eq, Show) +instance Exception BadRequestException + +type Outcome = Either CircuitBreakerException (Either CircuitBreakerTestException Int) + +failing :: IORef Int -> Int -> IO Int +failing ref _ = atomicModifyIORef' ref (\c -> (c + 1, ())) >> throwIO CircuitBreakerTestException + +callAll :: (Int -> IO Int) -> [Int] -> IO [Outcome] +callAll service = traverse (try . try . service) + +options :: VirtualClock -> CircuitBreakerOptions IO failure +options vc = defaultCircuitBreakerOptions { breakerClock = virtualClock vc } + +openFailure :: Outcome +openFailure = Left (CircuitBreakerException "Circuit breaker open.") + +-- | A scripted outcome for comparing the exception and value shapes. +data Step = Succeed Int | Fail Bool | Wait Int + deriving (Show) + +stepValue :: Step -> Int +stepValue (Succeed n) = n +stepValue (Wait n) = n +stepValue (Fail _) = 0 + +-- | A failure carrying whether it should count against the breaker. +newtype StepFailure = StepFailure Bool + deriving (Eq, Show) + +instance Exception StepFailure + +instance Arbitrary Step where + arbitrary = oneof [Succeed <$> arbitrary, Fail <$> arbitrary, Wait <$> choose (0, 30)] + spec :: Spec spec = do describe "circuitBreaker" $ do - it "Multiple failures prevent subsequent calls" $ - property $ \failureMax -> do - let positiveFailureMax = abs failureMax `mod` 5 - options = defaultCircuitBreakerOptions { maxBreakerFailures = positiveFailureMax } - ref <- newIORef (0 :: Int) - let service _ = atomicModifyIORef' ref (\c -> (c + 1, ())) >> throwIO CircuitBreakerTestException :: IO Int - (s, circuitBreakerService) <- circuitBreaker options service - results <- traverse (try . try . circuitBreakerService) requests :: IO [Either CircuitBreakerException (Either CircuitBreakerTestException Int)] - let expectedResults = replicate (positiveFailureMax + 1) (Right $ Left CircuitBreakerTestException) - ++ replicate (10 - positiveFailureMax - 1) (Left $ CircuitBreakerException "Circuit breaker open.") - results `shouldBe` expectedResults + it "Opens on the failure number given by the threshold" $ + property $ \(threshold' :: Int) -> do + let threshold = 1 + abs threshold' `mod` 5 + vc <- newVirtualClock + ref <- newIORef 0 + (s, service) <- circuitBreaker ((options vc) { breakerFailureThreshold = threshold }) (failing ref) + results <- callAll service [1 .. 10] + results `shouldBe` replicate threshold (Right (Left CircuitBreakerTestException)) ++ replicate (10 - threshold) openFailure isCircuitBreakerOpen s `shouldReturn` True isCircuitBreakerClosed s `shouldReturn` False - readIORef ref `shouldReturn` (positiveFailureMax + 1) + readIORef ref `shouldReturn` threshold it "Successful calls pass straight through" $ - property $ \(failureMax :: Int, requestIDs :: [Int]) -> do - let positiveFailureMax = abs failureMax - options = defaultCircuitBreakerOptions { maxBreakerFailures = positiveFailureMax } - service req = pure $ req + 1 - (s, circuitBreakerService) <- circuitBreaker options service - results <- traverse circuitBreakerService requestIDs - let expectedResults = fmap (+ 1) requestIDs - results `shouldBe` expectedResults - isCircuitBreakerOpen s `shouldReturn` False + property $ \(requestIDs :: [Int]) -> do + (s, service) <- circuitBreaker defaultCircuitBreakerOptions (\req -> pure (req + 1) :: IO Int) + traverse service requestIDs `shouldReturn` fmap (+ 1) requestIDs isCircuitBreakerClosed s `shouldReturn` True + it "Counts consecutive failures, a success resetting the count" $ do + vc <- newVirtualClock + (s, service) <- circuitBreaker (options vc) $ \(req :: Int) -> + if req > 0 then pure req else throwIO CircuitBreakerTestException + _ <- callAll service [0, 0, 1, 0, 0, 1, 0, 0] + isCircuitBreakerClosed s `shouldReturn` True + _ <- callAll service [0] + isCircuitBreakerOpen s `shouldReturn` True + + it "With a failure window, only failures within the window count" $ do + vc <- newVirtualClock + ref <- newIORef 0 + (s, service) <- circuitBreaker ((options vc) { breakerFailureWindowMs = Just 1000 }) (failing ref) + replicateM_ 5 $ do + _ <- callAll service [1, 1] + advanceClock vc 1000000 + isCircuitBreakerClosed s `shouldReturn` True + _ <- callAll service [1, 1] + advanceClock vc 999000 + _ <- callAll service [1] + isCircuitBreakerOpen s `shouldReturn` True + + it "A call cancelled by an asynchronous exception does not count as a failure" $ do + vc <- newVirtualClock + started <- newEmptyTMVarIO + (s, service) <- circuitBreaker ((options vc) { breakerFailureThreshold = 1 }) $ \(req :: Int) -> do + atomically $ putTMVar started () + clockSleep (virtualClock vc) 1000000 + pure req + call <- async (service 1) + atomically $ readTMVar started + cancel call + isCircuitBreakerClosed s `shouldReturn` True + + it "Stays open for the open duration, then admits a single trial" $ do + vc <- newVirtualClock + gate <- newEmptyTMVarIO + healthy <- newIORef False + let service (req :: Int) = do + ok <- readIORef healthy + if ok then atomically (readTMVar gate) >> pure req else throwIO CircuitBreakerTestException + (s, breaker) <- circuitBreaker ((options vc) { breakerFailureThreshold = 1, breakerOpenDurationMs = 500 }) service + _ <- callAll breaker [1] + advanceClock vc 499000 + callAll breaker [1] `shouldReturn` [openFailure] + advanceClock vc 1000 + writeIORef healthy True + trial <- async (breaker 7) + let waitForTrial = circuitBreakerPhases s >>= \p -> if p == [BreakerHalfOpen] then pure () else waitForTrial + waitForTrial + callAll breaker [1] `shouldReturn` [openFailure] + atomically $ putTMVar gate () + wait trial `shouldReturn` 7 + isCircuitBreakerClosed s `shouldReturn` True + + it "A failed trial opens the circuit again for the open duration" $ do + vc <- newVirtualClock + ref <- newIORef 0 + (s, breaker) <- circuitBreaker ((options vc) { breakerFailureThreshold = 1, breakerOpenDurationMs = 500 }) (failing ref) + _ <- callAll breaker [1] + advanceClock vc 500000 + callAll breaker [1, 1] `shouldReturn` [Right (Left CircuitBreakerTestException), openFailure] + circuitBreakerPhases s `shouldReturn` [BreakerOpen] + readIORef ref `shouldReturn` 2 + + it "Failures rejected by countsAsFailure do not count" $ do + vc <- newVirtualClock + let counts e = maybe True (const False) (fromException e :: Maybe BadRequestException) + (s, breaker) <- circuitBreaker ((options vc) { breakerFailureThreshold = 1, countsAsFailure = counts }) $ \(req :: Int) -> + throwIO (BadRequestException req) :: IO Int + results <- forM [1 .. 5] (try . breaker) + results `shouldBe` fmap (Left . BadRequestException) [1 .. 5] + isCircuitBreakerClosed s `shouldReturn` True + + it "Reports each transition to the observer" $ do + vc <- newVirtualClock + seen <- newIORef [] + ref <- newIORef 0 + healthy <- newIORef False + let service (req :: Int) = readIORef healthy >>= \ok -> if ok then pure req else failing ref req + observe t = atomicModifyIORef' seen (\ts -> (ts ++ [t], ())) + (_, breaker) <- circuitBreaker ((options vc) { breakerFailureThreshold = 2, breakerOpenDurationMs = 10, onBreakerTransition = observe }) service + _ <- callAll breaker [1, 1] + advanceClock vc 10000 + _ <- callAll breaker [1] + advanceClock vc 10000 + writeIORef healthy True + _ <- callAll breaker [1] + readIORef seen `shouldReturn` + [ BreakerTransition BreakerClosed BreakerOpen + , BreakerTransition BreakerOpen BreakerHalfOpen + , BreakerTransition BreakerHalfOpen BreakerOpen + , BreakerTransition BreakerOpen BreakerHalfOpen + , BreakerTransition BreakerHalfOpen BreakerClosed + ] + it "combineCircuitBreakerStates reports open if any breaker is open" $ do - let opts = defaultCircuitBreakerOptions { maxBreakerFailures = 0 } + let opts = (defaultCircuitBreakerOptions :: CircuitBreakerOptions IO e) { breakerFailureThreshold = 1 } (s1, svc1) <- circuitBreaker opts (\(_ :: Int) -> throwIO CircuitBreakerTestException :: IO Int) (s2, _) <- circuitBreaker opts (\(x :: Int) -> pure (x + 1) :: IO Int) let combined = combineCircuitBreakerStates [s1, s2] @@ -56,3 +184,45 @@ _ <- try (svc1 1) :: IO (Either CircuitBreakerTestException Int) isCircuitBreakerOpen combined `shouldReturn` True isCircuitBreakerClosed combined `shouldReturn` False + + describe "circuitBreakerWithStore" $ + it "Never calls the service when the store refuses" $ do + ref <- newIORef 0 + let store = BreakerStore (pure (Refused, Nothing)) (\_ _ -> pure Nothing) (pure BreakerOpen) + breaker = circuitBreakerWithStore defaultCircuitBreakerOptions store (failing ref) + callAll breaker [1, 2] `shouldReturn` [openFailure, openFailure] + readIORef ref `shouldReturn` 0 + + describe "circuitBreakerEither" $ do + it "Counts Left failures and refuses with the given value" $ do + vc <- newVirtualClock + (s, breaker) <- circuitBreakerEither ((options vc) { breakerFailureThreshold = 2 }) "open" $ \(_ :: Int) -> + pure (Left "down" :: Either String Int) + traverse breaker [1, 2, 3] `shouldReturn` [Left "down", Left "down", Left "open"] + isCircuitBreakerOpen s `shouldReturn` True + + it "Behaves like the exception shape for the same outcomes" $ + property $ \(steps :: [Step]) -> do + vc1 <- newVirtualClock + vc2 <- newVirtualClock + let opts vc = (options vc) { breakerFailureThreshold = 2, breakerOpenDurationMs = 20 } + countsException e = maybe True (\(StepFailure c) -> c) (fromException e) + asException (Fail c) = throwIO (StepFailure c) + asException step = pure (stepValue step) + asValue (Fail c) = pure (Left (Just c)) + asValue step = pure (Right (stepValue step)) + (_, exceptional) <- circuitBreaker ((opts vc1) { countsAsFailure = countsException }) asException + (_, valued) <- circuitBreakerEither ((opts vc2) { countsAsFailure = or }) Nothing asValue + let run step = do + e <- try (try (exceptional step)) + v <- valued step + case step of + Wait n -> advanceClock vc1 (n * 1000) >> advanceClock vc2 (n * 1000) + _ -> pure () + let normalised = case e of + Left (CircuitBreakerException _) -> Left Nothing + Right (Left (StepFailure c)) -> Left (Just c) + Right (Right n) -> Right n + pure (normalised, v) + pairs <- traverse run steps + fmap fst pairs `shouldBe` fmap snd pairs diff --git a/test/Glue/ConcurrencyLimitSpec.hs b/test/Glue/ConcurrencyLimitSpec.hs --- a/test/Glue/ConcurrencyLimitSpec.hs +++ b/test/Glue/ConcurrencyLimitSpec.hs @@ -2,18 +2,24 @@ module Glue.ConcurrencyLimitSpec (spec) where -import Control.Concurrent (threadDelay) -import Control.Concurrent.Async (async, mapConcurrently, wait) +import Control.Concurrent (threadDelay, yield) +import Control.Concurrent.Async (async, cancel, mapConcurrently, poll, + wait) import Data.IORef import Glue.ConcurrencyLimit import Test.Hspec import UnliftIO.Exception (Exception, throwIO, try) +import UnliftIO.STM (atomically, newEmptyTMVarIO, + putTMVar, readTMVar) data ConcurrencyTestException = ConcurrencyTestException deriving (Eq, Show) instance Exception ConcurrencyTestException +waitUntil :: IO Bool -> IO () +waitUntil condition = condition >>= \done -> if done then pure () else yield >> waitUntil condition + spec :: Spec spec = do describe "limitConcurrency" $ do @@ -40,21 +46,25 @@ , maxQueued = Just 1 , concurrencyLimitDescription = "Busy" } - service (req :: Int) = do - threadDelay (60 * 1000) + gate <- newEmptyTMVarIO + entered <- newIORef (0 :: Int) + let service (req :: Int) = do + atomicModifyIORef' entered (\n -> (n + 1, ())) + atomically $ readTMVar gate pure req limited <- limitConcurrency opts service a1 <- async (try (limited 1) :: IO (Either ConcurrencyLimitExceededException Int)) - threadDelay (10 * 1000) + waitUntil ((== 1) <$> readIORef entered) + -- With 1 in flight, one of these queues and the other is rejected at once. a2 <- async (try (limited 2) :: IO (Either ConcurrencyLimitExceededException Int)) - threadDelay (10 * 1000) - -- 3rd request should be rejected immediately since 1 is in-flight and 1 is queued - r3 <- try (limited 3) :: IO (Either ConcurrencyLimitExceededException Int) - r1 <- wait a1 - r2 <- wait a2 - r3 `shouldBe` Left (ConcurrencyLimitExceededException "Busy") - r1 `shouldBe` Right 1 - r2 `shouldBe` Right 2 + a3 <- async (try (limited 3) :: IO (Either ConcurrencyLimitExceededException Int)) + let rejected = fmap (maybe False (either (const False) (either (const True) (const False)))) . poll + waitUntil ((||) <$> rejected a2 <*> rejected a3) + atomically $ putTMVar gate () + wait a1 `shouldReturn` Right 1 + results <- traverse wait [a2, a3] + length [() | Left (ConcurrencyLimitExceededException "Busy") <- results] `shouldBe` 1 + length [r | Right r <- results, r `elem` [2, 3]] `shouldBe` 1 it "Releases slot when underlying service throws an exception" $ do let opts = defaultConcurrencyLimitOptions { maxInFlight = 1, maxQueued = Just 0 } @@ -63,3 +73,96 @@ limited <- limitConcurrency opts failingOrOk limited 0 `shouldThrow` (== ConcurrencyTestException) limited 5 `shouldReturn` 15 + + describe "limitConcurrencyEither" $ + it "Returns the saturation failure as a value" $ do + gate <- newEmptyTMVarIO + started <- newEmptyTMVarIO + let opts = defaultConcurrencyLimitOptions { maxInFlight = 1, maxQueued = Just 0 } + limited <- limitConcurrencyEither opts "busy" $ \(req :: Int) -> do + atomically $ putTMVar started () + atomically $ readTMVar gate + pure (Right req) + first <- async (limited 1) + atomically $ readTMVar started + limited 2 `shouldReturn` (Left "busy" :: Either String Int) + atomically $ putTMVar gate () + wait first `shouldReturn` Right 1 + + describe "limitConcurrencyBy" $ do + it "Refuses a second call on a key with a limit of one" $ do + store <- newInMemoryAdmissionStore (const 1) + gate <- newEmptyTMVarIO + let limited = limitConcurrencyByWithStore store (`div` 10) (\(req :: Int) -> atomically (readTMVar gate) >> pure req) + first <- async (limited 1) + waitUntil ((== 1) <$> slotsInUse store 0) + limited 2 `shouldReturn` Left Saturated + atomically $ putTMVar gate () + wait first `shouldReturn` Right 1 + limited 3 `shouldReturn` Right 3 + + it "Keeps keys isolated" $ do + store <- newInMemoryAdmissionStore (const 1) + gate <- newEmptyTMVarIO + let limited = limitConcurrencyByWithStore store (`div` 10) (\(req :: Int) -> atomically (readTMVar gate) >> pure req) + first <- async (limited 1) + waitUntil ((== 1) <$> slotsInUse store 0) + other <- async (limited 11) + waitUntil ((== 1) <$> slotsInUse store 1) + atomically $ putTMVar gate () + wait first `shouldReturn` Right 1 + wait other `shouldReturn` Right 11 + + it "Uses a separate limit per key" $ do + store <- newInMemoryAdmissionStore (\k -> if k == "wide" then 2 else 1) + gate <- newEmptyTMVarIO + let limited = limitConcurrencyByWithStore store fst (\(_, n :: Int) -> atomically (readTMVar gate) >> pure n) + a <- async (limited ("wide" :: String, 1)) + b <- async (limited ("wide", 2)) + waitUntil ((== 2) <$> slotsInUse store "wide") + limited ("wide", 3) `shouldReturn` Left Saturated + atomically $ putTMVar gate () + wait a `shouldReturn` Right 1 + wait b `shouldReturn` Right 2 + + it "Releases the slot on exception and on cancellation, leaving no keys behind" $ do + store <- newInMemoryAdmissionStore (const 1) + let failing (_ :: Int) = throwIO ConcurrencyTestException :: IO Int + limitConcurrencyByWithStore store id failing 1 `shouldThrow` (== ConcurrencyTestException) + slotsInUse store 1 `shouldReturn` 0 + gate <- newEmptyTMVarIO + blocked <- async (limitConcurrencyByWithStore store id (\(req :: Int) -> atomically (readTMVar gate) >> pure req) 2) + waitUntil ((== 1) <$> slotsInUse store 2) + cancel blocked + slotsInUse store 2 `shouldReturn` 0 + trackedKeys store `shouldReturn` 0 + + it "Never exceeds the limit for a key under concurrent load" $ do + current <- newIORef (0 :: Int) + peak <- newIORef (0 :: Int) + let service (req :: Int) = do + cur <- atomicModifyIORef' current (\c -> (c + 1, c + 1)) + atomicModifyIORef' peak (\p -> (max p cur, ())) + yield + atomicModifyIORef' current (\c -> (c - 1, ())) + pure req + limited <- limitConcurrencyBy (const ()) (const 3) service + results <- mapConcurrently limited [1 .. 200] + length [r | Right r <- results] `shouldSatisfy` (> 0) + readIORef peak >>= (`shouldSatisfy` (<= 3)) + + describe "withSlot" $ + it "Holds the slot until the work finishes, beyond any service call" $ do + store <- newInMemoryAdmissionStore (const 1) + gate <- newEmptyTMVarIO + streaming <- async (withSlot store "stream" (atomically (readTMVar gate) >> pure "consumed")) + waitUntil ((== 1) <$> slotsInUse store ("stream" :: String)) + withSlot store "stream" (pure "second") `shouldReturn` Left Saturated + atomically $ putTMVar gate () + wait streaming `shouldReturn` Right "consumed" + withSlot store "stream" (pure "second") `shouldReturn` Right "second" + + describe "limitConcurrencyByEither" $ + it "Returns the given failure when the key is full" $ do + limited <- limitConcurrencyByEither (const ()) (const 0) "full" (\(req :: Int) -> pure (Right req)) + limited 1 `shouldReturn` (Left "full" :: Either String Int) diff --git a/test/Glue/DogpileProtectionSpec.hs b/test/Glue/DogpileProtectionSpec.hs --- a/test/Glue/DogpileProtectionSpec.hs +++ b/test/Glue/DogpileProtectionSpec.hs @@ -2,12 +2,16 @@ module Glue.DogpileProtectionSpec (spec) where -import Control.Concurrent (threadDelay) +import Control.Concurrent (yield) import Control.Concurrent.Async (async, wait) import Data.IORef +import Glue.Clock +import Glue.Clock.Virtual import Glue.DogpileProtection import Test.Hspec import UnliftIO.Exception (Exception, throwIO) +import UnliftIO.STM (atomically, newEmptyTMVarIO, + putTMVar, readTMVar) data DogpileProtectionTestException = DogpileProtectionTestException deriving (Eq, Show) @@ -17,30 +21,55 @@ requests :: [Int] requests = replicate 10 1 -delayTime :: Int -delayTime = 50 * 1000 +waitForCount :: IORef Int -> Int -> IO () +waitForCount ref n = readIORef ref >>= \c -> if c >= n then pure () else yield >> waitForCount ref n spec :: Spec spec = do describe "dogpileProtect" $ do it "With multiple calls to a slow service only one actually gets through" $ do counter <- newIORef (0 :: Int) + gate <- newEmptyTMVarIO let service request = do atomicModifyIORef' counter (\n -> (n + 1, ())) - threadDelay delayTime + atomically $ readTMVar gate pure (request * 2) protectedService <- dogpileProtect service asyncResults <- traverse (async . protectedService) requests + waitForCount counter 1 + atomically $ putTMVar gate () traverse wait asyncResults `shouldReturn` fmap (* 2) requests readIORef counter `shouldReturn` 1 it "With multiple calls to a slow failing service only one actually gets through" $ do counter <- newIORef (0 :: Int) + gate <- newEmptyTMVarIO let service _ = do atomicModifyIORef' counter (\n -> (n + 1, ())) - threadDelay delayTime + atomically $ readTMVar gate throwIO DogpileProtectionTestException :: IO Int protectedService <- dogpileProtect service asyncResults <- traverse (async . protectedService) requests + waitForCount counter 1 + atomically $ putTMVar gate () traverse wait asyncResults `shouldThrow` (== DogpileProtectionTestException) readIORef counter `shouldReturn` 1 + + describe "dogpileProtectWith" $ + it "Retains a completed result for exactly the cache time" $ do + vc <- newVirtualClock + counter <- newIORef (0 :: Int) + let clock = virtualClock vc :: Clock IO + options = (defaultDogpileOptions :: DogpileOptions IO) { dogpileCacheTimeMs = 100, dogpileClock = clock } + service (request :: Int) = atomicModifyIORef' counter (\n -> (n + 1, request * 2)) + protectedService <- dogpileProtectWith options service + protectedService 1 `shouldReturn` 2 + waitForSleepers vc 1 + protectedService 1 `shouldReturn` 2 + readIORef counter `shouldReturn` 1 + advanceClock vc (99 * 1000) + protectedService 1 `shouldReturn` 2 + readIORef counter `shouldReturn` 1 + advanceClock vc 1000 + protectedService 1 `shouldReturn` 2 + readIORef counter `shouldReturn` 2 diff --git a/test/Glue/FailoverSpec.hs b/test/Glue/FailoverSpec.hs --- a/test/Glue/FailoverSpec.hs +++ b/test/Glue/FailoverSpec.hs @@ -1,17 +1,41 @@ -{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} module Glue.FailoverSpec (spec) where +import qualified Control.Exception as E +import Data.IORef +import Data.List.NonEmpty (NonEmpty (..)) import Glue.Failover import Test.Hspec import Test.QuickCheck -import UnliftIO.Exception (Exception, throwIO) +import UnliftIO.Exception (Exception, fromException, throwIO, try) data FailoverTestException = FailoverTestException deriving (Eq, Show) instance Exception FailoverTestException +-- | A backend failure, carrying the backend index and whether it should advance. +data BackendException = BackendException Int Bool + deriving (Eq, Show) + +instance Exception BackendException + +advances :: E.SomeException -> Bool +advances e = maybe True (\(BackendException _ advance) -> advance) (fromException e) + +-- | Backends that record their calls, each succeeding or failing (advancing or not) as scripted. +backends :: IORef [Int] -> NonEmpty (Maybe Bool) -> NonEmpty (Int -> IO Int) +backends calls = fmap backend . indexed + where + indexed (x :| xs) = (0, x) :| zip [1 ..] xs + backend (index, script) req = do + atomicModifyIORef' calls (\c -> (c ++ [index], ())) + case script of + Nothing -> pure (req + index) + Just advance -> throwIO (BackendException index advance) + spec :: Spec spec = do describe "failover" $ do @@ -24,3 +48,43 @@ successCase = failOverService request `shouldReturn` ((if request <= 10 then 10 else request) + 100) failureCase = failOverService request `shouldThrow` (== FailoverTestException) in if request + positiveFailureMax >= 10 then successCase else failureCase + + describe "failoverAcross" $ do + it "Tries backends in order and returns the first success with its index" $ do + calls <- newIORef [] + let services = backends calls (Just True :| [Just True, Nothing, Nothing]) + failoverAcrossIndexed defaultFailoverAcrossOptions services 100 `shouldReturn` (2, 102) + readIORef calls `shouldReturn` [0, 1, 2] + + it "Stops at a failure that does not advance" $ do + calls <- newIORef [] + let services = backends calls (Just True :| [Just False, Nothing]) + options = defaultFailoverAcrossOptions { advanceOn = advances } + failoverAcross options services 1 `shouldThrow` (== BackendException 1 False) + readIORef calls `shouldReturn` [0, 1] + + it "Reports the last or first failure when every backend fails" $ do + calls <- newIORef [] + let services = backends calls (Just True :| [Just True, Just True]) + failoverAcross defaultFailoverAcrossOptions services 1 `shouldThrow` (== BackendException 2 True) + failoverAcross defaultFailoverAcrossOptions { reportedFailure = ReportFirstFailure } services 1 + `shouldThrow` (== BackendException 0 True) + + it "Propagates an asynchronous exception without advancing" $ do + calls <- newIORef ([] :: [Int]) + let interrupted _ = atomicModifyIORef' calls (\c -> (c ++ [0], ())) >> E.throwIO E.UserInterrupt :: IO Int + services = interrupted :| [\req -> atomicModifyIORef' calls (\c -> (c ++ [1], req))] + failoverAcross defaultFailoverAcrossOptions services 1 `shouldThrow` (== E.UserInterrupt) + readIORef calls `shouldReturn` [0] + + it "Behaves like the Either shape for the same outcomes" $ + property $ \(script :: [Maybe Bool], first :: Bool) -> do + calls <- newIORef [] + let scripted = Just True :| script + options :: FailoverAcrossOptions e + options = defaultFailoverAcrossOptions { reportedFailure = if first then ReportFirstFailure else ReportLastFailure } + valued = fmap (\(index, s) req -> pure (maybe (Right (req + index)) (\advance -> Left (index, advance)) s :: Either (Int, Bool) Int)) + ((0, Just True) :| zip [1 ..] script) + viaException <- try (failoverAcrossIndexed options { advanceOn = advances } (backends calls scripted) 10) + viaValue <- failoverAcrossIndexedEither options { advanceOn = snd } valued 10 + either (\(BackendException i a) -> Left (i, a)) Right viaException `shouldBe` viaValue diff --git a/test/Glue/HedgingSpec.hs b/test/Glue/HedgingSpec.hs --- a/test/Glue/HedgingSpec.hs +++ b/test/Glue/HedgingSpec.hs @@ -2,41 +2,189 @@ module Glue.HedgingSpec (spec) where -import Control.Concurrent (threadDelay) +import Control.Concurrent (yield) +import Control.Concurrent.Async (async, wait) import Data.IORef +import Data.List.NonEmpty (NonEmpty (..)) +import Glue.Clock +import Glue.Clock.Virtual +import Glue.Failure import Glue.Hedging +import Glue.Race import Test.Hspec -import UnliftIO.Exception (Exception, throwIO) +import UnliftIO.Exception (Exception, SomeException, finally, + fromException, throwIO) +import UnliftIO.STM (atomically, newEmptyTMVarIO, + putTMVar, readTMVar) -data HedgingTestException = HedgingTestException +data HedgingTestException = HedgingTestException Int deriving (Eq, Show) instance Exception HedgingTestException +-- | How an attempt ended, without the payload, for comparing reports. +data Ended = Answered Int | Failed Int | Cancelled | Crashed + deriving (Eq, Show) + +summarise :: AttemptOutcome SomeException Int -> Ended +summarise (AttemptAnswered n) = Answered n +summarise (AttemptFailed e) = maybe Crashed (\(HedgingTestException n) -> Failed n) (fromException e) +summarise AttemptCancelled = Cancelled +summarise (AttemptCrashed _) = Crashed + +-- | Options on a virtual clock that record launches, attempt endings and the winner. +recording :: VirtualClock -> (Int -> IO Bool) -> IO (HedgeOptions IO SomeException Int, IORef [Int], IORef [(Int, Ended)], IORef [Int]) +recording vc admit = do + launches <- newIORef [] + endings <- newIORef [] + winners <- newIORef [] + let record ref x = atomicModifyIORef' ref (\xs -> (xs ++ [x], ())) + options = (defaultHedgeOptions :: HedgeOptions IO SomeException Int) + { hedgeDelayMs = 100 + , admitLaunch = \i -> record launches i >> admit i + , onAttemptEnded = \i o -> record endings (i, summarise o) + , onHedgeWon = record winners + , hedgeClock = virtualClock vc + } + pure (options, launches, endings, winners) + spec :: Spec spec = do + describe "hedgeAcross" $ do + it "Returns the primary's answer without launching a backup inside the delay" $ do + vc <- newVirtualClock + gate <- newEmptyTMVarIO + (options, launches, endings, winners) <- recording vc (const (pure True)) + call <- async $ hedgeAcross options ((\x -> atomically (readTMVar gate) >> pure x) :| [pure . negate]) 5 + waitForSleepers vc 1 + atomically $ putTMVar gate () + wait call `shouldReturn` 5 + readIORef launches `shouldReturn` [] + readIORef endings `shouldReturn` [(0, Answered 5)] + readIORef winners `shouldReturn` [0] + + it "Launches the backup after the delay and cancels the slow primary" $ do + vc <- newVirtualClock + closed <- newIORef False + (options, launches, endings, winners) <- recording vc (const (pure True)) + let primary x = (clockSleep (virtualClock vc) 500000 >> pure x) `finally` writeIORef closed True + call <- async $ hedgeAcross options (primary :| [pure . negate]) 5 + waitForSleepers vc 2 + advanceClock vc 100000 + wait call `shouldReturn` (-5) + readIORef closed `shouldReturn` True + readIORef launches `shouldReturn` [1] + readIORef endings `shouldReturn` [(1, Answered (-5)), (0, Cancelled)] + readIORef winners `shouldReturn` [1] + + it "Keeps waiting on the primary when the backup's launch is declined" $ do + vc <- newVirtualClock + (options, launches, endings, _) <- recording vc (const (pure False)) + let primary x = clockSleep (virtualClock vc) 500000 >> pure x + call <- async $ hedgeAcross options (primary :| [pure . negate]) 5 + waitForSleepers vc 2 + advanceClock vc 100000 + let waitForDecision = readIORef launches >>= \l -> if null l then yield >> waitForDecision else pure () + waitForDecision + advanceClock vc 400000 + wait call `shouldReturn` 5 + readIORef launches `shouldReturn` [1] + readIORef endings `shouldReturn` [(0, Answered 5)] + + it "Launches the backup immediately when the primary fails early" $ do + vc <- newVirtualClock + (options, launches, endings, _) <- recording vc (const (pure True)) + hedgeAcross options ((\_ -> throwIO (HedgingTestException 0)) :| [pure . negate]) 5 `shouldReturn` (-5) + readIORef launches `shouldReturn` [1] + readIORef endings `shouldReturn` [(0, Failed 0), (1, Answered (-5))] + + it "Rethrows the first failure when every attempt fails" $ do + vc <- newVirtualClock + (options, _, endings, winners) <- recording vc (const (pure True)) + let failing n _ = throwIO (HedgingTestException n) :: IO Int + hedgeAcross options (failing 0 :| [failing 1, failing 2]) (5 :: Int) `shouldThrow` (== HedgingTestException 0) + readIORef endings `shouldReturn` [(0, Failed 0), (1, Failed 1), (2, Failed 2)] + readIORef winners `shouldReturn` [] + + it "Rethrows the primary's failure when the backup is declined" $ do + vc <- newVirtualClock + (options, _, _, _) <- recording vc (const (pure False)) + hedgeAcross options ((\_ -> throwIO (HedgingTestException 0)) :| [pure . negate]) 5 + `shouldThrow` (== HedgingTestException 0) + + describe "hedgeAcrossEither" $ + it "Treats Left as a failed attempt and returns the first Left if all fail" $ do + vc <- newVirtualClock + let options = (defaultHedgeOptions :: HedgeOptions IO String (Either String Int)) { hedgeClock = virtualClock vc } + hedgeAcrossEither options ((\_ -> pure (Left "primary")) :| [pure . Right]) 3 `shouldReturn` Right 3 + hedgeAcrossEither options ((\_ -> pure (Left "primary")) :| [\_ -> pure (Left "backup")]) (3 :: Int) + `shouldReturn` (Left "primary" :: Either String Int) + + describe "Glue.Race" $ do + it "Reports attempts that ended together in index order" $ do + result <- withRace exceptionFailures systemClock $ \race -> do + _ <- launch race (pure (1 :: Int)) + _ <- launch race (pure 2) + let settle = finishedAttempts race >>= \f -> if length f == 2 then pure () else yield >> settle + settle + first <- awaitNext race Nothing + second <- awaitNext race Nothing + none <- awaitNext race Nothing + pure (fmap eventText [first, second, none]) + result `shouldBe` ["0 answered 1", "1 answered 2", "nothing running"] + + it "Abandoning an attempt that already completed reports its result" $ do + result <- withRace exceptionFailures systemClock $ \race -> do + index <- launch race (pure (7 :: Int)) + let settle = finishedAttempts race >>= \f -> if f == [index] then pure () else yield >> settle + settle + summarise <$> abandon race index + result `shouldBe` Answered 7 + + it "Times out when no attempt ends within the limit" $ do + vc <- newVirtualClock + gate <- newEmptyTMVarIO + let clock = virtualClock vc :: Clock IO + waiting <- async $ withRace exceptionFailures clock $ \race -> do + _ <- launch race (atomically (readTMVar gate) >> pure (1 :: Int)) + eventText <$> awaitNext race (Just 50) + waitForSleepers vc 1 + advanceClock vc 50000 + wait waiting `shouldReturn` "timed out" + atomically $ putTMVar gate () + describe "hedgingService" $ do it "Completes with the fast hedged call when the primary call is slow" $ do + vc <- newVirtualClock ref <- newIORef (0 :: Int) - let service (req :: Int) = do + let clock = virtualClock vc :: Clock IO + service (req :: Int) = do attemptNum <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) if attemptNum == 1 - then threadDelay (500 * 1000) >> pure (req + 100) + then clockSleep clock (500 * 1000) >> pure (req + 100) else pure (req * 2) - hedged = hedgingService (HedgingOptions { hedgingCount = 2, hedgingDelayMs = 5 }) service - hedged 21 `shouldReturn` 42 + hedged = hedgingService ((defaultHedgingOptions :: HedgingOptions IO) { hedgingCount = 2, hedgingDelayMs = 5, hedgingClock = clock }) service + call <- async (hedged 21) + waitForSleepers vc 2 + advanceClock vc 5000 + wait call `shouldReturn` 42 it "Succeeds if one attempt fails immediately and another succeeds" $ do ref <- newIORef (0 :: Int) let service (req :: Int) = do attemptNum <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) if attemptNum == 1 - then throwIO HedgingTestException + then throwIO (HedgingTestException 1) else pure (req * 3) - hedged = hedgingService (HedgingOptions { hedgingCount = 2, hedgingDelayMs = 0 }) service + hedged = hedgingService (defaultHedgingOptions { hedgingCount = 2, hedgingDelayMs = 0 }) service hedged 10 `shouldReturn` 30 it "Rethrows the exception if all hedged attempts fail" $ do - let service (_ :: Int) = throwIO HedgingTestException :: IO Int - hedged = hedgingService (HedgingOptions { hedgingCount = 3, hedgingDelayMs = 0 }) service - hedged 1 `shouldThrow` (== HedgingTestException) + let service (_ :: Int) = throwIO (HedgingTestException 1) :: IO Int + hedged = hedgingService (defaultHedgingOptions { hedgingCount = 3, hedgingDelayMs = 0 }) service + hedged 1 `shouldThrow` (== HedgingTestException 1) + where + eventText (AttemptEnded i (AttemptAnswered n)) = show i ++ " answered " ++ show (n :: Int) + eventText (AttemptEnded i _) = show i ++ " ended" + eventText RaceTimedOut = "timed out" + eventText NothingRunning = "nothing running" diff --git a/test/Glue/PeriodicSpec.hs b/test/Glue/PeriodicSpec.hs new file mode 100644 --- /dev/null +++ b/test/Glue/PeriodicSpec.hs @@ -0,0 +1,134 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module Glue.PeriodicSpec (spec) where + +import Control.Concurrent (yield) +import Control.Concurrent.Async (async, wait) +import Data.IORef +import Glue.Clock +import Glue.Clock.Virtual +import Glue.Periodic +import Test.Hspec +import UnliftIO.Exception (Exception, SomeException, + fromException, throwIO) +import UnliftIO.STM (atomically, newEmptyTMVarIO, + putTMVar, readTMVar) + +data PassFailed = PassFailed + deriving (Eq, Show) + +instance Exception PassFailed + +waitUntil :: IO Bool -> IO () +waitUntil condition = condition >>= \done -> if done then pure () else yield >> waitUntil condition + +options :: VirtualClock -> PeriodicOptions IO +options vc = defaultPeriodicOptions { periodMs = 100, periodicGracePeriodMs = 50, periodicClock = virtualClock vc } + +spec :: Spec +spec = do + describe "periodically" $ do + it "Runs a pass every period, measured from the end of the previous pass" $ do + vc <- newVirtualClock + starts <- newIORef [] + let pass = do + now <- virtualNow vc + atomicModifyIORef' starts (\s -> (s ++ [now], ())) + clockSleep (virtualClock vc) 30000 + handle <- periodically (options vc) pass + waitForSleepers vc 1 + advanceClock vc 100000 + waitForSleepers vc 1 + advanceClock vc 30000 + waitUntil ((== 1) <$> passesCompleted handle) + waitForSleepers vc 1 + advanceClock vc 100000 + waitForSleepers vc 1 + advanceClock vc 30000 + waitUntil ((== 2) <$> passesCompleted handle) + readIORef starts `shouldReturn` [100000, 230000] + stopPeriodic handle `shouldReturn` True + + it "Runs the first pass immediately when asked" $ do + vc <- newVirtualClock + handle <- periodically (options vc) { runImmediately = True } (pure ()) + waitUntil ((== 1) <$> passesCompleted handle) + virtualNow vc `shouldReturn` 0 + stopPeriodic handle `shouldReturn` True + + it "Adds jitter from the random source to the period" $ do + vc <- newVirtualClock + handle <- periodically (options vc) { periodJitterMs = 40, periodicRandom = \n -> pure (n `div` 2) } (pure ()) + waitForSleepers vc 1 + advanceClock vc 119000 + passesCompleted handle `shouldReturn` 0 + advanceClock vc 1000 + waitUntil ((== 1) <$> passesCompleted handle) + stopPeriodic handle `shouldReturn` True + + it "Reports a failed pass and keeps running" $ do + vc <- newVirtualClock + failures <- newIORef (0 :: Int) + calls <- newIORef (0 :: Int) + let pass = do + n <- atomicModifyIORef' calls (\c -> (c + 1, c + 1)) + if n == 1 then throwIO PassFailed else pure () + onFailed (e :: SomeException) = case fromException e of + Just PassFailed -> atomicModifyIORef' failures (\f -> (f + 1, ())) + Nothing -> pure () + handle <- periodically (options vc) { onPassFailed = onFailed } pass + runNow handle `shouldReturn` True + runNow handle `shouldReturn` True + readIORef failures `shouldReturn` 1 + readIORef calls `shouldReturn` 2 + passesCompleted handle `shouldReturn` 2 + stopPeriodic handle `shouldReturn` True + + it "runNow queues behind a running pass" $ do + vc <- newVirtualClock + gate <- newEmptyTMVarIO + calls <- newIORef (0 :: Int) + let pass = do + n <- atomicModifyIORef' calls (\c -> (c + 1, c + 1)) + if n == 1 then atomically (readTMVar gate) else pure () + handle <- periodically (options vc) { runImmediately = True } pass + waitUntil ((== 1) <$> readIORef calls) + queued <- async (runNow handle) + atomically $ putTMVar gate () + wait queued `shouldReturn` True + readIORef calls `shouldReturn` 2 + stopPeriodic handle `shouldReturn` True + + it "Stopping lets a running pass finish within the grace period" $ do + vc <- newVirtualClock + gate <- newEmptyTMVarIO + started <- newEmptyTMVarIO + finished <- newIORef False + let pass = do + atomically $ putTMVar started () + atomically $ readTMVar gate + writeIORef finished True + handle <- periodically (options vc) { runImmediately = True } pass + atomically $ readTMVar started + stopping <- async (stopPeriodic handle) + waitForSleepers vc 1 + atomically $ putTMVar gate () + wait stopping `shouldReturn` True + readIORef finished `shouldReturn` True + + it "Stopping cancels a pass that outlasts the grace period" $ do + vc <- newVirtualClock + started <- newEmptyTMVarIO + finished <- newIORef False + let pass = do + atomically $ putTMVar started () + clockSleep (virtualClock vc) 1000000 + writeIORef finished True + handle <- periodically (options vc) { runImmediately = True } pass + atomically $ readTMVar started + stopping <- async (stopPeriodic handle) + waitForSleepers vc 2 + advanceClock vc 50000 + wait stopping `shouldReturn` False + readIORef finished `shouldReturn` False + runNow handle `shouldReturn` False diff --git a/test/Glue/RetrySpec.hs b/test/Glue/RetrySpec.hs --- a/test/Glue/RetrySpec.hs +++ b/test/Glue/RetrySpec.hs @@ -1,14 +1,19 @@ +{-# LANGUAGE ScopedTypeVariables #-} + module Glue.RetrySpec (spec) where -import Control.Concurrent (forkIO, myThreadId, threadDelay) -import Control.Exception.Base (AsyncException (UserInterrupt), - throwTo) +import Control.Concurrent (throwTo) +import Control.Concurrent.Async (async, asyncThreadId, waitCatch) +import Control.Exception.Base (AsyncException (UserInterrupt)) import Control.Monad.IO.Class (liftIO) import Data.IORef +import Glue.Clock import Glue.Retry import Test.Hspec import Test.QuickCheck -import UnliftIO.Exception (Exception, throwIO) +import UnliftIO.Exception (Exception, fromException, throwIO, try) +import UnliftIO.STM (atomically, newEmptyTMVarIO, putTMVar, + readTMVar) newtype SmallInt = SmallInt Int deriving (Eq, Show) @@ -23,31 +28,162 @@ instance Exception RetryTestException +newtype NotRetryableException = NotRetryableException Int + deriving (Eq, Show) + +instance Exception NotRetryableException + +-- | A clock that records each sleep (in milliseconds) instead of sleeping. +recordingClock :: IORef [Int] -> Clock IO +recordingClock ref = + Clock + { clockNow = pure 0 + , clockSleep = \us -> atomicModifyIORef' ref (\s -> (s ++ [us `div` 1000], ())) + , clockTimeout = \_ action -> Just <$> action + } + +-- | A service failing the given number of times before succeeding, counting calls. +failingTimes :: IORef Int -> Int -> Int -> IO Int +failingTimes ref failures req = do + counter <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) + if counter > failures then pure req else throwIO RetryTestException + +withSleeps :: (IORef [Int] -> IO ()) -> IO [Int] +withSleeps body = do + sleeps <- newIORef [] + body sleeps + readIORef sleeps + +alwaysFailing :: IO (Int -> IO Int) +alwaysFailing = do + ref <- newIORef 0 + pure (failingTimes ref maxBound) + spec :: Spec spec = do describe "retryingService" $ do it "Attempts a service call multiple times" $ property $ \(request, SmallInt failures, SmallInt retries) -> do ref <- liftIO $ newIORef 0 - let service req = do - counter <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) - if counter > failures then pure req else throwIO RetryTestException - options = defaultRetryOptions { maximumRetries = retries } - retryService = retryingService options service + let options = defaultRetryOptions { maximumRetries = retries } + retryService = retryingService options (failingTimes ref failures) successCase = retryService request `shouldReturn` (request :: Int) failureCase = retryService request `shouldThrow` (== RetryTestException) if retries >= failures then successCase else failureCase it "Asynchronous exceptions are rethrown" $ property $ \(request, retries) -> do - ref <- liftIO $ newIORef (0 :: Int) + ref <- liftIO $ newIORef (0 :: Int) + started <- newEmptyTMVarIO let service req = do atomicModifyIORef' ref (\c -> (c + 1, ())) - threadId <- myThreadId - _ <- forkIO (threadDelay 2000 >> throwTo threadId UserInterrupt) - threadDelay 1000000 + atomically $ putTMVar started () + never <- newEmptyTMVarIO + atomically $ readTMVar never :: IO () pure (req * 2 :: Int) options = defaultRetryOptions { maximumRetries = retries } retryService = retryingService options service - retryService request `shouldThrow` (== UserInterrupt) + call <- async (retryService request) + atomically $ readTMVar started + throwTo (asyncThreadId call) UserInterrupt + result <- waitCatch call + either fromException (const Nothing) result `shouldBe` Just UserInterrupt readIORef ref `shouldReturn` 1 + + it "Does not retry a failure rejected by retryOn" $ do + ref <- newIORef (0 :: Int) + let service req = atomicModifyIORef' ref (\c -> (c + 1, ())) >> throwIO (NotRetryableException req) :: IO Int + retryable e = maybe True (const False) (fromException e :: Maybe NotRetryableException) + retryService = retryingService defaultRetryOptions { retryOn = retryable } service + retryService 4 `shouldThrow` (== NotRetryableException 4) + readIORef ref `shouldReturn` 1 + + it "Does not retry a request rejected by retryAllowed" $ do + ref <- newIORef 0 + let retryService = retryingService defaultRetryOptions { retryAllowed = (> 0) } (failingTimes ref 1) + retryService 0 `shouldThrow` (== RetryTestException) + readIORef ref `shouldReturn` 1 + + it "Waits according to the backoff, capped by the maximum delay" $ do + service <- alwaysFailing + sleeps <- withSleeps $ \ref -> do + let options = defaultRetryOptions + { maximumRetries = 5 + , retryBackoff = ExponentialBackoff 100 2 + , retryMaxDelayMs = Just 300 + , retryClock = recordingClock ref + } + retryingService options service 1 `shouldThrow` (== RetryTestException) + sleeps `shouldBe` [100, 200, 300, 300, 300] + + it "Computes constant, linear and exponential backoffs" $ do + fmap (backoffDelayMs (ConstantBackoff 50)) [0 .. 3] `shouldBe` [50, 50, 50, 50] + fmap (backoffDelayMs (LinearBackoff 50 25)) [0 .. 3] `shouldBe` [50, 75, 100, 125] + fmap (backoffDelayMs (ExponentialBackoff 50 3)) [0 .. 3] `shouldBe` [50, 150, 450, 1350] + + it "Keeps jittered waits within their bounds" $ + property $ \(randoms :: [NonNegative Int], equal :: Bool) -> do + source <- newIORef (cycle (0 : fmap getNonNegative randoms)) + service <- alwaysFailing + let random n = atomicModifyIORef' source (\rs -> (drop 1 rs, head' rs `mod` (n + 1))) + head' = foldr const 0 + sleeps <- withSleeps $ \ref -> do + let options = defaultRetryOptions + { maximumRetries = 6 + , retryBackoff = ExponentialBackoff 10 2 + , retryMaxDelayMs = Just 200 + , retryJitter = if equal then EqualJitter else FullJitter + , retryRandom = random + , retryClock = recordingClock ref + } + retryingService options service 1 `shouldThrow` (== RetryTestException) + let caps = [10, 20, 40, 80, 160, 200] + lower cap = if equal then cap `div` 2 else 0 + sleeps `shouldSatisfy` (\s -> and (zipWith (\d cap -> d >= lower cap && d <= cap) s caps) && length s == 6) + + it "Skips a retry whose wait would pass the deadline, returning the last failure" $ do + ref <- newIORef (0 :: Int) + retries <- newIORef [] + let service req = do + n <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) + throwIO (NotRetryableException (req + n)) :: IO Int + sleeps <- withSleeps $ \sleepRef -> do + let options = defaultRetryOptions + { maximumRetries = 5 + , retryBackoff = LinearBackoff 100 100 + , retryRemainingMs = pure (Just 150) + , onRetry = \attempt _ delay -> atomicModifyIORef' retries (\r -> (r ++ [(attempt, delay)], ())) + , retryClock = recordingClock sleepRef + } + retryingService options service 10 `shouldThrow` (== NotRetryableException 12) + sleeps `shouldBe` [100] + readIORef retries `shouldReturn` [(1, 100)] + + it "Does not retry once the deadline has passed" $ do + ref <- newIORef 0 + let options = defaultRetryOptions { retryRemainingMs = pure (Just 0) } + retryingService options (failingTimes ref 1) 1 `shouldThrow` (== RetryTestException) + readIORef ref `shouldReturn` 1 + + describe "retryingServiceEither" $ do + it "Retries Left failures accepted by retryOn and returns the last one" $ do + ref <- newIORef (0 :: Int) + let service (_ :: Int) = atomicModifyIORef' ref (\c -> (c + 1, Left (c + 1) :: Either Int Int)) + options = defaultRetryOptions { maximumRetries = 5, retryOn = (< 3) } + retryingServiceEither options service 1 `shouldReturn` Left 3 + readIORef ref `shouldReturn` 3 + + it "Behaves like the exception shape for the same outcomes" $ + property $ \(outcomes :: [Bool], SmallInt retries) -> do + exceptionRef <- newIORef outcomes + valueRef <- newIORef outcomes + let next ref = atomicModifyIORef' ref (\o -> case o of { (x : xs) -> (xs, x); [] -> ([], True) }) + exceptional (req :: Int) = next exceptionRef >>= \ok -> if ok then pure req else throwIO RetryTestException + valued (req :: Int) = next valueRef >>= \ok -> pure (if ok then Right req else Left RetryTestException) + options :: RetryOptions IO Int f + options = defaultRetryOptions { maximumRetries = retries } + viaException <- try (retryingService options exceptional 7) + viaValue <- retryingServiceEither options valued 7 + viaException `shouldBe` viaValue + remaining <- readIORef valueRef + readIORef exceptionRef `shouldReturn` remaining diff --git a/test/Glue/ShutdownSpec.hs b/test/Glue/ShutdownSpec.hs --- a/test/Glue/ShutdownSpec.hs +++ b/test/Glue/ShutdownSpec.hs @@ -2,26 +2,35 @@ module Glue.ShutdownSpec (spec) where -import Control.Concurrent (threadDelay) +import Control.Concurrent (yield) import Control.Concurrent.Async (async, wait) +import Glue.Clock +import Glue.Clock.Virtual import Glue.Shutdown import Test.Hspec +import UnliftIO.STM (atomically, newEmptyTMVarIO, + putTMVar, takeTMVar) spec :: Spec spec = do describe "gracefulShutdown" $ do it "Allows in-flight requests to drain cleanly within the grace period and rejects new requests" $ do - let service (req :: Int) = do - threadDelay (30 * 1000) + vc <- newVirtualClock + release <- newEmptyTMVarIO + let clock = virtualClock vc :: Clock IO + service (req :: Int) = do + atomically $ takeTMVar release pure (req * 2) - (wrapped, handle) <- gracefulShutdown (defaultShutdownOptions { shutdownGracePeriodMs = 500 }) service + (wrapped, handle) <- gracefulShutdown ((defaultShutdownOptions :: ShutdownOptions IO Int) { shutdownGracePeriodMs = 500, shutdownClock = clock }) service isShuttingDown handle `shouldReturn` False inFlight <- async (wrapped 21) - threadDelay (5 * 1000) - inFlightCount handle `shouldReturn` 1 - drainedCleanly <- initiateShutdown handle - drainedCleanly `shouldBe` True + let waitInFlight = inFlightCount handle >>= \n -> if n == 1 then pure () else yield >> waitInFlight + waitInFlight + shutdown <- async (initiateShutdown handle) + waitForSleepers vc 1 isShuttingDown handle `shouldReturn` True + atomically $ putTMVar release () + wait shutdown `shouldReturn` True wait inFlight `shouldReturn` 42 wrapped 99 `shouldThrow` (== ServiceShuttingDownException "Service is shutting down.") @@ -33,11 +42,22 @@ wrapped 10 `shouldReturn` (-1) it "Cancels stuck in-flight requests when grace period expires" $ do - let opts = defaultShutdownOptions { shutdownFallback = Just (-999 :: Int), shutdownGracePeriodMs = 20 } - slowService (_ :: Int) = threadDelay (500 * 1000) >> pure 100 + vc <- newVirtualClock + let clock = virtualClock vc :: Clock IO + opts = (defaultShutdownOptions :: ShutdownOptions IO Int) { shutdownFallback = Just (-999), shutdownGracePeriodMs = 20, shutdownClock = clock } + slowService (_ :: Int) = clockSleep clock (500 * 1000) >> pure (100 :: Int) (wrapped, handle) <- gracefulShutdown opts slowService inFlight <- async (wrapped 1) - threadDelay (5 * 1000) - drainedCleanly <- initiateShutdown handle - drainedCleanly `shouldBe` False + waitForSleepers vc 1 + shutdown <- async (initiateShutdown handle) + waitForSleepers vc 2 + advanceClock vc (20 * 1000) + wait shutdown `shouldReturn` False wait inFlight `shouldReturn` (-999) + + describe "gracefulShutdownEither" $ + it "Returns the shutting-down failure as a value" $ do + (wrapped, handle) <- gracefulShutdownEither defaultShutdownOptions "shutting down" (\(x :: Int) -> pure (Right (x * 2))) + wrapped 4 `shouldReturn` (Right 8 :: Either String Int) + _ <- initiateShutdown handle + wrapped 4 `shouldReturn` Left "shutting down" diff --git a/test/Glue/TimeoutSpec.hs b/test/Glue/TimeoutSpec.hs --- a/test/Glue/TimeoutSpec.hs +++ b/test/Glue/TimeoutSpec.hs @@ -1,20 +1,48 @@ +{-# LANGUAGE ScopedTypeVariables #-} + module Glue.TimeoutSpec (spec) where -import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (async, wait) +import Glue.Clock +import Glue.Clock.Virtual import Glue.Timeout import Test.Hspec import Test.QuickCheck -timeoutOptions :: TimeoutOptions -timeoutOptions = defaultTimeoutOptions { timeoutLimitMs = 10 } - spec :: Spec spec = do describe "addTimeout" $ do it "Services taking too long should fail with a timeout" $ - property $ \request -> - let service req = (if req > 0 then pure () else threadDelay (50 * 1000)) >> pure (req + 100) - timeoutService = addTimeout timeoutOptions service - successCase = timeoutService request `shouldReturn` (request + 100) - failureCase = timeoutService request `shouldThrow` (== TimeoutException (timeoutDescription timeoutOptions)) - in if request > (0 :: Int) then successCase else failureCase + property $ \(request :: Int) -> do + vc <- newVirtualClock + let clock = virtualClock vc :: Clock IO + options = (defaultTimeoutOptions :: TimeoutOptions IO) { timeoutLimitMs = 10, timeoutClock = clock } + service req = (if req > 0 then pure () else clockSleep clock (50 * 1000)) >> pure (req + 100) + timeoutService = addTimeout options service + if request > 0 + then timeoutService request `shouldReturn` (request + 100) + else do + call <- async (timeoutService request) + waitForSleepers vc 2 + advanceClock vc (10 * 1000) + wait call `shouldThrow` (== TimeoutException (timeoutDescription options)) + + it "Does not time out a call that finishes just inside the limit" $ do + vc <- newVirtualClock + let clock = virtualClock vc :: Clock IO + options = (defaultTimeoutOptions :: TimeoutOptions IO) { timeoutLimitMs = 10, timeoutClock = clock } + call <- async (addTimeout options (\(x :: Int) -> clockSleep clock 9999 >> pure x) 7) + waitForSleepers vc 2 + advanceClock vc 9999 + wait call `shouldReturn` 7 + + describe "addTimeoutEither" $ + it "Returns the timeout failure as a value" $ do + vc <- newVirtualClock + let clock = virtualClock vc :: Clock IO + options = (defaultTimeoutOptions :: TimeoutOptions IO) { timeoutLimitMs = 10, timeoutClock = clock } + service (_ :: Int) = clockSleep clock 50000 >> pure (Right (1 :: Int)) + call <- async (addTimeoutEither options "timed out" service 1) + waitForSleepers vc 2 + advanceClock vc 10000 + wait call `shouldReturn` (Left ("timed out" :: String)) diff --git a/test/Glue/WeightedLoadBalancerSpec.hs b/test/Glue/WeightedLoadBalancerSpec.hs new file mode 100644 --- /dev/null +++ b/test/Glue/WeightedLoadBalancerSpec.hs @@ -0,0 +1,108 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module Glue.WeightedLoadBalancerSpec (spec) where + +import qualified Data.ByteString as BS +import qualified Data.ByteString.Char8 as BC +import Data.Either (isRight) +import Data.List.NonEmpty (NonEmpty (..)) +import qualified Data.Map.Strict as M +import Glue.LoadBalancer +import Test.Hspec +import Test.QuickCheck +import UnliftIO.Exception (Exception, throwIO) + +data BackendDown = BackendDown + deriving (Eq, Show) + +instance Exception BackendDown + +referenceKey :: (Integer, Integer) +referenceKey = (0x0706050403020100, 0x0f0e0d0c0b0a0908) + +referenceHash :: Int -> Integer +referenceHash n = toInteger (sipHash24 (fromInteger (fst referenceKey)) (fromInteger (snd referenceKey)) (BS.pack (fmap fromIntegral [0 .. n - 1]))) + +keys :: [BS.ByteString] +keys = fmap (BC.pack . ("session-" ++) . show) [1 .. 20000 :: Int] + +spec :: Spec +spec = do + describe "sipHash24" $ + it "Matches the reference vectors (key 00..0f, message 00..n-1)" $ do + referenceHash 0 `shouldBe` 0x726fdb47dd0e0e31 + referenceHash 1 `shouldBe` 0x74f839c593dc67fd + referenceHash 7 `shouldBe` 0xab0200f58b01d137 + referenceHash 8 `shouldBe` 0x93f5f5799a932462 + referenceHash 15 `shouldBe` 0xa129ca6149be45e5 + referenceHash 63 `shouldBe` 0x958a324ceb064572 + + describe "validateWeights" $ do + it "Accepts non-negative weights summing to 10000, including zero weights" $ do + validateWeights (10000 :| []) `shouldBe` Right () + validateWeights (8000 :| [2000, 0]) `shouldBe` Right () + + it "Rejects negative weights and wrong sums" $ do + validateWeights (10500 :| [-500]) `shouldBe` Left (NegativeWeight 1 (-500)) + validateWeights (5000 :| [4000]) `shouldBe` Left (WeightsDoNotSumToTotal 9000) + + describe "chooseIndex" $ do + it "Is deterministic for a key and salt" $ + property $ \(key :: String) -> do + let salt = saltFromBytes (BC.pack "split") + chooseIndex salt (BC.pack key) (5000 :| [5000]) `shouldBe` chooseIndex salt (BC.pack key) (5000 :| [5000]) + + it "Never chooses a zero-weight backend and gives a full-weight backend everything" $ + property $ \(key :: String) -> do + let salt = saltFromBytes (BC.pack "split") + chooseIndex salt (BC.pack key) (0 :| [10000, 0]) `shouldBe` 1 + chooseIndex salt (BC.pack key) (10000 :| [0]) `shouldBe` 0 + + it "Maps buckets to backends by their cumulative ranges" $ + property $ \(key :: String) -> do + let salt = saltFromBytes (BC.pack "split") + bucket = bucketOf salt (BC.pack key) + index = chooseIndex salt (BC.pack key) (2500 :| [2500, 5000]) + index `shouldBe` (if bucket < 2500 then 0 else if bucket < 5000 then 1 else 2) + + it "Keeps a key's backend when unrelated weights change order" $ + property $ \(key :: String) -> do + let salt = saltFromBytes (BC.pack "split") + original = chooseIndex salt (BC.pack key) (5000 :| [3000, 2000]) + reordered = chooseIndex salt (BC.pack key) (5000 :| [2000, 3000]) + -- Keys in the first backend's range are unaffected by reordering the others. + (original == 0) `shouldBe` (reordered == 0) + + it "Splits traffic in proportion to the weights" $ do + let salt = saltFromBytes (BC.pack "split") + counts = M.fromListWith (+) [(chooseIndex salt key (8000 :| [2000]), 1 :: Int) | key <- keys] + share index = fromIntegral (M.findWithDefault 0 index counts) / fromIntegral (length keys) :: Double + share 0 `shouldSatisfy` (\s -> abs (s - 0.8) < 0.015) + share 1 `shouldSatisfy` (\s -> abs (s - 0.2) < 0.015) + + it "Gives independent assignments for different salts" $ do + let a = saltFromBytes (BC.pack "split-a") + b = saltFromBytes (BC.pack "split-b") + both = length [() | key <- keys, chooseIndex a key (5000 :| [5000]) == 0, chooseIndex b key (5000 :| [5000]) == 0] + ratio = fromIntegral both / fromIntegral (length keys) :: Double + ratio `shouldSatisfy` (\r -> abs (r - 0.25) < 0.015) + + describe "weightedBy" $ do + it "Rejects invalid weights up front" $ do + let backends = (6000, pure . (+ 1)) :| [(6000, pure . (+ 2))] :: NonEmpty (Int, Int -> IO Int) + either Just (const Nothing) (weightedBy (BC.pack . show) (Salt 1 2) backends) `shouldBe` Just (WeightsDoNotSumToTotal 12000) + + it "Routes every request to the backend chooseIndex picks" $ + property $ \(request :: Int) -> do + let salt = Salt 1 2 + backends = (3000, \r -> pure (0, r)) :| [(7000, \r -> pure (1, r))] :: NonEmpty (Int, Int -> IO (Int, Int)) + keyOf = BC.pack . show + case weightedBy keyOf salt backends of + Left err -> expectationFailure (show err) + Right service -> service request `shouldReturn` (chooseIndex salt (keyOf request) (3000 :| [7000]), request) + + it "Propagates a failure of the chosen backend" $ do + let backends = (10000, \_ -> throwIO BackendDown) :| [(0, pure)] :: NonEmpty (Int, Int -> IO Int) + built = weightedBy (BC.pack . show) (Salt 1 2) backends + isRight built `shouldBe` True + either (const (pure ())) (\service -> service 1 `shouldThrow` (== BackendDown)) built