{-# LANGUAGE OverloadedStrings #-}
{-# OPTIONS_HADDOCK not-home #-}

-- | Internal to the arbiter packages. Not covered by the PVP.
-- 'runMaintenancePass' and 'MaintenancePace' are public through "Arbiter.Worker".
--
-- 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.Enum (showText)
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.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.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
Ops.micros (MaintenancePace -> NominalDiffTime
paceWindow MaintenancePace
pace))

-- | Gaps between runs of each kind of work, and the bucket idle age. A zero gap
-- runs that work every pass.
data MaintenancePace = MaintenancePace
  { MaintenancePace -> NominalDiffTime
paceWindow :: NominalDiffTime
  -- ^ Gap in seconds between runs of one ordinary operation.
  , MaintenancePace -> NominalDiffTime
paceSparseWindow :: NominalDiffTime
  -- ^ Gap in seconds between runs of 'PruneRateLimitBuckets' and
  -- 'ReconcilePruneConcurrency'.
  , MaintenancePace -> NominalDiffTime
paceBucketIdle :: NominalDiffTime
  -- ^ Idle age in seconds at which a prune collects a full 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.
-- The pass covers every queue in the schema ('Ops.schemaQueues'), not only the
-- caller's registry. Which caller wins a gate does not change the result.
-- Returns the operations that threw. A failure for one queue is logged and not
-- returned. A failure does not stop the pass. When the schema's queues cannot be
-- read, no operation runs and every operation is returned.
runMaintenancePass
  :: forall m
   . ( Arb.RegistryAdmissionPolicies (RegistryOf m)
     , MonadArbiter m
     , RegistryTables (RegistryOf m)
     )
  => LogConfig
  -> (MaintenanceOp -> Int64 -> m ())
  -> MaintenancePace
  -> NominalDiffTime
  -- ^ Statement timeout
  -> 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
  schema <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
Arb.getSchema
  tryAny (Ops.schemaQueues schema (registryTableNames (Proxy @(RegistryOf m))))
    >>= either
      (\SomeException
exception -> [MaintenanceOp
forall a. Bounded a => a
minBound .. MaintenanceOp
forall a. Bounded a => a
maxBound] [MaintenanceOp] -> m () -> m [MaintenanceOp]
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 queue discovery failed, pass skipped" SomeException
exception)
      (maintainQueues logCfg report pace stmtTimeout schema)

-- | 'runMaintenancePass' over the given queues.
maintainQueues
  :: forall m
   . ( Arb.RegistryAdmissionPolicies (RegistryOf m)
     , MonadArbiter m
     )
  => LogConfig
  -> (MaintenanceOp -> Int64 -> m ())
  -> MaintenancePace
  -> NominalDiffTime
  -> SchemaName
  -> [Text]
  -> m [MaintenanceOp]
maintainQueues :: forall (m :: * -> *).
(RegistryAdmissionPolicies (RegistryOf m), MonadArbiter m) =>
LogConfig
-> (MaintenanceOp -> Int64 -> m ())
-> MaintenancePace
-> NominalDiffTime
-> Text
-> [Text]
-> m [MaintenanceOp]
maintainQueues LogConfig
logCfg MaintenanceOp -> Int64 -> m ()
report MaintenancePace
pace NominalDiffTime
stmtTimeout Text
schema [Text]
queues = 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
      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 RateLimitPolicy -> Bool
forall a. Set a -> Bool
Set.null (forall (registry :: JobPayloadRegistry).
RegistryRateLimitPolicies registry =>
Set RateLimitPolicy
registryRateLimitPolicies @(RegistryOf m)))
      gated :: MaintenanceOp -> (Maybe a -> m ()) -> NominalDiffTime -> m a -> m (Maybe MaintenanceOp)
      gated :: forall a.
MaintenanceOp
-> (Maybe a -> m ())
-> NominalDiffTime
-> m a
-> m (Maybe MaintenanceOp)
gated MaintenanceOp
operation Maybe a -> m ()
emit NominalDiffTime
every m a
work =
        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 (MaintenanceOp -> Text
maintenanceOpName MaintenanceOp
operation) NominalDiffTime
every m a
work m (Either SomeException (Maybe a))
-> (Either SomeException (Maybe a) -> 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 a -> m ())
-> Either SomeException (Maybe a)
-> m (Maybe MaintenanceOp)
forall (m :: * -> *) a.
Monad m =>
MaintenanceOp
-> (a -> m ()) -> Either SomeException a -> m (Maybe MaintenanceOp)
failedOp MaintenanceOp
operation Maybe a -> m ()
emit
      gatedCount :: MaintenanceOp
-> NominalDiffTime -> m Int64 -> m (Maybe MaintenanceOp)
gatedCount MaintenanceOp
operation = MaintenanceOp
-> (Maybe Int64 -> m ())
-> NominalDiffTime
-> m Int64
-> m (Maybe MaintenanceOp)
forall a.
MaintenanceOp
-> (Maybe a -> m ())
-> NominalDiffTime
-> m a
-> m (Maybe MaintenanceOp)
gated 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))
      logSwept :: LogLevel -> Text -> Text -> Int64 -> m ()
