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

-- | Heartbeat guard, lease fence, and job deadline tests, instantiated for each backend.
module Arbiter.Test.Worker.Deadline (deadlineSpec) where

import Arbiter.Core.Exceptions (JobForceCancelled (..))
import Arbiter.Core.HighLevel (QueueOperation, RegistryAdmissionPolicies)
import Arbiter.Core.HighLevel qualified as HL
import Arbiter.Core.Job.DLQ (DLQJob (..))
import Arbiter.Core.Job.Types
  ( JobRead
  , JobWrite
  , ObservabilityHooks (..)
  , defaultJob
  , defaultObservabilityHooks
  , lastError
  , payload
  , primaryKey
  , setGroupKey
  , setMaxAttempts
  )
import Arbiter.Core.MonadArbiter (JobHandler, RegistryOf, ResultOf)
import Arbiter.Core.QueueRegistry (RegistryTables)
import Arbiter.Core.Trace (capturingContextIO)
import Arbiter.Worker (runWorkerPool)
import Arbiter.Worker.BackoffStrategy (Jitter (NoJitter))
import Arbiter.Worker.Config
  ( BatchCallbacks (..)
  , WorkerConfig (..)
  , ackAll
  , batchedWorkerConfig
  , transactionalWorkerConfig
  )
import Arbiter.Worker.Heartbeat (PoolGuard, newHeartbeatGuard)
import Arbiter.Worker.Heartbeat.Guard (Batch (..), guardBatch, leaseExpiredReason, reclaimedReason, runHeartbeatGuard)
import Arbiter.Worker.Logger (LogConfig (..), LogDestination (..), defaultLogConfig, silentLogConfig)
import Control.Concurrent (threadDelay)
import Control.Exception (SomeException, fromException, uninterruptibleMask_)
import Control.Monad (void, when)
import Control.Monad.IO.Class (MonadIO, liftIO)
import Data.Either (isRight)
import Data.Foldable (toList)
import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef, writeIORef)
import Data.Int (Int64)
import Data.List (partition)
import Data.List.NonEmpty (NonEmpty ((:|)))
import Data.Maybe (fromMaybe, isJust)
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (UTCTime, getCurrentTime)
import GHC.Clock (getMonotonicTime)
import Test.Hspec (Spec, before, describe, it, shouldBe, shouldReturn, shouldSatisfy)
import UnliftIO (MonadUnliftIO, finally, mask_, tryAny, withRunInIO)
import UnliftIO.Async (async, poll, waitCatch, withAsync)

import Arbiter.Test.Poll (bumpRef, waitUntil)
import Arbiter.Test.Worker.Backend (TestBackend (..))
import Arbiter.Test.Worker.Rows (flagCancelled, holdRowLock, reclaimJob, releaseRow, rowCount, takeClaimHolder)

-- | Longer than any deadline under test.
handlerSleepMicros :: Int
handlerSleepMicros :: Int
handlerSleepMicros = Int
30_000_000

-- | Cleanup slow enough that the heartbeat outlives the fence throw.
unwindMicros :: Int
unwindMicros :: Int
unwindMicros = Int
4_000_000

-- | Masked interval longer than the deadline.
maskedMicros :: Int
maskedMicros :: Int
maskedMicros = Int
4_000_000

-- | A heartbeat hook slow enough to cover another batch's whole deadline.
slowHookMicros :: Int
slowHookMicros :: Int
slowHookMicros = Int
8_000_000

-- | Longer than 'maskedMicros'.
stuckRegistrationMillis :: Int
stuckRegistrationMillis :: Int
stuckRegistrationMillis = Int
8_000

-- | A pause after the registration returns.
afterRegistrationMicros :: Int
afterRegistrationMicros :: Int
afterRegistrationMicros = Int
500_000

-- | How long two sibling handlers run.
siblingRunMicros :: Int
siblingRunMicros :: Int
siblingRunMicros = Int
6_000_000

-- | A pause for both sibling batches to register before the cancel.
beforeCancelMicros :: Int
beforeCancelMicros :: Int
beforeCancelMicros = Int
700_000

-- | How long a sibling batch's row stays locked.
rowLockMicros :: Int
rowLockMicros :: Int
rowLockMicros = Int
4_000_000

-- | The window, in seconds after the lock, in which the free batch has to beat.
lockedBeatFrom, lockedBeatTo :: Double
lockedBeatFrom :: Double
lockedBeatFrom = Double
1
lockedBeatTo :: Double
lockedBeatTo = Double
3.5

-- | How long a grouped handler runs after enqueuing into its group.
groupedHandlerMicros :: Int
groupedHandlerMicros :: Int
groupedHandlerMicros = Int
2_500_000

-- | A pause into the first handler's run before the fenced job is inserted.
beforeFencedInsertMicros :: Int
beforeFencedInsertMicros :: Int
beforeFencedInsertMicros = Int
1_300_000

-- | The most seconds a fenced job may take to reach the DLQ.
fenceBudgetSeconds :: Double
fenceBudgetSeconds :: Double
fenceBudgetSeconds = Double
5

-- | How long a poll waits for the worker to reach a state.
waitMillis :: Int
waitMillis :: Int
waitMillis = Int
10_000

-- | 'waitMillis' for a state a backoff or a fence delays.
slowWaitMillis :: Int
slowWaitMillis :: Int
slowWaitMillis = Int
15_000

-- | 'waitMillis' for a DLQ move.
dlqWaitMillis :: Int
dlqWaitMillis :: Int
dlqWaitMillis = Int
20_000

-- | Run @action@ with the batch registered with the guard, as the pool does.
withJobsHeartbeat
  :: (MonadUnliftIO m)
  => PoolGuard payload
  -> UTCTime
  -> NonEmpty (JobRead payload)
  -> m [JobRead payload]
  -> m a
  -> m a
withJobsHeartbeat :: forall (m :: * -> *) payload a.
MonadUnliftIO m =>
PoolGuard payload
-> UTCTime
-> NonEmpty (JobRead payload)
-> m [JobRead payload]
-> m a
-> m a
withJobsHeartbeat PoolGuard payload
guard UTCTime
startTime NonEmpty (JobRead payload)
jobs m [JobRead payload]
pending m a
action = do
  inherit <- m (IO () -> IO ())
