capataz 0.1.0.1 → 0.2.0.0

raw patch · 18 files changed

+1250/−724 lines, 18 filesdep +prettyprinterdep +riodep −data-defaultdep −microlensdep −protoludedep ~basedep ~bytestringdep ~pretty-showPVP ok

version bump matches the API change (PVP)

Dependencies added: prettyprinter, rio

Dependencies removed: data-default, microlens, protolude, safe-exceptions, stm, tasty-rerun, text, unordered-containers, vector

Dependency ranges changed: base, bytestring, pretty-show, tasty, tasty-hunit, tasty-smallcheck, teardown, time, uuid

API changes (from Hackage documentation)

Files

• CHANGELOG.md +24/−0
• README.md +12/−64
• capataz.cabal +38/−46
• src/Capataz.hs +13/−0
• src/Control/Concurrent/Capataz.hs +6/−3
• src/Control/Concurrent/Capataz/Internal/Core.hs +70/−48
• src/Control/Concurrent/Capataz/Internal/Process.hs +66/−51
• src/Control/Concurrent/Capataz/Internal/Supervisor.hs +104/−76
• src/Control/Concurrent/Capataz/Internal/Types.hs +484/−150
• src/Control/Concurrent/Capataz/Internal/Types/Lens.hs +36/−36
• src/Control/Concurrent/Capataz/Internal/Util.hs +55/−52
• src/Control/Concurrent/Capataz/Internal/Worker.hs +13/−14
• src/Control/Concurrent/Capataz/Lens.hs +1/−2
• src/Control/Concurrent/Capataz/Util.hs +141/−0
• test/testsuite/Control/Concurrent/Capataz/SupervisorTest.hs +7/−7
• test/testsuite/Control/Concurrent/CapatazTest.hs +99/−99
• test/testsuite/Main.hs +2/−3
• test/testsuite/Test/Util.hs +79/−73

@@ -6,6 +6,30 @@ [1]: http://semver.org/spec/v2.0.0.html [2]: https://github.com/roman/capataz/releases +## v0.2.0.0 Bumblebee release++**BREAKING CHANGES**++* Allow the creation of Supervision Trees on Monad Transformer stacks 🎉+* Move to latest stable stackage snapshot (lts-11.10)+* Remove `protolude` in favor of `rio` (closes #9)+* Add convinience module `Capataz`+* Add `Control.Concurrent.Capataz.Util` which provide `ProcessSpec` for normal use-cases+* Add `buildLogWorkerSpec` and `buildLogWorkerOptions` which create a dedicated supervised+  thread that logs messages (sits on top of `rio` logger utilities)+* Add `terminateCapataz` and `terminateCapataz_` , these functions replace the+  re-exported `runTeardown`+* Add re-export of `TeardownResult` from the `teardown` package+* Update the `capataz-simple-example` project to use `typed-process` and `rio`+* Add export for `WorkerName` and `SupervisorName` aliases+* Add new dependency `prettyprinter`+* Add new dependency `pretty-show` to prettify exceptions+* Add `Pretty` instance for `CapatazEvent`+* Add `Display` instance for `CapatazEvent`+* Add `ProcessThreadId` newtype to better render `ThreadId` values+* Add `buildWorkerOptions1` and `workerSpec1` (closes #17)+* Move to CircleCI+ ## v0.1.0.1  * Bump bounds of `async` dependency

@@ -4,68 +4,10 @@  ## Table Of Contents -* [Raison d'etre](#raison-detre)+* [Installation](#installation) * [Documentation](#documentation) * [Development](#development) -## Raison d'être--As time progresses, I've come to love developing concurrent applications in-Haskell, its API (STM, MVars, etc.) and light threading RTS bring a lot to the-table. There is another technology that is more famous than Haskell in-regards to concurrency, and that is Erlang, more specifically its OTP library.--If you wonder why that is, you may need to look into the OTP library design,-actors systems (in general) provide an architecture that enables applications to-be tolerant to failure through the enforcement of communication via message-passing and by making use of a critical infrastructure piece called a *Supervisor*.--After trying to replicate Erlang's behavior on Haskell applications by using the-[distributed-process](https://hackage.haskell.org/package/distributed-process)-library (a clone of OTP), and after implementing several (disposable) iterations-of actor systems in Haskell, I've settled with just this library, one that-provides a simple Supervisor API.--This library is intended to be a drop-in replacement to `forkIO` invocations-throughout your codebase, the difference being, you'll need to do a bit more of-setup specifying supervision rules, and also pass along a reference of a-supervisor for every thread you fork.--### Why not [distributed-process](https://hackage.haskell.org/package/distributed-process)?--`distributed-process` is an impressive library, and brings many great utilities-if you need to develop applications that need to be distributed and reliable.-However, it is a heavyweight solution that will enforce serious changes to your-application. This library is intended to provide the reliability benefits of-`distributed-process`, without the constraints imposed by the *distributed*-part.--### Why not a complete actor system?--Actor systems are very pervasive, they impose specific design constraints on-your application which can be rather expensive. This library attempts to bring-some of the reliability benefits of actor systems without the "change all your-application to work with actors" part of the equation.--That said, this library can serve as a basis for a more prominent library that-provides an opinionated Inter-Process communication scheme. If you happen to-attempt at doing exactly that, please let me know, I would love to learn about-such initiatives.--### Why not [async](https://hackage.haskell.org/package/async)?--`async` is a fabulous library that allows Applicative composition of small-asynchronous sub-routines into bigger ones and link errors between them. Given-this, `async` fits the bill perfectly for small operations that happen-concurrently, not necessarily for long living threads that need to be restarted.-This library attempts not to replace async's forte, but rather provides other-benefits not found in `async`, like automatically restarting threads with a given-recipe under error situations.--## Documentation--Documentation can be found [here](https://romanandreg.gitbooks.io/capataz/content/)- ## Installation  [![Hackage](https://img.shields.io/hackage/v/capataz.svg)](https://img.shields.io/hackage/v/capataz.svg)@@ -88,18 +30,24 @@ - capataz ``` +## Documentation++The documentation website can be found+[here](https://romanandreg.gitbooks.io/capataz/content/). Some relevant links:++* [Purpose](https://romanandreg.gitbooks.io/capataz/content/purpose.html)+* [FAQ](https://romanandreg.gitbooks.io/capataz/content/FAQ.html)+ ## Development -[![Build Status](https://travis-ci.org/roman/Haskell-capataz.svg?branch=master)](https://travis-ci.org/roman/Haskell-capataz)-[![Github](https://img.shields.io/github/commits-since/roman/haskell-capataz/v0.1.0.0.svg)](https://img.shields.io/github/commits-since/roman/haskell-capataz/v0.1.0.0.svg)+![Build Status](https://circleci.com/gh/roman/Haskell-teardown.svg?style=svg)+[![Github](https://img.shields.io/github/commits-since/roman/haskell-capataz/v0.2.0.0.svg)](https://img.shields.io/github/commits-since/roman/haskell-capataz/v0.2.0.0.svg) [![Hackage Dependencies](https://img.shields.io/hackage-deps/v/capataz.svg)](http://packdeps.haskellers.com/feed?needle=capataz) -Follow the [developer guidelines](https://romanandreg.gitbooks.io/capataz/content/developer-guidelines.html)+Follow the [developer guidelines](https://romanandreg.gitbooks.io/capataz/content/CONTRIBUTING.html)  ## In future releases  * Replace Protolude in favor of RIO * Documentation of performance analysis-* Documentation improvements-* capataz-dashboard package that provides web-ui with Supervisor statistics * Ensure unit tests always finish on all concurrent scenarios (dejafu experiment)

@@ -1,12 +1,13 @@-name: capataz-version: 0.1.0.1 cabal-version: >=1.10-build-type: Simple+name: capataz+version: 0.2.0.0 license: MIT license-file: LICENSE copyright: © 2018 Roman Gonzalez maintainer: open-source@roman-gonzalez.info+author: Roman Gonzalez stability: alpha (experimental)+tested-with: ghc ==8.0.1 ghc ==8.0.2 ghc ==8.2.1 homepage: https://github.com/roman/Haskell-capataz#readme bug-reports: https://github.com/roman/Haskell-capataz/issues synopsis: OTP-like supervision trees in Haskell@@ -28,8 +29,7 @@     * Complete telemetry on the sub-routine lifecycle of your application (start,     error, restarts, shutdown) category: Control, Concurrency-author: Roman Gonzalez-tested-with: GHC ==8.0.1 GHC ==8.0.2 GHC ==8.2.1+build-type: Simple extra-source-files:     CHANGELOG.md     README.md@@ -40,6 +40,7 @@  library     exposed-modules:+        Capataz         Control.Concurrent.Capataz         Control.Concurrent.Capataz.Event         Control.Concurrent.Capataz.Lens@@ -49,57 +50,48 @@         Control.Concurrent.Capataz.Internal.Supervisor         Control.Concurrent.Capataz.Internal.Util         Control.Concurrent.Capataz.Internal.Worker-    build-depends:-        async >=2.1.1.1 && <2.3,-        base >=4.10.1.0 && <4.11,-        bytestring >=0.10.8.2 && <0.11,-        data-default >=0.7.1.1 && <0.8,-        microlens >=0.4.8.3 && <0.5,-        protolude >=0.2.1 && <0.3,-        safe-exceptions >=0.1.7.0 && <0.2,-        stm >=2.4.4.1 && <2.5,-        teardown >=0.3.0.0 && <0.4,-        text >=1.2.3.0 && <1.3,-        time >=1.8.0.2 && <1.9,-        unordered-containers >=0.2.8.0 && <0.3,-        uuid >=1.3.13 && <1.4,-        vector >=0.12.0.1 && <0.13-    default-language: Haskell2010     hs-source-dirs: src     other-modules:         Control.Concurrent.Capataz.Internal.Process+        Control.Concurrent.Capataz.Util         Paths_capataz-    ghc-options: -Wall -Wincomplete-uni-patterns -Wincomplete-record-updates+    default-language: Haskell2010+    ghc-options: -Wall -Wincomplete-uni-patterns+                 -Wincomplete-record-updates+    build-depends:+        async >=2.1.1.1 && <2.3,+        base ==4.*,+        bytestring >=0.10.8,+        pretty-show >=1.6.13,+        prettyprinter >=1.1,+        rio >=0.1.2.0,+        teardown >=0.5.0.0,+        time >=1.6.0,+        uuid >=1.3 -test-suite  capataz-test+test-suite capataz-test     type: exitcode-stdio-1.0     main-is: Main.hs-    build-depends:-        async >=2.1.1.1 && <2.3,-        base >=4.10.1.0 && <4.11,-        bytestring >=0.10.8.2 && <0.11,-        capataz -any,-        data-default >=0.7.1.1 && <0.8,-        microlens >=0.4.8.3 && <0.5,-        pretty-show >=1.6.16 && <1.7,-        protolude >=0.2.1 && <0.3,-        safe-exceptions >=0.1.7.0 && <0.2,-        stm >=2.4.4.1 && <2.5,-        tasty >=0.12.0.1 && <0.13,-        tasty-hunit >=0.10.0.1 && <0.11,-        tasty-rerun >=1.1.9 && <1.2,-        tasty-smallcheck >=0.8.1 && <0.9,-        teardown >=0.3.0.0 && <0.4,-        text >=1.2.3.0 && <1.3,-        time >=1.8.0.2 && <1.9,-        unordered-containers >=0.2.8.0 && <0.3,-        uuid >=1.3.13 && <1.4,-        vector >=0.12.0.1 && <0.13-    default-language: Haskell2010     hs-source-dirs: test/testsuite     other-modules:         Control.Concurrent.Capataz.SupervisorTest         Control.Concurrent.CapatazTest         Test.Util         Paths_capataz-    ghc-options: -Wall -Wincomplete-uni-patterns -Wincomplete-record-updates -threaded+    default-language: Haskell2010+    ghc-options: -Wall -Wincomplete-uni-patterns+                 -Wincomplete-record-updates -threaded+    build-depends:+        async >=2.1.1.1 && <2.3,+        base ==4.*,+        bytestring >=0.10.8,+        capataz -any,+        pretty-show >=1.7,+        prettyprinter >=1.1,+        rio >=0.1.2.0,+        tasty >=1.0.1.1,+        tasty-hunit >=0.10.0.1,+        tasty-smallcheck >=0.8.1,+        teardown >=0.5.0.0,+        time >=1.6.0,+        uuid >=1.3

@@ -0,0 +1,13 @@+-- | Convinience module that re-exports modules:+--+-- * "Control.Concurrent.Capataz"+-- * "Control.Concurrent.Capataz.Util"+--+--  @since 0.2.0.0+module Capataz+  (+    module X+  ) where++import Control.Concurrent.Capataz      as X+import Control.Concurrent.Capataz.Util as X

@@ -12,6 +12,7 @@ , Control.Concurrent.Capataz.Internal.Types.CallbackType (..)  , Control.Concurrent.Capataz.Internal.Types.WorkerId+, Control.Concurrent.Capataz.Internal.Types.WorkerName , Control.Concurrent.Capataz.Internal.Types.WorkerRestartStrategy (..) , Control.Concurrent.Capataz.Internal.Types.WorkerTerminationPolicy (..) , Control.Concurrent.Capataz.Internal.Types.WorkerOptions@@ -23,6 +24,7 @@ , Control.Concurrent.Capataz.Internal.Types.ProcessError (..)  , Control.Concurrent.Capataz.Internal.Types.SupervisorId+, Control.Concurrent.Capataz.Internal.Types.SupervisorName , Control.Concurrent.Capataz.Internal.Types.Supervisor , Control.Concurrent.Capataz.Internal.Types.SupervisorRestartStrategy (..) , Control.Concurrent.Capataz.Internal.Types.SupervisorStatus (..)@@ -61,6 +63,8 @@ , Control.Concurrent.Capataz.Internal.Core.forkWorker , Control.Concurrent.Capataz.Internal.Core.forkSupervisor , Control.Concurrent.Capataz.Internal.Core.forkCapataz+, Control.Concurrent.Capataz.Internal.Core.terminateCapataz+, Control.Concurrent.Capataz.Internal.Core.terminateCapataz_ , Control.Concurrent.Capataz.Internal.Core.terminateProcess  -- * Utility functions@@ -70,10 +74,9 @@ , Control.Concurrent.Capataz.Internal.Core.getCapatazTeardown  -- * Teardown (re-exported)-, Control.Teardown.teardown+, Control.Teardown.TeardownResult  -- * Lens (re-exported)-, (.~) , (&) , Control.Concurrent.Capataz.Lens.set )@@ -81,6 +84,6 @@  import qualified Control.Concurrent.Capataz.Internal.Core import qualified Control.Concurrent.Capataz.Internal.Types-import           Control.Concurrent.Capataz.Lens           ((&), (.~))+import           Control.Concurrent.Capataz.Lens           ((&)) import qualified Control.Concurrent.Capataz.Lens import qualified Control.Teardown

