{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TypeApplications #-}
{-# LANGUAGE TypeFamilies #-}

-- | Schema-wide maintenance coordinated across worker pools.
module Arbiter.Worker.Reaper
  ( reaperLoop
  , runMaintenancePass
  , MaintenancePace (..)
  , runReaperOp
  ) where

import Arbiter.Core.Concurrency.Spec (registryConcurrencyPolicies)
import Arbiter.Core.HighLevel qualified as Arb
import Arbiter.Core.Job.Schema (SchemaName)
import Arbiter.Core.MonadArbiter (MonadArbiter, RegistryOf)
import Arbiter.Core.Operations qualified as Ops
import Arbiter.Core.QueueRegistry (RegistryTables (..))
import Arbiter.Core.RateLimit.Spec (registryRateLimitPolicies)
import Control.Monad (forever, void, when)
import Data.Aeson (FromJSON, ToJSON)
import Data.Either (fromRight)
import Data.Foldable (fold, traverse_)
import Data.Int (Int64)
import Data.Maybe (catMaybes)
import Data.Proxy (Proxy (..))
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (NominalDiffTime)
import UnliftIO (MonadUnliftIO, SomeException, tryAny)
import UnliftIO.Concurrent (threadDelay)

import Arbiter.Worker.Config (MaintenanceOp (..), maintenanceOpName)
import Arbiter.Worker.Logger (LogConfig, LogLevel (..), tryLog, warnEx)
import Arbiter.Worker.Logger.Internal (runHook)

-- | Run schema-wide maintenance, with each operation independently gated
-- across all worker pools.
reaperLoop
  :: forall m
   . ( Arb.RegistryAdmissionPolicies (RegistryOf m)
     , MonadArbiter m
     , RegistryTables (RegistryOf m)
     )
  => LogConfig
  -> (MaintenanceOp -> Int64 -> m ())
  -> MaintenancePace
  -> NominalDiffTime
  -> m ()
reaperLoop :: forall (m :: * -> *).
(RegistryAdmissionPolicies (RegistryOf m), MonadArbiter m,
 RegistryTables (RegistryOf m)) =>
LogConfig
-> (MaintenanceOp -> Int64 -> m ())
-> MaintenancePace
-> NominalDiffTime
-> m ()
reaperLoop LogConfig
logCfg MaintenanceOp -> Int64 -> m ()
report MaintenancePace
pace NominalDiffTime
stmtTimeout =
  m () -> m ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
    m [MaintenanceOp] -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m [MaintenanceOp] -> m ()) -> m [MaintenanceOp] -> m ()
forall a b. (a -> b) -> a -> b
$ LogConfig
-> (MaintenanceOp -> Int64 -> m ())
-> MaintenancePace
-> NominalDiffTime
-> m [MaintenanceOp]
forall (m :: * -> *).
(RegistryAdmissionPolicies (RegistryOf m), MonadArbiter m,
 RegistryTables (RegistryOf m)) =>
LogConfig
-> (MaintenanceOp -> Int64 -> m ())
-> MaintenancePace
-> NominalDiffTime
-> m [MaintenanceOp]
runMaintenancePass LogConfig
logCfg MaintenanceOp -> Int64 -> m ()
report MaintenancePace
pace NominalDiffTime
stmtTimeout
    Int -> m ()
forall (m :: * -> *). MonadIO m => Int -> m ()
threadDelay (NominalDiffTime -> Int
forall b. Integral b => NominalDiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
ceiling (MaintenancePace -> NominalDiffTime
paceWindow MaintenancePace
pace) Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
1_000_000)

-- | Gaps a caller holds between runs of each kind of work. A zero gap runs it every pass.
data MaintenancePace = MaintenancePace
  { MaintenancePace -> NominalDiffTime
paceWindow :: NominalDiffTime
  -- ^ Gap between runs of one ordinary operation.
  , MaintenancePace -> NominalDiffTime
paceSparseWindow :: NominalDiffTime
  -- ^ Gap between runs of one whole-schema operation.
  , MaintenancePace -> NominalDiffTime
paceBucketIdle :: NominalDiffTime
  -- ^ Idle age at which a prune collects a rate-limit bucket.
  }
  deriving stock (MaintenancePace -> MaintenancePace -> Bool
(MaintenancePace -> MaintenancePace -> Bool)
-> (MaintenancePace -> MaintenancePace -> Bool)
-> Eq MaintenancePace
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: MaintenancePace -> MaintenancePace -> Bool
== :: MaintenancePace -> MaintenancePace -> Bool
$c/= :: MaintenancePace -> MaintenancePace -> Bool
/= :: MaintenancePace -> MaintenancePace -> Bool
Eq, Int -> MaintenancePace -> ShowS
[MaintenancePace] -> ShowS
MaintenancePace -> String
(Int -> MaintenancePace -> ShowS)
-> (MaintenancePace -> String)
-> ([MaintenancePace] -> ShowS)
-> Show MaintenancePace
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> MaintenancePace -> ShowS
showsPrec :: Int -> MaintenancePace -> ShowS
$cshow :: MaintenancePace -> String
show :: MaintenancePace -> String
$cshowList :: [MaintenancePace] -> ShowS
showList :: [MaintenancePace] -> ShowS
Show)

-- | One pass of the maintenance the reaper runs, with each operation independently
-- gated across all callers. An operation whose window has not elapsed is skipped.
-- Returns the operations that failed. A failure does not stop the pass.
runMaintenancePass
  :: forall m
   . ( Arb.RegistryAdmissionPolicies (RegistryOf m)
     , MonadArbiter m
     , RegistryTables (RegistryOf m)
     )
  => LogConfig
  -> (MaintenanceOp -> Int64 -> m ())
  -> MaintenancePace
  -> NominalDiffTime
  -> m [MaintenanceOp]
runMaintenancePass :: forall (m :: * -> *).
(RegistryAdmissionPolicies (RegistryOf m), MonadArbiter m,
 RegistryTables (RegistryOf m)) =>
LogConfig
-> (MaintenanceOp -> Int64 -> m ())
-> MaintenancePace
-> NominalDiffTime
-> m [MaintenanceOp]
runMaintenancePass LogConfig
logCfg MaintenanceOp -> Int64 -> m ()
report MaintenancePace
pace NominalDiffTime
stmtTimeout = do
  let reaped :: MaintenanceOp -> Int64 -> m ()
reaped MaintenanceOp
operation Int64
count = LogConfig -> Text -> m () -> m ()
forall (m :: * -> *).
MonadUnliftIO m =>
LogConfig -> Text -> m () -> m ()
runHook LogConfig
logCfg Text
"onMaintenance" (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ MaintenanceOp -> Int64 -> m ()
report MaintenanceOp
operation Int64
count
      queues :: [Text]
queues = Proxy (RegistryOf m) -> [Text]
forall (registry :: JobPayloadRegistry).
RegistryTables registry =>
Proxy registry -> [Text]
registryTableNames (forall (t :: JobPayloadRegistry). Proxy t
forall {k} (t :: k). Proxy t
Proxy @(RegistryOf m))
      window :: NominalDiffTime
window = MaintenancePace -> NominalDiffTime
paceWindow MaintenancePace
pace
      sparseWindow :: NominalDiffTime
sparseWindow = MaintenancePace -> NominalDiffTime
paceSparseWindow MaintenancePace
pace
      hasConcurrency :: Bool
hasConcurrency = Bool -> Bool
not (Set ConcurrencyPolicy -> Bool
forall a. Set a -> Bool
Set.null (forall (registry :: JobPayloadRegistry).
RegistryConcurrencyPolicies registry =>
Set ConcurrencyPolicy
registryConcurrencyPolicies @(RegistryOf m)))
      hasRateLimit :: Bool
hasRateLimit = Bool -> Bool
not (Set Policy -> Bool
forall a. Set a -> Bool
Set.null (forall (registry :: JobPayloadRegistry).
RegistryRateLimitPolicies registry =>
Set Policy
registryRateLimitPolicies @(RegistryOf m)))
  schema <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
Arb.getSchema
  let gatedCount MaintenanceOp
operation NominalDiffTime
every m Int64
work =
        LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> m Int64
-> m (Either SomeException (Maybe Int64))
forall (m :: * -> *) a.
MonadArbiter m =>
LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> m a
-> m (Either SomeException (Maybe a))
tryReaperOp LogConfig
logCfg Text
schema NominalDiffTime
stmtTimeout (MaintenanceOp -> Text
maintenanceOpName MaintenanceOp
operation) NominalDiffTime
every m Int64
work
          m (Either SomeException (Maybe Int64))
-> (Either SomeException (Maybe Int64) -> m (Maybe MaintenanceOp))
-> m (Maybe MaintenanceOp)
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= MaintenanceOp
-> (Maybe Int64 -> m ())
-> Either SomeException (Maybe Int64)
-> m (Maybe MaintenanceOp)
forall (m :: * -> *) a.
Monad m =>
MaintenanceOp
-> (a -> m ()) -> Either SomeException a -> m (Maybe MaintenanceOp)
reportOutcome MaintenanceOp
operation ((Int64 -> m ()) -> Maybe Int64 -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (MaintenanceOp -> Int64 -> m ()
reaped MaintenanceOp
operation))
      reportFailed MaintenanceOp
operation =
        (Text -> m ()) -> [Text] -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (\Text
queue -> LogConfig -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadUnliftIO m =>
LogConfig -> LogLevel -> Text -> m ()
tryLog LogConfig
logCfg LogLevel
Warning (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$ MaintenanceOp -> Text
maintenanceOpName MaintenanceOp
operation Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" failed for queue: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
queue)
      reportSwept MaintenanceOp
operation Int64 -> m ()
done =
        ((Int64, [Text]) -> m ()) -> Maybe (Int64, [Text]) -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (\(Int64
count, [Text]
failed) -> MaintenanceOp -> Int64 -> m ()
reaped MaintenanceOp
operation Int64
count m () -> m () -> m ()
forall a b. m a -> m b -> m b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> MaintenanceOp -> [Text] -> m ()
reportFailed MaintenanceOp
operation [Text]
failed m () -> m () -> m ()
forall a b. m a -> m b -> m b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int64
count Int64 -> Int64 -> Bool
forall a. Ord a => a -> a -> Bool
> Int64
0) (Int64 -> m ()
done Int64
count))
      sweep MaintenanceOp
operation NominalDiffTime
every Int64 -> m ()
done m (Int64, [Text])
work =
        LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> m (Int64, [Text])
-> m (Either SomeException (Maybe (Int64, [Text])))
forall (m :: * -> *) a.
MonadArbiter m =>
LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> m a
-> m (Either SomeException (Maybe a))
tryReaperOp LogConfig
logCfg Text
schema NominalDiffTime
stmtTimeout (MaintenanceOp -> Text
maintenanceOpName MaintenanceOp
operation) NominalDiffTime
every m (Int64, [Text])
work
          m (Either SomeException (Maybe (Int64, [Text])))
-> (Either SomeException (Maybe (Int64, [Text]))
    -> m (Maybe MaintenanceOp))
-> m (Maybe MaintenanceOp)
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= MaintenanceOp
-> (Maybe (Int64, [Text]) -> m ())
-> Either SomeException (Maybe (Int64, [Text]))
-> m (Maybe MaintenanceOp)
forall (m :: * -> *) a.
Monad m =>
MaintenanceOp
-> (a -> m ()) -> Either SomeException a -> m (Maybe MaintenanceOp)
reportOutcome MaintenanceOp
operation (MaintenanceOp -> (Int64 -> m ()) -> Maybe (Int64, [Text]) -> m ()
reportSwept MaintenanceOp
operation Int64 -> m ()
done)
      refreshGroups =
        LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> (Maybe (Map Text GroupsCursor)
    -> m ((Int64, [Text]), Map Text GroupsCursor))
-> m (Either SomeException (Maybe (Int64, [Text])))
forall s (m :: * -> *) a.
(FromJSON s, MonadArbiter m, ToJSON s) =>
LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> (Maybe s -> m (a, s))
-> m (Either SomeException (Maybe a))
runReaperStateOp
          LogConfig
logCfg
          Text
schema
          NominalDiffTime
stmtTimeout
          (MaintenanceOp -> Text
maintenanceOpName MaintenanceOp
RefreshGroups)
          NominalDiffTime
window
          (Text
-> [Text]
-> Map Text GroupsCursor
-> m ((Int64, [Text]), Map Text GroupsCursor)
forall (m :: * -> *).
MonadArbiter m =>
Text
-> [Text]
-> Map Text GroupsCursor
-> m ((Int64, [Text]), Map Text GroupsCursor)
Ops.refreshAllGroups Text
schema [Text]
queues (Map Text GroupsCursor
 -> m ((Int64, [Text]), Map Text GroupsCursor))
-> (Maybe (Map Text GroupsCursor) -> Map Text GroupsCursor)
-> Maybe (Map Text GroupsCursor)
-> m ((Int64, [Text]), Map Text GroupsCursor)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Maybe (Map Text GroupsCursor) -> Map Text GroupsCursor
forall m. Monoid m => Maybe m -> m
forall (t :: * -> *) m. (Foldable t, Monoid m) => t m -> m
fold)
          m (Either SomeException (Maybe (Int64, [Text])))
-> (Either SomeException (Maybe (Int64, [Text]))
    -> m (Maybe MaintenanceOp))
-> m (Maybe MaintenanceOp)
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= MaintenanceOp
-> (Maybe (Int64, [Text]) -> m ())
-> Either SomeException (Maybe (Int64, [Text]))
-> m (Maybe MaintenanceOp)
forall (m :: * -> *) a.
Monad m =>
MaintenanceOp
-> (a -> m ()) -> Either SomeException a -> m (Maybe MaintenanceOp)
reportOutcome MaintenanceOp
RefreshGroups (MaintenanceOp -> (Int64 -> m ()) -> Maybe (Int64, [Text]) -> m ()
reportSwept MaintenanceOp
RefreshGroups (m () -> Int64 -> m ()
forall a b. a -> b -> a
const (() -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())))
  fmap catMaybes . sequence $
    [ refreshGroups
    , gatedCount SweepStaleWorkers window $ Ops.sweepStaleWorkers schema
    , sweep
        SweepExhaustedJobs
        window
        (\Int64
count -> LogConfig -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadUnliftIO m =>
LogConfig -> LogLevel -> Text -> m ()
tryLog LogConfig
logCfg LogLevel
Warning (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$ Text
"Reaper moved " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int64 -> String
forall a. Show a => a -> String
show Int64
count) Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" exhausted job(s) to the DLQ")
        $ Ops.sweepExhaustedJobs schema queues
    , sweep
        SweepCancelledJobs
        window
        (\Int64
count -> LogConfig -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadUnliftIO m =>
LogConfig -> LogLevel -> Text -> m ()
tryLog LogConfig
logCfg LogLevel
Info (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$ Text
"Reaper deleted " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int64 -> String
forall a. Show a => a -> String
show Int64
count) Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" orphaned cancelled job(s)")
        $ Ops.sweepCancelledJobs schema queues
    ]
      <> [gatedCount PruneRateLimitBuckets sparseWindow (Arb.pruneRateLimitBuckets @m (paceBucketIdle pace)) | hasRateLimit]
      <> [gatedCount ReconcileConcurrencyStale window (Arb.reconcileConcurrencyCountsIfStale @m) | hasConcurrency]
      <> [gatedCount ReconcilePruneConcurrency sparseWindow (Arb.reconcileAndPruneConcurrency @m) | hasConcurrency]
      <> [ sweep
             PurgeArchives
             window
             (\Int64
count -> LogConfig -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadUnliftIO m =>
LogConfig -> LogLevel -> Text -> m ()
tryLog LogConfig
logCfg LogLevel
Info (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$ Text
"Reaper purged " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int64 -> String
forall a. Show a => a -> String
show Int64
count) Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" archived job(s)")
             $ Ops.purgeArchives schema queues
         ]

-- | Name a failed operation. Report a completed one through @emit@.
reportOutcome
  :: (Monad m)
  => MaintenanceOp
  -> (a -> m ())
  -> Either SomeException a
  -> m (Maybe MaintenanceOp)
reportOutcome :: forall (m :: * -> *) a.
Monad m =>
MaintenanceOp
-> (a -> m ()) -> Either SomeException a -> m (Maybe MaintenanceOp)
reportOutcome MaintenanceOp
operation a -> m ()
emit = (SomeException -> m (Maybe MaintenanceOp))
-> (a -> m (Maybe MaintenanceOp))
-> Either SomeException a
-> m (Maybe MaintenanceOp)
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (m (Maybe MaintenanceOp) -> SomeException -> m (Maybe MaintenanceOp)
forall a b. a -> b -> a
const (Maybe MaintenanceOp -> m (Maybe MaintenanceOp)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (MaintenanceOp -> Maybe MaintenanceOp
forall a. a -> Maybe a
Just MaintenanceOp
operation))) (\a
ran -> Maybe MaintenanceOp
forall a. Maybe a
Nothing Maybe MaintenanceOp -> m () -> m (Maybe MaintenanceOp)
forall a b. a -> m b -> m a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ a -> m ()
emit a
ran)

-- | Run one gated maintenance operation. Database statement timeouts bound
-- individual statements. Failures are logged and do not stop the loop.
runReaperOp
  :: (MonadArbiter m)
  => LogConfig
  -> SchemaName
  -> NominalDiffTime
  -> Text
  -> NominalDiffTime
  -> m a
  -> m (Maybe a)
runReaperOp :: forall (m :: * -> *) a.
MonadArbiter m =>
LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> m a
-> m (Maybe a)
runReaperOp LogConfig
logCfg Text
schema NominalDiffTime
stmtTimeout Text
task NominalDiffTime
every m a
work =
  Maybe a -> Either SomeException (Maybe a) -> Maybe a
forall b a. b -> Either a b -> b
fromRight Maybe a
forall a. Maybe a
Nothing (Either SomeException (Maybe a) -> Maybe a)
-> m (Either SomeException (Maybe a)) -> m (Maybe a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> m a
-> m (Either SomeException (Maybe a))
forall (m :: * -> *) a.
MonadArbiter m =>
LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> m a
-> m (Either SomeException (Maybe a))
tryReaperOp LogConfig
logCfg Text
schema NominalDiffTime
stmtTimeout Text
task NominalDiffTime
every m a
work

-- | 'runReaperOp', keeping the failure. @Right Nothing@ is an operation already running.
tryReaperOp
  :: (MonadArbiter m)
  => LogConfig
  -> SchemaName
  -> NominalDiffTime
  -> Text
  -> NominalDiffTime
  -> m a
  -> m (Either SomeException (Maybe a))
tryReaperOp :: forall (m :: * -> *) a.
MonadArbiter m =>
LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> m a
-> m (Either SomeException (Maybe a))
tryReaperOp LogConfig
logCfg Text
schema NominalDiffTime
stmtTimeout Text
task NominalDiffTime
every m a
work =
  LogConfig
-> Text -> m (Maybe a) -> m (Either SomeException (Maybe a))
forall (m :: * -> *) a.
MonadUnliftIO m =>
LogConfig -> Text -> m a -> m (Either SomeException a)
reaperGate LogConfig
logCfg Text
task (m (Maybe a) -> m (Either SomeException (Maybe a)))
-> m (Maybe a) -> m (Either SomeException (Maybe a))
forall a b. (a -> b) -> a -> b
$
    Text
-> Text -> NominalDiffTime -> NominalDiffTime -> m a -> m (Maybe a)
forall (m :: * -> *) a.
MonadArbiter m =>
Text
-> Text -> NominalDiffTime -> NominalDiffTime -> m a -> m (Maybe a)
Ops.runGatedBounded Text
schema Text
task NominalDiffTime
every NominalDiffTime
stmtTimeout m a
work

runReaperStateOp
  :: (FromJSON s, MonadArbiter m, ToJSON s)
  => LogConfig
  -> SchemaName
  -> NominalDiffTime
  -> Text
  -> NominalDiffTime
  -> (Maybe s -> m (a, s))
  -> m (Either SomeException (Maybe a))
runReaperStateOp :: forall s (m :: * -> *) a.
(FromJSON s, MonadArbiter m, ToJSON s) =>
LogConfig
-> Text
-> NominalDiffTime
-> Text
-> NominalDiffTime
-> (Maybe s -> m (a, s))
-> m (Either SomeException (Maybe a))
runReaperStateOp LogConfig
logCfg Text
schema NominalDiffTime
stmtTimeout Text
task NominalDiffTime
every Maybe s -> m (a, s)
work =
  LogConfig
-> Text -> m (Maybe a) -> m (Either SomeException (Maybe a))
forall (m :: * -> *) a.
MonadUnliftIO m =>
LogConfig -> Text -> m a -> m (Either SomeException a)
reaperGate LogConfig
logCfg Text
task (m (Maybe a) -> m (Either SomeException (Maybe a)))
-> m (Maybe a) -> m (Either SomeException (Maybe a))
forall a b. (a -> b) -> a -> b
$
    Text
-> Text
-> NominalDiffTime
-> NominalDiffTime
-> (Maybe s -> m (a, s))
-> m (Maybe a)
forall s (m :: * -> *) a.
(FromJSON s, MonadArbiter m, ToJSON s) =>
Text
-> Text
-> NominalDiffTime
-> NominalDiffTime
-> (Maybe s -> m (a, s))
-> m (Maybe a)
Ops.runGatedStateBounded Text
schema Text
task NominalDiffTime
every NominalDiffTime
stmtTimeout Maybe s -> m (a, s)
work

reaperGate :: (MonadUnliftIO m) => LogConfig -> Text -> m a -> m (Either SomeException a)
reaperGate :: forall (m :: * -> *) a.
MonadUnliftIO m =>
LogConfig -> Text -> m a -> m (Either SomeException a)
reaperGate LogConfig
logCfg Text
task m a
action =
  m a -> m (Either SomeException a)
forall (m :: * -> *) a.
MonadUnliftIO m =>
m a -> m (Either SomeException a)
tryAny m a
action
    m (Either SomeException a)
-> (Either SomeException a -> m (Either SomeException a))
-> m (Either SomeException a)
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (SomeException -> m (Either SomeException a))
-> (a -> m (Either SomeException a))
-> Either SomeException a
-> m (Either SomeException a)
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (\SomeException
exception -> SomeException -> Either SomeException a
forall a b. a -> Either a b
Left SomeException
exception Either SomeException a -> m () -> m (Either SomeException a)
forall a b. a -> m b -> m a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ LogConfig -> Text -> SomeException -> m ()
forall (m :: * -> *).
MonadUnliftIO m =>
LogConfig -> Text -> SomeException -> m ()
warnEx LogConfig
logCfg (Text
"Reaper op failed: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
task) SomeException
exception) (Either SomeException a -> m (Either SomeException a)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either SomeException a -> m (Either SomeException a))
-> (a -> Either SomeException a) -> a -> m (Either SomeException a)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> Either SomeException a
forall a b. b -> Either a b
Right)