forall (m :: * -> *) a. MonadUnliftIO m => m (IO a -> IO a)
capturingContextIO
  withRunInIO $ \forall a. m a -> IO a
run -> PoolGuard payload -> Batch IO (JobRead payload) -> IO a -> IO a
forall (n :: * -> *) job a.
(MonadFork n, MonadMask n, MonadMonotonicTime n, MonadSTM n,
 MonadTime n) =>
HeartbeatGuard n job -> Batch n job -> n a -> n a
guardBatch PoolGuard payload
guard (NonEmpty (JobRead payload)
-> IO [JobRead payload]
-> UTCTime
-> (IO () -> IO ())
-> Batch IO (JobRead payload)
forall (n :: * -> *) job.
NonEmpty job -> n [job] -> UTCTime -> (n () -> n ()) -> Batch n job
Batch NonEmpty (JobRead payload)
jobs (m [JobRead payload] -> IO [JobRead payload]
forall a. m a -> IO a
run m [JobRead payload]
pending) UTCTime
startTime IO () -> IO ()
inherit) (m a -> IO a
forall a. m a -> IO a
run m a
action)

-- | Deadline suite. The queue under test declares @()@ as its result type.
deadlineSpec
  :: forall payload m env
   . ( Eq payload
     , QueueOperation m payload
     , RegistryAdmissionPolicies (RegistryOf m)
     , RegistryTables (RegistryOf m)
     , ResultOf m payload ~ ()
     )
  => TestBackend payload m env
  -> Spec
deadlineSpec :: forall payload (m :: * -> *) env.
(Eq payload, QueueOperation m payload,
 RegistryAdmissionPolicies (RegistryOf m),
 RegistryTables (RegistryOf m), ResultOf m payload ~ ()) =>