logSwept LogLevel
level Text
verb Text
noun Int64
count =
        LogConfig -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadUnliftIO m =>
LogConfig -> LogLevel -> Text -> m ()
tryLog LogConfig
logCfg LogLevel
level (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$ Text
"Reaper " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
verb Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int64 -> Text
forall a. Show a => a -> Text
showText Int64
count Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
noun
      reportFailed :: MaintenanceOp -> [Text] -> m ()
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 -> (Int64 -> m ()) -> Maybe (Int64, [Text]) -> m ()
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
-> NominalDiffTime
-> (Int64 -> m ())
-> m (Int64, [Text])
-> m (Maybe MaintenanceOp)
sweep MaintenanceOp
operation NominalDiffTime
every Int64 -> m ()
done = MaintenanceOp
-> (Maybe (Int64, [Text]) -> m ())
-> NominalDiffTime
-> m (Int64, [Text])
-> m (Maybe MaintenanceOp)
forall a.
MaintenanceOp
-> (Maybe a -> m ())
-> NominalDiffTime
-> m a
-> m (Maybe MaintenanceOp)
gated MaintenanceOp
operation (MaintenanceOp -> (Int64 -> m ()) -> Maybe (Int64, [Text]) -> m ()
reportSwept MaintenanceOp
operation Int64 -> m ()
done) NominalDiffTime
every
      refreshGroups :: m (Maybe MaintenanceOp)
refreshGroups =
        let task :: Text
task = MaintenanceOp -> Text
maintenanceOpName MaintenanceOp
RefreshGroups
         in LogConfig
-> Text
-> m (Maybe (Int64, [Text]))
-> m (Either SomeException (Maybe (Int64, [Text])))
forall (m :: * -> *) a.
MonadUnliftIO m =>
LogConfig -> Text -> m a -> m (Either SomeException a)
reaperGate
              LogConfig
logCfg
              Text
task
              (Text
-> Text
-> NominalDiffTime
-> NominalDiffTime
-> (Maybe (Map Text GroupsCursor)
    -> m ((Int64, [Text]), Map Text GroupsCursor))
-> m (Maybe (Int64, [Text]))
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
window NominalDiffTime
stmtTimeout (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)
failedOp 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 ())))
  ([Maybe MaintenanceOp] -> [MaintenanceOp])
