glue-core 0.7.0.1 → 0.8.0.0

raw patch · 26 files changed

+2377/−338 lines, 26 filesdep +bytestringdep +containersdep +randomdep −timedep ~glue-commonPVP ok

version bump matches the API change (PVP)

Dependencies added: bytestring, containers, random

Dependencies removed: time

Dependency ranges changed: glue-common

API changes (from Hackage documentation)

Files

• CHANGELOG.md +74/−0
• glue-core.cabal +13/−5
• src/Glue/Batcher.hs +13/−10
• src/Glue/CircuitBreaker.hs +239/−78
• src/Glue/ConcurrencyLimit.hs +148/−14
• src/Glue/DogpileProtection.hs +11/−9
• src/Glue/Failover.hs +104/−2
• src/Glue/Hedging.hs +144/−49
• src/Glue/LoadBalancer.hs +118/−1
• src/Glue/Periodic.hs +137/−0
• src/Glue/Preload.hs +11/−7
• src/Glue/Race.hs +144/−0
• src/Glue/Retry.hs +119/−43
• src/Glue/Shutdown.hs +38/−16
• src/Glue/Timeout.hs +28/−8
• test/Glue/BatcherSpec.hs +2/−2
• test/Glue/CircuitBreakerSpec.hs +195/−25
• test/Glue/ConcurrencyLimitSpec.hs +116/−13
• test/Glue/DogpileProtectionSpec.hs +34/−5
• test/Glue/FailoverSpec.hs +66/−2
• test/Glue/HedgingSpec.hs +160/−12
• test/Glue/PeriodicSpec.hs +134/−0
• test/Glue/RetrySpec.hs +150/−14
• test/Glue/ShutdownSpec.hs +33/−13
• test/Glue/TimeoutSpec.hs +38/−10
• test/Glue/WeightedLoadBalancerSpec.hs +108/−0

@@ -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.

@@ -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

@@ -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.

@@ -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

@@ -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

@@ -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, ()))

@@ -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

@@ -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)

@@ -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)

@@ -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+      }

@@ -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 ()

@@ -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

@@ -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 <https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/>.+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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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)

@@ -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

@@ -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

@@ -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"

@@ -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

@@ -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

@@ -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"

@@ -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))

@@ -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