TestBackend payload m env -> Spec
deadlineSpec TestBackend {Text
schema :: Text
schema :: forall payload (m :: * -> *) env. TestBackend payload m env -> Text
schema, Text
table :: Text
table :: forall payload (m :: * -> *) env. TestBackend payload m env -> Text
table, ByteString
connStr :: ByteString
connStr :: forall payload (m :: * -> *) env.
TestBackend payload m env -> ByteString
connStr, Text -> payload
mkSimple :: Text -> payload
mkSimple :: forall payload (m :: * -> *) env.
TestBackend payload m env -> Text -> payload
mkSimple, IO env
mkEnv :: IO env
mkEnv :: forall payload (m :: * -> *) env.
TestBackend payload m env -> IO env
mkEnv, (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler :: (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler :: forall payload (m :: * -> *) env.
TestBackend payload m env
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler, forall a. env -> m a -> IO a
runM :: forall a. env -> m a -> IO a
runM :: forall payload (m :: * -> *) env.
TestBackend payload m env -> forall a. env -> m a -> IO a
runM} =
  IO env -> SpecWith env -> Spec
forall a. IO a -> SpecWith a -> Spec
before IO env
mkEnv (SpecWith env -> Spec) -> SpecWith env -> Spec
forall a b. (a -> b) -> a -> b
$ do
    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Guard registration" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"interrupts a blocked claim hook at the duration deadline before the handler starts" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        handlerCalls <- Int -> IO (IORef Int)
forall a. a -> IO (IORef a)
newIORef (Int
0 :: Int)
        hookCalls <- newIORef (0 :: Int)
        let handler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler ((JobRead payload -> m (ResultOf m payload))
 -> JobHandler m payload (ResultOf m payload))
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ \JobRead payload
_ -> IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
handlerCalls
            hooks =
              ObservabilityHooks m payload
forall (m :: * -> *) payload.
Applicative m =>
ObservabilityHooks m payload
defaultObservabilityHooks
                { onJobClaimed = \JobRead payload
_ UTCTime
_ -> IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ do
                    IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
hookCalls
                    Int -> IO ()
threadDelay Int
handlerSleepMicros
                }
        void (insertedId env)
        base :: WorkerConfig m payload <- transactionalWorkerConfig 1 handler
        let config =
              WorkerConfig m payload
base
                { pollInterval = 0.05
                , visibilityTimeout = 3
                , jobHeartbeatInterval = 1
                , maxJobDuration = Just 0.5
                , observabilityHooks = hooks
                , logConfig = silentLogConfig
                }
        withAsync (runM env $ runWorkerPool config) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
5_000 (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ Bool -> Bool
not (Bool -> Bool)
-> ([DLQJob payload] -> Bool) -> [DLQJob payload] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [DLQJob payload] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null ([DLQJob payload] -> Bool) -> IO [DLQJob payload] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> env -> IO [DLQJob payload]
listDLQ env
env
          IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef IORef Int
hookCalls IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
1
          IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef IORef Int
handlerCalls IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
0

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"returns once a signal in flight meets the unregister" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        job <- env -> JobWrite payload -> IO (JobRead payload)
inserted env
env (payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"slow"))
        config <- transactionalWorkerConfig 1 idleHandler
        guard <- runM env (newHeartbeatGuard config {maxJobDuration = Just 1, logConfig = silentLogConfig})
        startTime <- getCurrentTime
        withAsync (runHeartbeatGuard guard) $ \Async Void
_ -> do
          registration <-
            IO () -> IO (Async ())
forall (m :: * -> *) a. MonadUnliftIO m => m a -> m (Async a)
async (IO () -> IO (Async ()))
-> (m () -> IO ()) -> m () -> IO (Async ())
forall b c a. (b -> c) -> (a -> b) -> a -> c
. env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> (m () -> m ()) -> m () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. m () -> m ()
forall (m :: * -> *) a. MonadUnliftIO m => m a -> m a
mask_ (m () -> IO (Async ())) -> m () -> IO (Async ())
forall a b. (a -> b) -> a -> b
$ do
              PoolGuard payload
-> UTCTime
-> NonEmpty (JobRead payload)
-> m [JobRead payload]
-> m ()
-> m ()
forall (m :: * -> *) payload a.
MonadUnliftIO m =>
PoolGuard payload
-> UTCTime
-> NonEmpty (JobRead payload)
-> m [JobRead payload]
-> m a
-> m a
withJobsHeartbeat PoolGuard payload
guard UTCTime
startTime (JobRead payload
job JobRead payload -> [JobRead payload] -> NonEmpty (JobRead payload)
forall a. a -> [a] -> NonEmpty a
:| []) ([JobRead payload] -> m [JobRead payload]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []) (IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> IO ()
forall a. IO a -> IO a
uninterruptibleMask_ (Int -> IO ()
threadDelay Int
maskedMicros)))
              IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (Int -> IO ()
threadDelay Int
afterRegistrationMicros)
          waitUntil stuckRegistrationMillis (isJust <$> poll registration)
          outcome <- waitCatch registration
          outcome `shouldSatisfy` isRight

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Shared extend" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"stops only the batch whose job another worker reclaimed" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        startedRef <- [Int64] -> IO (IORef [Int64])
forall a. a -> IO (IORef a)
newIORef ([] :: [Int64])
        finishedRef <- newIORef ([] :: [Int64])
        reasonsRef <- newIORef ([] :: [(Int64, Text)])
        beatsRef <- newIORef ([] :: [(Int64, Double)])
        let handler :: JobHandler m payload ()
            handler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler ((JobRead payload -> m (ResultOf m payload))
 -> JobHandler m payload (ResultOf m payload))
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ \JobRead payload
job -> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (ResultOf m payload) -> m (ResultOf m payload))
-> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ do
              IORef [Int64] -> ([Int64] -> ([Int64], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Int64]
startedRef (\[Int64]
started -> (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job Int64 -> [Int64] -> [Int64]
forall a. a -> [a] -> [a]
: [Int64]
started, ()))
              Int -> IO ()
threadDelay Int
siblingRunMicros
              IORef [Int64] -> ([Int64] -> ([Int64], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Int64]
finishedRef (\[Int64]
finished -> (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job Int64 -> [Int64] -> [Int64]
forall a. a -> [a] -> [a]
: [Int64]
finished, ()))
            hooks :: ObservabilityHooks m payload
            hooks =
              ObservabilityHooks m payload
forall (m :: * -> *) payload.
Applicative m =>
ObservabilityHooks m payload
defaultObservabilityHooks
                { onJobHeartbeat = \JobRead payload
job UTCTime
_ UTCTime
_ ->
                    IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IORef [(Int64, Double)] -> JobRead payload -> IO ()
forall payload. IORef [(Int64, Double)] -> JobRead payload -> IO ()
stamp IORef [(Int64, Double)]
beatsRef JobRead payload
job)
                , onJobUnavailable = \JobRead payload
job Text
reason ->
                    IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ IORef [(Int64, Text)]
-> ([(Int64, Text)] -> ([(Int64, Text)], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [(Int64, Text)]
reasonsRef (\[(Int64, Text)]
seen -> ((JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job, Text
reason) (Int64, Text) -> [(Int64, Text)] -> [(Int64, Text)]
forall a. a -> [a] -> [a]
: [(Int64, Text)]
seen, ()))
                }
        stolenId <- insertedId env
        keptId <- insertedId env

        config <- transactionalWorkerConfig 2 handler
        let workerConfig =
              WorkerConfig m payload
config
                { pollInterval = 0.2
                , jitter = NoJitter
                , visibilityTimeout = 20
                , jobHeartbeatInterval = 1
                , observabilityHooks = hooks
                , logConfig = silentLogConfig
                }

        withAsync (runM env $ runWorkerPool workerConfig) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ (Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
2) (Int -> Bool) -> ([Int64] -> Int) -> [Int64] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Int64] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length ([Int64] -> Bool) -> IO [Int64] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef [Int64] -> IO [Int64]
forall a. IORef a -> IO a
readIORef IORef [Int64]
startedRef
          ByteString -> Text -> Text -> Int64 -> IO ()
reclaimJob ByteString
connStr Text
schema Text
table Int64
stolenId
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ Bool -> Bool
not (Bool -> Bool)
-> ([(Int64, Text)] -> Bool) -> [(Int64, Text)] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [(Int64, Text)] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null ([(Int64, Text)] -> Bool) -> IO [(Int64, Text)] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef [(Int64, Text)] -> IO [(Int64, Text)]
forall a. IORef a -> IO a
readIORef IORef [(Int64, Text)]
reasonsRef
          reclaimedAt <- IO Double
getMonotonicTime
          waitUntil slowWaitMillis $ not . null <$> readIORef finishedRef
          reasons <- readIORef reasonsRef
          reasons `shouldBe` [(stolenId, reclaimedReason)]
          finished <- readIORef finishedRef
          finished `shouldBe` [keptId]
          beats <- readIORef beatsRef
          [() | (jobId, at) <- beats, jobId == keptId, at > reclaimedAt] `shouldSatisfy` (not . null)

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Shared guard" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"stops only the batch whose live claim the cancel names" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        config <- Int
-> JobHandler m payload (ResultOf m payload)
-> IO (WorkerConfig m payload)
forall (n :: * -> *) (m :: * -> *) payload.
(MonadArbiter n, MonadIO m) =>
Int
-> JobHandler n payload (ResultOf n payload)
-> m (WorkerConfig n payload)
transactionalWorkerConfig Int
1 JobHandler m payload ()
JobHandler m payload (ResultOf m payload)
idleHandler
        let guardConfig = WorkerConfig m payload
config {jobHeartbeatInterval = 0.5, visibilityTimeout = 20, logConfig = silentLogConfig}
        handedId <- insertedPlainId env
        _ <- insertedPlainId env
        batch <- runM env (HL.claimNextVisibleJobsAs @payload 2 20 (workerId config))
        (handed, kept) <- case partition ((== handedId) . primaryKey) batch of
          ([JobRead payload
job], [JobRead payload
sibling]) -> (JobRead payload, JobRead payload)
-> IO (JobRead payload, JobRead payload)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (JobRead payload
job, JobRead payload
sibling)
          ([JobRead payload], [JobRead payload])
_ -> String -> IO (JobRead payload, JobRead payload)
forall a. HasCallStack => String -> IO a
forall (m :: * -> *) a.
(MonadFail m, HasCallStack) =>
String -> m a
fail String
"expected two claimed jobs"
        releaseRow connStr schema table handedId
        reclaimed <- runM env (HL.claimNextVisibleJobsAs @payload 1 20 (workerId config)) >>= single
        guard <- runM env (newHeartbeatGuard guardConfig)
        startTime <- getCurrentTime
        withAsync (runHeartbeatGuard guard) $ \Async Void
_ -> do
          keeper <-
            IO () -> IO (Async ())
forall (m :: * -> *) a. MonadUnliftIO m => m a -> m (Async a)
async (IO () -> IO (Async ()))
-> (m () -> IO ()) -> m () -> IO (Async ())
forall b c a. (b -> c) -> (a -> b) -> a -> c
. env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO (Async ())) -> m () -> IO (Async ())
forall a b. (a -> b) -> a -> b
$
              PoolGuard payload
-> UTCTime
-> NonEmpty (JobRead payload)
-> m [JobRead payload]
-> m ()
-> m ()
forall (m :: * -> *) payload a.
MonadUnliftIO m =>
PoolGuard payload
-> UTCTime
-> NonEmpty (JobRead payload)
-> m [JobRead payload]
-> m a
-> m a
withJobsHeartbeat PoolGuard payload
guard UTCTime
startTime (JobRead payload
handed JobRead payload -> [JobRead payload] -> NonEmpty (JobRead payload)
forall a. a -> [a] -> NonEmpty a
:| [JobRead payload
kept]) ([JobRead payload] -> m [JobRead payload]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [JobRead payload
kept]) (IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (Int -> IO ()
threadDelay Int
siblingRunMicros))
          holder <-
            async . runM env $
              withJobsHeartbeat guard startTime (reclaimed :| []) (pure [reclaimed]) (liftIO (threadDelay siblingRunMicros))
          threadDelay beforeCancelMicros
          flagCancelled connStr schema table handedId
          holderOutcome <- waitCatch holder
          keeperOutcome <- waitCatch keeper
          either isForceCancelled (const False) holderOutcome `shouldBe` True
          keeperOutcome `shouldSatisfy` isRight

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"beats a batch while a sibling batch's row is locked" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        startedRef <- [Int64] -> IO (IORef [Int64])
forall a. a -> IO (IORef a)
newIORef ([] :: [Int64])
        beatsRef <- newIORef ([] :: [(Int64, Double)])
        let handler :: JobHandler m payload ()
            handler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler ((JobRead payload -> m (ResultOf m payload))
 -> JobHandler m payload (ResultOf m payload))
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ \JobRead payload
job -> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (ResultOf m payload) -> m (ResultOf m payload))
-> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ do
              IORef [Int64] -> ([Int64] -> ([Int64], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Int64]
startedRef (\[Int64]
started -> (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job Int64 -> [Int64] -> [Int64]
forall a. a -> [a] -> [a]
: [Int64]
started, ()))
              Int -> IO ()
threadDelay Int
handlerSleepMicros
            hooks :: ObservabilityHooks m payload
            hooks = ObservabilityHooks m payload
forall (m :: * -> *) payload.
Applicative m =>
ObservabilityHooks m payload
defaultObservabilityHooks {onJobHeartbeat = \JobRead payload
job UTCTime
_ UTCTime
_ -> IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IORef [(Int64, Double)] -> JobRead payload -> IO ()
forall payload. IORef [(Int64, Double)] -> JobRead payload -> IO ()
stamp IORef [(Int64, Double)]
beatsRef JobRead payload
job)}
        lockedId <- insertedId env
        freeId <- insertedId env
        config <- transactionalWorkerConfig 2 handler
        let workerConfig =
              WorkerConfig m payload