-> m [Maybe MaintenanceOp] -> m [MaintenanceOp]
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap [Maybe MaintenanceOp] -> [MaintenanceOp]
forall a. [Maybe a] -> [a]
catMaybes (m [Maybe MaintenanceOp] -> m [MaintenanceOp])
-> ([m (Maybe MaintenanceOp)] -> m [Maybe MaintenanceOp])
-> [m (Maybe MaintenanceOp)]
-> m [MaintenanceOp]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [m (Maybe MaintenanceOp)] -> m [Maybe MaintenanceOp]
forall (t :: * -> *) (m :: * -> *) a.
(Traversable t, Monad m) =>
t (m a) -> m (t a)
forall (m :: * -> *) a. Monad m => [m a] -> m [a]
sequence ([m (Maybe MaintenanceOp)] -> m [MaintenanceOp])
-> [m (Maybe MaintenanceOp)] -> m [MaintenanceOp]
forall a b. (a -> b) -> a -> b
$
    [ m (Maybe MaintenanceOp)
refreshGroups
    , MaintenanceOp
-> NominalDiffTime -> m Int64 -> m (Maybe MaintenanceOp)
gatedCount MaintenanceOp
SweepStaleWorkers NominalDiffTime
window (m Int64 -> m (Maybe MaintenanceOp))
-> m Int64 -> m (Maybe MaintenanceOp)
forall a b. (a -> b) -> a -> b
$ Text -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> m Int64
Ops.sweepStaleWorkers Text
schema
    , MaintenanceOp
-> NominalDiffTime
-> (Int64 -> m ())
-> m (Int64, [Text])
-> m (Maybe MaintenanceOp)
sweep
        MaintenanceOp
SweepExhaustedJobs
        NominalDiffTime
window
        (LogLevel -> Text -> Text -> Int64 -> m ()
logSwept LogLevel
Warning Text
"moved" Text
"exhausted job(s) to the DLQ")
        (m (Int64, [Text]) -> m (Maybe MaintenanceOp))
-> m (Int64, [Text]) -> m (Maybe MaintenanceOp)
forall a b. (a -> b) -> a -> b
$ Text -> [Text] -> m (Int64, [Text])
forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m (Int64, [Text])
Ops.sweepExhaustedJobs Text
schema [Text]
queues
    , MaintenanceOp
-> NominalDiffTime
-> (Int64 -> m ())
-> m (Int64, [Text])
-> m (Maybe MaintenanceOp)
sweep
        MaintenanceOp
SweepCancelledJobs
        NominalDiffTime
window
        (LogLevel -> Text -> Text -> Int64 -> m ()
logSwept LogLevel
Info Text
"deleted" Text
"orphaned cancelled job(s)")
        (m (Int64, [Text]) -> m (Maybe MaintenanceOp))
-> m (Int64, [Text]) -> m (Maybe MaintenanceOp)
forall a b. (a -> b) -> a -> b
$ Text -> [Text] -> m (Int64, [Text])
forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m (Int64, [Text])
Ops.sweepCancelledJobs Text
schema [Text]
queues
    ]
      [m (Maybe MaintenanceOp)]
-> [m (Maybe MaintenanceOp)] -> [m (Maybe MaintenanceOp)]
forall a. Semigroup a => a -> a -> a
<> [MaintenanceOp
-> NominalDiffTime -> m Int64 -> m (Maybe MaintenanceOp)
gatedCount MaintenanceOp
PruneRateLimitBuckets NominalDiffTime
sparseWindow (forall (m :: * -> *). MonadArbiter m => NominalDiffTime -> m Int64
Arb.pruneRateLimitBuckets @m (MaintenancePace -> NominalDiffTime
paceBucketIdle MaintenancePace
pace)) | Bool
hasRateLimit]
      [m (Maybe MaintenanceOp)]
-> [m (Maybe MaintenanceOp)] -> [m (Maybe MaintenanceOp)]
forall a. Semigroup a => a -> a -> a
<> [MaintenanceOp
-> NominalDiffTime -> m Int64 -> m (Maybe MaintenanceOp)
gatedCount MaintenanceOp
ReconcileConcurrencyStale NominalDiffTime
window (Text -> [Text] -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m Int64
Ops.reconcileConcurrencyCountsIfStale Text
schema [Text]
queues) | Bool
hasConcurrency]
      [m (Maybe MaintenanceOp)]