@@ -17,6 +17,8 @@ , forkSupervisor , forkCapataz , terminateProcess+, terminateCapataz+, terminateCapataz_ , joinCapatazThread , getSupervisorProcessId , getSupervisorAsync@@ -24,11 +26,10 @@ ) where -import Protolude+import RIO -import Control.Concurrent.MVar (newEmptyMVar, takeMVar)-import Control.Teardown        (Teardown, newTeardown)-import Data.Time.Clock         (getCurrentTime)+import Control.Teardown (Teardown, TeardownResult, newTeardown, runTeardown, runTeardown_)+import RIO.Time         (getCurrentTime)  import qualified Data.UUID.V4 as UUID (nextRandom) @@ -43,22 +44,26 @@ -- that contain a supervisor (e.g. Capataz record). class HasSupervisor a where   -- | Fetches a supervisor from a record internals.-  getSupervisor :: a -> Supervisor+  getSupervisor :: a m -> Supervisor m  instance HasSupervisor Capataz where   getSupervisor Capataz {capatazSupervisor} = capatazSupervisor  instance HasSupervisor Supervisor where-  getSupervisor = identity+  getSupervisor = id  -- | Creates a Capataz record, which holds both a root supervisor and a--- "Teardown" to shut down the system. The root supervisor monitors failures on--- process threads defined with "supervisorProcessSpecList" or created--- dynamically using "forkWorker" or "forkSupervisor".-forkCapataz :: Text -> (CapatazOptions -> CapatazOptions) -> IO Capataz+-- 'Teardown' to shut down the system. The root supervisor monitors failures on+-- process threads defined with 'supervisorProcessSpecList' or created+-- dynamically using 'forkWorker' or 'forkSupervisor'.+forkCapataz+  :: (MonadUnliftIO m, MonadIO m)+  => Text+  -> (CapatazOptions m -> CapatazOptions m)+  -> m (Capataz m) forkCapataz capatazName modOptionsFn = do-  capatazId    <- UUID.nextRandom-  supervisorId <- UUID.nextRandom+  capatazId    <- liftIO UUID.nextRandom+  supervisorId <- liftIO UUID.nextRandom   let     capatazOptions@CapatazOptions { notifyEvent } =       defCapatazOptions capatazName modOptionsFn@@ -85,14 +90,14 @@             }            MonitorEvent ProcessCompleted'{} ->-            panic "Capataz completed; this should never happen"+            error "Capataz completed; this should never happen"            MonitorEvent ProcessForcedRestart{} ->-            panic+            error               "Capataz was restarted from a OneForAll strategy; this should never happen"            ControlAction{} ->-            panic "Capataz received a ControlAction message; bad implementation"+            error "Capataz received a ControlAction message; bad implementation"       , notifyEvent       } @@ -100,11 +105,11 @@     parentSupervisorEnv     supervisorOptions     supervisorId-    0+    0 -- initial restart count -  capatazTeardown <- newTeardown+  capatazTeardown <- withRunInIO $ \run -> newTeardown     "capataz"-    ( do+    (run $ do       Supervisor.haltSupervisor "capataz system shutdown" supervisorEnv       eventTime <- getCurrentTime       notifyEvent CapatazTerminated {supervisorId , supervisorName , eventTime }@@ -112,25 +117,25 @@    return Capataz {capatazSupervisor , capatazTeardown } --- | Creates a green thread from an "IO ()" sub-routine. Depending in options--- defined in the "WorkerOptions" record, it will automatically restart this+-- | Creates a green thread from an @IO ()@ sub-routine. Depending in options+-- defined in the 'WorkerOptions' record, it will automatically restart this -- sub-routine in case of failures. -- -- See documentation of related functions: ----- * "buildWorkerOptionsWithDefault"--- * "buildWorkerOptions"+-- * 'buildWorkerOptionsWithDefaults'+-- * 'buildWorkerOptions' -- forkWorker-  :: HasSupervisor supervisor-  => WorkerOptions -- ^ Worker options (restart, name, callbacks, etc)-  -> supervisor    -- ^ "Supervisor" that supervises the worker-  -> IO WorkerId   -- ^ An identifier that can be used to terminate the "Worker"+  :: (MonadIO m, HasSupervisor supervisor)+  => WorkerOptions m -- ^ Worker options (restart, name, callbacks, etc)+  -> supervisor m   -- ^ 'Supervisor' that supervises the worker+  -> m WorkerId   -- ^ An identifier that can be used to terminate the 'Worker' forkWorker workerOptions sup = do   let Supervisor { supervisorNotify } = getSupervisor sup   workerIdVar <- newEmptyMVar   supervisorNotify-    ( ControlAction ForkWorker+    (ControlAction ForkWorker       { workerOptions       , returnWorkerId = putMVar workerIdVar       }@@ -138,24 +143,24 @@   takeMVar workerIdVar  -- | Creates a green thread which monitors other green threads for failures and--- restarts them using settings defined on "SupervisorOptions".+-- restarts them using settings defined on 'SupervisorOptions'. -- -- See documentation of related functions: ----- * "buildSupervisorOptionsWithDefault"--- * "buildSupervisorOptions"+-- * 'buildSupervisorOptionsWithDefault'+-- * 'buildSupervisorOptions' -- forkSupervisor-  :: HasSupervisor parentSupervisor-  => SupervisorOptions -- ^ Supervisor options-  -> parentSupervisor  -- ^ Parent supervisor instance that supervises new supervisor-  -> IO Supervisor     -- ^ A record used to dynamically create and supervise+  :: (MonadIO m, HasSupervisor parentSupervisor)+  => SupervisorOptions m -- ^ Supervisor options+  -> parentSupervisor m  -- ^ Parent supervisor instance that supervises new supervisor+  -> m (Supervisor m)     -- ^ A record used to dynamically create and supervise                        -- other processes forkSupervisor supervisorOptions parentSup = do   let Supervisor { supervisorNotify } = getSupervisor parentSup   supervisorVar <- newEmptyMVar   supervisorNotify-    ( ControlAction ForkSupervisor+    (ControlAction ForkSupervisor       { supervisorOptions       , returnSupervisor  = putMVar supervisorVar       }@@ -165,16 +170,21 @@ -- | Stops the execution of a green thread being supervised by the given -- supervisor. ----- NOTE: If "ProcessId" maps to a worker that is configured with a "Permanent"--- worker restart strategy, the worker green thread __will be restarted again__.+-- __IMPORTANT__ If 'ProcessId' maps to a worker that is configured with a+-- 'Permanent' worker restart strategy, the worker green thread __will be+-- restarted again__. -- terminateProcess-  :: HasSupervisor supervisor => Text -> ProcessId -> supervisor -> IO Bool+  :: (MonadIO m, HasSupervisor supervisor)+  => Text+  -> ProcessId+  -> supervisor m+  -> m Bool terminateProcess processTerminationReason processId supervisor = do   let Supervisor { supervisorNotify } = getSupervisor supervisor   result <- newEmptyMVar   supervisorNotify-    ( ControlAction TerminateProcess+    (ControlAction TerminateProcess       { processId       , processTerminationReason       , notifyProcessTermination = putMVar result@@ -184,22 +194,34 @@  -- | Joins the thread of the root supervisor of the given capataz system to the -- current thread.-joinCapatazThread :: Capataz -> IO ()+joinCapatazThread :: MonadIO m => Capataz m -> m () joinCapatazThread Capataz { capatazSupervisor } =   let Supervisor { supervisorAsync } = capatazSupervisor   in  wait supervisorAsync --- | Gets "Teardown" record of this capataz system.-getCapatazTeardown :: Capataz -> Teardown+-- | Terminates a 'Capataz' system (all supervised threads) and returns a 'TeardownResult'+--+-- @since 0.2.0.0+terminateCapataz :: MonadIO m => Capataz m -> m TeardownResult+terminateCapataz = liftIO . runTeardown++-- | Terminates a 'Capataz' system (all supervised threads)+--+-- @since 0.2.0.0+terminateCapataz_ :: MonadIO m => Capataz m -> m ()+terminateCapataz_ = liftIO . runTeardown_++-- | Gets 'Teardown' record of this capataz system.+getCapatazTeardown :: Capataz m -> Teardown getCapatazTeardown Capataz { capatazTeardown } = capatazTeardown --- | Gets the "Async" of a Supervisor thread.+-- | Gets the 'Async' of a Supervisor thread. ----- NOTE: There is no way to get the "Async" value of the root supervisor; this--- is to avoid error scenarios.-getSupervisorAsync :: Supervisor -> Async ()+-- NOTE: There is no way to get the 'Async' value of the root supervisor; this+-- is done on-purpose to avoid error scenarios.+getSupervisorAsync :: Supervisor m -> Async () getSupervisorAsync Supervisor { supervisorAsync } = supervisorAsync --- | Gets the process identifier of a Supervisor; normally used for termination.-getSupervisorProcessId :: Supervisor -> ProcessId+-- | Gets the process identifier of a 'Supervisor'; normally used for termination.+getSupervisorProcessId :: Supervisor m -> ProcessId getSupervisorProcessId Supervisor { supervisorId } = supervisorId