config
                { pollInterval = 0.2
                , jitter = NoJitter
                , visibilityTimeout = 20
                , jobHeartbeatInterval = 1
                , observabilityHooks = hooks
                , logConfig = silentLogConfig
                }
        withAsync (runM env $ runWorkerPool workerConfig) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ (Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
2) (Int -> Bool) -> ([Int64] -> Int) -> [Int64] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Int64] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length ([Int64] -> Bool) -> IO [Int64] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef [Int64] -> IO [Int64]
forall a. IORef a -> IO a
readIORef IORef [Int64]
startedRef
          lockedFrom <- IO Double
getMonotonicTime
          holdRowLock connStr schema table lockedId rowLockMicros
          beats <- readIORef beatsRef
          [() | (jobId, at) <- beats, jobId == freeId, at > lockedFrom + lockedBeatFrom, at < lockedFrom + lockedBeatTo]
            `shouldSatisfy` (not . null)

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"extends a grouped job whose handler enqueued into its group" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        loggedRef <- [Text] -> IO (IORef [Text])
forall a. a -> IO (IORef a)
newIORef ([] :: [Text])
        finishedRef <- newIORef (0 :: Int)
        let handler :: JobHandler m payload ()
            handler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler ((JobRead payload -> m (ResultOf m payload))
 -> JobHandler m payload (ResultOf m payload))
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ \JobRead payload
job ->
              Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (JobRead payload -> payload
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> payload
payload JobRead payload
job payload -> payload -> Bool
forall a. Eq a => a -> a -> Bool
== Text -> payload
mkSimple Text
"slow") (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
                m (Maybe (JobRead payload)) -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (JobWrite payload -> m (Maybe (JobRead payload))
forall payload (m :: * -> *).
QueueOperation m payload =>
JobWrite payload -> m (Maybe (JobRead payload))
HL.insertJob (Maybe Text -> JobWrite payload -> JobWrite payload
forall payload. Maybe Text -> JobWrite payload -> JobWrite payload
setGroupKey (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
"shared") (payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"child"))))
                IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (Int -> IO ()
threadDelay Int
groupedHandlerMicros)
                IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
finishedRef)
            capture LogLevel