-> [m (Maybe MaintenanceOp)] -> [m (Maybe MaintenanceOp)]
forall a. Semigroup a => a -> a -> a
<> [MaintenanceOp
-> NominalDiffTime -> m Int64 -> m (Maybe MaintenanceOp)
gatedCount MaintenanceOp
ReconcilePruneConcurrency NominalDiffTime
sparseWindow (Text -> [Text] -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m Int64
Ops.reconcileAndPruneConcurrency Text
schema [Text]
queues) | Bool
hasConcurrency]
      [m (Maybe MaintenanceOp)]
-> [m (Maybe MaintenanceOp)] -> [m (Maybe MaintenanceOp)]
forall a. Semigroup a => a -> a -> a
<> [ MaintenanceOp
-> NominalDiffTime
-> (Int64 -> m ())
-> m (Int64, [Text])
-> m (Maybe MaintenanceOp)
sweep
             MaintenanceOp
PurgeArchives
             NominalDiffTime
window
             (LogLevel -> Text -> Text -> Int64 -> m ()
logSwept LogLevel
Info Text
"purged" Text
"archived job(s)")
             (m (Int64, [Text]) -> m (Maybe MaintenanceOp))
-> m (Int64, [Text]) -> m (Maybe MaintenanceOp)
forall a b. (a -> b) -> a -> b
$ Text -> [Text] -> m (Int64, [Text])
forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m (Int64, [Text])
Ops.purgeArchives Text
schema [Text]
queues
         ]
      [m (Maybe MaintenanceOp)]
-> [m (Maybe MaintenanceOp)] -> [m (Maybe MaintenanceOp)]
forall a. Semigroup a => a -> a -> a
<> [ MaintenanceOp
-> NominalDiffTime
-> (Int64 -> m ())
-> m (Int64, [Text])
-> m (Maybe MaintenanceOp)
sweep
             MaintenanceOp
PurgeDLQ
             NominalDiffTime
window
             (LogLevel -> Text -> Text -> Int64 -> m ()
logSwept LogLevel
Info Text
"purged" Text
"DLQ job(s)")
             (m (Int64, [Text]) -> m (Maybe MaintenanceOp))
-> m (Int64, [Text]) -> m (Maybe MaintenanceOp)
forall a b. (a -> b) -> a -> b
$ Text -> [Text] -> m (Int64, [Text])
forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m (Int64, [Text])
Ops.purgeDLQ Text
schema [Text]
queues
         ]

-- | Name a failed operation. Report a completed one through @emit@.
failedOp
  :: (Monad m)
  => MaintenanceOp
  -> (a -> m ())
  -> Either SomeException a
  -> m (Maybe MaintenanceOp)
failedOp :: forall (m :: * -> *) a.
Monad m =>
MaintenanceOp
-> (a -> m ()) -> Either SomeException a -> m (Maybe MaintenanceOp)
failedOp 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. 'Nothing' when it failed, is running
-- elsewhere, or its window has not elapsed. Failures are logged.
runReaperOp
  :: (MonadArbiter m)
  => LogConfig
  -> NominalDiffTime
  -- ^ Statement timeout
  -> Text
  -- ^ Task name
  -> NominalDiffTime
  -- ^ Gate window: minimum gap between runs across all callers
  -> m a
  -> m (Maybe a)
runReaperOp :: forall (m :: * -> *) a.
MonadArbiter m =>
LogConfig
-> NominalDiffTime -> Text -> NominalDiffTime -> m a -> m (Maybe a)
runReaperOp LogConfig
logCfg NominalDiffTime
stmtTimeout Text
task NominalDiffTime
every m a
work = do
  schema <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
Arb.getSchema
  fromRight Nothing <$> tryReaperOp logCfg schema stmtTimeout task every work

-- | 'runReaperOp', keeping the failure. @Right Nothing@ is a gate that ran too recently or is held elsewhere.
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

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)