glue-core 0.6.3 → 0.7.0

raw patch · 28 files changed

+1343/−627 lines, 28 filesdep +unliftiodep −ekg-coredep −lifted-basedep −monad-controldep ~basedep ~glue-commonsetup-changedPVP ok

version bump matches the API change (PVP)

Dependencies added: unliftio

Dependencies removed: ekg-core, lifted-base, monad-control, text, transformers, transformers-base

Dependency ranges changed: base, glue-common

API changes (from Hackage documentation)

Files

• Setup.hs +0/−2
• glue-core.cabal +86/−72
• src/Glue/Batcher.hs +70/−55
• src/Glue/Caching.hs +42/−32
• src/Glue/CircuitBreaker.hs +88/−65
• src/Glue/ConcurrencyLimit.hs +87/−0
• src/Glue/DogpileProtection.hs +81/−44
• src/Glue/Failover.hs +29/−20
• src/Glue/Hedging.hs +81/−0
• src/Glue/LoadBalancer.hs +51/−0
• src/Glue/Preload.hs +109/−70
• src/Glue/Retry.hs +52/−36
• src/Glue/Shutdown.hs +128/−0
• src/Glue/Switching.hs +15/−13
• src/Glue/Timeout.hs +33/−28
• test/Glue/BatcherSpec.hs +15/−10
• test/Glue/CachingSpec.hs +29/−31
• test/Glue/CircuitBreakerSpec.hs +39/−28
• test/Glue/ConcurrencyLimitSpec.hs +65/−0
• test/Glue/DogpileProtectionSpec.hs +22/−23
• test/Glue/FailoverSpec.hs +14/−16
• test/Glue/HedgingSpec.hs +42/−0
• test/Glue/LoadBalancerSpec.hs +46/−0
• test/Glue/PreloadSpec.hs +24/−24
• test/Glue/RetrySpec.hs +38/−39
• test/Glue/ShutdownSpec.hs +43/−0
• test/Glue/SwitchingSpec.hs +6/−8
• test/Glue/TimeoutSpec.hs +8/−11

@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain

@@ -1,78 +1,92 @@-name:                   glue-core-version:                0.6.3-synopsis:               Make better services and clients.-description:            Combinator library to enhance the general functionality of services and clients.-license:                BSD3-license-file:           LICENSE-author:                 Sean Parsons-maintainer:             github@futurenotfound.com-category:               Network-build-type:             Simple-cabal-version:          >=1.10+cabal-version:      3.0+name:               glue-core+version:            0.7.0+synopsis:           Make better services and clients.+description:        Combinator library to enhance the general functionality of services and clients.+license:            BSD-3-Clause+license-file:       LICENSE+author:             Sean Parsons+maintainer:         github@futurenotfound.com+category:           Network+build-type:         Simple+tested-with:        GHC == 9.6.7, GHC == 9.8.4, GHC == 9.10.3, GHC == 9.12.3  source-repository head-  type:                 git-  location:             git://github.com/seanparsons/glue.git+  type:     git+  location: https://github.com/seanparsons/glue.git +common common-options+  default-language: Haskell2010+  ghc-options:+    -Wall+    -Wcompat+    -Widentities+    -Wincomplete-record-updates+    -Wincomplete-uni-patterns+    -Wmissing-home-modules+    -Wpartial-fields+    -Wredundant-constraints+ library-  exposed-modules:      Glue.Failover-                        Glue.Caching-                        Glue.CircuitBreaker-                        Glue.DogpileProtection-                        Glue.Timeout-                        Glue.Retry-                        Glue.Batcher-                        Glue.Preload-                        Glue.Switching-  -- other-extensions:-  build-depends:        base >=4.6 && <5,-                        glue-common,-                        transformers,-                        transformers-base,-                        lifted-base,-                        time,-                        monad-control,-                        unordered-containers,-                        hashable,-                        text-  ghc-options:          -Wall-  hs-source-dirs:       src-  default-language:     Haskell2010+  import:         common-options+  hs-source-dirs: src+  exposed-modules:+    Glue.Batcher+    Glue.Caching+    Glue.CircuitBreaker+    Glue.ConcurrencyLimit+    Glue.DogpileProtection+    Glue.Failover+    Glue.Hedging+    Glue.LoadBalancer+    Glue.Preload+    Glue.Retry+    Glue.Shutdown+    Glue.Switching+    Glue.Timeout+  build-depends:+    , base                  >= 4.16 && < 5+    , glue-common           ^>= 0.7.0+    , hashable+    , time+    , unliftio              >= 0.2+    , unordered-containers  test-suite glue-core-tests-  build-depends:        base >=4.6 && <5,-                        glue-common ,-                        glue-core,-                        QuickCheck -any,-                        quickcheck-instances,-                        hspec,-                        transformers,-                        transformers-base,-                        lifted-base,-                        time,-                        monad-control,-                        unordered-containers,-                        hashable,-                        ekg-core,-                        text,-                        async-  other-modules:        Glue.CachingSpec-                        Glue.FailoverSpec-                        Glue.RetrySpec-                        Glue.DogpileProtectionSpec-                        Glue.TimeoutSpec-                        Glue.CircuitBreakerSpec-                        Glue.BatcherSpec-                        Glue.PreloadSpec-                        Glue.SwitchingSpec-                        Spec-  ghc-options:          -rtsopts-                        -Wall-                        -O2-                        -threaded-  build-tools:          hspec-discover-  type:                 exitcode-stdio-1.0-  main-is:              Main.hs-  buildable:            True-  default-language:     Haskell2010-  hs-source-dirs:       test+  import:             common-options+  type:               exitcode-stdio-1.0+  hs-source-dirs:     test+  main-is:            Main.hs+  other-modules:+    Glue.BatcherSpec+    Glue.CachingSpec+    Glue.CircuitBreakerSpec+    Glue.ConcurrencyLimitSpec+    Glue.DogpileProtectionSpec+    Glue.FailoverSpec+    Glue.HedgingSpec+    Glue.LoadBalancerSpec+    Glue.PreloadSpec+    Glue.RetrySpec+    Glue.ShutdownSpec+    Glue.SwitchingSpec+    Glue.TimeoutSpec+    Spec+  build-tool-depends: hspec-discover:hspec-discover+  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N+    -O2+  build-depends:+    , async+    , base                  >= 4.16 && < 5+    , glue-common+    , glue-core+    , hashable+    , hspec+    , QuickCheck+    , quickcheck-instances+    , time+    , unliftio              >= 0.2+    , unordered-containers