_level Text
msg [Pair]
_ctx = IORef [Text] -> ([Text] -> ([Text], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Text]
loggedRef (\[Text]
messages -> (Text
msg Text -> [Text] -> [Text]
forall a. a -> [a] -> [a]
: [Text]
messages, ()))
        parent <- inserted env (setGroupKey (Just "shared") (setMaxAttempts (Just 1) (defaultJob (mkSimple "slow"))))
        config :: WorkerConfig m payload <- transactionalWorkerConfig 1 handler
        let workerConfig =
              WorkerConfig m payload
config
                { pollInterval = 0.2
                , jitter = NoJitter
                , visibilityTimeout = 20
                , jobHeartbeatInterval = 1
                , logConfig = defaultLogConfig {logDestination = LogCallback capture}
                }
        withAsync (runM env $ runWorkerPool workerConfig) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ (Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
1) (Int -> Bool) -> IO Int -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef IORef Int
finishedRef
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ (Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0) (Int -> Bool) -> IO Int -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ByteString -> Text -> Text -> Int64 -> IO Int
rowCount ByteString
connStr Text
schema Text
table (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
parent)
          logged <- IORef [Text] -> IO [Text]
forall a. IORef a -> IO a
readIORef IORef [Text]
loggedRef
          filter (T.isInfixOf "deadlock" . T.toLower) logged `shouldBe` []
          dlq <- listDLQ env
          map (primaryKey . jobSnapshot) dlq `shouldBe` []

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Job deadline" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"interrupts a handler that outruns the maximum job duration" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        startedRef <- Int -> IO (IORef Int)
forall a. a -> IO (IORef a)
newIORef (Int
0 :: Int)
        finishedRef <- newIORef (0 :: Int)
        let handler :: JobHandler m payload ()
            handler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler ((JobRead payload -> m (ResultOf m payload))
 -> JobHandler m payload (ResultOf m payload))
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ \JobRead payload
_job -> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (ResultOf m payload) -> m (ResultOf m payload))
-> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ do
              IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
startedRef
              Int -> IO ()
threadDelay Int
handlerSleepMicros
              IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
finishedRef

        void (insertedId env)

        config :: WorkerConfig m payload <- transactionalWorkerConfig 1 handler
        let workerConfig =
              WorkerConfig m payload
config
                { pollInterval = 0.2
                , jitter = NoJitter
                , maxJobDuration = Just 1
                , logConfig = silentLogConfig
                }

        withAsync (runM env $ runWorkerPool workerConfig) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ (Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
1) (Int -> Bool) -> IO Int -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef IORef Int
startedRef
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
dlqWaitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ Bool -> Bool
not (Bool -> Bool)
-> ([DLQJob payload] -> Bool) -> [DLQJob payload] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [DLQJob payload] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null ([DLQJob payload] -> Bool) -> IO [DLQJob payload] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> env -> IO [DLQJob payload]
listDLQ env
env
          dlq <- env -> IO [DLQJob payload]
listDLQ env
env
          finished <- readIORef finishedRef
          finished `shouldBe` 0
          map (lastError . jobSnapshot) dlq
            `shouldBe` [Just "claim hooks and handler ran past the maximum job duration of 1s"]

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"fences a batch on time while another batch's heartbeat hook is slow" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        startedRef <- [(Int64, Double)] -> IO (IORef [(Int64, Double)])
forall a. a -> IO (IORef a)
newIORef ([] :: [(Int64, Double)])
        let handler :: JobHandler m payload ()
            handler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler ((JobRead payload -> m (ResultOf m payload))
 -> JobHandler m payload (ResultOf m payload))
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ \JobRead payload
job -> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (ResultOf m payload) -> m (ResultOf m payload))
-> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ do
              now <- IO Double
getMonotonicTime
              atomicModifyIORef' startedRef (\[(Int64, Double)]
started -> ((JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job, Double
now) (Int64, Double) -> [(Int64, Double)] -> [(Int64, Double)]
forall a. a -> [a] -> [a]
: [(Int64, Double)]
started, ()))
              threadDelay handlerSleepMicros
        slowId <- insertedId env
        let hooks :: ObservabilityHooks m payload
            hooks =
              ObservabilityHooks m payload
forall (m :: * -> *) payload.
Applicative m =>
ObservabilityHooks m payload
defaultObservabilityHooks
                { onJobHeartbeat = \JobRead payload
job UTCTime
_ UTCTime
_ ->
                    Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job Int64 -> Int64 -> Bool
forall a. Eq a => a -> a -> Bool
== Int64
slowId) (IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (Int -> IO ()
threadDelay Int
slowHookMicros))
                }

        config <- transactionalWorkerConfig 2 handler
        let workerConfig =
              WorkerConfig m payload
config
                { pollInterval = 0.2
                , jitter = NoJitter
                , visibilityTimeout = 20
                , jobHeartbeatInterval = 1
                , maxJobDuration = Just 2
                , observabilityHooks = hooks
                , logConfig = silentLogConfig
                }

        withAsync (runM env $ runWorkerPool workerConfig) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ (Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
1) (Int -> Bool)
-> ([(Int64, Double)] -> Int) -> [(Int64, Double)] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [(Int64, Double)] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length ([(Int64, Double)] -> Bool) -> IO [(Int64, Double)] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef [(Int64, Double)] -> IO [(Int64, Double)]
forall a. IORef a -> IO a
readIORef IORef [(Int64, Double)]
startedRef
          Int -> IO ()
threadDelay Int
beforeFencedInsertMicros
          fencedId <- env -> IO Int64
insertedId env
env
          waitUntil waitMillis $ (== 2) . length <$> readIORef startedRef
          waitUntil slowWaitMillis $ any ((== fencedId) . primaryKey . jobSnapshot) <$> listDLQ env
          fencedAt <- getMonotonicTime
          started <- readIORef startedRef
          let startedAt = Double -> Maybe Double -> Double
forall a. a -> Maybe a -> a
fromMaybe Double
0 (Int64 -> [(Int64, Double)] -> Maybe Double
forall a b. Eq a => a -> [(a, b)] -> Maybe b
lookup Int64
fencedId [(Int64, Double)]
started)
          (fencedAt - startedAt) `shouldSatisfy` (< fenceBudgetSeconds)

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"reports stopping a handler that had already finalized its batch" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        loggedRef <- [Text] -> IO (IORef [Text])
forall a. a -> IO (IORef a)
newIORef ([] :: [Text])
        finishedRef <- newIORef (0 :: Int)
        let handler
              :: NonEmpty (JobRead payload)
              -> BatchCallbacks m payload ()
              -> m ()
            handler NonEmpty (JobRead payload)