@@ -6,43 +6,43 @@ -- Worker process types. module Control.Concurrent.Capataz.Internal.Process where -import Protolude+import RIO  import Control.Concurrent.Capataz.Internal.Types import Control.Concurrent.Capataz.Internal.Util     (readProcessMap, sortProcessesByTerminationOrder)-import Data.Time.Clock                           (UTCTime, getCurrentTime)+import RIO.Time                                  (UTCTime, getCurrentTime)  -- | Gets "Async" from a given Process.-getProcessAsync :: Process -> Async ()+getProcessAsync :: Process m -> Async () getProcessAsync process = case process of   WorkerProcess     Worker { workerAsync }         -> workerAsync   SupervisorProcess Supervisor { supervisorAsync } -> supervisorAsync  -- | Gets "ThreadId" from a given Process.-getProcessThreadId :: Process -> ThreadId-getProcessThreadId = asyncThreadId . getProcessAsync+getProcessThreadId :: Process m -> ProcessThreadId+getProcessThreadId = PTID . asyncThreadId . getProcessAsync  -- | Gets "ProcessId" from a given Process.-getProcessId :: Process -> ProcessId+getProcessId :: Process m -> ProcessId getProcessId process = case process of   WorkerProcess     Worker { workerId }         -> workerId   SupervisorProcess Supervisor { supervisorId } -> supervisorId  -- | Gets "ProcessName" from a given "ProcessSpec".-getProcessName :: ProcessSpec -> ProcessName+getProcessName :: ProcessSpec m -> ProcessName getProcessName procSpec = case procSpec of   WorkerSpec     WorkerOptions { workerName }         -> workerName   SupervisorSpec SupervisorOptions { supervisorName } -> supervisorName  -- | Gets "ProcessType" from a given "ProcessSpec".-getProcessType :: ProcessSpec -> ProcessType+getProcessType :: ProcessSpec m -> ProcessType getProcessType processSpec = case processSpec of   WorkerSpec{}     -> WorkerType   SupervisorSpec{} -> SupervisorType  -- | Gets "ProcessSpec" of a given "Process".-getProcessSpec :: Process -> ProcessSpec+getProcessSpec :: Process m -> ProcessSpec m getProcessSpec process = case process of   WorkerProcess Worker { workerOptions } -> WorkerSpec workerOptions   SupervisorProcess Supervisor { supervisorOptions } ->@@ -50,7 +50,8 @@  -- | Utility function to send notifications to the capataz system callback when -- a Process fails.-notifyProcessFailed :: SupervisorEnv -> Process -> SomeException -> IO ()+notifyProcessFailed+  :: MonadIO m => SupervisorEnv m -> Process m -> SomeException -> m () notifyProcessFailed SupervisorEnv { supervisorId, supervisorName, notifyEvent } process processError   = do     eventTime <- getCurrentTime@@ -67,7 +68,8 @@  -- | Utility function to send notifications to the capataz system callback when -- a Process is terminated.-notifyProcessTerminated :: SupervisorEnv -> Process -> Text -> IO ()+notifyProcessTerminated+  :: MonadIO m => SupervisorEnv m -> Process m -> Text -> m () notifyProcessTerminated SupervisorEnv { supervisorId, supervisorName, notifyEvent } process terminationReason   = do     eventTime <- getCurrentTime@@ -85,7 +87,11 @@ -- | Utility function to send notifications to the capataz system callback when -- a Process is started or restarted. notifyProcessStarted-  :: Maybe (ProcessId, RestartCount) -> ParentSupervisorEnv -> Process -> IO ()+  :: MonadIO m+  => Maybe (ProcessId, RestartCount)+  -> ParentSupervisorEnv m+  -> Process m+  -> m () notifyProcessStarted mRestartInfo ParentSupervisorEnv { supervisorId, supervisorName, notifyEvent } process   = do     eventTime <- getCurrentTime@@ -112,7 +118,7 @@  -- | Utility function to send notifications when a Process sub-routine completes -- without errors.-notifyProcessCompleted :: SupervisorEnv -> Process -> UTCTime -> IO ()+notifyProcessCompleted :: SupervisorEnv m -> Process m -> UTCTime -> m () notifyProcessCompleted SupervisorEnv { supervisorId, supervisorName, notifyEvent } process eventTime   = notifyEvent ProcessCompleted     { supervisorId@@ -126,39 +132,41 @@   -- | Utility function to execute a Process onCompletion sub-routine.-callProcessOnCompletion :: ProcessSpec -> IO ()+callProcessOnCompletion :: Monad m => ProcessSpec m -> m () callProcessOnCompletion procSpec = case procSpec of   WorkerSpec WorkerOptions { workerOnCompletion } -> workerOnCompletion   _                                               -> return ()  -- | Utility function to execute a Process onFailure sub-routine.-callProcessOnFailure :: ProcessSpec -> SomeException -> IO ()+callProcessOnFailure :: ProcessSpec m -> SomeException -> m () callProcessOnFailure procSpec err = case procSpec of   WorkerSpec WorkerOptions { workerOnFailure } -> workerOnFailure err   SupervisorSpec SupervisorOptions { supervisorOnFailure } ->     supervisorOnFailure err  -- | Utility function to execute a Process onTermination sub-routine.-callProcessOnTermination :: ProcessSpec -> IO ()+callProcessOnTermination :: Monad m => ProcessSpec m -> m () callProcessOnTermination procSpec = case procSpec of   WorkerSpec WorkerOptions { workerOnTermination } -> workerOnTermination   _                                                -> return ()  -- | Handles errors produced - or thrown to - a process thread. handleProcessException-  :: (IO () -> IO a)-  -> ParentSupervisorEnv-  -> ProcessSpec+  :: (MonadUnliftIO m)+  => (m () -> m a)+  -> ParentSupervisorEnv m+  -> ProcessSpec m   -> ProcessId   -> RestartCount   -> SomeException-  -> IO MonitorEvent+  -> m MonitorEvent handleProcessException unmask ParentSupervisorEnv { supervisorId, supervisorName, notifyEvent } procSpec processId restartCount err   = do     let processName = getProcessName procSpec-    processThreadId  <- myThreadId+    processThreadId  <- PTID <$> myThreadId     monitorEventTime <- getCurrentTime-    case fromException err of++    case fromAnyException err of       Just RestartProcessException -> return ProcessForcedRestart         { processId         , processName@@ -166,7 +174,7 @@         }        Just TerminateProcessException { processTerminationReason } -> do-        eErrResult <- try $ unmask $ callProcessOnTermination procSpec+        eErrResult <- unsafeTry $ unmask $ callProcessOnTermination procSpec          notifyEvent ProcessCallbackExecuted           { supervisorId@@ -180,6 +188,7 @@           , eventTime            = monitorEventTime           } +         case eErrResult of           Left processCallbackError -> return ProcessFailed'             { processId@@ -210,9 +219,8 @@           , processRestartCount      = restartCount           } -      -- This exception was an error from the given sub-routine       _ -> do-        eErrResult <- try $ unmask $ callProcessOnFailure procSpec err+        eErrResult <- unsafeTry $ unmask $ callProcessOnFailure procSpec err          notifyEvent ProcessCallbackExecuted           { supervisorId@@ -249,18 +257,19 @@  -- | Handles completion of a Process sub-routine. handleProcessCompletion-  :: (IO () -> IO a)-  -> ParentSupervisorEnv-  -> ProcessSpec+  :: (MonadUnliftIO m)+  => (m () -> m a)+  -> ParentSupervisorEnv m+  -> ProcessSpec m   -> ProcessId   -> RestartCount-  -> IO MonitorEvent+  -> m MonitorEvent handleProcessCompletion unmask ParentSupervisorEnv { supervisorId, supervisorName, notifyEvent } procSpec processId restartCount   = do     let processName = getProcessName procSpec-    processThreadId  <- myThreadId+    processThreadId  <- PTID <$> myThreadId     monitorEventTime <- getCurrentTime-    eCompResult      <- try $ unmask $ callProcessOnCompletion procSpec+    eCompResult      <- unsafeTry $ unmask $ callProcessOnCompletion procSpec      notifyEvent ProcessCallbackExecuted       { supervisorId@@ -296,10 +305,11 @@ -- function gets executed on the supervisor's thread. -- terminateProcess-  :: Text -- ^ Description that indicates _why_ there is a termination-  -> SupervisorEnv-  -> Process-  -> IO ()+  :: (MonadUnliftIO m)+  => Text -- ^ Description that indicates _why_ there is a termination+  -> SupervisorEnv m+  -> Process m+  -> m () terminateProcess processTerminationReason env process = do   case process of     WorkerProcess worker -> terminateWorker processTerminationReason worker@@ -310,7 +320,7 @@  -- | Internal utility function that manages execution of a termination policy -- for a Worker.-terminateWorker :: Text -> Worker -> IO ()+terminateWorker :: (MonadUnliftIO m) => Text -> Worker m -> m () terminateWorker processTerminationReason Worker { workerId, workerOptions, workerAsync }   = do     let processId = workerId@@ -327,24 +337,29 @@           , processTerminationReason           } -      TimeoutMillis millis -> race_-        ( do-          threadDelay (millis * 1000)-          cancelWith+      TimeoutMillis millis -> do+        -- NOTE: Given Teardown executes teardown operations in an uninterruptible mask+        -- we need to run asyncWithUnmask to come back to a unmasked state, sadly, race+        -- doesn't use asyncWithUnmask_, so we need to use it here+        result <- asyncWithUnmask $ \unmask -> unmask $ race_+          (do+            threadDelay (millis * 1000)+            cancelWith+              workerAsync+              BrutallyTerminateProcessException+                { processId+                , processTerminationReason+                }+          )+          (cancelWith             workerAsync-            BrutallyTerminateProcessException-              { processId-              , processTerminationReason-              }-        )-        ( cancelWith-          workerAsync-          TerminateProcessException {processId , processTerminationReason }-        )+            TerminateProcessException {processId , processTerminationReason }+          )+        wait result  -- | Internal utility function that manages execution of a termination policy -- for a Supervisor.-terminateSupervisor :: Text -> Supervisor -> IO ()+terminateSupervisor :: MonadIO m => Text -> Supervisor m -> m () terminateSupervisor processTerminationReason Supervisor { supervisorId = processId, supervisorAsync }   = cancelWith     supervisorAsync@@ -352,7 +367,7 @@  -- | Internal sub-routine that terminates workers of a supervisor, used when a -- supervisor instance is terminated.-terminateProcessMap :: Text -> SupervisorEnv -> IO ()+terminateProcessMap :: (MonadUnliftIO m) => Text -> SupervisorEnv m -> m () terminateProcessMap terminationReason env@SupervisorEnv { supervisorId, supervisorName, supervisorProcessTerminationOrder, notifyEvent }   = do     eventTime  <- getCurrentTime

@@ -7,16 +7,13 @@ {-# LANGUAGE ScopedTypeVariables   #-} module Control.Concurrent.Capataz.Internal.Supervisor where -import Control.Concurrent.Async      (asyncWithUnmask)-import Control.Concurrent.STM.TQueue (newTQueueIO, readTQueue, writeTQueue)-import Control.Concurrent.STM.TVar   (newTVarIO)-import Data.IORef                    (newIORef)-import Data.Time.Clock               (NominalDiffTime, UTCTime, diffUTCTime, getCurrentTime)-import Protolude+import           RIO+import qualified RIO.HashMap as HashMap -import qualified Data.HashMap.Strict as HashMap-import qualified Data.UUID.V4        as UUID+import RIO.Time (NominalDiffTime, UTCTime, diffUTCTime, getCurrentTime) +import qualified Data.UUID.V4 as UUID+ import Control.Concurrent.Capataz.Internal.Types  import qualified Control.Concurrent.Capataz.Internal.Process as Process@@ -27,14 +24,15 @@ -- from the public @forkSupervisor@ function which sends a message to the -- supervisor loop. forkSupervisor-  :: ParentSupervisorEnv-  -> SupervisorOptions+  :: (MonadUnliftIO m)+  => ParentSupervisorEnv m+  -> SupervisorOptions m   -> Maybe (ProcessId, RestartCount)-  -> IO Supervisor+  -> m (Supervisor m) forkSupervisor parentEnv supervisorOptions mRestartInfo = do   (supervisorId, restartCount) <- case mRestartInfo of     Just (supervisorId, restartCount) -> pure (supervisorId, restartCount)-    Nothing                           -> (,) <$> UUID.nextRandom <*> pure 0+    Nothing                           -> (,) <$> liftIO UUID.nextRandom <*> pure 0     supervisor <- supervisorMain parentEnv@@ -50,12 +48,13 @@ -- | Utility function that builds an utility record which is used on all -- internal APIs of the supervision logic. buildSupervisorEnv-  :: (CapatazEvent -> IO ())-  -> (SupervisorMessage -> IO ())-  -> STM SupervisorMessage+  :: MonadIO m+  => (CapatazEvent -> m ())+  -> (SupervisorMessage m -> m ())+  -> STM (SupervisorMessage m)   -> SupervisorId-  -> SupervisorOptions-  -> IO SupervisorEnv+  -> SupervisorOptions m+  -> m (SupervisorEnv m) buildSupervisorEnv notifyEvent supervisorNotify supervisorGetNotification supervisorId supervisorOptions@SupervisorOptions {..}   = do     supervisorProcessMap <- newIORef mempty@@ -63,7 +62,8 @@     return SupervisorEnv {..}  -- | Handles an event produced by one of the processes this supervisor monitors.-handleMonitorEvent :: SupervisorEnv -> MonitorEvent -> IO Bool+handleMonitorEvent+  :: (MonadUnliftIO m) => SupervisorEnv m -> MonitorEvent -> m Bool handleMonitorEvent env monitorEv = do   case monitorEv of     ProcessForcedRestart{} ->@@ -87,7 +87,8 @@   return True  -- | Handles an action triggered by the public Capataz API.-handleControlAction :: SupervisorEnv -> ControlAction -> IO Bool+handleControlAction+  :: (MonadUnliftIO m) => SupervisorEnv m -> ControlAction m -> m Bool handleControlAction env controlAction = case controlAction of   ForkWorker { workerOptions, returnWorkerId } -> do     worker@Worker { workerId } <- Worker.forkWorker@@ -120,7 +121,7 @@  -- | Executes the shutdown operation of a Supervisor, including the termination -- of its supervised processes.-haltSupervisor :: Text -> SupervisorEnv -> IO ()+haltSupervisor :: (MonadUnliftIO m) => Text -> SupervisorEnv m -> m () haltSupervisor reason env = do   Util.writeSupervisorStatus  env    Halting   Process.terminateProcessMap reason env@@ -130,40 +131,33 @@  -- | Handles all messages that a Supervisor can receive from its monitored -- processes or from the public API.-handleSupervisorMessage :: SupervisorEnv -> SupervisorMessage -> IO Bool+handleSupervisorMessage+  :: (MonadUnliftIO m) => SupervisorEnv m -> SupervisorMessage m -> m Bool handleSupervisorMessage env message = case message of   ControlAction controlAction -> handleControlAction env controlAction   MonitorEvent  monitorEvent  -> handleMonitorEvent env monitorEvent  -- | This sub-routine executes the main thread loop of a "Supervisor" instance. supervisorLoop-  :: (forall b . IO b -> IO b)-  -> ParentSupervisorEnv-  -> SupervisorEnv+  :: (MonadUnliftIO m)+  => (forall b . m b -> m b)+  -> ParentSupervisorEnv m+  -> SupervisorEnv m   -> RestartCount-  -> IO ()+  -> m () supervisorLoop unmask parentEnv@ParentSupervisorEnv { supervisorId, supervisorName, supervisorNotify = notifyParentSupervisor } env@SupervisorEnv { supervisorId = processId, supervisorName = processName, supervisorOptions, supervisorStatusVar, supervisorGetNotification, notifyEvent } restartCount   = do-    processThreadId <- myThreadId+    processThreadId <- PTID <$> myThreadId     loopResult      <--      unmask-      $   try+      unsafeTry+      $   unmask       $   atomically       $   (,)       <$> Util.readSupervisorStatusSTM supervisorStatusVar       <*> supervisorGetNotification      case loopResult of-      Left supervisorError -> do-        haltSupervisor (show supervisorError) env-        result <- Process.handleProcessException-          unmask-          parentEnv-          (SupervisorSpec supervisorOptions)-          processId-          restartCount-          supervisorError-        notifyParentSupervisor (MonitorEvent result)+      Left  supervisorError   -> handleSupervisorLoopError supervisorError        Right (status, message) -> case status of         Initializing -> do@@ -176,18 +170,11 @@           supervisorLoop unmask parentEnv env restartCount          Running -> do-          eContinueLoop <- try $ unmask $ handleSupervisorMessage env message+          eContinueLoop <- unsafeTry $ unmask $ handleSupervisorMessage+            env+            message           case eContinueLoop of-            Left supervisorError -> do-              haltSupervisor (show supervisorError) env-              result <- Process.handleProcessException-                unmask-                parentEnv-                (SupervisorSpec supervisorOptions)-                processId-                restartCount-                supervisorError-              notifyParentSupervisor (MonitorEvent result)+            Left supervisorError -> handleSupervisorLoopError supervisorError              Right continueLoop               | continueLoop -> supervisorLoop unmask parentEnv env restartCount@@ -211,15 +198,27 @@         Halted ->           -- Discard messages when halted           return ()+ where+  handleSupervisorLoopError err = do+    haltSupervisor (tshow err) env+    result <- Process.handleProcessException+      unmask+      parentEnv+      (SupervisorSpec supervisorOptions)+      processId+      restartCount+      err+    notifyParentSupervisor (MonitorEvent result)  -- | This sub-routine starts a Supervisor thread and initializes its -- processList. supervisorMain-  :: ParentSupervisorEnv-  -> SupervisorOptions+  :: (MonadUnliftIO m)+  => ParentSupervisorEnv m+  -> SupervisorOptions m   -> SupervisorId   -> RestartCount-  -> IO Supervisor+  -> m (Supervisor m) supervisorMain parentEnv@ParentSupervisorEnv { notifyEvent } supervisorOptions@SupervisorOptions { supervisorName, supervisorProcessSpecList } supervisorId restartCount   = do     supervisorCreationTime <- getCurrentTime@@ -241,7 +240,7 @@      forM_       supervisorProcessSpecList-      ( \processSpec -> case processSpec of+      (\processSpec -> case processSpec of         WorkerSpec workerOptions -> do           worker <- Worker.forkWorker             (Util.toParentSupervisorEnv supervisorEnv)@@ -273,7 +272,7 @@  -- | Tracks difference between two timestamps so that we keep track of a -- Supervisor error intensity.-calcDiffSeconds :: UTCTime -> IO NominalDiffTime+calcDiffSeconds :: MonadIO m => UTCTime -> m NominalDiffTime calcDiffSeconds creationTime = do   currentTime <- getCurrentTime   return $ diffUTCTime currentTime creationTime@@ -282,7 +281,7 @@ -- error intensity has been breached, see "ProcessRestartAction" for possible -- outcomes. calcRestartAction-  :: SupervisorEnv -> Int -> NominalDiffTime -> ProcessRestartAction+  :: SupervisorEnv m -> Int -> NominalDiffTime -> ProcessRestartAction calcRestartAction SupervisorEnv { supervisorIntensity, supervisorPeriodSeconds } restartCount diffSeconds   = case () of     _@@ -298,7 +297,12 @@  -- | Sub-routine responsible of executing a "SupervisorRestartStrategy". execCapatazRestartStrategy-  :: SupervisorEnv -> ProcessId -> ProcessSpec -> Int -> IO ()+  :: (MonadUnliftIO m)+  => SupervisorEnv m+  -> ProcessId+  -> ProcessSpec m+  -> Int+  -> m () execCapatazRestartStrategy supervisorEnv@SupervisorEnv { supervisorRestartStrategy } processId processSpec processRestartCount   = case supervisorRestartStrategy of     AllForOne -> do@@ -330,13 +334,14 @@ -- | Executes a restart action returned from the invokation of -- "calcRestartAction". execRestartAction-  :: SupervisorEnv+  :: (MonadUnliftIO m)+  => SupervisorEnv m   -> ProcessId-  -> ProcessSpec+  -> ProcessSpec m   -> Text   -> UTCTime   -> Int-  -> IO ()+  -> m () execRestartAction supervisorEnv@SupervisorEnv { supervisorOnIntensityReached } processId processSpec processName processCreationTime processRestartCount   = do     restartAction <- calcRestartAction supervisorEnv processRestartCount@@ -345,7 +350,7 @@     case restartAction of       HaltSupervisor -> do         -- skip exceptions on callback-        (_ :: Either SomeException ()) <- try supervisorOnIntensityReached+        (_ :: Either SomeException ()) <- unsafeTry supervisorOnIntensityReached         throwIO SupervisorIntensityReached           { processId           , processName@@ -359,7 +364,7 @@         supervisorEnv         processId         processSpec-        (succ processRestartCount)+        (processRestartCount + 1)   --------------------------------------------------------------------------------@@ -367,7 +372,12 @@ -- | Restarts _all_ processes that are supervised by Supervisor, invoked when -- one worker green thread fails and causes sibling process threads to get -- restarted as well (e.g. "AllForOne" supervisor restart strategy).-restartProcessList :: SupervisorEnv -> WorkerId -> RestartCount -> IO [Process]+restartProcessList+  :: (MonadUnliftIO m)+  => SupervisorEnv m+  -> WorkerId+  -> RestartCount+  -> m [Process m] restartProcessList supervisorEnv@SupervisorEnv { supervisorProcessTerminationOrder } failingProcessId restartCount   = do     processMap <- Util.readProcessMap supervisorEnv@@ -404,7 +414,7 @@  -- | Sub-routine that is used when there is a restart request sent to a Process -- caused by an "AllForOne" restart from a failing sibling process.-forceRestartProcess :: SupervisorEnv -> Process -> IO ()+forceRestartProcess :: MonadIO m => SupervisorEnv m -> Process m -> m () forceRestartProcess env process = do   Process.notifyProcessTerminated env process "forced restart"   cancelWith (Process.getProcessAsync process) RestartProcessException@@ -412,7 +422,12 @@ -- | Starts a new worker thread taking into account an existing "WorkerId" and -- keeping a "RestartCount" to manage the Supervisor error intensity. restartWorker-  :: SupervisorEnv -> WorkerOptions -> WorkerId -> RestartCount -> IO Process+  :: (MonadUnliftIO m)+  => SupervisorEnv m+  -> WorkerOptions m+  -> WorkerId+  -> RestartCount+  -> m (Process m) restartWorker supervisorEnv workerOptions workerId restartCount =   WorkerProcess <$> Worker.forkWorker     (Util.toParentSupervisorEnv supervisorEnv)@@ -423,11 +438,12 @@ -- "SupervisorId" and keeping a "RestartCount" to manage the parent Supervisor -- error intensity. restartSupervisor-  :: ParentSupervisorEnv-  -> SupervisorOptions+  :: (MonadUnliftIO m)+  => ParentSupervisorEnv m+  -> SupervisorOptions m   -> ProcessId   -> RestartCount-  -> IO Process+  -> m (Process m) restartSupervisor parentEnv supervisorOptions processId restartCount =   SupervisorProcess <$> forkSupervisor parentEnv                                        supervisorOptions@@ -438,7 +454,8 @@  -- | Executes restart strategy for when a worker finishes it execution because -- of a completion (e.g. worker sub-routine finished without any errors).-handleWorkerCompleted :: SupervisorEnv -> Worker -> IO ()+handleWorkerCompleted+  :: (MonadUnliftIO m) => SupervisorEnv m -> Worker m -> m () handleWorkerCompleted env worker = do   let Worker { workerId, workerOptions, workerCreationTime } = worker       WorkerOptions { workerName, workerRestartStrategy }    = workerOptions@@ -461,7 +478,8 @@  -- | Executes restart strategy for when a process finishes it execution because -- of a completion (e.g. worker sub-routine finished without any errors).-handleProcessCompleted :: SupervisorEnv -> ProcessId -> UTCTime -> IO ()+handleProcessCompleted+  :: (MonadUnliftIO m) => SupervisorEnv m -> ProcessId -> UTCTime -> m () handleProcessCompleted env processId completionTime = do   mProcess <- Util.fetchProcess env processId   case mProcess of@@ -472,14 +490,15 @@       case process of         WorkerProcess worker -> handleWorkerCompleted env worker         _ ->-          panic+          error             $  "ERROR: Supervisor ("             <> show (Process.getProcessId process)             <> ") should never complete"  -- | Executes restart strategy for when a worker finishes it execution because -- of a failure.-handleWorkerFailed :: SupervisorEnv -> Worker -> Int -> IO ()+handleWorkerFailed+  :: (MonadUnliftIO m) => SupervisorEnv m -> Worker m -> Int -> m () handleWorkerFailed env worker restartCount = do   let Worker { workerId, workerCreationTime, workerOptions } = worker       WorkerOptions { workerName, workerRestartStrategy }    = workerOptions@@ -494,7 +513,8 @@  -- | Executes restart strategy for when a supervisor finishes it execution because -- of a failure.-handleSupervisorFailed :: SupervisorEnv -> Supervisor -> Int -> IO ()+handleSupervisorFailed+  :: (MonadUnliftIO m) => SupervisorEnv m -> Supervisor m -> Int -> m () handleSupervisorFailed env supervisor restartCount = do   let Supervisor { supervisorId, supervisorCreationTime, supervisorOptions } =         supervisor@@ -509,7 +529,12 @@ -- | Executes restart strategy for when a process finishes it execution because -- of a failure. handleProcessFailed-  :: SupervisorEnv -> WorkerId -> SomeException -> Int -> IO ()+  :: (MonadUnliftIO m)+  => SupervisorEnv m+  -> WorkerId+  -> SomeException+  -> Int+  -> m () handleProcessFailed env processId processError restartCount = do   mProcess <- Util.fetchProcess env processId   case mProcess of@@ -524,7 +549,8 @@  -- | Executes restart strategy for when a worker finishes it execution because -- of a termination from its supervisor.-handleWorkerTerminated :: SupervisorEnv -> Worker -> Int -> IO ()+handleWorkerTerminated+  :: (MonadUnliftIO m) => SupervisorEnv m -> Worker m -> Int -> m () handleWorkerTerminated env worker restartCount = do   let Worker { workerId, workerCreationTime, workerOptions } = worker       WorkerOptions { workerName, workerRestartStrategy }    = workerOptions@@ -541,7 +567,8 @@  -- | Executes restart strategy for when a supervisor finishes it execution -- because of a termination from its parent supervisor.-handleSupervisorTerminated :: SupervisorEnv -> Supervisor -> Int -> IO ()+handleSupervisorTerminated+  :: (MonadUnliftIO m) => SupervisorEnv m -> Supervisor m -> Int -> m () handleSupervisorTerminated env supervisor restartCount = do   let Supervisor { supervisorId, supervisorCreationTime, supervisorOptions } =         supervisor@@ -555,7 +582,8 @@  -- | Executes restart strategy for when a process finishes it execution because -- of a termination from its supervisor.-handleProcessTerminated :: SupervisorEnv -> ProcessId -> Text -> Int -> IO ()+handleProcessTerminated+  :: (MonadUnliftIO m) => SupervisorEnv m -> ProcessId -> Text -> Int -> m () handleProcessTerminated env processId terminationReason restartCount = do   mProcess <- Util.fetchProcess env processId   case mProcess of

@@ -2,38 +2,57 @@ {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE NamedFieldPuns        #-} {-# LANGUAGE NoImplicitPrelude     #-}+{-# LANGUAGE OverloadedStrings     #-}  {-| This module contains all the types used across all the other modules -} module Control.Concurrent.Capataz.Internal.Types where -import Protolude+import RIO+import RIO.Time (NominalDiffTime, UTCTime) -import Control.Concurrent.STM.TVar (TVar)-import Control.Teardown            (ITeardown (..), Teardown)-import Data.Default                (Default (..))-import Data.HashMap.Strict         (HashMap)-import Data.IORef                  (IORef)-import Data.Time.Clock             (NominalDiffTime, UTCTime)-import Data.UUID                   (UUID)+import Control.Teardown (HasTeardown (..), Teardown)+import Data.UUID        (UUID) +import qualified Control.Exception as UnsafeE+import           Data.Typeable     (cast)++import           Data.Text.Prettyprint.Doc (Pretty (..), (<+>))+import qualified Data.Text.Prettyprint.Doc as Pretty+import           Text.Show.Pretty          (ppShow)+ type CapatazId = UUID type WorkerId = UUID type SupervisorId = UUID type ProcessId = UUID-type WorkerAction = IO ()-type ProcessThreadId = ThreadId+type WorkerAction m = WorkerId -> m () type ProcessName = Text type CapatazName = Text type SupervisorName = Text type WorkerName = Text type RestartCount = Int-type ProcessMap = HashMap ProcessId Process+type ProcessMap m = HashMap ProcessId (Process m) type ParentSupervisor = Supervisor --- | Event delivered to the "notifyEvent" callback sub-routine; these events can+-- | Wrapper for 'ThreadId'+--+-- @since 0.2.0.0+newtype ProcessThreadId+  = PTID ThreadId+  deriving (Generic, Eq, Show)++-- | @since 0.2.0.0+instance Pretty ProcessThreadId where+  pretty (PTID tid) =+    case words (show tid) of+      (_:threadNumber:_) -> pretty threadNumber+      _                  -> "unknown"++-- | Event delivered to the 'notifyEvent' callback sub-routine; these events can -- be used to monitor the capataz system and track what is doing, providing high -- levels of telemetry for all supervisors and workers of a capataz system, -- ergo, should be used for logging, monitoring and testing purposes.+--+-- @since 0.1.0.0 data CapatazEvent   = InvalidSupervisorStatusReached {     supervisorId   :: !SupervisorId@@ -131,8 +150,212 @@   }   deriving (Generic, Show) --- | Defines how a "Worker" process termination should be handled by its+-- | @since 0.2.0.0+instance Pretty CapatazEvent where+  pretty ev =+    case ev of+      InvalidSupervisorStatusReached {supervisorId, supervisorName} ->+        "Supervisor got into an state that should never have happened, please"+        <+> "report a ticket to"+        <+> "https://github.com/roman/Haskell-capataz/issues/new with title:"+        <+> Pretty.dquotes "InvalidSupervisorStatusReached error"+        <> prettyAttributes [("Supervisor ID"+                             , prettySupervisorId supervisorId supervisorName)]++      SupervisorStatusChanged {+          supervisorId+        , supervisorName+        , prevSupervisorStatus+        , newSupervisorStatus+        } ->+        "Supervisor changed state"+        <> prettyAttributes [ ("Supervisor ID"+                              , prettySupervisorId supervisorId supervisorName)+                            , ("Previous State"+                              , Pretty.dquotes $ Pretty.pretty prevSupervisorStatus)+                            , ("New State"+                              , Pretty.dquotes (Pretty.pretty newSupervisorStatus))+                            ]++      ProcessStarted {+          supervisorId+        , supervisorName+        , processId+        , processThreadId+        , processName+        , processType+        } ->+        "Supervisor spawned new process"+        <> prettyAttributes [ ("Supervisor ID"+                              , prettySupervisorId supervisorId supervisorName)+                            , ( "Process ID"+                              , prettyProcessId processId processName processThreadId)+                            , ( "Process Type",+                                pretty processType)+                            ]++      ProcessFailed {+          processId+        , processThreadId+        , processName+        , processType+        , processError+        } ->+        "Process failed with error"+        <> prettyAttributes [ ( "Process ID"+                              , prettyProcessId processId processName processThreadId)+                            , ( "Process Type", pretty processType)+                            , ( "Error"+                              , Pretty.nest 2 (Pretty.hardline+                                               <> "|" <+> pretty (ppShow processError)))+                            ]++      ProcessRestarted {+          supervisorId+        , supervisorName+        , processId+        , processThreadId+        , processName+        , processType+        } ->+        "Supervisor restarted failed process"+        <> prettyAttributes [ ("Supervisor ID"+                              , prettySupervisorId supervisorId supervisorName)+                            , ( "Process ID"+                              , prettyProcessId processId processName processThreadId)+                            , ( "Process Type", pretty processType)+                            ]++      ProcessTerminated {+          supervisorId+        , supervisorName+        , processId+        , processThreadId+        , processName+        , processType+        , terminationReason+        } ->+        "Supervisor terminated process"+        <> prettyAttributes [ ("Supervisor ID"+                              , prettySupervisorId supervisorId supervisorName)+                            , ( "Process ID"+                              , prettyProcessId processId processName processThreadId)+                            , ( "Process Type", pretty processType)+                            , ( "Reason"+                              , Pretty.nest 2+                                  (Pretty.fillBreak 10 $ pretty terminationReason))+                            ]+++      ProcessCompleted {+          processId+        , processThreadId+        , processName+        , processType+        } ->+        "Process completed execution with no errors"+        <> prettyAttributes [ ( "Process ID"+                              , prettyProcessId processId processName processThreadId)+                            , ( "Process Type", pretty processType)+                            ]++      ProcessCallbackExecuted {+          processId+        , processThreadId+        , processName+        , processType+        , processCallbackError+        , processCallbackType+        } ->+        case processCallbackError of+          Nothing ->+            "Process executed callback"+             <> prettyAttributes [ ( "Process ID"+                                   , prettyProcessId processId processName processThreadId)+                                 , ( "Process Type", pretty processType)+                                 , ( "Callback", pretty processCallbackType)+                                 ]+          Just err ->+            "Process executed callback and it failed"+             <> prettyAttributes [ ( "Process ID"+                                   , prettyProcessId processId processName processThreadId)+                                 , ( "Process Type", pretty processType)+                                 , ( "Callback", pretty processCallbackType)+                                 , ( "Error"+                                   , Pretty.nest 2+                                       (Pretty.hardline+                                         <> "|"+                                         <+> pretty (ppShow err)))+                                 ]++      ProcessTerminationStarted {+          supervisorId+        , supervisorName+        , terminationReason+        } ->+        "Supervisor started termination of its children"+        <> prettyAttributes [ ("Supervisor ID"+                              , prettySupervisorId supervisorId supervisorName)+                            , ( "Reason"+                              , Pretty.nest 2+                                  (Pretty.fillBreak 10 $ pretty terminationReason))+                            ]++      ProcessTerminationFinished {+          supervisorId+        , supervisorName+        } ->+        "Supervisor finished termination of its children"+        <> prettyAttributes [ ("Supervisor ID"+                              , prettySupervisorId supervisorId supervisorName)+                            ]++      CapatazFailed {+          supervisorId+        , supervisorName+        , supervisorError+        } ->+        "Root Supervisor had a fatal failure"+        <> prettyAttributes [ ("Supervisor ID"+                              , prettySupervisorId supervisorId supervisorName)+                            , ( "Error"+                              , Pretty.nest 2+                                  (Pretty.hardline+                                    <> "|"+                                    <+> pretty (ppShow supervisorError)))+                            ]++      CapatazTerminated {+          supervisorId+        , supervisorName+        } ->+        "Root supervisor was terminated"+        <> prettyAttributes [ ("Supervisor ID"+                              , prettySupervisorId supervisorId supervisorName)+                            ]+    where+      prettyAttributes attrList =+        Pretty.nest 4+          (Pretty.hardline+           <> Pretty.vsep (map (\(k, v) -> Pretty.fill 20 (k <> ":") <+> v)+                                attrList))++      prettySupervisorId supId supName =+        Pretty.angles+        $ pretty (show supId) <> "/" <> pretty supName++      prettyProcessId procId procName procTid =+        Pretty.angles+        $ pretty (show procId) <> "/" <> pretty procName <> "/" <> pretty procTid++-- | @since 0.2.0.0+instance Display CapatazEvent where+  display = displayShow . pretty++-- | Defines how a 'Worker' process termination should be handled by its -- supervisor.+--+-- @since 0.0.0.0 data WorkerTerminationPolicy   -- | Supervisor waits until infinity for the worker termination callback to   -- finish execution.@@ -148,9 +371,11 @@   | TimeoutMillis !Int   deriving (Generic, Show, Eq, Ord) -instance Default WorkerTerminationPolicy where-  -- | Default worker termination is a timeout of three (3) seconds.-  def = TimeoutMillis 3000+-- | Default worker termination is a timeout of three (3) seconds.+--+-- @since 0.2.0.0+defWorkerTerminationPolicy :: WorkerTerminationPolicy+defWorkerTerminationPolicy = TimeoutMillis 3000  instance NFData WorkerTerminationPolicy @@ -174,6 +399,8 @@  -- | Specifies the order in which supervised process should be terminated by a -- Supervisor in case of a restart or shutdown.+--+-- @since 0.0.0.0 data ProcessTerminationOrder   -- | Supervisor terminates supervised process from most recent to oldest.   = NewestFirst@@ -181,13 +408,17 @@   | OldestFirst   deriving (Generic, Show, Eq, Ord) -instance Default ProcessTerminationOrder where-  -- | default is "OldestFirst".-  def = OldestFirst+-- | Default termination order is 'OldestFirst'.+--+-- @since 0.2.0.0+defProcessTerminationOrder :: ProcessTerminationOrder+defProcessTerminationOrder = OldestFirst  instance NFData ProcessTerminationOrder  -- | Specifies how a Supervisor restarts a failing process.+--+-- @since 0.0.0.0 data SupervisorRestartStrategy   -- | Supervisor terminates all sibling supervised processes that didn't fail,   -- and then restarts all of them together. This strategy serves best when all@@ -198,31 +429,34 @@   | OneForOne   deriving (Generic, Show, Eq, Ord) -instance Default SupervisorRestartStrategy where-  -- | Default restart strategy is "OneForOne".-  def = OneForOne+-- | Default restart strategy is 'OneForOne'.+--+-- @since 0.2.0.0+defSupervisorRestartStategy :: SupervisorRestartStrategy+defSupervisorRestartStategy = OneForOne  instance NFData SupervisorRestartStrategy  -- | Allows to: ----- * Specify options for The root supervisor of a capataz system.+-- * Specify options for the root 'Supervisor' of a capataz system. ----- * Provie a "notifyEvent" callback to monitor or log a capataz system.+-- * Provide a 'notifyEvent' callback to monitor or log a capataz system. ---data CapatazOptions+-- @since 0.1.0.0+data CapatazOptions m   = CapatazOptions {     supervisorName                    :: !SupervisorName   , supervisorIntensity               :: !Int   , supervisorPeriodSeconds           :: !NominalDiffTime   , supervisorRestartStrategy         :: !SupervisorRestartStrategy-  , supervisorProcessSpecList         :: ![ProcessSpec]+  , supervisorProcessSpecList         :: ![ProcessSpec m]   , supervisorProcessTerminationOrder :: !ProcessTerminationOrder-  , supervisorOnIntensityReached      :: !(IO ())+  , supervisorOnIntensityReached      :: !(m ())     -- | Callback sub-routine that gets executed when the root supervisor fails.-  , supervisorOnFailure               :: !(SomeException -> IO ())+  , supervisorOnFailure               :: !(SomeException -> m ())     -- | Callback used for telemetry purposes.-  , notifyEvent                       :: !(CapatazEvent -> IO ())+  , notifyEvent                       :: !(CapatazEvent -> m ())   }  @@ -243,26 +477,29 @@   deriving (Generic, Show, Eq)  instance NFData WorkerRestartStrategy-instance Default WorkerRestartStrategy where-  -- |  A worker default restart strategy is "Transient".-  def = Transient +-- |  A worker default restart strategy is "Transient".+defWorkerRestartStrategy :: WorkerRestartStrategy+defWorkerRestartStrategy = Transient+ -- | Specifies all options that can be used to create a Worker Process. You may--- create a record of this type via the smart constructor "buildWorkerOptions".-data WorkerOptions+-- create a record of this type via the smart constructor 'buildWorkerOptions'.+--+-- @since 0.1.0.0+data WorkerOptions m   = WorkerOptions {     -- | An @IO ()@ sub-routine that will be executed when the worker     -- thread is created, this attribute is lazy given we want to this     -- value on a worker thread environment.-    workerAction            :: WorkerAction+    workerAction            :: WorkerAction m     -- | Name of the Worker (present on "CapatazEvent" records)   , workerName              :: !WorkerName     -- | Callback used when the worker fails with an error-  , workerOnFailure         :: !(SomeException -> IO ())+  , workerOnFailure         :: !(SomeException -> m ())     -- | Callback used when the worker completes execution without error-  , workerOnCompletion      :: !(IO ())+  , workerOnCompletion      :: !(m ())     -- | Callback used when the worker is terminated-  , workerOnTermination     :: !(IO ())+  , workerOnTermination     :: !(m ())     -- | Indicates how a worker should be terminated   , workerTerminationPolicy :: !WorkerTerminationPolicy     -- | Indicates how a worker should be restarted@@ -271,7 +508,9 @@   deriving (Generic)  -- | Record that contains the "Async" record (thread reference) of a worker-data Worker+--+-- @since 0.1.0.0+data Worker m   = Worker {     -- | Unique identifier for a worker that is executing     workerId           :: !WorkerId@@ -283,7 +522,7 @@   , workerName         :: !WorkerName     -- | "WorkerOptions" contains all the options around restart and termination     -- policies-  , workerOptions      :: !WorkerOptions+  , workerOptions      :: !(WorkerOptions m)   }  data ProcessEnv@@ -295,7 +534,7 @@   , processRestartStrategy :: !WorkerRestartStrategy   } -data SupervisorOptions+data SupervisorOptions m   = SupervisorOptions {     -- | Name of the Supervisor (present on "CapatazEvent" records)     supervisorName                    :: Text@@ -308,40 +547,40 @@     -- | What is the "SupervisorRestartStrategy" for this Capataz   , supervisorRestartStrategy         :: !SupervisorRestartStrategy     -- | Static set of workers that start as soon as the "Capataz" is created-  , supervisorProcessSpecList         :: ![ProcessSpec]+  , supervisorProcessSpecList         :: ![ProcessSpec m]     -- | In which order the "Supervisor" record is going to terminate it's workers   , supervisorProcessTerminationOrder :: !ProcessTerminationOrder     -- | Callback used when the error intensity is reached-  , supervisorOnIntensityReached      :: !(IO ())-  , supervisorOnFailure               :: !(SomeException -> IO ())+  , supervisorOnIntensityReached      :: !(m ())+  , supervisorOnFailure               :: !(SomeException -> m ())   } -data Supervisor+data Supervisor m   = Supervisor {     supervisorId           :: !SupervisorId   , supervisorName         :: !SupervisorName-  , supervisorOptions      :: !SupervisorOptions+  , supervisorOptions      :: !(SupervisorOptions m)   , supervisorCreationTime :: !UTCTime   , supervisorAsync        :: !(Async ())-  , supervisorNotify       :: SupervisorMessage -> IO ()-  , supervisorEnv          :: !SupervisorEnv+  , supervisorNotify       :: SupervisorMessage m -> m ()+  , supervisorEnv          :: !(SupervisorEnv m)   }  -- | Internal record that represents an action being sent from threads using -- the Capataz public API.-data ControlAction+data ControlAction m   = ForkWorker {-    workerOptions  :: !WorkerOptions-  , returnWorkerId :: !(WorkerId -> IO ())+    workerOptions  :: !(WorkerOptions m)+  , returnWorkerId :: !(WorkerId -> m ())   }   | ForkSupervisor {-    supervisorOptions :: !SupervisorOptions-  , returnSupervisor  :: !(Supervisor -> IO ())+    supervisorOptions :: !(SupervisorOptions m)+  , returnSupervisor  :: !(Supervisor m -> m ())   }   | TerminateProcess {     processId                :: !ProcessId   , processTerminationReason :: !Text-  , notifyProcessTermination :: !(Bool -> IO ())+  , notifyProcessTermination :: !(Bool -> m ())   }   deriving (Generic) @@ -384,11 +623,20 @@   | OnTermination   deriving (Generic, Show, Eq) +instance Pretty CallbackType where+  pretty = Pretty.pretty . show+ data ProcessType   = SupervisorType   | WorkerType   deriving (Show, Eq) +instance Pretty ProcessType  where+  pretty ty =+    case ty of+      SupervisorType -> "Supervisor"+      WorkerType     -> "Worker"+ -- | Internal exception triggered when a callback of a Worker fails data ProcessError   = ProcessCallbackFailed {@@ -448,97 +696,108 @@  instance NFData SupervisorStatus +instance Pretty SupervisorStatus where+  pretty =+    pretty . show+ -- | Internal message delivered to a supervisor process that can either be a -- call from public API or an event from its monitored worker process.-data SupervisorMessage+data SupervisorMessage m   -- | Represents a request from done to the supervisor thread from another   -- thread using the public API-  = ControlAction !ControlAction+  = ControlAction !(ControlAction m)   -- | Represents an event (failure, completion, etc) from a monitored worker   -- process to the supervisor   | MonitorEvent !MonitorEvent   deriving (Generic) --- | Internal Type to manage both Worker and Supervisor processes-data Process-  = WorkerProcess  Worker-  | SupervisorProcess Supervisor+-- | Internal Type to manage both 'Worker' and 'Supervisor' processes+--+-- @since 0.1.0.0+data Process m+  = WorkerProcess  !(Worker m)+  | SupervisorProcess !(Supervisor m) --- | Record used to specify how to __build__ a runtime "Process" in a static+-- | Record used to specify how to __build__ a runtime 'Process' in a static -- supervision tree; to create values of this type, you must use: ----- * "workerSpec" or "workerSpecWithDefaults" to build a worker process+-- * 'workerSpec' or 'workerSpecWithDefaults' to build a worker process ----- * "supervisorSpec" or "supervisorSpecWithDefaults" to build a supervisor+-- * 'supervisorSpec' or 'supervisorSpecWithDefaults' to build a supervisor -- process ---data ProcessSpec-  = WorkerSpec WorkerOptions-  | SupervisorSpec SupervisorOptions+-- @since 0.1.0.0+data ProcessSpec m+  = WorkerSpec (WorkerOptions m)+  | SupervisorSpec (SupervisorOptions m)  -- | Record that contains the environment of a capataz monitor, this is used as -- the main record to create workers and to stop the supervisor thread.-data Capataz+--+-- @since 0.0.0.0+data Capataz m   = Capataz {-    capatazSupervisor :: !Supervisor+    capatazSupervisor :: !(Supervisor m)   , capatazTeardown   :: !Teardown   } -instance ITeardown Capataz where-  teardown Capataz {capatazTeardown} =-    teardown capatazTeardown+instance HasTeardown (Capataz m) where+  getTeardown Capataz {capatazTeardown} =+    capatazTeardown  -- | Internal utility record used to hold part of the runtime information of a -- supervisor that acts as a parent of another supervisor.-data ParentSupervisorEnv+data ParentSupervisorEnv m   = ParentSupervisorEnv {     supervisorId     :: !SupervisorId   , supervisorName   :: !SupervisorName-  , supervisorNotify :: !(SupervisorMessage -> IO ())-  , notifyEvent      :: !(CapatazEvent -> IO ())+  , supervisorNotify :: !(SupervisorMessage m -> m ())+  , notifyEvent      :: !(CapatazEvent -> m ())   }  -- | Convenience internal utility record that contains all values related to a -- supervisor process.-data SupervisorEnv+data SupervisorEnv m   = SupervisorEnv {     supervisorId                      :: !SupervisorId   , supervisorName                    :: !SupervisorName-  , supervisorNotify                  :: !(SupervisorMessage -> IO ())-  , supervisorGetNotification         :: !(STM SupervisorMessage)-  , supervisorProcessMap              :: !(IORef ProcessMap)+  , supervisorNotify                  :: !(SupervisorMessage m -> m ())+  , supervisorGetNotification         :: !(STM (SupervisorMessage m))+  , supervisorProcessMap              :: !(IORef (ProcessMap m))   , supervisorStatusVar               :: !(TVar SupervisorStatus)-  , supervisorOptions                 :: !SupervisorOptions+  , supervisorOptions                 :: !(SupervisorOptions m)   , supervisorIntensity               :: !Int     -- ^ http://erlang.org/doc/design_principles/sup_princ.html#max_intensity   , supervisorPeriodSeconds           :: !NominalDiffTime   , supervisorRestartStrategy         :: !SupervisorRestartStrategy   , supervisorProcessTerminationOrder :: !ProcessTerminationOrder-  , supervisorOnIntensityReached      :: !(IO ())-  , supervisorOnIntensityReached      :: !(SomeException -> IO ())-  , notifyEvent                       :: !(CapatazEvent -> IO ())+  , supervisorOnIntensityReached      :: !(m ())+  , supervisorOnIntensityReached      :: !(SomeException -> m ())+  , notifyEvent                       :: !(CapatazEvent -> m ())   } --- | Builds a "CapatazOptions" record with defaults on how to create a capataz+-- | Builds a 'CapatazOptions' record with defaults on how to create a capataz -- root supervisor, these defaults are: -- -- * Intensity error tolerance is set to 1 error every 5 seconds ----- * A "SupervisorRestartStrategy" of "OneForOne"+-- * A 'SupervisorRestartStrategy' of 'OneForOne' ----- * A "ProcessTerminationOrder" of "OldestFirst"+-- * A 'ProcessTerminationOrder' of 'OldestFirst' ----- This function is intended to be used in combination with "forkCapataz".+-- This function is intended to be used in combination with 'forkCapataz'. --+-- @since 0.1.0.0 defCapatazOptions-  :: Text-  -> (CapatazOptions -> CapatazOptions) -- ^ Function to modify root supervisor-  -> CapatazOptions+  :: Monad m+  => Text+  -> (CapatazOptions m -> CapatazOptions m) -- ^ Function to modify root supervisor+  -> CapatazOptions m defCapatazOptions supervisorName modFn = modFn CapatazOptions   { supervisorName   , supervisorIntensity               = 2   , supervisorPeriodSeconds           = 5-  , supervisorRestartStrategy         = def+  , supervisorRestartStrategy         = defSupervisorRestartStategy   , supervisorProcessSpecList         = []   , supervisorProcessTerminationOrder = OldestFirst   , supervisorOnIntensityReached      = return ()@@ -546,78 +805,106 @@   , notifyEvent                       = const $ return ()   } --- | Builds a "ProcessSpec" record for a supervisor process with defaults from--- "supervisorSpecWithDefaults". This function allows overrides of these+-- | Builds a 'ProcessSpec' record for a supervisor process with defaults from+-- 'supervisorSpecWithDefaults'. This function allows overrides of these -- defaults using lenses. -- -- This function is used when building a supervisor branch in a static -- supervision trees. --+-- @since 0.1.0.0 supervisorSpec-  :: SupervisorName -- ^ Name used for telemetry purposes-  -> (SupervisorOptions -> SupervisorOptions) -- ^ Function to modify default-                                              -- supervisor options-  -> ProcessSpec+  :: Monad m+  => SupervisorName -- ^ Name used for telemetry purposes+  -> (SupervisorOptions m -> SupervisorOptions m) -- ^ 'SupervisorOptions' modifier+  -> ProcessSpec m supervisorSpec sName modFn =   SupervisorSpec (buildSupervisorOptions sName modFn) {-# INLINE supervisorSpec #-} --- | Builds a "ProcessSpec" record for a supervisor process with defaults from---  "buildSupervisorOptionsWithDefaults".+-- | Builds a 'ProcessSpec' record for a supervisor process with defaults from+--  'buildSupervisorOptionsWithDefaults'. -- -- This function is used when building a supervisor branch in a static -- supervision trees. --+-- @since 0.1.0.0 supervisorSpecWithDefaults-  :: SupervisorName -- ^ Name used for telemetry purposes-  -> ProcessSpec-supervisorSpecWithDefaults sName = supervisorSpec sName identity+  :: Monad m+  => SupervisorName -- ^ Name used for telemetry purposes+  -> ProcessSpec m+supervisorSpecWithDefaults sName = supervisorSpec sName id {-# INLINE supervisorSpecWithDefaults #-} --- | Builds a "ProcessSpec" record for a worker process with defaults from--- "workerSpecWithDefaults". This function allows overrides of these+-- | Builds a 'ProcessSpec' record for a worker process with defaults from+-- 'workerSpecWithDefaults'. This function allows overrides of these -- defaults using lenses. -- -- This function is used when building a worker in a static supervision tree. --+-- @since 0.1.0.0 workerSpec-  :: WorkerName -- ^ Name used for telemetry purposes-  -> IO () -- ^ IO sub-routine to be supervised-  -> (WorkerOptions -> WorkerOptions) -- ^ Function to modify default worker+  :: Monad m+  => WorkerName -- ^ Name used for telemetry purposes+  -> m () -- ^ 'IO' sub-routine to be supervised+  -> (WorkerOptions m -> WorkerOptions m) -- ^ Function to modify default worker                                       -- options-  -> ProcessSpec+  -> ProcessSpec m workerSpec wName wAction modFn =   WorkerSpec (buildWorkerOptions wName wAction modFn) {-# INLINE workerSpec #-} --- | Builds a "ProcessSpec" record for a worker process with defaults from--- "buildSupervisorOptionsWithDefaults".+-- | Builds a 'ProcessSpec' record for a worker process with defaults from+-- 'workerSpecWithDefaults'. This function allows overrides of these+-- defaults using lenses. -- -- This function is used when building a worker in a static supervision tree.+-- The given sub-routine will receive the 'WorkerId' as a parameter --+-- @since 0.2.0.0+workerSpec1+  :: Monad m+  => WorkerName -- ^ Name used for telemetry purposes+  -> (WorkerId -> m ()) -- ^ sub-routine to be supervised+  -> (WorkerOptions m -> WorkerOptions m) -- ^ Function to modify default worker+                                      -- options+  -> ProcessSpec m+workerSpec1 wName wAction modFn =+  WorkerSpec (buildWorkerOptions1 wName wAction modFn)+{-# INLINE workerSpec1 #-}++-- | Builds a 'ProcessSpec' record for a worker process with defaults from+-- 'buildSupervisorOptionsWithDefaults'.+--+-- This function is used when building a worker in a static supervision tree.+--+-- @since 0.1.0.0 workerSpecWithDefaults-  :: WorkerName -- ^ Name used for telemetry purposes-  -> IO () -- ^ IO sub-routine to be supervised-  -> ProcessSpec-workerSpecWithDefaults wName wAction = workerSpec wName wAction identity+  :: Monad m+  => WorkerName -- ^ Name used for telemetry purposes+  -> m () -- ^ IO sub-routine to be supervised+  -> ProcessSpec m+workerSpecWithDefaults wName wAction = workerSpec wName wAction id {-# INLINE workerSpecWithDefaults #-} --- | Builds a "SupervisorOptions" record with defaults from--- "buildSupervisorOptionsWithDefaults". This function allows overrides of these+-- | Builds a 'SupervisorOptions' record with defaults from+-- 'buildSupervisorOptionsWithDefaults'. This function allows overrides of these -- defaults using lenses. ----- This function is intended to be used in combination with "forkSupervisor".+-- This function is intended to be used in combination with 'forkSupervisor'. --+-- @since 0.1.0.0 buildSupervisorOptions-  :: SupervisorName -- ^ Name used for telemetry purposes-  -> (SupervisorOptions -> SupervisorOptions) -- ^ Function to modify default+  :: Monad m+  => SupervisorName -- ^ Name used for telemetry purposes+  -> (SupervisorOptions m -> SupervisorOptions m) -- ^ Function to modify default                                               -- supervisor options-  -> SupervisorOptions+  -> SupervisorOptions m buildSupervisorOptions supervisorName modFn = modFn SupervisorOptions   { supervisorName   , supervisorIntensity               = 2   , supervisorPeriodSeconds           = 5-  , supervisorRestartStrategy         = def+  , supervisorRestartStrategy         = defSupervisorRestartStategy   , supervisorProcessSpecList         = []   , supervisorProcessTerminationOrder = OldestFirst   , supervisorOnIntensityReached      = return ()@@ -625,52 +912,77 @@   } {-# INLINE buildSupervisorOptions #-} --- | Builds a "SupervisorOptions" record with defaults to create a supervisor+-- | Builds a 'SupervisorOptions' record with defaults to create a supervisor -- process, these defaults are: -- -- * Intensity error tolerance is set to 1 error every 5 seconds ----- * A "SupervisorRestartStrategy" of "OneForOne"+-- * A 'SupervisorRestartStrategy' of 'OneForOne' ----- * A "ProcessTerminationOrder" of "OldestFirst"+-- * A 'ProcessTerminationOrder' of 'OldestFirst' ----- This function is intended to be used in combination with "forkSupervisor".+-- This function is intended to be used in combination with 'forkSupervisor'. --+-- @since 0.1.0.0 buildSupervisorOptionsWithDefaults-  :: SupervisorName -- ^ Name used for telemetry purposes-  -> SupervisorOptions-buildSupervisorOptionsWithDefaults = flip buildSupervisorOptions identity+  :: Monad m+  => SupervisorName -- ^ Name used for telemetry purposes+  -> SupervisorOptions m+buildSupervisorOptionsWithDefaults = flip buildSupervisorOptions id {-# INLINE buildSupervisorOptionsWithDefaults #-} --- | Builds a "WorkerOptions" record, keeps the defaults from---   "buildWorkerOptionsWithDefaults" but allows overrides using lenses.+-- | Builds a 'WorkerOptions' record, keeps the defaults from+--   'buildWorkerOptionsWithDefaults' but allows overrides using lenses. ----- This function is intended to be used in combination with "forkWorker". See--- the ... example in the examples directory for a demonstration.+-- This function is intended to be used in combination with 'forkWorker'. See the+-- capataz-simple-example project in the examples directory for a demonstration. --+-- @since 0.1.0.0 buildWorkerOptions-  :: WorkerName -- ^ Name used for telemetry purposes-  -> IO () -- ^ IO sub-routine to be supervised-  -> (WorkerOptions -> WorkerOptions) -- ^ Function to modify default worker+  :: Monad m+  => WorkerName -- ^ Name used for telemetry purposes+  -> m () -- ^ Process sub-routine to be supervised+  -> (WorkerOptions m -> WorkerOptions m) -- ^ Function to modify default worker                                       -- options-  -> WorkerOptions-buildWorkerOptions workerName workerAction f = f WorkerOptions+  -> WorkerOptions m+buildWorkerOptions workerName workerAction f =+  buildWorkerOptions1 workerName (const workerAction) f+{-# INLINE buildWorkerOptions #-}++-- | Builds a 'WorkerOptions' record, keeps the defaults from+--   'buildWorkerOptionsWithDefaults' but allows overrides using lenses.+--+-- This function is intended to be used in combination with 'forkWorker'. See the+-- capataz-simple-example project in the examples directory for a demonstration.+--+-- The given sub-routine will receive the 'WorkerId' as a parameter.+--+-- @since 0.2.0.0+buildWorkerOptions1+  :: Monad m+  => WorkerName -- ^ Name used for telemetry purposes+  -> (WorkerId -> m ()) -- ^ Process sub-routine to be supervised+  -> (WorkerOptions m -> WorkerOptions m) -- ^ Function to modify default worker+                                      -- options+  -> WorkerOptions m+buildWorkerOptions1 workerName workerAction f = f WorkerOptions   { workerName-  , workerAction+  , workerAction            = workerAction   , workerOnFailure         = const $ return ()   , workerOnCompletion      = return ()   , workerOnTermination     = return ()-  , workerTerminationPolicy = def-  , workerRestartStrategy   = def+  , workerTerminationPolicy = defWorkerTerminationPolicy+  , workerRestartStrategy   = defWorkerRestartStrategy   }-{-# INLINE buildWorkerOptions #-}+{-# INLINE buildWorkerOptions1 #-} --- | Builds a "WorkerOptions" record with defaults to create a worker process,++-- | Builds a 'WorkerOptions' record with defaults to create a worker process, -- the defaults are: ----- * A "Transient" "WorkerRestartStrategy"+-- * A 'Transient' 'WorkerRestartStrategy' ----- * A "WorkerTerminationPolicy" of a 3 seconds timeout+-- * A 'WorkerTerminationPolicy' of a 3 seconds timeout -- -- * A _completion_ callback that just returns unit --@@ -678,14 +990,36 @@ -- -- * A _failure_ callback that just returns unit ----- This function is intended to be used in combination with "forkWorker", for--- creating a worker in an static supervision tree, use "workerSpecWithDefaults"--- instead. See the ... example for a demonstration.+-- This function is intended to be used in combination with 'forkWorker', for creating a+-- worker in an static supervision tree, use 'workerSpecWithDefaults' instead. See the+-- capataz-simple-example project in the examples directory for a demonstration. --+-- @since 0.1.0.0 buildWorkerOptionsWithDefaults-  :: WorkerName -- ^ Name used for telemetry purposes-  -> IO () -- ^ IO sub-routine to be supervised-  -> WorkerOptions+  :: Monad m+  => WorkerName -- ^ Name used for telemetry purposes+  -> m () -- ^ 'IO' sub-routine to be supervised+  -> WorkerOptions m buildWorkerOptionsWithDefaults wName wAction =-  buildWorkerOptions wName wAction identity+  buildWorkerOptions wName wAction id {-# INLINE buildWorkerOptionsWithDefaults #-}++-- | Used for debugging purposes+getMaskingState :: MonadIO m => m UnsafeE.MaskingState+getMaskingState = liftIO UnsafeE.getMaskingState++-- | Given we want to capture async exceptions to send them back to a supervisor+-- and we are running on masked states, we need to have a try that catches all+-- kinds of exceptions+unsafeTry :: (Exception e, MonadUnliftIO m) => m a -> m (Either e a)+unsafeTry action = withRunInIO $ \run -> UnsafeE.try (run action)++-- | Given unliftio wraps exceptions in 3 layers of Exceptions, and we are using+-- vanilla exceptions, we need to make sure that we account for all different+-- exception types+fromAnyException :: Exception e => SomeException -> Maybe e+fromAnyException ex = case UnsafeE.fromException ex of+  Just (UnsafeE.SomeAsyncException innerEx1) -> case cast innerEx1 of+    Just (AsyncExceptionWrapper innerEx2) -> cast innerEx2+    Nothing                               -> cast innerEx1+  Nothing -> fromException ex

@@ -3,9 +3,10 @@ {-# LANGUAGE RecordWildCards   #-} module Control.Concurrent.Capataz.Internal.Types.Lens where +import RIO+ import Control.Concurrent.Capataz.Internal.Types-import Data.Time.Clock                           (NominalDiffTime)-import Protolude+import RIO.Time                                  (NominalDiffTime)  -------------------------------------------------------------------------------- @@ -40,9 +41,9 @@   -- supervisor.   supervisorProcessSpecListL     :: Functor f-    => ([ProcessSpec] -> f [ProcessSpec])-    -> s-    -> f s+    => ([ProcessSpec m] -> f [ProcessSpec m])+    -> s m+    -> f (s m)  class HasSupervisorProcessTerminationOrder s where   -- | Specifies order in which a supervisor is going to terminate its@@ -58,31 +59,31 @@   -- breach in a supervisor's error intensity.   supervisorOnIntensityReachedL     :: Functor f-    => (IO () -> f (IO ()))-    -> s-    -> f s+    => (m () -> f (m ()))+    -> s m+    -> f (s m)  class HasSupervisorFailureCallback s where   -- | Specifies callback sub-routine that gets executed when a supervisor   -- fails.   supervisorOnFailureL     :: Functor f-    => ((SomeException -> IO ()) -> f (SomeException -> IO ()))-    -> s-    -> f s+    => ((SomeException -> m ()) -> f (SomeException -> m ()))+    -> s m+    -> f (s m)  -instance HasSupervisorIntensity SupervisorOptions where+instance HasSupervisorIntensity (SupervisorOptions m) where   supervisorIntensityL k SupervisorOptions {supervisorIntensity, ..} =     fmap (\newSupIntensity -> SupervisorOptions { supervisorIntensity = newSupIntensity, .. })          (k supervisorIntensity) -instance HasSupervisorPeriodSeconds SupervisorOptions where+instance HasSupervisorPeriodSeconds (SupervisorOptions m) where   supervisorPeriodSecondsL k SupervisorOptions {supervisorPeriodSeconds, ..} =     fmap (\newSupPeriodSeconds -> SupervisorOptions { supervisorPeriodSeconds = newSupPeriodSeconds, .. })          (k supervisorPeriodSeconds) -instance HasSupervisorRestartStrategy SupervisorOptions where+instance HasSupervisorRestartStrategy (SupervisorOptions m) where   supervisorRestartStrategyL k SupervisorOptions {supervisorRestartStrategy, ..} =     fmap (\newSupRestartStrategy ->             SupervisorOptions { supervisorRestartStrategy = newSupRestartStrategy, .. })@@ -94,7 +95,7 @@             SupervisorOptions { supervisorProcessSpecList = newSupProcessSpecList, .. })          (k supervisorProcessSpecList) -instance HasSupervisorProcessTerminationOrder SupervisorOptions where+instance HasSupervisorProcessTerminationOrder (SupervisorOptions m) where   supervisorProcessTerminationOrderL k SupervisorOptions {supervisorProcessTerminationOrder, ..} =     fmap (\newSupProcessTerminationOrder ->             SupervisorOptions { supervisorProcessTerminationOrder = newSupProcessTerminationOrder, .. })@@ -112,17 +113,17 @@  -------------------- -instance HasSupervisorIntensity CapatazOptions where+instance HasSupervisorIntensity (CapatazOptions m) where   supervisorIntensityL k CapatazOptions {supervisorIntensity, ..} =     fmap (\newSupIntensity -> CapatazOptions { supervisorIntensity = newSupIntensity, .. })          (k supervisorIntensity) -instance HasSupervisorPeriodSeconds CapatazOptions where+instance HasSupervisorPeriodSeconds (CapatazOptions m) where   supervisorPeriodSecondsL k CapatazOptions {supervisorPeriodSeconds, ..} =     fmap (\newSupPeriodSeconds -> CapatazOptions { supervisorPeriodSeconds = newSupPeriodSeconds, .. })          (k supervisorPeriodSeconds) -instance HasSupervisorRestartStrategy CapatazOptions where+instance HasSupervisorRestartStrategy (CapatazOptions m) where   supervisorRestartStrategyL k CapatazOptions {supervisorRestartStrategy, ..} =     fmap (\newSupRestartStrategy ->             CapatazOptions { supervisorRestartStrategy = newSupRestartStrategy, .. })@@ -134,7 +135,7 @@             CapatazOptions { supervisorProcessSpecList = newSupProcessSpecList, .. })          (k supervisorProcessSpecList) -instance HasSupervisorProcessTerminationOrder CapatazOptions where+instance HasSupervisorProcessTerminationOrder (CapatazOptions m) where   supervisorProcessTerminationOrderL k CapatazOptions {supervisorProcessTerminationOrder, ..} =     fmap (\newSupProcessTerminationOrder ->             CapatazOptions { supervisorProcessTerminationOrder = newSupProcessTerminationOrder, .. })@@ -155,9 +156,9 @@   -- telemetry purposes (e.g. logging, monitoring, etc). onSystemEventL   :: Functor f-  => ((CapatazEvent -> IO ()) -> f (CapatazEvent -> IO ()))-  -> CapatazOptions-  -> f CapatazOptions+  => ((CapatazEvent -> m ()) -> f (CapatazEvent -> m ()))+  -> CapatazOptions m+  -> f (CapatazOptions m) onSystemEventL k CapatazOptions { notifyEvent, ..} = fmap   (\newNotifyEvent -> CapatazOptions {notifyEvent = newNotifyEvent, ..})   (k notifyEvent)@@ -171,10 +172,10 @@ -- worker "WorkerTerminationPolicy". -- workerOnFailureL-  :: Functor f-  => ((SomeException -> IO ()) -> f (SomeException -> IO ()))-  -> WorkerOptions-  -> f WorkerOptions+  :: (Functor f)+  => ((SomeException -> m ()) -> f (SomeException -> m ()))+  -> WorkerOptions m+  -> f (WorkerOptions m) workerOnFailureL k WorkerOptions { workerOnFailure, ..} = fmap   (\newWorkerAction -> WorkerOptions {workerOnFailure = newWorkerAction, ..})   (k workerOnFailure)@@ -186,7 +187,7 @@ -- worker "WorkerTerminationPolicy". -- workerOnCompletionL-  :: Functor f => (IO () -> f (IO ())) -> WorkerOptions -> f WorkerOptions+  :: (Functor f) => (m () -> f (m ())) -> WorkerOptions m -> f (WorkerOptions m) workerOnCompletionL k WorkerOptions { workerOnCompletion, ..} = fmap   (\newWorkerAction -> WorkerOptions {workerOnCompletion = newWorkerAction, ..})   (k workerOnCompletion)@@ -199,10 +200,9 @@ -- worker "WorkerTerminationPolicy". -- workerOnTerminationL-  :: Functor f => (IO () -> f (IO ())) -> WorkerOptions -> f WorkerOptions+  :: Functor f => (m () -> f (m ())) -> WorkerOptions m -> f (WorkerOptions m) workerOnTerminationL k WorkerOptions { workerOnTermination, ..} = fmap-  ( \newWorkerAction ->-    WorkerOptions {workerOnTermination = newWorkerAction, ..}+  (\newWorkerAction -> WorkerOptions {workerOnTermination = newWorkerAction, ..}   )   (k workerOnTermination) @@ -211,10 +211,10 @@ workerTerminationPolicyL   :: Functor f   => (WorkerTerminationPolicy -> f WorkerTerminationPolicy)-  -> WorkerOptions-  -> f WorkerOptions+  -> WorkerOptions m+  -> f (WorkerOptions m) workerTerminationPolicyL k WorkerOptions { workerTerminationPolicy, ..} = fmap-  ( \newWorkerAction ->+  (\newWorkerAction ->     WorkerOptions {workerTerminationPolicy = newWorkerAction, ..}   )   (k workerTerminationPolicy)@@ -224,10 +224,10 @@ workerRestartStrategyL   :: Functor f   => (WorkerRestartStrategy -> f WorkerRestartStrategy)-  -> WorkerOptions-  -> f WorkerOptions+  -> WorkerOptions m+  -> f (WorkerOptions m) workerRestartStrategyL k WorkerOptions { workerRestartStrategy, ..} = fmap-  ( \newWorkerAction ->+  (\newWorkerAction ->     WorkerOptions {workerRestartStrategy = newWorkerAction, ..}   )   (k workerRestartStrategy)

@@ -12,30 +12,28 @@ -} module Control.Concurrent.Capataz.Internal.Util where -import Protolude+import           RIO+import qualified RIO.HashMap as HashMap+import qualified RIO.List    as List+import qualified RIO.Text    as T -import           Control.Concurrent.STM      (STM, atomically, retry)-import           Control.Concurrent.STM.TVar (TVar, readTVar, writeTVar)-import           Data.IORef                  (atomicModifyIORef', readIORef)-import qualified Data.Text                   as T-import           Data.Time.Clock             (getCurrentTime)+import RIO.Time (getCurrentTime) -import qualified Data.HashMap.Strict as HashMap-import           GHC.Conc            (labelThread)+import GHC.Conc (labelThread)  import Control.Concurrent.Capataz.Internal.Types  -- | Returns only the number of the ThreadId. getTidNumber :: ThreadId -> Maybe Text-getTidNumber tid = case T.words $ show tid of-  (_:tidNumber:_) -> Just tidNumber-  _               -> Nothing+getTidNumber tid = case T.words $ tshow tid of+  (_ : tidNumber : _) -> Just tidNumber+  _                   -> Nothing  --------------------------------------------------------------------------------  -- | Internal functions that overwrites the GHC thread name, for increasing -- traceability on GHC internals.-setProcessThreadName :: WorkerId -> WorkerName -> IO ()+setProcessThreadName :: MonadIO m => WorkerId -> WorkerName -> m () setProcessThreadName workerId workerName = do   tid <- myThreadId   let workerIdentifier =@@ -43,55 +41,57 @@           ""           T.unpack           (getTidNumber tid)-  labelThread tid workerIdentifier+  liftIO $ labelThread tid workerIdentifier --- | Gets the "ProcessId" of both a Worker or Supervisor process.-getProcessId :: Process -> ProcessId+-- | Gets the 'ProcessId' of both a Worker or Supervisor process.+getProcessId :: Process m -> ProcessId getProcessId process = case process of   WorkerProcess     Worker { workerId }         -> workerId   SupervisorProcess Supervisor { supervisorId } -> supervisorId --- | Gets a supervised "Process" from a "Supervisor" instance.-fetchProcess :: SupervisorEnv -> ProcessId -> IO (Maybe Process)+-- | Gets a supervised 'Process' from a 'Supervisor' instance.+fetchProcess+  :: MonadIO m => SupervisorEnv m -> ProcessId -> m (Maybe (Process m)) fetchProcess SupervisorEnv { supervisorProcessMap } processId = do   processMap <- readIORef supervisorProcessMap   case HashMap.lookup processId processMap of     Just process -> return $ Just process     _            -> return Nothing --- | Appends a new "Process" to the "Supervisor" existing process map.-appendProcessToMap :: SupervisorEnv -> Process -> IO ()+-- | Appends a new 'Process' to the 'Supervisor' existing process map.+appendProcessToMap :: MonadIO m => SupervisorEnv m -> Process m -> m () appendProcessToMap SupervisorEnv { supervisorProcessMap } process =   atomicModifyIORef' supervisorProcessMap                      (\processMap -> (appendProcess processMap, ()))  where   appendProcess = HashMap.alter (const $ Just process) (getProcessId process) --- | Removes a "Process" from a "Supervisor" existing process map.-removeProcessFromMap :: SupervisorEnv -> ProcessId -> IO ()+-- | Removes a 'Process' from a 'Supervisor' existing process map.+removeProcessFromMap :: MonadIO m => SupervisorEnv m -> ProcessId -> m () removeProcessFromMap SupervisorEnv { supervisorProcessMap } processId =   atomicModifyIORef'     supervisorProcessMap-    ( \processMap -> maybe (processMap, ())-                           (const (HashMap.delete processId processMap, ()))-                           (HashMap.lookup processId processMap)+    (\processMap -> maybe (processMap, ())+                          (const (HashMap.delete processId processMap, ()))+                          (HashMap.lookup processId processMap)     ) --- | Function to modify a "Supervisor" process map using a pure function.-resetProcessMap :: SupervisorEnv -> (ProcessMap -> ProcessMap) -> IO ()+-- | Function to modify a 'Supervisor' process map using a pure function.+resetProcessMap+  :: MonadIO m => SupervisorEnv m -> (ProcessMap m -> ProcessMap m) -> m () resetProcessMap SupervisorEnv { supervisorProcessMap } processMapFn =   atomicModifyIORef' supervisorProcessMap                      (\processMap -> (processMapFn processMap, ())) --- | Function to get a snapshot of a "Supervisor" process map.-readProcessMap :: SupervisorEnv -> IO ProcessMap+-- | Function to get a snapshot of a 'Supervisor' process map.+readProcessMap :: MonadIO m => SupervisorEnv m -> m (ProcessMap m) readProcessMap SupervisorEnv { supervisorProcessMap } =   readIORef supervisorProcessMap --- | Returns all processes of a "Supervisor" by "ProcessTerminationOrder". This--- is used on "AllForOne" restarts and shutdown operations.+-- | Returns all processes of a 'Supervisor' by 'ProcessTerminationOrder'. This+-- is used on 'AllForOne' restarts and shutdown operations. sortProcessesByTerminationOrder-  :: ProcessTerminationOrder -> ProcessMap -> [Process]+  :: ProcessTerminationOrder -> ProcessMap m -> [Process m] sortProcessesByTerminationOrder terminationOrder processMap =   case terminationOrder of     OldestFirst -> workers@@ -103,28 +103,30 @@   processCreationTime (SupervisorProcess Supervisor { supervisorCreationTime })     = supervisorCreationTime -  workers = sortBy (comparing processCreationTime) (HashMap.elems processMap)+  workers =+    List.sortBy (comparing processCreationTime) (HashMap.elems processMap)  -------------------------------------------------------------------------------- --- | Returns the "SupervisorStatus", this sub-routine will retry transaction--- until its associated "Supervisor" has a status different from "Initializing".+-- | Returns the 'SupervisorStatus', this sub-routine will retry transaction+-- until its associated 'Supervisor' has a status different from 'Initializing'. readSupervisorStatusSTM :: TVar SupervisorStatus -> STM SupervisorStatus readSupervisorStatusSTM statusVar = do   status <- readTVar statusVar-  if status == Initializing then retry else return status+  if status == Initializing then retrySTM else return status --- | Executes transaction that returns the "SupervisorStatus".-readSupervisorStatus :: SupervisorEnv -> IO SupervisorStatus+-- | Executes transaction that returns the 'SupervisorStatus'.+readSupervisorStatus :: MonadIO m => SupervisorEnv m -> m SupervisorStatus readSupervisorStatus SupervisorEnv { supervisorStatusVar } =   atomically $ readTVar supervisorStatusVar --- | Modifes the "Supervisor" status.+-- | Modifes the 'Supervisor' status. ----- IMPORTANT: this is the only function that should be used for this purpose+-- __IMPORTANT__ This is the only function that should be used for this purpose -- given it has the side-effect of notifying a status change via the--- "notifyEvent" sub-routine, orginally given in the "CapatazOption" record.-writeSupervisorStatus :: SupervisorEnv -> SupervisorStatus -> IO ()+-- 'notifyEvent' sub-routine, orginally given in the 'CapatazOption' record.+writeSupervisorStatus+  :: MonadIO m => SupervisorEnv m -> SupervisorStatus -> m () writeSupervisorStatus SupervisorEnv { supervisorId, supervisorName, supervisorStatusVar, notifyEvent } newSupervisorStatus   = do @@ -142,9 +144,9 @@       , eventTime       } --- | Used from public API functions to send "ControlAction" messages to a+-- | Used from public API functions to send 'ControlAction' messages to a -- Supervisor thread loop.-sendControlMsg :: SupervisorEnv -> ControlAction -> IO ()+sendControlMsg :: SupervisorEnv m -> ControlAction m -> m () sendControlMsg SupervisorEnv { supervisorNotify } ctrlMsg =   supervisorNotify (ControlAction ctrlMsg) @@ -152,22 +154,23 @@ -- Supervisor thread loop, it receives an IO sub-routine that expects an IO -- operation that blocks a thread until the message is done. sendSyncControlMsg-  :: SupervisorEnv-  -> (IO () -> ControlAction) -- ^ Blocking sub-routine used from the caller-  -> IO ()+  :: MonadIO m+  => SupervisorEnv m+  -> (m () -> ControlAction m) -- ^ Blocking sub-routine used from the caller+  -> m () sendSyncControlMsg SupervisorEnv { supervisorNotify } mkCtrlMsg = do   result <- newEmptyMVar   supervisorNotify (ControlAction $ mkCtrlMsg (putMVar result ()))   takeMVar result --- | Utility function to transform a "CapatazOptions" record to a--- "SupervisorOptions" record.-capatazOptionsToSupervisorOptions :: CapatazOptions -> SupervisorOptions+-- | Utility function to transform a 'CapatazOptions' record to a+-- 'SupervisorOptions' record.+capatazOptionsToSupervisorOptions :: CapatazOptions m -> SupervisorOptions m capatazOptionsToSupervisorOptions CapatazOptions {..} = SupervisorOptions {..} --- | Utility function to transform a "SupervisorEnv" record to a--- "ParentSupervisorEnv" record; used on functions where supervision of+-- | Utility function to transform a 'SupervisorEnv' record to a+-- 'ParentSupervisorEnv' record; used on functions where supervision of -- supervisors is managed.-toParentSupervisorEnv :: SupervisorEnv -> ParentSupervisorEnv+toParentSupervisorEnv :: SupervisorEnv m -> ParentSupervisorEnv m toParentSupervisorEnv SupervisorEnv { supervisorId, supervisorName, supervisorNotify, notifyEvent }   = ParentSupervisorEnv {..}

@@ -7,10 +7,9 @@ -} module Control.Concurrent.Capataz.Internal.Worker where -import Protolude+import RIO -import Control.Concurrent.Async (asyncWithUnmask)-import Data.Time.Clock          (getCurrentTime)+import RIO.Time (getCurrentTime)  import qualified Data.UUID.V4 as UUID @@ -21,19 +20,18 @@  -- | Decorates the given @IO ()@ sub-routine with failure handling workerMain-  :: ParentSupervisorEnv-  -> WorkerOptions+  :: (MonadUnliftIO m)+  => ParentSupervisorEnv m+  -> WorkerOptions m   -> WorkerId   -> RestartCount-  -> IO Worker+  -> m (Worker m) workerMain env@ParentSupervisorEnv { supervisorNotify } workerOptions@WorkerOptions { workerName, workerAction } workerId restartCount   = do     workerCreationTime <- getCurrentTime     workerAsync        <- asyncWithUnmask $ \unmask -> do--      eResult <- try $ do-        Util.setProcessThreadName workerId workerName-        unmask workerAction+      Util.setProcessThreadName workerId workerName+      eResult     <- unsafeTry $ unmask (workerAction workerId)        resultEvent <- case eResult of         Left err -> Process.handleProcessException unmask@@ -62,14 +60,15 @@ -- this is different from the public @forkWorker@ function which sends a message -- to the capataz loop forkWorker-  :: ParentSupervisorEnv-  -> WorkerOptions+  :: (MonadUnliftIO m)+  => ParentSupervisorEnv m+  -> WorkerOptions m   -> Maybe (WorkerId, RestartCount)-  -> IO Worker+  -> m (Worker m) forkWorker env workerOptions mRestartInfo = do   (workerId, restartCount) <- case mRestartInfo of     Just (workerId, restartCount) -> pure (workerId, restartCount)-    Nothing                       -> (,) <$> UUID.nextRandom <*> pure 0+    Nothing                       -> (,) <$> liftIO UUID.nextRandom <*> pure 0    worker <- workerMain env workerOptions workerId restartCount   Process.notifyProcessStarted mRestartInfo env (WorkerProcess worker)

@@ -17,5 +17,4 @@     , workerRestartStrategyL     , workerTerminationPolicyL     )-import Lens.Micro                                     as X (set, (&), (.~), (^.))-import Lens.Micro.Extras                              as X (view)+import RIO                                            as X (set, view, (&), (^.))

@@ -0,0 +1,141 @@+{-# LANGUAGE NamedFieldPuns    #-}++{-# LANGUAGE NoImplicitPrelude #-}+module Control.Concurrent.Capataz.Util (+    buildLogWorkerSpec+  , buildLogWorkerOptions+  ) where++import Control.Concurrent.Capataz+    ( ProcessSpec+    , WorkerName+    , WorkerOptions+    , WorkerRestartStrategy (Permanent)+    , buildWorkerOptions+    , workerRestartStrategyL+    , workerSpec+    )+import RIO++data LogMsg+  = LogMsg+  {+    lmCallStack :: !CallStack+  , lmLogSource :: !LogSource+  , lmLogLevel  :: !LogLevel+  , lmPayload   :: !Utf8Builder+  }+++runLoggerThread :: MonadUnliftIO m => LogOptions -> TBQueue LogMsg -> m a+runLoggerThread logOptions inputQueue = withLogFunc logOptions $ \logFunc ->+  runRIO logFunc $ forever $ do+    logMsg <- atomically $ readTBQueue inputQueue+    let LogMsg { lmLogSource, lmLogLevel, lmPayload } = logMsg+    -- TODO: create a ticket for a logGenericWithStack function in rio+    logGeneric lmLogSource lmLogLevel lmPayload++-- | Builds a 'ProcessSpec' that spawns a thread that logs messages written with+-- the returned 'LogFunc'. Use this function when your want your logger to be+-- part of a static supervision tree.+--+--+-- __IMPORTANT__ If you use the returned 'LogFunc' to log functions and the+-- 'ProcessSpec' is not used in a supervision tree, your logging won't work and+-- your application will eventually block the current thread when logging.+--+-- A minimal example:+--+-- @+-- {-# LANGUAGE OverloadedStrings #-}+-- {-# LANGUAGE NoImplicitPrelude #-}+-- import RIO+-- import Capataz+--+-- main :: IO ()+-- main = do+--   logOptions <- logOptionsHandle stdout True+--   (loggerSpec, logFunc) <- buildLogWorkerSpec logOptions "app-logger" 100 id+--   runRIO logFunc $ do+--     bracket (forkCapataz "application" (set supervisorProcessSpecListL [loggerSpec]))+--             terminateCapataz_ $ \_capataz -> do+--       logInfo "this log message is written by a dedicated supervised thread"+--       threadDelay 1000100+-- @+--+-- @since 0.2.0.0+buildLogWorkerSpec+  :: (MonadUnliftIO m, MonadIO m0)+  => LogOptions  -- ^ options for the 'LogFunc' instance+  -> WorkerName  -- ^ name of the logger worker process+  -> Int         -- ^ how many log messages can be in-flight when writer is slow?+  -> (WorkerOptions m -> WorkerOptions m) -- ^ worker process modifier+  -> m0 (ProcessSpec m, LogFunc)+buildLogWorkerSpec logOptions procName bufferSize modOptions = do+  inputQueue <- newTBQueueIO bufferSize+  let myLogFunc =+        mkLogFunc $ \lmCallStack lmLogSource lmLogLevel lmPayload -> atomically+          (writeTBQueue+            inputQueue+            LogMsg {lmCallStack , lmLogSource , lmLogLevel , lmPayload }+          )++      loggerSpec = workerSpec+        procName+        (runLoggerThread logOptions inputQueue)+        (modOptions . set workerRestartStrategyL Permanent)++  return (loggerSpec, myLogFunc)++-- | Builds a 'WorkerOptions' record that spawns a thread that logs messages+-- written with the returned 'LogFunc'. Use this function if you want to build a+-- logger thread dynamically via 'forkWorker'.+--+--+-- __IMPORTANT__ If you use the returned 'LogFunc' to log functions and the+-- 'WorkerOptions' is not used in a 'forkWorker' call, your logging won't work+-- and your application will eventually block the current thread when logging.+--+-- A minimal example:+--+-- @+-- {-# LANGUAGE OverloadedStrings #-}+-- {-# LANGUAGE NoImplicitPrelude #-}+-- import RIO+-- import Capataz+--+-- main :: IO ()+-- main = do+--   logOptions <- logOptionsHandle stdout True+--   (loggerOptions, logFunc) <- buildLogWorkerOptions logOptions "app-logger" 100 id+--   runRIO logFunc $ do+--     bracket (forkCapataz "application" id)+--             terminateCapataz_ $ \capataz -> do+--       _workerId <- forkWorker loggerOptions capataz+--       logInfo "this log message is written by a dedicated supervised thread"+--       threadDelay 1000100+-- @+--+-- @since 0.2.0.0+buildLogWorkerOptions+  :: (MonadUnliftIO m, MonadIO m0)+  => LogOptions+  -> WorkerName+  -> Int+  -> (WorkerOptions m -> WorkerOptions m)+  -> m0 (WorkerOptions m, LogFunc)+buildLogWorkerOptions logOptions procName bufferSize modOptions = do+  inputQueue <- newTBQueueIO bufferSize+  let myLogFunc =+        mkLogFunc $ \lmCallStack lmLogSource lmLogLevel lmPayload -> atomically+          (writeTBQueue+            inputQueue+            LogMsg {lmCallStack , lmLogSource , lmLogLevel , lmPayload }+          )++      loggerSpec = buildWorkerOptions+        procName+        (runLoggerThread logOptions inputQueue)+        (modOptions . set workerRestartStrategyL Permanent)++  return (loggerSpec, myLogFunc)

@@ -3,7 +3,7 @@ {-# LANGUAGE OverloadedStrings     #-} module Control.Concurrent.Capataz.SupervisorTest where -import Protolude+import RIO  import           Control.Concurrent.Capataz (set) import qualified Control.Concurrent.Capataz as SUT@@ -17,11 +17,11 @@   "supervision trees"   [ testCase "initialize and teardown of supervision tree works as expected"     $ testCapatazStreamWithOptions-        ( & set+        (& set           SUT.supervisorProcessSpecListL           [ SUT.supervisorSpec             "tree-1"-            ( set+            (set               SUT.supervisorProcessSpecListL               [ SUT.workerSpecWithDefaults "1-A"                                            (forever $ threadDelay 10001000)@@ -31,7 +31,7 @@             )           , SUT.supervisorSpec             "tree-2"-            ( set+            (set               SUT.supervisorProcessSpecListL               [ SUT.workerSpecWithDefaults "2-A"                                            (forever $ threadDelay 10001000)@@ -85,7 +85,7 @@   , testCase "supervision sub-tree gets restarted on failure" $ do     failingAction <- mkFailingSubRoutine 2     testCapatazStreamWithOptions-      ( set SUT.supervisorIntensityL 3 . set+      (set SUT.supervisorIntensityL 3 . set         SUT.supervisorProcessSpecListL         [ SUT.supervisorSpec             "tree-1"@@ -123,11 +123,11 @@             )           , SUT.supervisorSpec             "tree-2"-            ( set+            (set               SUT.supervisorProcessSpecListL               [ SUT.workerSpec "stable-worker"                                (forever $ threadDelay 1000100)-                               identity+                               id               ]             )           ]

@@ -20,13 +20,13 @@ -} module Control.Concurrent.CapatazTest (tests) where -import Protolude+import RIO +import Control.Exception (ErrorCall (..))+ import Test.Tasty       (TestTree, testGroup) import Test.Tasty.HUnit (testCase) -import Control.Concurrent.STM.TVar (modifyTVar', newTVarIO, readTVar)- import           Control.Concurrent.Capataz (set) import qualified Control.Concurrent.Capataz as SUT @@ -63,10 +63,10 @@     "capataz with processSpecList"     [ testCase "initialize and teardown of workers works as expected"       $ testCapatazStreamWithOptions-          ( set+          (set             SUT.supervisorProcessSpecListL-            [ SUT.workerSpec "A" (forever $ threadDelay 10001000) identity-            , SUT.workerSpec "B" (forever $ threadDelay 10001000) identity+            [ SUT.workerSpec "A" (forever $ threadDelay 10001000) id+            , SUT.workerSpec "B" (forever $ threadDelay 10001000) id             ]           )           [ assertWorkerStarted "A"@@ -100,11 +100,11 @@           testCapatazStreamWithOptions             (set SUT.supervisorOnIntensityReachedL signalIntensityReached)             []-            ( \capataz -> do+            (\capataz -> do               _workerId <- SUT.forkWorker-                ( SUT.buildWorkerOptions "test-worker"-                                         (throwIO RestartingWorkerError)-                                         identity+                (SUT.buildWorkerOptions "test-worker"+                                        (throwIO RestartingWorkerError)+                                        id                 )                 capataz               waitTillIntensityReached@@ -126,9 +126,9 @@               "does execute callback when worker sub-routine is completed"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   _workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (return ())                       (set SUT.workerRestartStrategyL SUT.Temporary)@@ -147,11 +147,11 @@           , testCase "does not execute callback when worker sub-routine fails"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   _workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions "test-worker"-                                             (throwIO RestartingWorkerError)-                                             identity+                    (SUT.buildWorkerOptions "test-worker"+                                            (throwIO RestartingWorkerError)+                                            id                     )                     capataz                   return ()@@ -163,7 +163,7 @@                 , assertEventType ProcessFailed                 ]                 [assertEventType CapatazTerminated]-                ( Just $ not . andP+                (Just $ not . andP                   [ assertEventType ProcessCallbackExecuted                   , assertCallbackType SUT.OnCompletion                   ]@@ -172,9 +172,9 @@               "does not execute callback when worker sub-routine is terminated"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (forever $ threadDelay 1000100)                       (set SUT.workerRestartStrategyL SUT.Temporary)@@ -189,7 +189,7 @@                 )                 [assertEventType ProcessTerminated]                 [assertEventType CapatazTerminated]-                ( Just $ not . andP+                (Just $ not . andP                   [ assertEventType ProcessCallbackExecuted                   , assertCallbackType SUT.OnCompletion                   ]@@ -197,9 +197,9 @@           , testCase "treats as worker sub-routine failed if callback fails"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   _workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (return ())                       ( set SUT.workerRestartStrategyL SUT.Temporary@@ -228,9 +228,9 @@           [ testCase "does execute callback when worker sub-routine fails"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   _workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (throwIO RestartingWorkerError)                       (set SUT.workerRestartStrategyL SUT.Temporary)@@ -250,9 +250,9 @@               "does not execute callback when worker sub-routine is completed"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   _workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (return ())                       (set SUT.workerRestartStrategyL SUT.Temporary)@@ -278,9 +278,9 @@               "does not execute callback when worker sub-routine is terminated"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (forever $ threadDelay 1000100)                       (set SUT.workerRestartStrategyL SUT.Temporary)@@ -303,9 +303,9 @@           , testCase "treats as worker sub-routine failed if callback fails"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   _workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (throwIO RestartingWorkerError)                       ( set SUT.workerRestartStrategyL SUT.Temporary@@ -337,9 +337,9 @@               "gets brutally killed when TimeoutSeconds termination policy is not met"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (forever $ threadDelay 10001000)                       ( set SUT.workerRestartStrategyL   SUT.Temporary@@ -372,9 +372,9 @@               "does execute callback when worker sub-routine is terminated"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (forever $ threadDelay 1000100)                       (set SUT.workerRestartStrategyL SUT.Temporary)@@ -398,9 +398,9 @@               "does not execute callback when worker sub-routine is completed"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   _workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (return ())                       (set SUT.workerRestartStrategyL SUT.Temporary)@@ -410,7 +410,7 @@                 )                 [assertEventType ProcessCompleted]                 []-                ( Just $ not . andP+                (Just $ not . andP                   [ assertEventType ProcessCallbackExecuted                   , assertCallbackType SUT.OnTermination                   ]@@ -418,9 +418,9 @@           , testCase "does not execute callback when worker sub-routine fails"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   _workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (throwIO (ErrorCall "surprise!"))                       (set SUT.workerRestartStrategyL SUT.Temporary)@@ -430,7 +430,7 @@                 )                 [assertEventType ProcessFailed]                 []-                ( Just $ not . andP+                (Just $ not . andP                   [ assertEventType ProcessCallbackExecuted                   , assertCallbackType SUT.OnTermination                   ]@@ -438,9 +438,9 @@           , testCase "treats as worker sub-routine failed if callback fails"             $ testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (forever $ threadDelay 10001000)                       ( set SUT.workerRestartStrategyL SUT.Temporary@@ -472,9 +472,9 @@         "with transient strategy"         [ testCase "does not restart worker on completion" $ testCapatazStream           []-          ( \capataz -> do+          (\capataz -> do             _workerId <- SUT.forkWorker-              ( SUT.buildWorkerOptions+              (SUT.buildWorkerOptions                 "test-worker"                 (return ())                 (set SUT.workerRestartStrategyL SUT.Transient)@@ -487,9 +487,9 @@           (Just $ not . assertEventType ProcessRestarted)         , testCase "does not restart worker on termination" $ testCapatazStream           []-          ( \capataz -> do+          (\capataz -> do             workerId <- SUT.forkWorker-              ( SUT.buildWorkerOptions+              (SUT.buildWorkerOptions                 "test-worker"                 (forever $ threadDelay 1000100)                 (set SUT.workerRestartStrategyL SUT.Transient)@@ -502,10 +502,10 @@           (Just $ not . assertEventType ProcessRestarted)         , testCase "does restart on failure" $ testCapatazStream           []-          ( \capataz -> do+          (\capataz -> do             subRoutineAction <- mkFailingSubRoutine 1             _workerId        <- SUT.forkWorker-              ( SUT.buildWorkerOptions+              (SUT.buildWorkerOptions                 "test-worker"                 subRoutineAction                 (set SUT.workerRestartStrategyL SUT.Transient)@@ -522,10 +522,10 @@         , testCase "does increase restart count on multiple worker failures"           $ testCapatazStream               []-              ( \capataz -> do+              (\capataz -> do                 subRoutineAction <- mkFailingSubRoutine 2                 _workerId        <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "test-worker"                     subRoutineAction                     (set SUT.workerRestartStrategyL SUT.Transient)@@ -545,10 +545,10 @@         "with permanent strategy"         [ testCase "does restart worker on completion" $ testCapatazStream           []-          ( \capataz -> do+          (\capataz -> do             subRoutineAction <- mkCompletingOnceSubRoutine             _workerId        <- SUT.forkWorker-              ( SUT.buildWorkerOptions+              (SUT.buildWorkerOptions                 "test-worker"                 subRoutineAction                 (set SUT.workerRestartStrategyL SUT.Permanent)@@ -566,13 +566,13 @@             "does not increase restart count on multiple worker completions"           $ testCapatazStream               []-              ( \capataz -> do+              (\capataz -> do               -- Note the number is two (2) given the assertion list has two `ProcessRestarted` assertions                 let expectedRestartCount = 2                 subRoutineAction <- mkCompletingBeforeNRestartsSubRoutine                   expectedRestartCount                 _workerId <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "test-worker"                     subRoutineAction                     (set SUT.workerRestartStrategyL SUT.Permanent)@@ -589,9 +589,9 @@               Nothing         , testCase "does restart on worker termination" $ testCapatazStream           []-          ( \capataz -> do+          (\capataz -> do             workerId <- SUT.forkWorker-              ( SUT.buildWorkerOptions+              (SUT.buildWorkerOptions                 "test-worker"                 (forever $ threadDelay 10001000)                 (set SUT.workerRestartStrategyL SUT.Permanent)@@ -611,12 +611,12 @@                     atomically (modifyTVar' terminationCountVar (+ 1))                   waitWorkerTermination i = atomically $ do                     n <- readTVar terminationCountVar-                    when (n /= i) retry+                    when (n /= i) retrySTM               testCapatazStream                 []-                ( \capataz -> do+                (\capataz -> do                   workerId <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "test-worker"                       (forever $ threadDelay 10001000)                       ( set SUT.workerRestartStrategyL SUT.Permanent@@ -645,10 +645,10 @@                 Nothing         , testCase "does restart on worker failure" $ testCapatazStream           []-          ( \capataz -> do+          (\capataz -> do             subRoutineAction <- mkFailingSubRoutine 1             _workerId        <- SUT.forkWorker-              ( SUT.buildWorkerOptions+              (SUT.buildWorkerOptions                 "test-worker"                 subRoutineAction                 (set SUT.workerRestartStrategyL SUT.Permanent)@@ -665,10 +665,10 @@         , testCase "does increase restart count on multiple worker failures"           $ testCapatazStream               []-              ( \capataz -> do+              (\capataz -> do                 subRoutineAction <- mkFailingSubRoutine 2                 _workerId        <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "test-worker"                     subRoutineAction                     (set SUT.workerRestartStrategyL SUT.Permanent)@@ -688,9 +688,9 @@         "with temporary strategy"         [ testCase "does not restart on worker completion" $ testCapatazStream           []-          ( \capataz -> do+          (\capataz -> do             _workerId <- SUT.forkWorker-              ( SUT.buildWorkerOptions+              (SUT.buildWorkerOptions                 "test-worker"                 (return ())                 (set SUT.workerRestartStrategyL SUT.Temporary)@@ -703,9 +703,9 @@           (Just $ not . assertEventType ProcessRestarted)         , testCase "does not restart on worker termination" $ testCapatazStream           []-          ( \capataz -> do+          (\capataz -> do             workerId <- SUT.forkWorker-              ( SUT.buildWorkerOptions+              (SUT.buildWorkerOptions                 "test-worker"                 (forever $ threadDelay 1000100)                 (set SUT.workerRestartStrategyL SUT.Temporary)@@ -719,11 +719,11 @@           (Just $ not . assertEventType ProcessRestarted)         , testCase "does not restart on worker failure" $ testCapatazStream           []-          ( \capataz -> do+          (\capataz -> do             _workerId <- SUT.forkWorker-              ( SUT.buildWorkerOptions+              (SUT.buildWorkerOptions                 "failing-worker"-                (panic "worker failed!")+                (error "worker failed!")                 (set SUT.workerRestartStrategyL SUT.Temporary)               )               capataz@@ -739,9 +739,9 @@       [ testCase "terminates all supervised worker sub-routines on teardown"         $ testCapatazStream             []-            ( \capataz -> do+            (\capataz -> do               _workerA <- SUT.forkWorker-                ( SUT.buildWorkerOptions+                (SUT.buildWorkerOptions                   "A"                   (forever $ threadDelay 1000100)                   (set SUT.workerRestartStrategyL SUT.Permanent)@@ -750,7 +750,7 @@                 _workerB <- SUT.forkWorker-                ( SUT.buildWorkerOptions+                (SUT.buildWorkerOptions                   "B"                   (forever $ threadDelay 1000100)                   (set SUT.workerRestartStrategyL SUT.Permanent)@@ -773,9 +773,9 @@             $ testCapatazStreamWithOptions                 (set SUT.supervisorRestartStrategyL SUT.OneForOne)                 []-                ( \capataz -> do+                (\capataz -> do                   _workerA <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "A"                       (forever $ threadDelay 1000100)                       (set SUT.workerRestartStrategyL SUT.Temporary)@@ -785,7 +785,7 @@                   ioB      <- mkFailingSubRoutine 1                    _workerB <- SUT.forkWorker-                    ( SUT.buildWorkerOptions+                    (SUT.buildWorkerOptions                       "B"                       (forever $ ioB >> threadDelay 1000100)                       (set SUT.workerRestartStrategyL SUT.Permanent)@@ -796,7 +796,7 @@                 )                 [andP [assertEventType ProcessRestarted, assertProcessName "B"]]                 []-                ( Just $ not . andP+                (Just $ not . andP                   [assertEventType ProcessRestarted, assertProcessName "A"]                 )         ]@@ -804,20 +804,20 @@         "with all for one supervisor restart strategy with newest first order"         [ testCase "does terminate all other workers that did not fail"           $ testCapatazStreamWithOptions-              ( \supOptions ->+              (\supOptions ->                 supOptions                   & set SUT.supervisorRestartStrategyL         SUT.AllForOne                   & set SUT.supervisorProcessTerminationOrderL SUT.OldestFirst               )               []-              ( \capataz -> do+              (\capataz -> do               -- This lockVar guarantees that workerB executes before workerA                 lockVar  <- newEmptyMVar                  ioA      <- mkFailingSubRoutine 1                  _workerA <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "A"                     (forever $ readMVar lockVar >> ioA)                     (set SUT.workerRestartStrategyL SUT.Permanent)@@ -825,7 +825,7 @@                   capataz                  _workerB <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "B"                     (putMVar lockVar () >> forever (threadDelay 10))                     (set SUT.workerRestartStrategyL SUT.Permanent)@@ -845,19 +845,19 @@               Nothing         , testCase "does not restart workers that are temporary"           $ testCapatazStreamWithOptions-              ( \supOptions ->+              (\supOptions ->                 supOptions                   & set SUT.supervisorRestartStrategyL         SUT.AllForOne                   & set SUT.supervisorProcessTerminationOrderL SUT.OldestFirst               )               []-              ( \capataz -> do+              (\capataz -> do                 lockVar  <- newEmptyMVar                  ioA      <- mkFailingSubRoutine 1                  _workerA <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "A"                     (forever $ readMVar lockVar >> ioA)                     (set SUT.workerRestartStrategyL SUT.Permanent)@@ -865,7 +865,7 @@                   capataz                  _workerB <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "B"                     (putMVar lockVar () >> forever (threadDelay 10))                     (set SUT.workerRestartStrategyL SUT.Temporary)@@ -881,24 +881,24 @@               , andP [assertEventType ProcessTerminated, assertProcessName "B"]               ]               []-              ( Just $ not . andP+              (Just $ not . andP                 [assertEventType ProcessRestarted, assertProcessName "B"]               )         , testCase "restarts workers that are not temporary"           $ testCapatazStreamWithOptions-              ( \supOptions ->+              (\supOptions ->                 supOptions                   & set SUT.supervisorRestartStrategyL         SUT.AllForOne                   & set SUT.supervisorProcessTerminationOrderL SUT.NewestFirst               )               []-              ( \capataz -> do+              (\capataz -> do                 ioA      <- mkFailingSubRoutine 1                  lockVar  <- newEmptyMVar                  _workerA <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "A"                     (forever $ readMVar lockVar >> ioA)                     (set SUT.workerRestartStrategyL SUT.Permanent)@@ -906,7 +906,7 @@                   capataz                  _workerB <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "B"                     (putMVar lockVar () >> forever (threadDelay 10))                     (set SUT.workerRestartStrategyL SUT.Transient)@@ -925,20 +925,20 @@         "with all for one capataz restart strategy with oldest first order"         [ testCase "does not restart workers that are temporary"           $ testCapatazStreamWithOptions-              ( \supOptions ->+              (\supOptions ->                 supOptions                   & set SUT.supervisorRestartStrategyL         SUT.AllForOne                   & set SUT.supervisorProcessTerminationOrderL SUT.OldestFirst               )               []-              ( \capataz -> do+              (\capataz -> do                 ioA      <- mkFailingSubRoutine 1                  -- This lockVar guarantees that workerB executes before workerA                 lockVar  <- newEmptyMVar                  _workerA <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "A"                     (forever $ readMVar lockVar >> ioA)                     (set SUT.workerRestartStrategyL SUT.Permanent)@@ -946,7 +946,7 @@                   capataz                  _workerB <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "B"                     (putMVar lockVar () >> forever (threadDelay 10))                     (set SUT.workerRestartStrategyL SUT.Temporary)@@ -957,25 +957,25 @@               )               [andP [assertEventType ProcessRestarted, assertProcessName "A"]]               []-              ( Just $ not . andP+              (Just $ not . andP                 [assertEventType ProcessRestarted, assertProcessName "B"]               )         , testCase "restarts workers that are not temporary"           $ testCapatazStreamWithOptions-              ( \supOptions ->+              (\supOptions ->                 supOptions                   & set SUT.supervisorRestartStrategyL         SUT.AllForOne                   & set SUT.supervisorProcessTerminationOrderL SUT.OldestFirst               )               []-              ( \capataz -> do+              (\capataz -> do                 ioA      <- mkFailingSubRoutine 1                  -- This lockVar guarantees that workerB executes before workerA                 lockVar  <- newEmptyMVar                  _workerA <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "A"                     (forever $ readMVar lockVar >> ioA)                     (set SUT.workerRestartStrategyL SUT.Permanent)@@ -983,7 +983,7 @@                   capataz                  _workerB <- SUT.forkWorker-                  ( SUT.buildWorkerOptions+                  (SUT.buildWorkerOptions                     "B"                     (putMVar lockVar () >> forever (threadDelay 10))                     (set SUT.workerRestartStrategyL SUT.Transient)

@@ -5,17 +5,16 @@ {-# LANGUAGE ScopedTypeVariables   #-} module Main where -import Protolude+import RIO  import qualified Control.Concurrent.Capataz.SupervisorTest as Supervisor import qualified Control.Concurrent.CapatazTest            as Capataz import           Test.Tasty     (defaultMainWithIngredients, testGroup)-import           Test.Tasty.Ingredients.Rerun              (rerunningTests) import           Test.Tasty.Runners     (consoleTestReporter, listingTests)  main :: IO () main = defaultMainWithIngredients-  [rerunningTests [listingTests, consoleTestReporter]]+  [listingTests, consoleTestReporter]   (testGroup "capataz" [Capataz.tests, Supervisor.tests])

@@ -4,14 +4,12 @@ {-# LANGUAGE OverloadedStrings     #-} module Test.Util where -import Protolude+import           RIO+import qualified RIO.Text as T -import Control.Concurrent.STM.TQueue (newTQueueIO, readTQueue, writeTQueue)-import Data.IORef                    (atomicModifyIORef', newIORef, readIORef, writeIORef)-import Test.Tasty.HUnit              (assertBool, assertFailure)-import Text.Show.Pretty              (ppShow)+import Test.Tasty.HUnit (assertBool, assertFailure)+import Text.Show.Pretty (ppShow) -import qualified Data.Text as T  import qualified Control.Concurrent.Capataz       as SUT import qualified Control.Concurrent.Capataz.Event as SUT@@ -22,7 +20,7 @@ -- | Utility function that gets the type name of a Record through it's Show -- output. fetchRecordName :: Show a => a -> Text-fetchRecordName = T.takeWhile (/= ' ') . show+fetchRecordName = T.takeWhile (/= ' ') . tshow  -- | Composes two predicate functions together with a boolean AND andP :: [a -> Bool] -> a -> Bool@@ -235,14 +233,15 @@ -- | Utility function to create a Worker sub-routine that fails at least a -- number of times mkFailingSubRoutine-  :: Int  -- ^ Number of times the Worker sub-routine will fail-  -> IO (IO ()) -- ^ Sub-routine used on worker creation+  :: (MonadIO m, MonadThrow m1, MonadIO m1)+  => Int  -- ^ Number of times the Worker sub-routine will fail+  -> m (m1 ()) -- ^ Sub-routine used on worker creation mkFailingSubRoutine failCount = do   countRef <- newIORef failCount   let subRoutine = do         shouldFail <- atomicModifyIORef' countRef-                                         (\count -> (pred count, count > 0))-        when shouldFail (throwIO RestartingWorkerError)+                                         (\count -> (count - 1, count > 0))+        when shouldFail (throwM RestartingWorkerError)    return subRoutine @@ -250,19 +249,19 @@ -- function works great when testing `Permanent` strategies, as you would like -- to assert restart events once (if it keeps completing it will fill up the log -- with restart events)-mkCompletingBeforeNRestartsSubRoutine :: Int -> IO (IO ())+mkCompletingBeforeNRestartsSubRoutine :: MonadIO m => Int -> m (m ()) mkCompletingBeforeNRestartsSubRoutine initCount = do   countRef <- newIORef initCount   let subRoutine = do         shouldStop <- atomicModifyIORef' countRef-                                         (\count -> (pred count, count > 0))+                                         (\count -> (count - 1, count > 0))         if shouldStop then return () else forever $ threadDelay 1000100   return subRoutine  -- | A sub-routine that will complete once. This function works great when -- testing `Permanent` strategies, as you would like to assert restart events -- once (if it keeps completing it will fill up the log with restart events)-mkCompletingOnceSubRoutine :: IO (IO ())+mkCompletingOnceSubRoutine :: MonadIO m => m (m ()) mkCompletingOnceSubRoutine = mkCompletingBeforeNRestartsSubRoutine 1  -- | Utility function to build a test environment for a Capataz execution.@@ -274,11 +273,12 @@ -- * A function to modify the default "CapatazOptions", this utility function injects --   a special @notifyEvent@ callback to execute given assertions. testCapatazStreamWithOptions-  :: (SUT.CapatazOptions -> SUT.CapatazOptions) -- ^ Function to modify default+  :: (MonadUnliftIO m)+  => (SUT.CapatazOptions (RIO LogFunc) -> SUT.CapatazOptions (RIO LogFunc)) -- ^ Function to modify default                                                 -- @CapatazOptions@   -> [SUT.CapatazEvent -> Bool] -- ^ Assertions happening before setup function                                 -- is called-  -> (SUT.Capataz -> IO ()) -- ^ Function used to test public the supervisor+  -> (SUT.Capataz (RIO LogFunc) -> RIO LogFunc ()) -- ^ Function used to test public the supervisor                             -- public API (a.k.a setup function)   -> [SUT.CapatazEvent -> Bool] -- ^ Assertions happening after the setup                                 -- function@@ -288,69 +288,75 @@                                       -- @CapatazEvents@ that happened in a                                       -- test, great when testing that an event                                       -- __did not__ happen-  -> IO ()+  -> m () testCapatazStreamWithOptions optionModFn preSetupAssertion setupFn postSetupAssertions postTeardownAssertions mAllEventsAssertion   = do      eventStream     <- newTQueueIO     accRef          <- newIORef []     pendingCountVar <- newIORef-      ( sum $ fmap+      (sum $ fmap         length         [preSetupAssertion, postSetupAssertions, postTeardownAssertions]       ) -    capataz <- SUT.forkCapataz-      rootSupervisorName-      (SUT.set SUT.onSystemEventL (trackEvent accRef eventStream) . optionModFn)+    logOptions <- logOptionsHandle stdout False+    withLogFunc logOptions $ \logFunc -> runRIO logFunc $ do+      capataz <- SUT.forkCapataz+        rootSupervisorName+        ( set+            SUT.onSystemEventL+            (\ev -> logDebug (display ev) >> trackEvent accRef eventStream ev)+        . optionModFn+        ) -    -- We check preSetup assertions are met before we execute the setup-    -- function. This serves to test initialization of capataz instance-    runAssertions "PRE-SETUP"-                  (eventStream, accRef)-                  pendingCountVar-                  preSetupAssertion-                  capataz+      -- We check preSetup assertions are met before we execute the setup+      -- function. This serves to test initialization of capataz instance+      runAssertions "PRE-SETUP"+                    (eventStream, accRef)+                    pendingCountVar+                    preSetupAssertion+                    capataz -    -- We execute the setup sub-routine, which is going to use the Capataz public-    -- API to assert events-    setupResult <- try (setupFn capataz)+      -- We execute the setup sub-routine, which is going to use the Capataz public+      -- API to assert events+      setupResult <- try (setupFn capataz) -    case setupResult of-      -- If the sub-routine fails, show exception-      Left  err -> assertFailure (show (err :: SomeException))-      Right _   -> do-        -- We now run post-setup assertions-        runAssertions "POST-SETUP"-                      (eventStream, accRef)-                      pendingCountVar-                      postSetupAssertions-                      capataz+      case setupResult of+        -- If the sub-routine fails, show exception+        Left  err -> liftIO $ assertFailure (show (err :: SomeException))+        Right _   -> do+          -- We now run post-setup assertions+          runAssertions "POST-SETUP"+                        (eventStream, accRef)+                        pendingCountVar+                        postSetupAssertions+                        capataz -        -- We now shutdown the capataz instance-        void $ SUT.teardown capataz+          -- We now shutdown the capataz instance+          SUT.terminateCapataz_ capataz -        -- We run assertions for after the capataz has been shut down-        runAssertions "POST-TEARDOWN"-                      (eventStream, accRef)-                      pendingCountVar-                      postTeardownAssertions-                      capataz+          -- We run assertions for after the capataz has been shut down+          runAssertions "POST-TEARDOWN"+                        (eventStream, accRef)+                        pendingCountVar+                        postTeardownAssertions+                        capataz -        -- Lastly, we check if there is a function that we want to execute-        -- across all events that happened in the test, this is to assert the-        -- absence of an event-        case mAllEventsAssertion of-          Nothing                 -> return ()-          Just allEventsAssertion -> do-            events <- reverse <$> readIORef accRef-            assertBool-              ( "On AFTER-TEST, expected all events to match predicate, but didn't ("-              <> show (length events)-              <> " events tried)\n"-              <> ppShow (zip ([0 ..] :: [Int]) events)-              )-              (all allEventsAssertion events)+          -- Lastly, we check if there is a function that we want to execute+          -- across all events that happened in the test, this is to assert the+          -- absence of an event+          case mAllEventsAssertion of+            Nothing                 -> return ()+            Just allEventsAssertion -> do+              events <- reverse <$> readIORef accRef+              liftIO $ assertBool+                ("On AFTER-TEST, expected all events to match predicate, but didn't ("+                <> show (length events)+                <> " events tried)\n"+                <> ppShow (zip ([0 ..] :: [Int]) events)+                )+                (all allEventsAssertion events)  where   -- Utility functions that runs the readEventLoop function with a timeout   -- of a second, this way we can guarantee assertions are met without having@@ -358,14 +364,14 @@   runAssertions stageName (eventStream, accRef) pendingCountVar assertions capataz     = do       raceResult <- race-        (threadDelay 1000100)+        (liftIO $ threadDelay 1000100)         (readEventLoop eventStream pendingCountVar assertions)       case raceResult of         Left _ -> do           events       <- reverse <$> readIORef accRef           pendingCount <- readIORef pendingCountVar-          void $ SUT.teardown capataz-          assertFailure+          SUT.terminateCapataz_ capataz+          liftIO $ assertFailure             (  "On "             <> stageName             <> " stage, expected all assertions to match, but didn't ("@@ -389,8 +395,8 @@   readEventLoop eventStream pendingCount assertions = do     writeIORef pendingCount (length assertions)     case assertions of-      []                        -> return ()-      (assertionFn:assertions1) -> do+      []                          -> return ()+      (assertionFn : assertions1) -> do         event <- atomically $ readTQueue eventStream         if assertionFn event           then readEventLoop eventStream pendingCount assertions1@@ -400,9 +406,10 @@ -- | A version of "testCapatazStreamWithOptions" that does not receive the -- function that modifies a "CapatazOptions" record. testCapatazStream-  :: [SUT.CapatazEvent -> Bool] -- ^ Assertions happening before setup function+  :: (MonadUnliftIO m)+  => [SUT.CapatazEvent -> Bool] -- ^ Assertions happening before setup function                                 -- is called-  -> (SUT.Capataz -> IO ()) -- ^ Function used to test public the supervisor+  -> (SUT.Capataz (RIO LogFunc) -> RIO LogFunc ()) -- ^ Function used to test public the supervisor                             -- public API (a.k.a setup function)   -> [SUT.CapatazEvent -> Bool] -- ^ Assertions happening after the setup                                 -- function@@ -412,6 +419,5 @@                                       -- @CapatazEvents@ that happened in a                                       -- test, great when testing that an event                                       -- __did not__ happen-  -> IO ()-testCapatazStream preSetupAssertions =-  testCapatazStreamWithOptions identity preSetupAssertions+  -> m ()+testCapatazStream = testCapatazStreamWithOptions id