@@ -2,85 +2,100 @@ {-# LANGUAGE ScopedTypeVariables #-}  -- | Module for creating a service that batches calls across requests.-module Glue.Batcher(-    BatchingOptions+module Glue.Batcher+  ( BatchingOptions (..)   , batchingService   , defaultBatchingOptions-  , batchWindowMs-) where+  ) where -import           Control.Concurrent.Lifted-import           Control.Exception.Lifted-import           Control.Monad-import           Control.Monad.Trans.Control-import           Data.Foldable-import           Data.Hashable-import qualified Data.HashMap.Strict         as M-import qualified Data.HashSet                as S-import           Data.IORef.Lifted+import           Control.Monad       (join, void)+import           Data.Foldable       (traverse_)+import           Data.Hashable       (Hashable)+import qualified Data.HashMap.Strict as M+import qualified Data.HashSet        as S import           Glue.Types+import           UnliftIO            (MonadIO, MonadUnliftIO, SomeException)+import           UnliftIO.Concurrent (forkIO, threadDelay)+import           UnliftIO.IORef      (IORef, atomicModifyIORef', newIORef)+import           UnliftIO.MVar       (newEmptyMVar, putMVar)  -- | Options for configuring the batching.-data BatchingOptions = BatchingOptions {-    batchWindowMs :: Int           -- ^ Window in milliseconds over which to batch.-} deriving (Eq, Show)+newtype BatchingOptions = BatchingOptions+  { batchWindowMs :: Int+  -- ^ Window in milliseconds over which to batch.+  }+  deriving (Eq, Show) --- | Default instance of 'BatchingOptions' with a batching window of 10ms.+-- | Default instance of t'BatchingOptions' with a batching window of 10ms. defaultBatchingOptions :: BatchingOptions-defaultBatchingOptions = BatchingOptions {-   batchWindowMs = 10-}+defaultBatchingOptions =+  BatchingOptions+    { batchWindowMs = 10+    }  -- | Cumulative request batch of all the outstanding calls. data RequestBatch a b = RequestBatch [PendingRequest a b] (MultiGetRequest a)  -- | Individual request to satisfy, for either a single key or a multi request.-data PendingRequest a b =-  SingleRequest a (ResultVar (Maybe b)) |-  MultiRequest (MultiGetRequest a) (ResultVar (MultiGetResponse a b))+data PendingRequest a b+  = SingleRequest a (ResultVar (Maybe b))+  | MultiRequest (MultiGetRequest a) (ResultVar (MultiGetResponse a b)) -applyToPending :: (Eq a, Hashable a, MonadBaseControl IO m) => Either SomeException (M.HashMap a b) -> PendingRequest a b -> m ()-applyToPending (Left e) (SingleRequest _ var)         = putMVar var $ Left e-applyToPending (Right results) (SingleRequest a var)  = putMVar var $ Right $ M.lookup a results-applyToPending (Left e) (MultiRequest _ var)          = putMVar var $ Left e-applyToPending (Right results) (MultiRequest as var)  = putMVar var $ Right $ M.filterWithKey (\k -> \_ -> S.member k as) results+applyToPending :: (Hashable a, MonadIO m) => Either SomeException (M.HashMap a b) -> PendingRequest a b -> m ()+applyToPending (Left e) (SingleRequest _ var)        = putMVar var $ Left e+applyToPending (Right results) (SingleRequest a var) = putMVar var $ Right $ M.lookup a results+applyToPending (Left e) (MultiRequest _ var)         = putMVar var $ Left e+applyToPending (Right results) (MultiRequest as var) = putMVar var $ Right $ M.filterWithKey (\k _ -> S.member k as) results -emptyBatch :: (Eq a, Hashable a) => RequestBatch a b+emptyBatch :: RequestBatch a b emptyBatch = RequestBatch [] S.empty -processCalls :: (Eq a, Hashable a, MonadBaseControl IO m) => MultiGetService m a b -> RequestBatch a b -> m ()+processCalls :: (Hashable a, MonadUnliftIO m) => MultiGetService m a b -> RequestBatch a b -> m () processCalls service (RequestBatch pendings requests) = do-                                                          result <- makeCall service requests-                                                          traverse_ (applyToPending result) pendings+  result <- makeCall service requests+  traverse_ (applyToPending result) pendings -startBatch :: (Eq a, Hashable a, MonadBaseControl IO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> m ()-startBatch options service ref = fmap (\_ -> ()) $ fork $ do-                                                            threadDelay (1000 * batchWindowMs options)-                                                            pendings <- atomicModifyIORef' ref (\p -> (emptyBatch, p))-                                                            processCalls service pendings+startBatch :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> m ()+startBatch options service ref =+  void $ forkIO $ do+    threadDelay (1000 * batchWindowMs options)+    pendings <- atomicModifyIORef' ref (\p -> (emptyBatch, p))+    processCalls service pendings -addPending :: (Eq a, Hashable a, MonadBaseControl IO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> PendingRequest a b -> m ()-addPending options service ref p@(SingleRequest request _) = join $ atomicModifyIORef' ref (\(RequestBatch ps rs) -> (RequestBatch (p : ps) (S.insert request rs), if null ps then startBatch options service ref else return ()))-addPending options service ref p@(MultiRequest requests _) = join $ atomicModifyIORef' ref (\(RequestBatch ps rs) -> (RequestBatch (p : ps) (S.union rs requests), if null ps then startBatch options service ref else return ()))+addPending :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> PendingRequest a b -> m ()+addPending options service ref p@(SingleRequest request _) =+  join $+    atomicModifyIORef'+      ref+      (\(RequestBatch ps rs) -> (RequestBatch (p : ps) (S.insert request rs), if null ps then startBatch options service ref else pure ()))+addPending options service ref p@(MultiRequest requests _) =+  join $+    atomicModifyIORef'+      ref+      (\(RequestBatch ps rs) -> (RequestBatch (p : ps) (S.union rs requests), if null ps then startBatch options service ref else pure ())) -singleService :: (Eq a, Hashable a, MonadBaseControl IO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> BasicService m a (Maybe b)+singleService :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> 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 :: (Eq a, Hashable a, MonadBaseControl IO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> MultiGetService m a b+  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 options service ref requests = do-                                              mvar <- newEmptyMVar-                                              let pending = MultiRequest requests mvar-                                              addPending options service ref pending-                                              getResult mvar+  mvar <- newEmptyMVar+  let pending = MultiRequest requests mvar+  addPending options service ref pending+  getResult mvar  -- | Function for constructing a batching service.-batchingService :: (Eq a, Hashable a, MonadBaseControl IO m, Applicative m, MonadBaseControl IO n)-                => BatchingOptions            -- ^ Options to configure the batched service.-                -> MultiGetService m a b      -- ^ Service around which to batch requests.-                -> n (BasicService m a (Maybe b), MultiGetService m a b)+batchingService+  :: (Hashable a, MonadUnliftIO m, MonadIO n)+  => BatchingOptions+  -- ^ Options to configure the batched service.+  -> MultiGetService m a b+  -- ^ Service around which to batch requests.+  -> n (BasicService m a (Maybe b), MultiGetService m a b) batchingService options service = do   ref <- newIORef emptyBatch-  return (singleService options service ref, multiService options service ref)+  pure (singleService options service ref, multiService options service ref)

@@ -2,48 +2,58 @@ {-# LANGUAGE ScopedTypeVariables #-}  -- | Module supporting the caching of a service.-module Glue.Caching(-    cacheWithBasic+module Glue.Caching+  ( cacheWithBasic   , cacheWithMulti-) where+  ) where -import           Data.Hashable+import           Data.Hashable       (Hashable) import qualified Data.HashMap.Strict as M import qualified Data.HashSet        as S import           Glue.Types --- | Caching of a `BasicService` instance, that defers to external functions for the actual caching.--- | Note: Values within m will be lost for calls that hit the cache.-cacheWithBasic :: (Monad m)-               => (a -> m (Maybe b))  -- ^ Cache lookup function, used before potentially invoking the fallback service.-               -> (a -> b -> m ())    -- ^ Cache write function, used after invoking the fallback service to populate the cache.-               -> BasicService m a b  -- ^ The service to cache.-               -> BasicService m a b+-- | Caching of a 'BasicService' instance, that defers to external functions for the actual caching.+-- Note: Values within @m@ will be lost for calls that hit the cache.+cacheWithBasic+  :: Monad m+  => (a -> m (Maybe b))+  -- ^ Cache lookup function, used before potentially invoking the fallback service.+  -> (a -> b -> m ())+  -- ^ Cache write function, used after invoking the fallback service to populate the cache.+  -> BasicService m a b+  -- ^ The service to cache.+  -> BasicService m a b cacheWithBasic lookupWith insertWith service =   let fallback request = do-                            result <- service request-                            insertWith request result-                            return result+        result <- service request+        insertWith request result+        pure result       cachedService request = do-                                fromCache <- lookupWith request-                                maybe (fallback request) return fromCache-  in cachedService+        fromCache <- lookupWith request+        maybe (fallback request) pure fromCache+  in  cachedService --- | Caching of a `MultiGetService` instance, that defers to external functions for the actual caching.--- | Partial responses will result in partial fallback calls that get just the missing keys.--- | Values within m will be lost for calls that hit the cache.-cacheWithMulti :: (Monad m, Functor m, Eq a, Hashable a)-               => ((MultiGetRequest a) -> m (MultiGetResponse a b)) -- ^ Cache lookup function, used before potentially invoking the fallback service.-               -> ((MultiGetResponse a b) -> m ())                  -- ^ Cache write function, used after invoking the fallback service to populate the cache.-               -> MultiGetService m a b                             -- ^ The service to cache.-               -> MultiGetService m a b+-- | Caching of a 'MultiGetService' instance, that defers to external functions for the actual caching.+-- Partial responses will result in partial fallback calls that get just the missing keys.+-- Values within @m@ will be lost for calls that hit the cache.+cacheWithMulti+  :: (Monad m, Hashable a)+  => (MultiGetRequest a -> m (MultiGetResponse a b))+  -- ^ Cache lookup function, used before potentially invoking the fallback service.+  -> (MultiGetResponse a b -> m ())+  -- ^ Cache write function, used after invoking the fallback service to populate the cache.+  -> MultiGetService m a b+  -- ^ The service to cache.+  -> MultiGetService m a b cacheWithMulti lookupWith insertWith service =   let fallback request = do-                            result <- service request-                            insertWith result-                            return result+        result <- service request+        insertWith result+        pure result       cachedService request = do-                                fromCache <- lookupWith request-                                let uncachedKeys = S.difference request (S.fromList $ M.keys fromCache)-                                if (S.null uncachedKeys) then (return fromCache) else (fmap (M.union fromCache) $ fallback uncachedKeys)-  in cachedService+        fromCache <- lookupWith request+        let uncachedKeys = S.difference request (S.fromList $ M.keys fromCache)+        if S.null uncachedKeys+          then pure fromCache+          else M.union fromCache <$> fallback uncachedKeys+  in  cachedService

@@ -1,55 +1,68 @@-{-# LANGUAGE DeriveDataTypeable  #-} {-# LANGUAGE FlexibleContexts    #-} {-# LANGUAGE ScopedTypeVariables #-} --- | Module containing circuit breaker functionality, which is the ability to open a circuit once a number of failures have occurred, thereby preventing later calls from attempting 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.-module Glue.CircuitBreaker(-    CircuitBreakerOptions+-- | Module containing circuit breaker functionality, which is the ability to open a circuit+-- once a number of failures have occurred, thereby preventing later calls from attempting+-- 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.+module Glue.CircuitBreaker+  ( CircuitBreakerOptions (..)   , CircuitBreakerState-  , CircuitBreakerException(..)+  , CircuitBreakerException (..)   , combineCircuitBreakerStates   , isCircuitBreakerOpen   , isCircuitBreakerClosed   , defaultCircuitBreakerOptions   , circuitBreaker-  , maxBreakerFailures-  , resetTimeoutSecs-  , breakerDescription-) where+  ) where -import           Control.Exception.Lifted-import           Control.Monad.Base-import           Control.Monad.Trans.Control-import           Data.IORef.Lifted-import           Data.Time.Clock.POSIX-import           Data.Typeable+import           Data.Time.Clock.POSIX (getPOSIXTime) import           Glue.Types+import           UnliftIO              (Exception, MonadIO, MonadUnliftIO,+                                        liftIO)+import           UnliftIO.Exception    (bracketOnError, throwIO)+import           UnliftIO.IORef        (IORef, atomicModifyIORef', newIORef,+                                        readIORef, writeIORef)  -- | 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     -- ^ Description that is attached to the failure so as to identify the particular circuit.-}+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+  -- ^ Description that is attached to the failure so as to identify the particular circuit.+  }+  deriving (Eq, Show)  -- | Defaulted options for the circuit breaker with 3 failures over 60 seconds. defaultCircuitBreakerOptions :: CircuitBreakerOptions-defaultCircuitBreakerOptions = CircuitBreakerOptions { maxBreakerFailures = 3, resetTimeoutSecs = 60, breakerDescription = "Circuit breaker open." }+defaultCircuitBreakerOptions =+  CircuitBreakerOptions+    { maxBreakerFailures = 3+    , resetTimeoutSecs   = 60+    , breakerDescription = "Circuit breaker open."+    }  -- | Status indicating if the circuit is open.-data BreakerStatus = BreakerClosed Int | BreakerOpen Int deriving (Eq, Show)+data BreakerStatus+  = BreakerClosed Int+  | BreakerOpen Int+  deriving (Eq, Show)  -- | Representation of the state the circuit breaker is currently in.-data CircuitBreakerState = CircuitBreakerState [IORef BreakerStatus]+newtype CircuitBreakerState = CircuitBreakerState [IORef BreakerStatus]  -- | Exception thrown when the circuit is open.-data CircuitBreakerException = CircuitBreakerException String deriving (Eq, Show, Typeable)+newtype CircuitBreakerException = CircuitBreakerException String+  deriving (Eq, Show)+ instance Exception CircuitBreakerException  -- | Combines multiple states together.-combineCircuitBreakerStates :: (Foldable t) => t CircuitBreakerState -> CircuitBreakerState-combineCircuitBreakerStates states = CircuitBreakerState $ foldMap (\(CircuitBreakerState refs) -> refs) states+combineCircuitBreakerStates :: Foldable t => t CircuitBreakerState -> CircuitBreakerState+combineCircuitBreakerStates = CircuitBreakerState . foldMap (\(CircuitBreakerState refs) -> refs)  -- | Determines if a specific status is open. isStatusOpen :: BreakerStatus -> Bool@@ -62,45 +75,55 @@ isStatusClosed (BreakerClosed _) = True  -- | Determines if a circuit breaker is open.-isCircuitBreakerOpen :: (MonadBaseControl IO m) => CircuitBreakerState -> m Bool-isCircuitBreakerOpen (CircuitBreakerState states) = fmap or $ traverse (\ref -> fmap isStatusOpen $ readIORef ref) states+isCircuitBreakerOpen :: MonadIO m => CircuitBreakerState -> m Bool+isCircuitBreakerOpen (CircuitBreakerState states) =+  or <$> traverse (fmap isStatusOpen . readIORef) states  -- | Determines if a circuit breaker is closed.-isCircuitBreakerClosed :: (MonadBaseControl IO m) => CircuitBreakerState -> m Bool-isCircuitBreakerClosed (CircuitBreakerState states) = fmap and $ traverse (\ref -> fmap isStatusClosed $ readIORef ref) states+isCircuitBreakerClosed :: MonadIO m => CircuitBreakerState -> m Bool+isCircuitBreakerClosed (CircuitBreakerState states) =+  and <$> traverse (fmap isStatusClosed . readIORef) states --- TODO: Check that values within m aren't lost on a successful call. -- | Circuit breaking services can be constructed with this function.-circuitBreaker :: (MonadBaseControl IO m, MonadBaseControl IO n)-               => CircuitBreakerOptions       -- ^ Options for specifying the circuit breaker behaviour.-               -> BasicService m a b          -- ^ Service to protect with the circuit breaker.-               -> n (CircuitBreakerState, BasicService m a b)-circuitBreaker options service =-  let retrieveTime                = liftBase $ round `fmap` getPOSIXTime-      failureMax                  = maxBreakerFailures options-      callIfClosed request ref    = bracketOnError (return ()) (\_ -> incErrors ref) (\_ -> service request)-      canaryCall request ref      = do-                                      result <- callIfClosed request ref-                                      writeIORef ref $ BreakerClosed 0-                                      return 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                 = throw $ 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--  in do-        ref <- newIORef $ BreakerClosed 0-        return (CircuitBreakerState [ref], breakerService ref)+circuitBreaker+  :: (MonadUnliftIO m, MonadIO n)+  => CircuitBreakerOptions+  -- ^ Options for specifying the circuit breaker behaviour.+  -> BasicService m a b+  -- ^ Service to protect with the circuit breaker.+  -> n (CircuitBreakerState, 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)

@@ -0,0 +1,87 @@+{-# LANGUAGE FlexibleContexts    #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Module supporting concurrent request constraints (bulkhead pattern),+-- limiting the number of in-flight and optionally queued requests.+module Glue.ConcurrencyLimit+  ( ConcurrencyLimitOptions (..)+  , ConcurrencyLimitExceededException (..)+  , defaultConcurrencyLimitOptions+  , limitConcurrency+  ) where++import           Glue.Types+import           UnliftIO           (Exception, MonadIO, MonadUnliftIO)+import           UnliftIO.Exception (bracketOnError, finally, throwIO)+import           UnliftIO.STM       (atomically, modifyTVar', newTVarIO,+                                     readTVar, retrySTM, writeTVar)++-- | Options for configuring concurrent request limits.+data ConcurrencyLimitOptions = ConcurrencyLimitOptions+  { maxInFlight                 :: Int+  -- ^ Maximum number of concurrent requests allowed to execute the underlying service simultaneously.+  , maxQueued                   :: Maybe Int+  -- ^ Optional maximum number of requests allowed to wait in the queue (`Nothing` for unbounded queue, `Just 0` for no queueing).+  , concurrencyLimitDescription :: String+  -- ^ Description attached to t'ConcurrencyLimitExceededException' when a request is rejected.+  }+  deriving (Eq, Show)++-- | Default concurrency limit options allowing 10 in-flight requests and an unbounded queue.+defaultConcurrencyLimitOptions :: ConcurrencyLimitOptions+defaultConcurrencyLimitOptions =+  ConcurrencyLimitOptions+    { maxInFlight                 = 10+    , maxQueued                   = Nothing+    , concurrencyLimitDescription = "Concurrency limit exceeded."+    }++-- | Exception thrown when both in-flight and queue limits have been reached.+newtype ConcurrencyLimitExceededException = ConcurrencyLimitExceededException String+  deriving (Eq, Show)++instance Exception ConcurrencyLimitExceededException++-- | 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 service = do+  let maxActive = max 1 (maxInFlight options)+  inFlightVar <- newTVarIO (0 :: Int)+  queuedVar   <- newTVarIO (0 :: Int)+  let releaseInFlight = atomically $ modifyTVar' inFlightVar (subtract 1)+      releaseQueued   = atomically $ modifyTVar' queuedVar (subtract 1)+      waitForSlot     = atomically $ do+        inFlight <- readTVar inFlightVar+        if inFlight < maxActive+          then do+            modifyTVar' queuedVar (subtract 1)+            writeTVar inFlightVar (inFlight + 1)+          else retrySTM+      limitedService request = do+        admission <- atomically $ do+          inFlight <- readTVar inFlightVar+          queued   <- readTVar queuedVar+          if inFlight < maxActive && queued == 0+            then do+              writeTVar inFlightVar (inFlight + 1)+              pure (Right True)+            else case maxQueued options of+              Just maxQ | queued >= maxQ -> pure (Left ())+              _ -> do+                writeTVar queuedVar (queued + 1)+                pure (Right False)+        case admission of+          Left () ->+            throwIO $ ConcurrencyLimitExceededException (concurrencyLimitDescription options)+          Right True ->+            finally (service request) releaseInFlight+          Right False -> do+            bracketOnError (pure ()) (\_ -> releaseQueued) (\_ -> waitForSlot)+            finally (service request) releaseInFlight+  pure limitedService

@@ -1,57 +1,94 @@ {-# LANGUAGE FlexibleContexts    #-} {-# LANGUAGE ScopedTypeVariables #-} --- | Module supporting the dogpile protection of a service, see <http://en.wikipedia.org/wiki/Cache_stampede http://en.wikipedia.org/wiki/Cache_stampede>.-module Glue.DogpileProtection(-  dogpileProtect-) where+-- | Module supporting the dogpile protection of a service,+-- see <https://en.wikipedia.org/wiki/Cache_stampede>.+module Glue.DogpileProtection+  ( DogpileOptions (..)+  , defaultDogpileOptions+  , dogpileProtectWith+  , dogpileProtect+  ) where -import           Control.Concurrent-import           Control.Exception-import           Data.Hashable+import           Control.Monad       (void)+import           Data.Hashable       (Hashable) import qualified Data.HashMap.Strict as M-import           Data.IORef import           Glue.Types+import           UnliftIO            (MonadIO, MonadUnliftIO, SomeException,+                                      liftIO)+import           UnliftIO.Concurrent (forkIO, threadDelay)+import           UnliftIO.Exception  (onException, throwIO, toException, tryAny)+import           UnliftIO.IORef      (IORef, atomicModifyIORef', newIORef)+import           UnliftIO.MVar       (MVar, newEmptyMVar, putMVar, readMVar,+                                      tryPutMVar) -data DogpileResult b = CachedValue (Either SomeException b)-                     | RequestInProgress (IO b)+-- | Options for configuring dogpile protection.+newtype DogpileOptions = DogpileOptions+  { dogpileCacheTimeMs :: Int+  -- ^ How long (in milliseconds) to retain a completed result for subsequent callers.+  }+  deriving (Eq, Show) -type ResultMap a b = M.HashMap a (DogpileResult b)+-- | Default options for dogpile protection, caching completed values for 5 seconds (5000ms).+defaultDogpileOptions :: DogpileOptions+defaultDogpileOptions =+  DogpileOptions+    { dogpileCacheTimeMs = 5000+    } -type ResultRef a b = IORef (ResultMap a b)+data DogpileResult b+  = CachedValue (Either SomeException b)+  | RequestInProgress (MVar (Either SomeException b)) -getDogpileResult :: DogpileResult b -> IO b-getDogpileResult (RequestInProgress requestWait) = requestWait-getDogpileResult (CachedValue result)            = either throw return result+type ResultMap a b = M.HashMap a (DogpileResult b)+type ResultRef a b = IORef (ResultMap a b) -waitForMVar :: MVar (Either SomeException b) -> IO b-waitForMVar mvar = do-  result <- readMVar mvar-  either throw return result+getDogpileResult :: MonadIO m => DogpileResult b -> m b+getDogpileResult (RequestInProgress mvar) = readMVar mvar >>= either throwIO pure+getDogpileResult (CachedValue result)     = either throwIO pure result -getProtectedResult :: (Eq a, Hashable a)-                   => a -> BasicService IO a b -> MVar (Either SomeException b) -> ResultRef a b -> ResultMap a b -> (ResultMap a b, IO b)-getProtectedResult request service mvar mapRef resultMap =-  let mvarWait = waitForMVar mvar-      invokeService = do-        result <- try $ service request-        putMVar mvar result-        atomicModifyIORef' mapRef (\mapToUpdate -> (M.insert request (CachedValue result) mapToUpdate, ()))-        forkIO $ do-          threadDelay (5 * 1000 * 1000)-          atomicModifyIORef' mapRef (\mapToUpdate -> (M.delete request mapToUpdate, ()))-      inCache = M.lookup request resultMap-  in  maybe (M.insert request (RequestInProgress mvarWait) resultMap, invokeService >> mvarWait) (\r -> (resultMap, getDogpileResult r)) inCache+-- | Dogpile protection of a service with configurable options.+dogpileProtectWith+  :: forall m n a b.+     (MonadUnliftIO m, MonadIO n, Hashable a)+  => DogpileOptions+  -- ^ Options configuring dogpile protection.+  -> BasicService m a b+  -- ^ The service to protect.+  -> n (BasicService m a b)+dogpileProtectWith options service = do+  mapRef <- newIORef M.empty :: n (ResultRef a b)+  let ttlMs = dogpileCacheTimeMs options+      protectedService request = do+        newVar <- newEmptyMVar+        actionOrExisting <- atomicModifyIORef' mapRef $ \resultMap ->+          case M.lookup request resultMap of+            Just existing -> (resultMap, Right existing)+            Nothing       -> (M.insert request (RequestInProgress newVar) resultMap, Left newVar)+        case actionOrExisting of+          Right existing -> getDogpileResult existing+          Left mvar -> do+            let cleanupOnAbort = do+                  atomicModifyIORef' mapRef (\m -> (M.delete request m, ()))+                  void $ tryPutMVar mvar (Left (toException (userError "Dogpile leader aborted")))+            result <- tryAny (service request) `onException` cleanupOnAbort+            putMVar mvar result+            if ttlMs > 0+              then do+                atomicModifyIORef' mapRef (\m -> (M.insert request (CachedValue result) m, ()))+                void $ liftIO $ forkIO $ do+                  threadDelay (ttlMs * 1000)+                  atomicModifyIORef' mapRef (\m -> (M.delete request m, ()))+              else+                atomicModifyIORef' mapRef (\m -> (M.delete request m, ()))+            either throwIO pure result+  pure protectedService --- | Dogpile protection of a service, to prevent multiple calls for the same value being submitted.--- | Loses the values held within m.-dogpileProtect :: (Eq a, Hashable a)-               => BasicService IO a b   -- ^ The service to protect.-               -> IO (BasicService IO a b)-dogpileProtect service = do-  mapRef <- newIORef M.empty-  let protectedService request = do-        mvar <- newEmptyMVar-        resultAction <- atomicModifyIORef' mapRef $ getProtectedResult request service mvar mapRef-        resultAction-  return protectedService+-- | Dogpile protection of a service using 'defaultDogpileOptions',+-- preventing multiple concurrent calls for the same value from being submitted.+dogpileProtect+  :: (MonadUnliftIO m, MonadIO n, Hashable a)+  => BasicService m a b+  -- ^ The service to protect.+  -> n (BasicService m a b)+dogpileProtect = dogpileProtectWith defaultDogpileOptions

@@ -2,36 +2,45 @@ {-# 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 Glue.Failover(-    FailoverOptions+module Glue.Failover+  ( FailoverOptions (..)   , defaultFailoverOptions   , failover-  , maxFailovers-  , transformFailoverRequest-) where+  ) where -import           Control.Exception.Lifted    hiding (throw)-import           Control.Monad.Trans.Control import           Glue.Types+import           UnliftIO           (MonadUnliftIO)+import           UnliftIO.Exception (catchAny)  -- | Options for determining behaviour of failover services.-data FailoverOptions a = FailoverOptions {-    maxFailovers             :: Int      -- ^ The maximum numer of times the service will failover and make another attempt.-  , transformFailoverRequest :: a -> a   -- ^ Each time the service performs a failover, transform the request with this function.-}+data FailoverOptions a = FailoverOptions+  { maxFailovers             :: Int+  -- ^ The maximum number of times the service will failover and make another attempt.+  , transformFailoverRequest :: a -> a+  -- ^ Each time the service performs a failover, transform the request with this function.+  } --- | Simple defaults that results in retrying 3 times in a dumb way immediately without transforming the request.+-- | Simple defaults that result in retrying 3 times immediately without transforming the request. defaultFailoverOptions :: FailoverOptions a-defaultFailoverOptions = FailoverOptions { maxFailovers = 3, transformFailoverRequest = id }+defaultFailoverOptions =+  FailoverOptions+    { maxFailovers             = 3+    , transformFailoverRequest = id+    }  -- | Creates a service that can transform requests with each subsequent failure.-failover :: (MonadBaseControl IO m)-         => FailoverOptions a     -- ^ Instance of 'FailoverOptions' to configure the failover functionality.-         -> BasicService m a b    -- ^ The service to perform failover around.-         -> BasicService m a b+failover+  :: MonadUnliftIO m+  => FailoverOptions a+  -- ^ Instance of t'FailoverOptions' to configure the failover functionality.+  -> BasicService m a b+  -- ^ The service to perform failover around.+  -> BasicService m a b failover options service =   let invokeService failCount request =-                                        let afterFail   = (\(_ :: SomeException) -> invokeService (failCount + 1) ((transformFailoverRequest options) request))-                                            invoke      = service request-                                        in  if failCount >= (maxFailovers options) then invoke else catch invoke afterFail+        let afterFail _ = invokeService (failCount + 1) (transformFailoverRequest options request)+            invoke      = service request+        in  if failCount >= maxFailovers options+              then invoke+              else catchAny invoke afterFail   in  invokeService 0

@@ -0,0 +1,81 @@+{-# LANGUAGE FlexibleContexts    #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Module supporting request hedging (duplication), dispatching multiple concurrent+-- requests (optionally staggered) and completing with the first successful result.+module Glue.Hedging+  ( HedgingOptions (..)+  , defaultHedgingOptions+  , hedgingService+  ) where++import           Control.Monad       (when)+import           Data.Maybe          (fromMaybe)+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)++-- | Options for configuring hedged (duplicated) service calls.+data HedgingOptions = 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).+  }+  deriving (Eq, Show)++-- | Default hedging options that fire 2 concurrent requests with no stagger delay.+defaultHedgingOptions :: HedgingOptions+defaultHedgingOptions =+  HedgingOptions+    { hedgingCount   = 2+    , hedgingDelayMs = 0+    }++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+  -- ^ 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

@@ -0,0 +1,51 @@+{-# LANGUAGE FlexibleContexts    #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Module providing round-robin load balancing across multiple services.+module Glue.LoadBalancer+  ( NoServicesAvailableException (..)+  , roundRobinLoadBalancer+  , roundRobinLoadBalancerNE+  ) where++import           Data.Foldable      (toList)+import           Data.List.NonEmpty (NonEmpty (..))+import qualified Data.List.NonEmpty as NE+import           Glue.Types+import           UnliftIO           (Exception, MonadIO)+import           UnliftIO.Exception (throwIO)+import           UnliftIO.IORef     (atomicModifyIORef', newIORef)++-- | Exception thrown when attempting to call a load balancer configured with an empty set of services.+data NoServicesAvailableException = NoServicesAvailableException+  deriving (Eq, Show)++instance Exception NoServicesAvailableException++-- | Creates a service that distributes requests across a 'NonEmpty' list of backend services in round-robin order.+roundRobinLoadBalancerNE+  :: (MonadIO m, MonadIO n)+  => NonEmpty (BasicService m a b)+  -- ^ Non-empty collection of backend services to balance across.+  -> n (BasicService m a b)+roundRobinLoadBalancerNE services = do+  let svcList = NE.toList services+      count   = length svcList+  indexRef <- newIORef (0 :: Int)+  let balanced request = do+        idx <- atomicModifyIORef' indexRef (\i -> ((i + 1) `mod` count, i))+        let target = svcList !! idx+        target request+  pure balanced++-- | Creates a service that distributes requests across a foldable collection of backend services in round-robin order.+-- If the collection is empty, invoking the returned service throws t'NoServicesAvailableException'.+roundRobinLoadBalancer+  :: (MonadIO m, MonadIO n, Foldable t)+  => t (BasicService m a b)+  -- ^ Collection of backend services to balance across.+  -> n (BasicService m a b)+roundRobinLoadBalancer services =+  case toList services of+    []     -> pure $ const $ throwIO NoServicesAvailableException+    (x:xs) -> roundRobinLoadBalancerNE (x :| xs)

@@ -1,109 +1,148 @@ {-# LANGUAGE BangPatterns        #-}-{-# LANGUAGE DeriveDataTypeable  #-} {-# LANGUAGE FlexibleContexts    #-} {-# LANGUAGE RankNTypes          #-} {-# LANGUAGE RecordWildCards     #-} {-# LANGUAGE ScopedTypeVariables #-}  -- | Module containing a form of caching where values for given keys are preloaded ahead of time.--- | Once warmed up requests for preloaded keys will be instant, with the values refreshed in the background.-module Glue.Preload(-    PreloadedOptions+-- Once warmed up, requests for preloaded keys will be instant, with the values refreshed in the background.+module Glue.Preload+  ( PreloadedOptions (..)+  , PreloadStoppedBeforeExecutedException (..)   , defaultPreloadedOptions   , preloadingService   , preloadingCall-  , preloadedKeys-  , preloadingRefreshTimeMs-  , preloadingRun-) where+  ) where -import           Control.Concurrent.Lifted-import           Control.Exception.Base      hiding (throwIO)-import           Control.Exception.Lifted-import           Control.Monad.IO.Class-import           Control.Monad.Trans.Control-import           Data.Hashable-import qualified Data.HashMap.Strict         as M-import qualified Data.HashSet                as S-import           Data.IORef.Lifted-import           Data.Maybe-import           Data.Typeable+import           Control.Monad       (void)+import           Data.Hashable       (Hashable)+import qualified Data.HashMap.Strict as M+import qualified Data.HashSet        as S+import           Data.Maybe          (fromJust) import           Glue.Types+import           UnliftIO            (Exception, MonadIO, MonadUnliftIO,+                                      SomeException (..), liftIO)+import           UnliftIO.Async      (race_)+import           UnliftIO.Concurrent (forkIO, threadDelay)+import           UnliftIO.Exception  (throwIO)+import           UnliftIO.IORef      (IORef, atomicModifyIORef', newIORef,+                                      readIORef)+import           UnliftIO.MVar       (MVar, newEmptyMVar, readMVar, tryPutMVar)  -- | Options for determining behaviour of preloading services.-data PreloadedOptions m a b = PreloadedOptions {-  preloadedKeys           :: S.HashSet a,       -- ^ Keys to preload.-  preloadingRefreshTimeMs :: Int,               -- ^ Amount of time between refreshes.-  preloadingRun           :: MToIO m            -- ^ Get an IO of the response for the caching.-}+data PreloadedOptions m a b = PreloadedOptions+  { preloadedKeys           :: S.HashSet a+  -- ^ Keys to preload.+  , preloadingRefreshTimeMs :: Int+  -- ^ Amount of time between refreshes.+  , preloadingRun           :: MToIO m+  -- ^ Get an 'IO' of the response for the caching.+  } --- | Defaulted options for preloading a HashSet of keys with a 30 second refresh time.+-- | 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 toPreload pRun = PreloadedOptions {-  preloadedKeys           = toPreload,-  preloadingRefreshTimeMs = 30 * 1000,-  preloadingRun           = pRun-}+defaultPreloadedOptions toPreload pRun =+  PreloadedOptions+    { preloadedKeys           = toPreload+    , preloadingRefreshTimeMs = 30 * 1000+    , preloadingRun           = pRun+    } -data PreloadedState a b = PreloadedStarted-                        | PreloadedWithResult (FailOrSuccess a b) Bool+data PreloadedState a b+  = PreloadedStarted+  | PreloadedWithResult (FailOrSuccess a b) Bool -data PreloadStoppedBeforeExecutedException = PreloadStoppedBeforeExecutedException deriving (Eq, Show, Typeable)+-- | Exception thrown when preloading is stopped before the initial preload has executed.+data PreloadStoppedBeforeExecutedException = PreloadStoppedBeforeExecutedException+  deriving (Eq, Show)+ instance Exception PreloadStoppedBeforeExecutedException -modifyStateWithResult :: forall a b . Either SomeException (MultiGetResponse a b) -> PreloadedState a b -> (PreloadedState a b, Bool)-modifyStateWithResult result PreloadedStarted           = (PreloadedWithResult result True, True)-modifyStateWithResult result (PreloadedWithResult _ c)  = (PreloadedWithResult result c, c)+modifyStateWithResult :: Either SomeException (MultiGetResponse a b) -> PreloadedState a b -> (PreloadedState a b, Bool)+modifyStateWithResult result PreloadedStarted          = (PreloadedWithResult result True, True)+modifyStateWithResult result (PreloadedWithResult _ c) = (PreloadedWithResult result c, c) -applyResultToState :: forall m a b . MonadBaseControl IO m => IORef (PreloadedState a b) -> FailOrSuccess a b -> m Bool-applyResultToState stateRef result = atomicModifyIORef' stateRef $ modifyStateWithResult result+applyResultToState :: MonadIO m => IORef (PreloadedState a b) -> MVar () -> FailOrSuccess a b -> m Bool+applyResultToState stateRef readyVar result = do+  continue <- atomicModifyIORef' stateRef $ modifyStateWithResult result+  void $ tryPutMVar readyVar ()+  pure continue -waitForResult :: forall m a b . (MonadBaseControl IO m) => IORef (PreloadedState a b) -> m (MultiGetResponse a b)-waitForResult stateRef = do-                            state <- readIORef stateRef-                            let tryAgainLater = threadDelay 1000 >> waitForResult stateRef-                            case state of-                                          PreloadedStarted                      -> tryAgainLater-                                          PreloadedWithResult (Right success) _ -> return success-                                          PreloadedWithResult (Left failure)  _ -> throwIO failure+waitForResult :: MonadIO m => IORef (PreloadedState a b) -> MVar () -> m (MultiGetResponse a b)+waitForResult stateRef readyVar = do+  readMVar readyVar+  state <- readIORef stateRef+  case state of+    PreloadedStarted                      -> throwIO PreloadStoppedBeforeExecutedException+    PreloadedWithResult (Right success) _ -> pure success+    PreloadedWithResult (Left failure)  _ -> throwIO failure  markAsFinished :: PreloadedState a b -> (PreloadedState a b, ())-markAsFinished PreloadedStarted = (PreloadedWithResult (Left (SomeException PreloadStoppedBeforeExecutedException)) False, ())+markAsFinished PreloadedStarted          = (PreloadedWithResult (Left (SomeException PreloadStoppedBeforeExecutedException)) False, ()) markAsFinished (PreloadedWithResult r _) = (PreloadedWithResult r False, ())  -- | Preloads the results of calls for given keys.-preloadingService :: forall m n a b . (MonadIO m, MonadIO n, MonadBaseControl IO m, MonadBaseControl IO n, Eq a, Hashable a, Show a)-                  => PreloadedOptions m a b      -- ^ Instance of 'PreloadedOptions' to configure the preloading functionality.-                  -> MultiGetService m a b       -- ^ The service to perform preloading of.-                  -> n (MultiGetService m a b, () -> n ())+preloadingService+  :: forall m n a b.+     (MonadUnliftIO m, MonadIO n, Hashable a)+  => PreloadedOptions m a b+  -- ^ Instance of t'PreloadedOptions' to configure the preloading functionality.+  -> MultiGetService m a b+  -- ^ The service to perform preloading of.+  -> n (MultiGetService m a b, () -> n ()) preloadingService PreloadedOptions{..} service = do   !stateIORef <- newIORef PreloadedStarted-  let stop _ = atomicModifyIORef' stateIORef markAsFinished+  !readyVar   <- newEmptyMVar+  !stopVar    <- newEmptyMVar+  let stop _ = do+        atomicModifyIORef' stateIORef markAsFinished+        void $ tryPutMVar readyVar ()+        void $ tryPutMVar stopVar ()   let runUpdate = do-                    result <- makeCall service preloadedKeys-                    applyResultToState stateIORef result+        result <- makeCall service preloadedKeys+        applyResultToState stateIORef readyVar result   let updatePreloaded = do-                          continue <- liftIO $ preloadingRun $ runUpdate-                          if continue then threadDelay (preloadingRefreshTimeMs * 1000) >> updatePreloaded else return ()+        continue <- liftIO $ preloadingRun runUpdate+        if continue+          then do+            stopped <- liftIO $ raceWait stopVar (preloadingRefreshTimeMs * 1000)+            if stopped then pure () else updatePreloaded+          else pure ()   let plService request = do-                            let fromPreloadKeys = S.intersection request preloadedKeys-                            let fromServiceKeys = S.difference request preloadedKeys-                            !fromPreload <- if S.null fromPreloadKeys then return M.empty else fmap (M.filterWithKey (\k -> \_ -> S.member k fromPreloadKeys)) $ waitForResult stateIORef-                            !fromService <- if S.null fromServiceKeys then return M.empty else service fromServiceKeys-                            return $ M.union fromService fromPreload-  _ <- fork updatePreloaded-  return (plService, stop)+        let fromPreloadKeys = S.intersection request preloadedKeys+        let fromServiceKeys = S.difference request preloadedKeys+        !fromPreload <-+          if S.null fromPreloadKeys+            then pure M.empty+            else M.filterWithKey (\k _ -> S.member k fromPreloadKeys) <$> waitForResult stateIORef readyVar+        !fromService <-+          if S.null fromServiceKeys+            then pure M.empty+            else service fromServiceKeys+        pure $ M.union fromService fromPreload+  void $ liftIO $ forkIO updatePreloaded+  pure (plService, stop)+  where+    raceWait :: MVar () -> Int -> IO Bool+    raceWait stopVar delayUs = do+      race_ (readMVar stopVar) (threadDelay delayUs)+      not <$> tryPutMVar stopVar ()  -- | Preloads a single parameter-less call.-preloadingCall :: forall m n b . (MonadIO m, MonadIO n, MonadBaseControl IO m, MonadBaseControl IO n)-               => MToIO m                     -- ^ Get an IO of the call for the caching.-               -> Int                         -- ^ Amount of time between refreshes.-               -> m b                         -- ^ The call to perform preloading of.-               -> n (m b, () -> n ())+preloadingCall+  :: forall m n b.+     (MonadUnliftIO m, MonadIO n)+  => MToIO m+  -- ^ Get an 'IO' of the call for the caching.+  -> Int+  -- ^ Amount of time between refreshes.+  -> m b+  -- ^ The call to perform preloading of.+  -> n (m b, () -> n ()) preloadingCall toIO timeBetweenRefreshes call = do   let multiGetService = basicToMultiGet (\_ -> call) :: MultiGetService m () b   let options = PreloadedOptions (S.singleton ()) timeBetweenRefreshes toIO   (preloading, stop) <- preloadingService options multiGetService   let preloadingBasicService = multiGetToBasic preloading-  let preloadingCallResult = fmap (fromJust) $ preloadingBasicService ()-  return (preloadingCallResult, stop)+  let preloadingCallResult = fromJust <$> preloadingBasicService ()+  pure (preloadingCallResult, stop)

@@ -1,56 +1,72 @@-{-# LANGUAGE DeriveDataTypeable  #-} {-# LANGUAGE FlexibleContexts    #-} {-# LANGUAGE ScopedTypeVariables #-}  -- | Module containing retry functionality, allowing the construction of services that attempt multiple times in case of transient failure.-module Glue.Retry(-    RetryOptions+module Glue.Retry+  ( RetryOptions (..)   , defaultRetryOptions   , retryingService-  , retryAllowed-  , retryInitialWaitTimeMs-  , maximumRetries-  , retryWaitTimeMultiplier-) where+  ) where -import           Control.Concurrent.Lifted-import           Control.Exception.Lifted-import           Control.Monad.Trans.Control+import           Control.Monad       (when) import           Glue.Types+import           UnliftIO            (MonadUnliftIO, SomeAsyncException,+                                      SomeException, fromException)+import           UnliftIO.Concurrent (threadDelay)+import           UnliftIO.Exception  (catch, throwIO)  -- | Options for determining behaviour of retrying services.-data RetryOptions a = 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 to wait after the first failure.-  maximumRetries          :: Int,         -- ^ The upper bound on how many attempts to make when invoking the service.-  retryWaitTimeMultiplier :: Double       -- ^ How much to multiply 'retryInitialWaitTimeMs' by for each number of times the service has retried.-}+data RetryOptions a = 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+  -- ^ 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.+  }  -- | Defaulted options for retrying 3 times with no wait time. defaultRetryOptions :: RetryOptions a-defaultRetryOptions = RetryOptions {-    retryAllowed            = (\_ -> True)-  , retryInitialWaitTimeMs  = 0-  , maximumRetries          = 3-  , retryWaitTimeMultiplier = 0-  }+defaultRetryOptions =+  RetryOptions+    { retryAllowed            = const True+    , retryInitialWaitTimeMs  = 0+    , maximumRetries          = 3+    , retryWaitTimeMultiplier = 0+    }  possibleAsyncException :: SomeException -> Maybe SomeAsyncException possibleAsyncException = fromException  -- | 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 :: (MonadBaseControl IO m)-                => RetryOptions a             -- ^ Instance of 'RetryOptions' to configure the retry functionality.-                -> BasicService m a b         -- ^ The service to perform retries of.-                -> BasicService m a b+-- Asynchronous exceptions don't result in a retry; they are immediately rethrown.+retryingService+  :: forall m a b.+     MonadUnliftIO m+  => RetryOptions a+  -- ^ 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 -> throw ae-                                                                    Nothing -> wait (rc + 1) >> 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-      wait retryCount             = threadDelay $ round $ fromIntegral (retryInitialWaitTimeMs options) * ((retryWaitTimeMultiplier options) ^ retryCount)+  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

@@ -0,0 +1,128 @@+{-# LANGUAGE FlexibleContexts    #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Module supporting graceful service shutdown, rejecting new requests with either+-- a fallback value or t'ServiceShuttingDownException', waiting for in-flight requests+-- to drain within a configurable grace period, and cancelling any remaining calls if+-- the grace period expires.+module Glue.Shutdown+  ( ShutdownOptions (..)+  , ServiceShuttingDownException (..)+  , ShutdownHandle (..)+  , defaultShutdownOptions+  , gracefulShutdown+  ) where++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+  { shutdownFallback      :: Maybe b+  -- ^ Optional default value returned for rejected requests after shutdown is initiated.+  -- If 'Nothing', t'ServiceShuttingDownException' is thrown instead.+  , shutdownGracePeriodMs :: Int+  -- ^ Maximum time in milliseconds to wait for existing in-flight requests to complete before cancelling them.+  , shutdownDescription   :: String+  -- ^ Description attached to t'ServiceShuttingDownException'.+  }+  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 =+  ShutdownOptions+    { shutdownFallback      = Nothing+    , shutdownGracePeriodMs = 5000+    , shutdownDescription   = "Service is shutting down."+    }++-- | Exception thrown when a service is shutting down and no 'shutdownFallback' is configured.+newtype ServiceShuttingDownException = ServiceShuttingDownException String+  deriving (Eq, Show)++instance Exception ServiceShuttingDownException++-- | Handle used to control and inspect a service wrapped with 'gracefulShutdown'.+data ShutdownHandle m = ShutdownHandle+  { initiateShutdown :: m Bool+  -- ^ Initiates shutdown: immediately rejects new requests, waits up to 'shutdownGracePeriodMs'+  -- for in-flight requests to finish, and cancels any remaining calls if the grace period expires.+  -- Returns 'True' if all in-flight calls drained cleanly within the grace period, or 'False' if forced cancellation occurred.+  , isShuttingDown   :: m Bool+  -- ^ Returns 'True' if shutdown has been initiated.+  , inFlightCount    :: m Int+  -- ^ Returns the current number of in-flight requests executing the service.+  }++-- | Wraps a service with graceful shutdown support.+gracefulShutdown+  :: forall m n a b.+     (MonadUnliftIO m, MonadIO n)+  => ShutdownOptions 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+  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 ()+      waitForForceCancel = atomically $ do+        cancelled <- readTVar forceCancelVar+        if cancelled then pure () else retrySTM++      waitDrained :: m ()+      waitDrained = atomically $ do+        active <- readTVar inFlightVar+        if active <= 0 then pure () else retrySTM++      wrappedService :: BasicService m a b+      wrappedService request = do+        accepted <- atomically $ do+          shuttingDown <- readTVar shuttingDownVar+          if shuttingDown+            then pure False+            else do+              modifyTVar' inFlightVar (+ 1)+              pure True+        if not accepted+          then rejectRequest+          else do+            res <-+              race waitForForceCancel (service request)+                `finally` atomically (modifyTVar' inFlightVar (subtract 1))+            case res of+              Right val -> pure val+              Left ()   -> rejectRequest++      handle =+        ShutdownHandle+          { initiateShutdown = do+              atomically $ writeTVar shuttingDownVar True+              let graceUs = max 0 (shutdownGracePeriodMs options) * 1000+              drained <- timeout graceUs waitDrained+              case drained of+                Just () -> pure True+                Nothing -> do+                  atomically $ writeTVar forceCancelVar True+                  waitDrained+                  pure False+          , isShuttingDown = atomically $ readTVar shuttingDownVar+          , inFlightCount  = atomically $ readTVar inFlightVar+          }++  pure (wrappedService, handle)

@@ -1,26 +1,28 @@ {-# LANGUAGE BangPatterns        #-}-{-# LANGUAGE DeriveDataTypeable  #-} {-# LANGUAGE FlexibleContexts    #-}-{-# LANGUAGE RecordWildCards     #-} {-# LANGUAGE ScopedTypeVariables #-}  -- | Module containing a switchable service, allowing the possibility of changing--- | a service "live".-module Glue.Switching(-    switchingService-) where+-- a service "live".+module Glue.Switching+  ( switchingService+  ) where -import           Control.Monad.Base-import           Data.IORef.Lifted import           Glue.Types+import           UnliftIO       (MonadIO)+import           UnliftIO.IORef (newIORef, readIORef, writeIORef)  -- | Provides a switchable service.-switchingService :: (MonadBase IO m, MonadBase IO n) => MultiGetService m a b  -- ^ The service to initialise the switching service with.-                 -> n (MultiGetService m a b, (MultiGetService m a b) -> n ())+-- Works for both 'BasicService' and 'MultiGetService' (since 'MultiGetService' is a 'BasicService').+switchingService+  :: (MonadIO m, MonadIO n)+  => BasicService m a b+  -- ^ The service to initialise the switching service with.+  -> n (BasicService m a b, BasicService m a b -> n ()) switchingService service = do   !serviceRef <- newIORef service   let switching rs = do-                        serviceToUse <- readIORef serviceRef-                        serviceToUse rs+        serviceToUse <- readIORef serviceRef+        serviceToUse rs   let switchingUpdate newService = writeIORef serviceRef newService-  return (switching, switchingUpdate)+  pure (switching, switchingUpdate)

@@ -1,45 +1,50 @@-{-# LANGUAGE DeriveDataTypeable  #-} {-# LANGUAGE FlexibleContexts    #-} {-# LANGUAGE ScopedTypeVariables #-}  -- | Module supporting adding timeouts to a given service.-module Glue.Timeout(-    TimeoutOptions-  , TimeoutException(..)+module Glue.Timeout+  ( TimeoutOptions (..)+  , TimeoutException (..)   , defaultTimeoutOptions   , addTimeout-  , timeoutDescription-  , timeoutLimitMs-) where+  ) where -import           Control.Concurrent.Lifted-import           Control.Exception.Lifted-import           Control.Monad.Trans.Control-import           Data.Typeable 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 {-    timeoutDescription :: String       -- ^ Description added to the 'TimeoutException' thrown when the timeout is exceeded.-  , timeoutLimitMs     :: Int          -- ^ Timeout in milliseconds.-}+data TimeoutOptions = TimeoutOptions+  { timeoutDescription :: String+  -- ^ Description added to the t'TimeoutException' thrown when the timeout is exceeded.+  , timeoutLimitMs     :: Int+  -- ^ Timeout in milliseconds.+  }+  deriving (Eq, Show) --- | Default instance of 'TimeoutOptions' with a timeout of 30 seconds.+-- | Default instance of t'TimeoutOptions' with a timeout of 30 seconds. defaultTimeoutOptions :: TimeoutOptions-defaultTimeoutOptions = TimeoutOptions { timeoutDescription = "Service call timed out.", timeoutLimitMs = 30000 }+defaultTimeoutOptions =+  TimeoutOptions+    { timeoutDescription = "Service call timed out."+    , timeoutLimitMs     = 30000+    }  -- | Exception thrown when the timeout is exceeded.-data TimeoutException = TimeoutException String deriving (Eq, Show, Typeable)+newtype TimeoutException = TimeoutException String+  deriving (Eq, Show)+ instance Exception TimeoutException  -- | Function for producing services protected with a timeout.-addTimeout :: (MonadBaseControl IO m)-           => TimeoutOptions        -- ^ Options to configure the timeout.-           -> BasicService m a b    -- ^ Service to protect with a timeout.-           -> BasicService m a b-addTimeout options service = (\request -> do-  currentThreadId <- myThreadId-  timeoutThreadId <- fork $ do-                              threadDelay (1000 * timeoutLimitMs options)-                              throwTo currentThreadId (TimeoutException $ timeoutDescription options)-  finally (service request) (killThread timeoutThreadId))+addTimeout+  :: MonadUnliftIO m+  => TimeoutOptions+  -- ^ 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

@@ -1,10 +1,8 @@-{-# LANGUAGE DeriveDataTypeable  #-}-{-# LANGUAGE OverloadedStrings   #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.BatcherSpec where+module Glue.BatcherSpec (spec) where -import           Control.Concurrent.Async+import           Control.Concurrent.Async (mapConcurrently) import qualified Data.HashMap.Strict      as M import qualified Data.HashSet             as S import           Glue.Batcher@@ -15,12 +13,19 @@ spec :: Spec spec = do   describe "batchingService" $ do-    it "Requests should receive their full results" $ do+    it "Multi-get requests should receive their full results" $       property $ \(requests :: [[Int]]) -> do-        let requestsAsSets = fmap (S.fromList) requests-        let serviceFunctionality rs = M.fromList $ fmap (\r -> (r, r * 2)) $ S.toList rs-        let service = (return . serviceFunctionality) :: MultiGetService IO Int Int-        (_, multiService) <- batchingService defaultBatchingOptions service+        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         let expectedResults = fmap serviceFunctionality requestsAsSets-        (mapConcurrently multiService requestsAsSets) `shouldReturn` expectedResults+        mapConcurrently multiService requestsAsSets `shouldReturn` expectedResults +    it "Single-key requests should receive their individual lookup results" $+      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+        let expectedResults = fmap (\r -> Just (r * 2)) requests+        mapConcurrently singleService requests `shouldReturn` expectedResults

@@ -1,22 +1,20 @@-{-# LANGUAGE DeriveDataTypeable  #-}-{-# LANGUAGE OverloadedStrings   #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.CachingSpec where+module Glue.CachingSpec (spec) where -import           Control.Exception.Base    hiding (throw, throwIO)-import           Control.Exception.Lifted import qualified Data.HashMap.Strict       as M import qualified Data.HashSet              as S import           Data.IORef-import           Data.Typeable import           Glue.Caching import           Glue.Types import           Test.Hspec import           Test.QuickCheck import           Test.QuickCheck.Instances ()+import           UnliftIO.Exception        (Exception, throwIO) -data CachingTestException = CachingTestException Int deriving (Eq, Show, Typeable)+newtype CachingTestException = CachingTestException Int+  deriving (Eq, Show)+ instance Exception CachingTestException  type TestResp = MultiGetResponse Int Int@@ -24,35 +22,35 @@ spec :: Spec spec = do   describe "cacheWithBasic" $ do-    it "For a second request, the value comes from the cache" $ do+    it "For a second request, the value comes from the cache" $       property $ \(request :: Int, result :: Int) -> do         ref <- newIORef (0 :: Int)         let service req = do-                            if req == request then return () else throwIO (CachingTestException (-1))-                            callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1))-                            if callCount > 1 then throwIO (CachingTestException callCount) else return result+              if req == request then pure () else throwIO (CachingTestException (-1))+              callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1))+              if callCount > 1 then throwIO (CachingTestException callCount) else pure result         cache <- newIORef M.empty-        let lookupWith r = fmap (M.lookup r) $ readIORef cache+        let lookupWith r = M.lookup r <$> readIORef cache         let insertWith req resp = atomicModifyIORef' cache (\c -> (M.insert req resp c, ()))         let cachedService = cacheWithBasic lookupWith insertWith service-        (cachedService request) `shouldReturn` result-        (cachedService request) `shouldReturn` result+        cachedService request `shouldReturn` result+        cachedService request `shouldReturn` result   describe "cacheWithMulti" $ do-    it "For a second request, the value comes from the cache" $ do+    it "For a second request, the value comes from the cache" $       property $ \(result :: TestResp) -> do         let request = S.fromList $ M.keys result         ref <- newIORef (0 :: Int)         let service req = do-                            if req == request then return () else throwIO (CachingTestException (-1))-                            callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1))-                            if callCount > 1 then throwIO (CachingTestException callCount) else return result+              if req == request then pure () else throwIO (CachingTestException (-1))+              callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1))+              if callCount > 1 then throwIO (CachingTestException callCount) else pure result         cache <- newIORef M.empty-        let lookupWith rs = fmap (M.filterWithKey (\k -> \_ -> S.member k rs)) $ readIORef cache+        let lookupWith rs = M.filterWithKey (\k _ -> S.member k rs) <$> readIORef cache         let insertWith resp = atomicModifyIORef' cache (\c -> (M.union resp c, ()))         let cachedService = cacheWithMulti lookupWith insertWith service-        (cachedService request) `shouldReturn` result-        (cachedService request) `shouldReturn` result-    it "Merges the cached values as appropriate from subsequent requests" $ do+        cachedService request `shouldReturn` result+        cachedService request `shouldReturn` result+    it "Merges the cached values as appropriate from subsequent requests" $       property $ \(first :: TestResp, second :: TestResp, both :: TestResp) -> do         let uniqueFirst = M.difference first (M.union second both)         let uniqueSecond = M.difference second (M.union first both)@@ -64,15 +62,15 @@         let secondRequest = S.fromList $ M.keys secondResult         ref <- newIORef (0 :: Int)         let service req = do-                            callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1))-                            if callCount > 2 then throwIO (CachingTestException callCount) else return $ M.filterWithKey (\k -> \_ -> S.member k req) uniqueAll+              callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1))+              if callCount > 2+                then throwIO (CachingTestException callCount)+                else pure $ M.filterWithKey (\k _ -> S.member k req) uniqueAll         cache <- newIORef M.empty-        let lookupWith rs = fmap (M.filterWithKey (\k -> \_ -> S.member k rs)) $ readIORef cache+        let lookupWith rs = M.filterWithKey (\k _ -> S.member k rs) <$> readIORef cache         let insertWith resp = atomicModifyIORef' cache (\c -> (M.union resp c, ()))         let cachedService = cacheWithMulti lookupWith insertWith service-        (cachedService firstRequest) `shouldReturn` firstResult-        (cachedService secondRequest) `shouldReturn` secondResult-        (cachedService firstRequest) `shouldReturn` firstResult-        (cachedService secondRequest) `shouldReturn` secondResult--+        cachedService firstRequest `shouldReturn` firstResult+        cachedService secondRequest `shouldReturn` secondResult+        cachedService firstRequest `shouldReturn` firstResult+        cachedService secondRequest `shouldReturn` secondResult

@@ -1,47 +1,58 @@-{-# LANGUAGE DeriveDataTypeable  #-}-{-# LANGUAGE OverloadedStrings   #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.CircuitBreakerSpec where+module Glue.CircuitBreakerSpec (spec) where -import           Control.Exception.Base   hiding (throw, throwIO, try)-import           Control.Exception.Lifted import           Data.IORef-import           Data.Typeable import           Glue.CircuitBreaker import           Test.Hspec import           Test.QuickCheck+import           UnliftIO.Exception  (Exception, throwIO, try) -data CircuitBreakerTestException = CircuitBreakerTestException deriving (Eq, Show, Typeable)+data CircuitBreakerTestException = CircuitBreakerTestException+  deriving (Eq, Show)+ instance Exception CircuitBreakerTestException  requests :: [Int]-requests = [1..10]+requests = [1 .. 10]  spec :: Spec spec = do   describe "circuitBreaker" $ do-    it "Multiple failures prevent subsequent calls" $ do+    it "Multiple failures prevent subsequent calls" $       property $ \failureMax -> do-        let positiveFailureMax      = (abs failureMax) `mod` 5-        let 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 (\req -> try $ try $ circuitBreakerService req) requests :: IO [Either CircuitBreakerException (Either CircuitBreakerTestException Int)]-        let expectedResults         = (replicate (positiveFailureMax + 1) (Right $ Left $ CircuitBreakerTestException)) ++ (replicate (10 - positiveFailureMax - 1) (Left $ CircuitBreakerException "Circuit breaker open."))+        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-        (isCircuitBreakerOpen s) `shouldReturn` True-        (isCircuitBreakerClosed s) `shouldReturn` False-        (readIORef ref) `shouldReturn` (positiveFailureMax + 1)-    it "Successful calls pass straight through" $ do+        isCircuitBreakerOpen s `shouldReturn` True+        isCircuitBreakerClosed s `shouldReturn` False+        readIORef ref `shouldReturn` (positiveFailureMax + 1)++    it "Successful calls pass straight through" $       property $ \(failureMax :: Int, requestIDs :: [Int]) -> do-        let positiveFailureMax      = abs failureMax-        let options                 = defaultCircuitBreakerOptions { maxBreakerFailures = positiveFailureMax }-        let service req             = return $ req + 1-        (s, circuitBreakerService)  <- circuitBreaker options service-        results                     <- traverse (\req -> circuitBreakerService req) requestIDs-        let expectedResults         = fmap (+ 1) requestIDs+        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-        (isCircuitBreakerClosed s) `shouldReturn` True+        isCircuitBreakerOpen s `shouldReturn` False+        isCircuitBreakerClosed s `shouldReturn` True++    it "combineCircuitBreakerStates reports open if any breaker is open" $ do+      let opts = defaultCircuitBreakerOptions { maxBreakerFailures = 0 }+      (s1, svc1) <- circuitBreaker opts (\(_ :: Int) -> throwIO CircuitBreakerTestException :: IO Int)+      (s2, _)    <- circuitBreaker opts (\(x :: Int) -> pure (x + 1) :: IO Int)+      let combined = combineCircuitBreakerStates [s1, s2]+      isCircuitBreakerOpen combined `shouldReturn` False+      isCircuitBreakerClosed combined `shouldReturn` True+      _ <- try (svc1 1) :: IO (Either CircuitBreakerTestException Int)+      isCircuitBreakerOpen combined `shouldReturn` True+      isCircuitBreakerClosed combined `shouldReturn` False

@@ -0,0 +1,65 @@+{-# LANGUAGE ScopedTypeVariables #-}++module Glue.ConcurrencyLimitSpec (spec) where++import           Control.Concurrent       (threadDelay)+import           Control.Concurrent.Async (async, mapConcurrently, wait)+import           Data.IORef+import           Glue.ConcurrencyLimit+import           Test.Hspec+import           UnliftIO.Exception       (Exception, throwIO, try)++data ConcurrencyTestException = ConcurrencyTestException+  deriving (Eq, Show)++instance Exception ConcurrencyTestException++spec :: Spec+spec = do+  describe "limitConcurrency" $ do+    it "Never exceeds maxInFlight concurrent executions" $ do+      currentInFlight <- newIORef (0 :: Int)+      peakInFlight    <- newIORef (0 :: Int)+      let maxAllowed = 3+          service (req :: Int) = do+            cur <- atomicModifyIORef' currentInFlight (\c -> (c + 1, c + 1))+            atomicModifyIORef' peakInFlight (\p -> (max p cur, ()))+            threadDelay (20 * 1000)+            atomicModifyIORef' currentInFlight (\c -> (c - 1, ()))+            pure (req * 2)+      limited <- limitConcurrency (defaultConcurrencyLimitOptions { maxInFlight = maxAllowed }) service+      results <- mapConcurrently limited [1 .. 12]+      results `shouldBe` fmap (* 2) [1 .. 12]+      peak <- readIORef peakInFlight+      peak `shouldSatisfy` (<= maxAllowed)+      peak `shouldSatisfy` (> 1)++    it "Rejects requests when maxInFlight and maxQueued are full" $ do+      let opts = ConcurrencyLimitOptions+            { maxInFlight                 = 1+            , maxQueued                   = Just 1+            , concurrencyLimitDescription = "Busy"+            }+          service (req :: Int) = do+            threadDelay (60 * 1000)+            pure req+      limited <- limitConcurrency opts service+      a1 <- async (try (limited 1) :: IO (Either ConcurrencyLimitExceededException Int))+      threadDelay (10 * 1000)+      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++    it "Releases slot when underlying service throws an exception" $ do+      let opts = defaultConcurrencyLimitOptions { maxInFlight = 1, maxQueued = Just 0 }+          failingOrOk (req :: Int) =+            if req == 0 then throwIO ConcurrencyTestException else pure (req + 10)+      limited <- limitConcurrency opts failingOrOk+      limited 0 `shouldThrow` (== ConcurrencyTestException)+      limited 5 `shouldReturn` 15

@@ -1,47 +1,46 @@-{-# LANGUAGE DeriveDataTypeable  #-}-{-# LANGUAGE OverloadedStrings   #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.DogpileProtectionSpec where+module Glue.DogpileProtectionSpec (spec) where -import           Control.Concurrent-import           Control.Concurrent.Async-import           Control.Exception.Base   hiding (throw, throwIO)-import           Control.Exception.Lifted+import           Control.Concurrent       (threadDelay)+import           Control.Concurrent.Async (async, wait) import           Data.IORef-import           Data.Traversable-import           Data.Typeable import           Glue.DogpileProtection-import           Prelude                  hiding (sequence) import           Test.Hspec+import           UnliftIO.Exception       (Exception, throwIO) -data DogpileProtectionTestException = DogpileProtectionTestException deriving (Eq, Show, Typeable)+data DogpileProtectionTestException = DogpileProtectionTestException+  deriving (Eq, Show)+ instance Exception DogpileProtectionTestException  requests :: [Int]-requests = take 10 $ repeat 1+requests = replicate 10 1  delayTime :: Int-delayTime = 1000 * 1000+delayTime = 50 * 1000  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)-      let service request = atomicModifyIORef' counter (\n -> (n + 1, ())) >> threadDelay delayTime >> return (request * 2)+      let service request = do+            atomicModifyIORef' counter (\n -> (n + 1, ()))+            threadDelay delayTime+            pure (request * 2)       protectedService <- dogpileProtect service       asyncResults <- traverse (async . protectedService) requests-      let results = traverse wait asyncResults-      results `shouldReturn` (fmap (*2) requests)-      (readIORef counter) `shouldReturn` 1+      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)-      let service _ = atomicModifyIORef' counter (\n -> (n + 1, ())) >> threadDelay delayTime >> throwIO DogpileProtectionTestException :: IO Int+      let service _ = do+            atomicModifyIORef' counter (\n -> (n + 1, ()))+            threadDelay delayTime+            throwIO DogpileProtectionTestException :: IO Int       protectedService <- dogpileProtect service       asyncResults <- traverse (async . protectedService) requests-      let results = traverse wait asyncResults-      results `shouldThrow` (== DogpileProtectionTestException)-      (readIORef counter) `shouldReturn` 1--+      traverse wait asyncResults `shouldThrow` (== DogpileProtectionTestException)+      readIORef counter `shouldReturn` 1

@@ -1,28 +1,26 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE FlexibleContexts   #-}-{-# LANGUAGE OverloadedStrings  #-}+{-# LANGUAGE FlexibleContexts #-} -module Glue.FailoverSpec where+module Glue.FailoverSpec (spec) where -import           Control.Exception.Base   hiding (throw, throwIO)-import           Control.Exception.Lifted-import           Data.Typeable import           Glue.Failover import           Test.Hspec import           Test.QuickCheck+import           UnliftIO.Exception (Exception, throwIO) -data FailoverTestException = FailoverTestException deriving (Eq, Show, Typeable)+data FailoverTestException = FailoverTestException+  deriving (Eq, Show)+ instance Exception FailoverTestException  spec :: Spec spec = do   describe "failover" $ do-    it "Failover handles potentially multiple errors" $ do+    it "Failover handles potentially multiple errors" $       property $ \(request, failureMax) ->-        let positiveFailureMax  = (abs failureMax) `mod` 10-            service req         = if req >= 10 then return (req + 100) else throwIO FailoverTestException-            options             = defaultFailoverOptions { transformFailoverRequest = (+1), maxFailovers = positiveFailureMax }-            failOverService     = failover options service-            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+        let positiveFailureMax = abs failureMax `mod` 10+            service req        = if req >= 10 then pure (req + 100) else throwIO FailoverTestException+            options            = defaultFailoverOptions { transformFailoverRequest = (+ 1), maxFailovers = positiveFailureMax }+            failOverService    = failover options service+            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

@@ -0,0 +1,42 @@+{-# LANGUAGE ScopedTypeVariables #-}++module Glue.HedgingSpec (spec) where++import           Control.Concurrent (threadDelay)+import           Data.IORef+import           Glue.Hedging+import           Test.Hspec+import           UnliftIO.Exception (Exception, throwIO)++data HedgingTestException = HedgingTestException+  deriving (Eq, Show)++instance Exception HedgingTestException++spec :: Spec+spec = do+  describe "hedgingService" $ do+    it "Completes with the fast hedged call when the primary call is slow" $ do+      ref <- newIORef (0 :: Int)+      let service (req :: Int) = do+            attemptNum <- atomicModifyIORef' ref (\c -> (c + 1, c + 1))+            if attemptNum == 1+              then threadDelay (500 * 1000) >> pure (req + 100)+              else pure (req * 2)+          hedged = hedgingService (HedgingOptions { hedgingCount = 2, hedgingDelayMs = 5 }) service+      hedged 21 `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+              else pure (req * 3)+          hedged = hedgingService (HedgingOptions { 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)

@@ -0,0 +1,46 @@+{-# LANGUAGE ScopedTypeVariables #-}++module Glue.LoadBalancerSpec (spec) where++import           Control.Concurrent.Async (mapConcurrently)+import           Data.IORef+import           Data.List.NonEmpty       (NonEmpty (..))+import           Glue.LoadBalancer+import           Glue.Types+import           Test.Hspec+import           Test.QuickCheck++spec :: Spec+spec = do+  describe "roundRobinLoadBalancer" $ do+    it "Throws NoServicesAvailableException when given an empty list" $ do+      lb <- roundRobinLoadBalancer ([] :: [BasicService IO Int Int])+      lb 42 `shouldThrow` (== NoServicesAvailableException)++    it "Distributes requests evenly in round-robin order" $+      property $ \(reqs :: [Int]) -> do+        ref0 <- newIORef (0 :: Int)+        ref1 <- newIORef (0 :: Int)+        ref2 <- newIORef (0 :: Int)+        let mkSvc tag ref req = do+              atomicModifyIORef' ref (\c -> (c + 1, ()))+              pure (tag, req * 2)+        lb <- roundRobinLoadBalancerNE+          (  mkSvc (0 :: Int) ref0+          :| [mkSvc 1 ref1, mkSvc 2 ref2]+          )+        results <- traverse lb reqs+        let expectedTags = zipWith (\i r -> (i `mod` 3, r * 2)) [0 ..] reqs+        results `shouldBe` expectedTags++    it "Handles concurrent requests across all backends" $ do+      ref0 <- newIORef (0 :: Int)+      ref1 <- newIORef (0 :: Int)+      let svc ref req = atomicModifyIORef' ref (\c -> (c + 1, ())) >> pure (req + 1)+      lb <- roundRobinLoadBalancer [svc ref0, svc ref1]+      results <- mapConcurrently lb ([1 .. 20] :: [Int])+      results `shouldBe` [2 .. 21]+      c0 <- readIORef ref0+      c1 <- readIORef ref1+      c0 `shouldBe` 10+      c1 `shouldBe` 10

@@ -1,18 +1,15 @@-{-# LANGUAGE DeriveDataTypeable  #-}-{-# LANGUAGE OverloadedStrings   #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.PreloadSpec where+module Glue.PreloadSpec (spec) where -import           Control.Exception.Base import qualified Data.HashMap.Strict       as M import qualified Data.HashSet              as S-import           Data.Typeable import           Glue.Preload import           Glue.Types import           Test.Hspec import           Test.QuickCheck import           Test.QuickCheck.Instances ()+import           UnliftIO.Exception        (Exception, finally, throwIO)  serviceFunctionality :: MultiGetRequest Int -> MultiGetResponse Int Int serviceFunctionality rs = M.fromList $ fmap (\r -> (r, r * 2)) $ S.toList rs@@ -20,41 +17,44 @@ failingService :: MultiGetRequest Int -> IO (MultiGetResponse Int Int) failingService _ = throwIO PreloadTestException -successfullCall :: IO Int-successfullCall = return 12+successfulCall :: IO Int+successfulCall = pure 12  failingCall :: IO Int failingCall = throwIO PreloadTestException -data PreloadTestException = PreloadTestException deriving (Eq, Show, Typeable)+data PreloadTestException = PreloadTestException+  deriving (Eq, Show)+ instance Exception PreloadTestException  spec :: Spec spec = do   describe "preloadingService" $ do-    it "Requests should work the same regardless of whether or not parts are preloaded" $ do+    it "Requests should work the same regardless of whether or not parts are preloaded" $       property $ \(preload :: S.HashSet Int, nonPreload :: S.HashSet Int) -> do-        let service = (return . serviceFunctionality) :: MultiGetService IO Int Int+        let service = (pure . serviceFunctionality) :: MultiGetService IO Int Int         (preloadedService, disable) <- preloadingService (defaultPreloadedOptions preload id) service         let expectedResults = serviceFunctionality (S.union preload nonPreload)-        actualResults <- preloadedService (S.union preload nonPreload)-        disable ()+        actualResults <- preloadedService (S.union preload nonPreload) `finally` disable ()         actualResults `shouldBe` expectedResults-    it "Exceptions should propagate" $ do++    it "Exceptions should propagate" $       property $ \(preload :: S.HashSet Int, nonPreload :: S.HashSet Int) -> do-        (preloadedService, _) <- preloadingService (defaultPreloadedOptions preload id) failingService+        (preloadedService, disable) <- preloadingService (defaultPreloadedOptions preload id) failingService         let expectedResults = serviceFunctionality (S.union preload nonPreload)-        let goodPath = preloadedService (S.union preload nonPreload) `shouldReturn` expectedResults-        let badPath = preloadedService (S.union preload nonPreload) `shouldThrow` (== PreloadTestException)-        if (S.null preload && S.null nonPreload) then goodPath else badPath+            goodPath = preloadedService (S.union preload nonPreload) `shouldReturn` expectedResults+            badPath  = preloadedService (S.union preload nonPreload) `shouldThrow` (== PreloadTestException)+        (if S.null preload && S.null nonPreload then goodPath else badPath) `finally` disable ()+   describe "preloadingCall" $ do-    it "Requests should work" $ do+    it "Requests should work" $       property $ do-        (preloadedCall, disable) <- preloadingCall id 1000 successfullCall-        actualResults <- preloadedCall-        disable ()+        (preloadedCall, disable) <- preloadingCall id 1000 successfulCall+        actualResults <- preloadedCall `finally` disable ()         actualResults `shouldBe` 12-    it "Exceptions should propagate" $ do++    it "Exceptions should propagate" $       property $ do-        (preloadedCall, _) <- preloadingCall id 1000 failingCall-        preloadedCall `shouldThrow` (== PreloadTestException)+        (preloadedCall, disable) <- preloadingCall id 1000 failingCall+        (preloadedCall `shouldThrow` (== PreloadTestException)) `finally` disable ()

@@ -1,54 +1,53 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE OverloadedStrings  #-}--module Glue.RetrySpec where+module Glue.RetrySpec (spec) where -import           Control.Concurrent-import           Control.Exception.Base   hiding (throw, throwIO, throwTo)-import           Control.Exception.Lifted hiding (throwTo)-import           Control.Monad.IO.Class+import           Control.Concurrent     (forkIO, myThreadId, threadDelay)+import           Control.Exception.Base (AsyncException (UserInterrupt),+                                         throwTo)+import           Control.Monad.IO.Class (liftIO) import           Data.IORef-import           Data.Typeable import           Glue.Retry import           Test.Hspec import           Test.QuickCheck+import           UnliftIO.Exception     (Exception, throwIO) -newtype SmallInt = SmallInt Int deriving (Eq, Show)+newtype SmallInt = SmallInt Int+  deriving (Eq, Show)  instance Arbitrary SmallInt where   arbitrary = sized $ \s -> do-                              n <- choose (0, s `min` 10)-                              return $ SmallInt n+    n <- choose (0, s `min` 10)+    pure $ SmallInt n -data RetryTestException = RetryTestException deriving (Eq, Show, Typeable)+data RetryTestException = RetryTestException+  deriving (Eq, Show)+ instance Exception RetryTestException  spec :: Spec spec = do   describe "retryingService" $ do-    it "Attempts a service call multiple times" $ do-      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 return req else throwIO RetryTestException-          let options       = defaultRetryOptions { maximumRetries = retries }-          let retryService  = retryingService options service-          let successCase   = (retryService request) `shouldReturn` (request :: Int)-          let failureCase   = (retryService request) `shouldThrow` (== RetryTestException)-          if retries >= failures then successCase else failureCase-    it "Asynchronous exceptions are rethrown" $ do-      property $ \(request, retries) ->-        do-          ref <- liftIO $ newIORef (0 :: Int)-          let service req   = do-                                atomicModifyIORef' ref (\c -> (c + 1, ()))-                                threadId <- myThreadId-                                _ <- forkIO (threadDelay 10000 >> throwTo threadId UserInterrupt)-                                threadDelay 1000000-                                return (req * 2 :: Int)-          let options       = defaultRetryOptions { maximumRetries = retries }-          let retryService  = retryingService options service-          (retryService request) `shouldThrow` (== UserInterrupt)-          (readIORef ref) `shouldReturn` 1+    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+            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)+        let service req = do+              atomicModifyIORef' ref (\c -> (c + 1, ()))+              threadId <- myThreadId+              _ <- forkIO (threadDelay 2000 >> throwTo threadId UserInterrupt)+              threadDelay 1000000+              pure (req * 2 :: Int)+            options      = defaultRetryOptions { maximumRetries = retries }+            retryService = retryingService options service+        retryService request `shouldThrow` (== UserInterrupt)+        readIORef ref `shouldReturn` 1

@@ -0,0 +1,43 @@+{-# LANGUAGE ScopedTypeVariables #-}++module Glue.ShutdownSpec (spec) where++import           Control.Concurrent       (threadDelay)+import           Control.Concurrent.Async (async, wait)+import           Glue.Shutdown+import           Test.Hspec++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)+            pure (req * 2)+      (wrapped, handle) <- gracefulShutdown (defaultShutdownOptions { shutdownGracePeriodMs = 500 }) service+      isShuttingDown handle `shouldReturn` False+      inFlight <- async (wrapped 21)+      threadDelay (5 * 1000)+      inFlightCount handle `shouldReturn` 1+      drainedCleanly <- initiateShutdown handle+      drainedCleanly `shouldBe` True+      isShuttingDown handle `shouldReturn` True+      wait inFlight `shouldReturn` 42+      wrapped 99 `shouldThrow` (== ServiceShuttingDownException "Service is shutting down.")++    it "Returns shutdownFallback for rejected calls when configured" $ do+      let opts = defaultShutdownOptions { shutdownFallback = Just (-1 :: Int), shutdownGracePeriodMs = 100 }+      (wrapped, handle) <- gracefulShutdown opts (\x -> pure (x * 2))+      wrapped 10 `shouldReturn` 20+      _ <- initiateShutdown handle+      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+      (wrapped, handle) <- gracefulShutdown opts slowService+      inFlight <- async (wrapped 1)+      threadDelay (5 * 1000)+      drainedCleanly <- initiateShutdown handle+      drainedCleanly `shouldBe` False+      wait inFlight `shouldReturn` (-999)

@@ -1,8 +1,6 @@-{-# LANGUAGE DeriveDataTypeable  #-}-{-# LANGUAGE OverloadedStrings   #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.SwitchingSpec where+module Glue.SwitchingSpec (spec) where  import           Glue.Switching import           Glue.Types@@ -13,15 +11,15 @@ spec :: Spec spec = do   describe "switchingService" $ do-    it "Requests should be handled by the service used" $ do+    it "Requests should be handled by the service used" $       property $ \(firstResult :: MultiGetResponse Int Int, secondResult :: MultiGetResponse Int Int, request :: MultiGetRequest Int) -> do-        let first _ = return firstResult-        let second _ = return secondResult+        let first _  = pure firstResult+            second _ = pure secondResult         (switchedService, serviceUpdate) <- switchingService first-        actualFirstResult <- switchedService request+        actualFirstResult   <- switchedService request         expectedFirstResult <- first request         serviceUpdate second-        actualSecondResult <- switchedService request+        actualSecondResult   <- switchedService request         expectedSecondResult <- second request         actualFirstResult `shouldBe` expectedFirstResult         actualSecondResult `shouldBe` expectedSecondResult

@@ -1,9 +1,6 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE OverloadedStrings  #-}--module Glue.TimeoutSpec where+module Glue.TimeoutSpec (spec) where -import           Control.Concurrent+import           Control.Concurrent (threadDelay) import           Glue.Timeout import           Test.Hspec import           Test.QuickCheck@@ -14,10 +11,10 @@ spec :: Spec spec = do   describe "addTimeout" $ do-    it "Services taking too long should fail with a timeout" $ do+    it "Services taking too long should fail with a timeout" $       property $ \request ->-        let service req         = (if req > 0 then return () else threadDelay (100 * 1000)) >> return (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+        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