jobs BatchCallbacks m payload ()
callbacks = do
              BatchCallbacks m payload () -> [JobRead payload] -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> [JobRead payload] -> m ()
ackAll BatchCallbacks m payload ()
callbacks (NonEmpty (JobRead payload) -> [JobRead payload]
forall a. NonEmpty a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList NonEmpty (JobRead payload)
jobs)
              IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ do
                Int -> IO ()
threadDelay Int
handlerSleepMicros
                IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
finishedRef
            capture LogLevel
_level Text
msg [Pair]
_ctx = IORef [Text] -> ([Text] -> ([Text], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Text]
loggedRef (\[Text]
messages -> (Text
msg Text -> [Text] -> [Text]
forall a. a -> [a] -> [a]
: [Text]
messages, ()))

        void (insertedPlainId env)

        config <- batchedWorkerConfig 1 1 handler
        let workerConfig =
              WorkerConfig m payload
config
                { pollInterval = 0.2
                , jitter = NoJitter
                , maxJobDuration = Just 1
                , logConfig = defaultLogConfig {logDestination = LogCallback capture}
                }

        withAsync (runM env $ runWorkerPool workerConfig) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
slowWaitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ (Text -> Bool) -> [Text] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (Text
"finalized" Text -> Text -> Bool
`T.isInfixOf`) ([Text] -> Bool) -> IO [Text] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef [Text] -> IO [Text]
forall a. IORef a -> IO a
readIORef IORef [Text]
loggedRef
          finished <- IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef IORef Int
finishedRef
          finished `shouldBe` 0
          logged <- readIORef loggedRef
          filter ("maximum job duration" `T.isInfixOf`) logged `shouldSatisfy` (not . null)

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"keeps beating while a masked handler holds off the deadline" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        startedRef <- Maybe Double -> IO (IORef (Maybe Double))
forall a. a -> IO (IORef a)
newIORef (Maybe Double
forall a. Maybe a
Nothing :: Maybe Double)
        finishedRef <- newIORef (0 :: Int)
        beatsRef <- newIORef ([] :: [Double])
        let handler :: JobHandler m payload ()
            handler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler ((JobRead payload -> m (ResultOf m payload))
 -> JobHandler m payload (ResultOf m payload))
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ \JobRead payload
_job -> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (ResultOf m payload) -> m (ResultOf m payload))
-> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ do
              IO Double
getMonotonicTime IO Double -> (Double -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= IORef (Maybe Double) -> Maybe Double -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef (Maybe Double)
startedRef (Maybe Double -> IO ())
-> (Double -> Maybe Double) -> Double -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Double -> Maybe Double
forall a. a -> Maybe a
Just
              IO () -> IO ()
forall a. IO a -> IO a
uninterruptibleMask_ (Int -> IO ()
threadDelay Int
maskedMicros)
              Int -> IO ()
threadDelay Int
handlerSleepMicros
              IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
finishedRef
            hooks :: ObservabilityHooks m payload
            hooks = ObservabilityHooks m payload
forall (m :: * -> *) payload.
Applicative m =>
ObservabilityHooks m payload
defaultObservabilityHooks {onJobHeartbeat = \JobRead payload
_ UTCTime
_ UTCTime
_ -> IORef [Double] -> m ()
forall (m :: * -> *). MonadIO m => IORef [Double] -> m ()
recordBeat IORef [Double]
beatsRef}

        void (insertedId env)

        config <- transactionalWorkerConfig 1 handler
        let workerConfig =
              WorkerConfig m payload
config
                { pollInterval = 0.2
                , jitter = NoJitter
                , maxJobDuration = Just 1
                , visibilityTimeout = 6
                , jobHeartbeatInterval = 1
                , observabilityHooks = hooks
                , logConfig = silentLogConfig
                }

        withAsync (runM env $ runWorkerPool workerConfig) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ Maybe Double -> Bool
forall a. Maybe a -> Bool
isJust (Maybe Double -> Bool) -> IO (Maybe Double) -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef (Maybe Double) -> IO (Maybe Double)
forall a. IORef a -> IO a
readIORef IORef (Maybe Double)
startedRef
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
dlqWaitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ Bool -> Bool
not (Bool -> Bool)
-> ([DLQJob payload] -> Bool) -> [DLQJob payload] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [DLQJob payload] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null ([DLQJob payload] -> Bool) -> IO [DLQJob payload] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> env -> IO [DLQJob payload]
listDLQ env
env
          started <- Double -> (Double -> Double) -> Maybe Double -> Double
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Double
0 Double -> Double
forall a. a -> a
id (Maybe Double -> Double) -> IO (Maybe Double) -> IO Double
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef (Maybe Double) -> IO (Maybe Double)
forall a. IORef a -> IO a
readIORef IORef (Maybe Double)
startedRef
          beats <- readIORef beatsRef
          finished <- readIORef finishedRef
          finished `shouldBe` 0
          length (filter (< started + maskedSeconds) beats) `shouldSatisfy` (>= 2)

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"interrupts a handler that catches sync exceptions" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        startedRef <- Int -> IO (IORef Int)
forall a. a -> IO (IORef a)
newIORef (Int
0 :: Int)
        finishedRef <- newIORef (0 :: Int)
        let handler :: JobHandler m payload ()
            handler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler ((JobRead payload -> m (ResultOf m payload))
 -> JobHandler m payload (ResultOf m payload))
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ \JobRead payload
_job -> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (ResultOf m payload) -> m (ResultOf m payload))
-> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ do
              IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
startedRef
              IO (Either SomeException ()) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO () -> IO (Either SomeException ())
forall (m :: * -> *) a.
MonadUnliftIO m =>
m a -> m (Either SomeException a)
tryAny (Int -> IO ()
threadDelay Int
handlerSleepMicros))
              IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
finishedRef

        void (insertedId env)

        config :: WorkerConfig m payload <- transactionalWorkerConfig 1 handler
        let workerConfig =
              WorkerConfig m payload
config
                { pollInterval = 0.2
                , jitter = NoJitter
                , maxJobDuration = Just 1
                , logConfig = silentLogConfig
                }

        withAsync (runM env $ runWorkerPool workerConfig) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ (Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
1) (Int -> Bool) -> IO Int -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef IORef Int
startedRef
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
dlqWaitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ Bool -> Bool
not (Bool -> Bool)
-> ([DLQJob payload] -> Bool) -> [DLQJob payload] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [DLQJob payload] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null ([DLQJob payload] -> Bool) -> IO [DLQJob payload] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> env -> IO [DLQJob payload]
listDLQ env
env
          finished <- IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef IORef Int
finishedRef
          finished `shouldBe` 0

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"keeps the lease while a fenced handler unwinds" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        startedRef <- Maybe Double -> IO (IORef (Maybe Double))
forall a. a -> IO (IORef a)
newIORef (Maybe Double
forall a. Maybe a
Nothing :: Maybe Double)
        finishedRef <- newIORef (0 :: Int)
        beatsRef <- newIORef ([] :: [Double])
        reasonsRef <- newIORef ([] :: [Text])
        let handler :: JobHandler m payload ()
            handler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler ((JobRead payload -> m (ResultOf m payload))
 -> JobHandler m payload (ResultOf m payload))
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ \JobRead payload
_job ->
              IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO
                ( do
                    IO Double
getMonotonicTime IO Double -> (Double -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= IORef (Maybe Double) -> Maybe Double -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef (Maybe Double)
startedRef (Maybe Double -> IO ())
-> (Double -> Maybe Double) -> Double -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Double -> Maybe Double
forall a. a -> Maybe a
Just
                    Int -> IO ()
threadDelay Int
handlerSleepMicros
                    IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
finishedRef
                )
                m () -> m () -> m ()
forall (m :: * -> *) a b. MonadUnliftIO m => m a -> m b -> m a
`finally` IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (Int -> IO ()
threadDelay Int
unwindMicros)
            hooks :: ObservabilityHooks m payload
            hooks =
              ObservabilityHooks m payload
forall (m :: * -> *) payload.
Applicative m =>
ObservabilityHooks m payload
defaultObservabilityHooks
                { onJobHeartbeat = \JobRead payload
_ UTCTime
_ UTCTime
_ -> IORef [Double] -> m ()
forall (m :: * -> *). MonadIO m => IORef [Double] -> m ()
recordBeat IORef [Double]
beatsRef
                , onJobUnavailable = \JobRead payload
_ Text
reason ->
                    IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ IORef [Text] -> ([Text] -> ([Text], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Text]
reasonsRef (\[Text]
seen -> (Text
reason Text -> [Text] -> [Text]
forall a. a -> [a] -> [a]
: [Text]
seen, ()))
                }

        void (insertedId env)

        config <- transactionalWorkerConfig 1 handler
        let workerConfig =
              WorkerConfig m payload
config
                { pollInterval = 0.2
                , jitter = NoJitter
                , maxJobDuration = Just 1
                , visibilityTimeout = 3
                , jobHeartbeatInterval = 1
                , observabilityHooks = hooks
                , logConfig = silentLogConfig
                }

        withAsync (runM env $ runWorkerPool workerConfig) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ Maybe Double -> Bool
forall a. Maybe a -> Bool
isJust (Maybe Double -> Bool) -> IO (Maybe Double) -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef (Maybe Double) -> IO (Maybe Double)
forall a. IORef a -> IO a
readIORef IORef (Maybe Double)
startedRef
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
dlqWaitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ Bool -> Bool
not (Bool -> Bool)
-> ([DLQJob payload] -> Bool) -> [DLQJob payload] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [DLQJob payload] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null ([DLQJob payload] -> Bool) -> IO [DLQJob payload] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> env -> IO [DLQJob payload]
listDLQ env
env
          started <- Double -> (Double -> Double) -> Maybe Double -> Double
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Double
0 Double -> Double
forall a. a -> a
id (Maybe Double -> Double) -> IO (Maybe Double) -> IO Double
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef (Maybe Double) -> IO (Maybe Double)
forall a. IORef a -> IO a
readIORef IORef (Maybe Double)
startedRef
          beats <- readIORef beatsRef
          reasons <- readIORef reasonsRef
          finished <- readIORef finishedRef
          finished `shouldBe` 0
          reasons `shouldBe` []
          dlq <- listDLQ env
          map (lastError . jobSnapshot) dlq
            `shouldBe` [Just "claim hooks and handler ran past the maximum job duration of 1s"]
          length (filter (> started + 1) beats) `shouldSatisfy` (>= 2)

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Lease fence" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"does not carry the fence past a row its extend did not move" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        startedRef <- Int -> IO (IORef Int)
forall a. a -> IO (IORef a)
newIORef (Int
0 :: Int)
        finishedRef <- newIORef (0 :: Int)
        reasonsRef <- newIORef ([] :: [Text])
        let handler :: JobHandler m payload ()
            handler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler ((JobRead payload -> m (ResultOf m payload))
 -> JobHandler m payload (ResultOf m payload))
-> (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ \JobRead payload
_job -> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (ResultOf m payload) -> m (ResultOf m payload))
-> IO (ResultOf m payload) -> m (ResultOf m payload)
forall a b. (a -> b) -> a -> b
$ do
              IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
startedRef
              Int -> IO ()
threadDelay Int
handlerSleepMicros
              IORef Int -> IO ()
forall a. Num a => IORef a -> IO ()
bumpRef IORef Int
finishedRef
            hooks :: ObservabilityHooks m payload
            hooks =
              ObservabilityHooks m payload
forall (m :: * -> *) payload.
Applicative m =>
ObservabilityHooks m payload
defaultObservabilityHooks
                { onJobUnavailable = \JobRead payload
_ Text
reason ->
                    IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ IORef [Text] -> ([Text] -> ([Text], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Text]
reasonsRef (\[Text]
seen -> (Text
reason Text -> [Text] -> [Text]
forall a. a -> [a] -> [a]
: [Text]
seen, ()))
                }

        void (insertedPlainId env)

        config <- transactionalWorkerConfig 1 handler
        let workerConfig =
              WorkerConfig m payload
config
                { pollInterval = 0.2
                , jitter = NoJitter
                , visibilityTimeout = 5
                , jobHeartbeatInterval = 2
                , observabilityHooks = hooks
                , logConfig = silentLogConfig
                }

        withAsync (runM env $ runWorkerPool workerConfig) $ \Async ()
_ -> do
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
waitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ (Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
1) (Int -> Bool) -> IO Int -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef IORef Int
startedRef
          ByteString -> Text -> Text -> IO ()
takeClaimHolder ByteString
connStr Text
schema Text
table
          HasCallStack => Int -> IO Bool -> IO ()
Int -> IO Bool -> IO ()
waitUntil Int
slowWaitMillis (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ Bool -> Bool
not (Bool -> Bool) -> ([Text] -> Bool) -> [Text] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Text] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null ([Text] -> Bool) -> IO [Text] -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef [Text] -> IO [Text]
forall a. IORef a -> IO a
readIORef IORef [Text]
reasonsRef
          finished <- IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef IORef Int
finishedRef
          finished `shouldBe` 0
          reasons <- readIORef reasonsRef
          reasons `shouldBe` [leaseExpiredReason]
  where
    inserted :: env -> JobWrite payload -> IO (JobRead payload)
    inserted :: env -> JobWrite payload -> IO (JobRead payload)
inserted env
env JobWrite payload
job = env -> m (Maybe (JobRead payload)) -> IO (Maybe (JobRead payload))
forall a. env -> m a -> IO a
runM env
env (JobWrite payload -> m (Maybe (JobRead payload))
forall payload (m :: * -> *).
QueueOperation m payload =>
JobWrite payload -> m (Maybe (JobRead payload))
HL.insertJob JobWrite payload
job) IO (Maybe (JobRead payload))
-> (Maybe (JobRead payload) -> IO (JobRead payload))
-> IO (JobRead payload)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= IO (JobRead payload)
-> (JobRead payload -> IO (JobRead payload))
-> Maybe (JobRead payload)
-> IO (JobRead payload)
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (String -> IO (JobRead payload)
forall a. HasCallStack => String -> IO a
forall (m :: * -> *) a.
(MonadFail m, HasCallStack) =>
String -> m a
fail String
"insert returned no job") JobRead payload -> IO (JobRead payload)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    insertedPlainId :: env -> IO Int64
    insertedPlainId :: env -> IO Int64
insertedPlainId env
env = JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey (JobRead payload -> Int64) -> IO (JobRead payload) -> IO Int64
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> env -> JobWrite payload -> IO (JobRead payload)
inserted env
env (payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"slow"))
    insertedId :: env -> IO Int64
    insertedId :: env -> IO Int64
insertedId env
env = JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey (JobRead payload -> Int64) -> IO (JobRead payload) -> IO Int64
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> env -> JobWrite payload -> IO (JobRead payload)
inserted env
env (Maybe Int32 -> JobWrite payload -> JobWrite payload
forall payload. Maybe Int32 -> JobWrite payload -> JobWrite payload
setMaxAttempts (Int32 -> Maybe Int32
forall a. a -> Maybe a
Just Int32
1) (payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"slow")))
    idleHandler :: JobHandler m payload ()
    idleHandler :: JobHandler m payload ()
idleHandler = (JobRead payload -> m (ResultOf m payload))
-> JobHandler m payload (ResultOf m payload)
mkHandler (\JobRead payload
_job -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
    listDLQ :: env -> IO [DLQJob payload]
    listDLQ :: env -> IO [DLQJob payload]
listDLQ env
env = env -> m [DLQJob payload] -> IO [DLQJob payload]
forall a. env -> m a -> IO a
runM env
env (Int -> Int -> m [DLQJob payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> Int -> m [DLQJob payload]
HL.listDLQJobs Int
10 Int
0)

maskedSeconds :: Double
maskedSeconds :: Double
maskedSeconds = Int -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
maskedMicros Double -> Double -> Double
forall a. Fractional a => a -> a -> a
/ Double
1_000_000

recordBeat :: (MonadIO m) => IORef [Double] -> m ()
recordBeat :: forall (m :: * -> *). MonadIO m => IORef [Double] -> m ()
recordBeat IORef [Double]
beatsRef = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ IO Double
getMonotonicTime IO Double -> (Double -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \Double
now -> IORef [Double] -> ([Double] -> ([Double], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Double]
beatsRef (\[Double]
beats -> (Double
now Double -> [Double] -> [Double]
forall a. a -> [a] -> [a]
: [Double]
beats, ()))

single :: [a] -> IO a
single :: forall a. [a] -> IO a
single [a
x] = a -> IO a
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure a
x
single [a]
_ = String -> IO a
forall a. HasCallStack => String -> IO a
forall (m :: * -> *) a.
(MonadFail m, HasCallStack) =>
String -> m a
fail String
"expected exactly one job"

isForceCancelled :: SomeException -> Bool
isForceCancelled :: SomeException -> Bool
isForceCancelled SomeException
exc = Maybe JobForceCancelled -> Bool
forall a. Maybe a -> Bool
isJust (SomeException -> Maybe JobForceCancelled
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exc :: Maybe JobForceCancelled)

stamp :: IORef [(Int64, Double)] -> JobRead payload -> IO ()
stamp :: forall payload. IORef [(Int64, Double)] -> JobRead payload -> IO ()
stamp IORef [(Int64, Double)]
ref JobRead payload
job = IO Double
getMonotonicTime IO Double -> (Double -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \Double
now -> IORef [(Int64, Double)]
-> ([(Int64, Double)] -> ([(Int64, Double)], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [(Int64, Double)]
ref (\[(Int64, Double)]
seen -> ((JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job, Double
now) (Int64, Double) -> [(Int64, Double)] -> [(Int64, Double)]
forall a. a -> [a] -> [a]
: [(Int64, Double)]
seen, ()))