{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TypeFamilies #-}
{-# OPTIONS_GHC -Wno-x-partial #-}

-- | Concurrency tests for any 'MonadArbiter' backend.
module Arbiter.Test.Concurrency
  ( concurrencySpec
  , raceConditionSpec
  , drainAll
  , findDuplicates
  , installHolDetector
  , holViolations
  , removeHolDetector
  ) where

import Arbiter.Core.Codec (Col (..), col)
import Arbiter.Core.HighLevel qualified as HL
import Arbiter.Core.Job.Schema (jobQueueTable)
import Arbiter.Core.Job.Types
import Arbiter.Core.MonadArbiter (MonadArbiter, RegistryOf, executeQuery, getSchema, withDbTransaction)
import Arbiter.Core.QueueRegistry (TableForPayload)
import Arbiter.Core.Sql.Query (rawRows)
import Control.Concurrent (threadDelay)
import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
import Control.Monad (replicateM, replicateM_, void, when)
import Control.Monad.IO.Class (liftIO)
import Data.Foldable (for_)
import Data.IORef (atomicModifyIORef', newIORef, readIORef)
import Data.Int (Int64)
import Data.List (sort)
import Data.List.NonEmpty qualified as NE
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (addUTCTime, getCurrentTime)
import Data.Traversable (for)
import GHC.TypeLits (KnownSymbol)
import Test.Hspec
import UnliftIO.Async (mapConcurrently, replicateConcurrently_, withAsync)

import Arbiter.Test.Poll (bumpRef)
import Arbiter.Test.StateMachine (holViolations, installHolDetector, removeHolDetector)

-- | A pause after an empty claim, before the next claim.
drainIdleMicros :: Int
drainIdleMicros :: Int
drainIdleMicros = Int
10_000

-- | A pause into the lease, before the extension.
beforeExtendMicros :: Int
beforeExtendMicros :: Int
beforeExtendMicros = Int
1_000_000

-- | Past the original lease.
pastOriginalLeaseMicros :: Int
pastOriginalLeaseMicros :: Int
pastOriginalLeaseMicros = Int
2_000_000

-- | Past a 0.2 s lease.
pastShortLeaseMicros :: Int
pastShortLeaseMicros :: Int
pastShortLeaseMicros = Int
400_000

-- | Past a 1 s lease.
pastOneSecondLeaseMicros :: Int
pastOneSecondLeaseMicros :: Int
pastOneSecondLeaseMicros = Int
1_200_000

-- | How long a claimer holds a job before the ack.
holdBeforeAckMicros :: Int
holdBeforeAckMicros :: Int
holdBeforeAckMicros = Int
50_000

-- | Claim and ack jobs in a loop until no more are available.
-- Backs off with 10ms delay between empty claim attempts, gives up
-- after @maxStreak@ retries of an empty claim.
claimAckLoop
  :: IO [a]
  -> Int
  -- ^ Max retries of an empty claim
  -> (a -> IO ())
  -- ^ Per-job action
  -> IO ()
claimAckLoop :: forall a. IO [a] -> Int -> (a -> IO ()) -> IO ()
claimAckLoop IO [a]
claim Int
maxStreak a -> IO ()
onJob = Int -> IO ()
go Int
0
  where
    go :: Int -> IO ()
go Int
streak = do
      jobs <- IO [a]
claim
      if null jobs
        then when (streak < maxStreak) $ threadDelay drainIdleMicros >> go (streak + 1)
        else for_ jobs onJob >> go 0

-- | Claim and run jobs until a claim returns none.
drainAll :: IO [a] -> (a -> IO ()) -> IO ()
drainAll :: forall a. IO [a] -> (a -> IO ()) -> IO ()
drainAll IO [a]
claim a -> IO ()
onJob = do
  jobs <- IO [a]
claim
  if null jobs then pure () else for_ jobs onJob >> drainAll claim onJob

-- | Find duplicate entries in a list. O(n log n).
findDuplicates :: (Ord a) => [a] -> [a]
findDuplicates :: forall a. Ord a => [a] -> [a]
findDuplicates = Set a -> Set a -> [a] -> [a]
forall {a}. Ord a => Set a -> Set a -> [a] -> [a]
go Set a
forall a. Set a
Set.empty Set a
forall a. Set a
Set.empty
  where
    go :: Set a -> Set a -> [a] -> [a]
go Set a
_ Set a
dups [] = Set a -> [a]
forall a. Set a -> [a]
Set.toList Set a
dups
    go Set a
seen Set a
dups (a
item : [a]
rest)
      | a -> Set a -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member a
item Set a
seen = Set a -> Set a -> [a] -> [a]
go Set a
seen (a -> Set a -> Set a
forall a. Ord a => a -> Set a -> Set a
Set.insert a
item Set a
dups) [a]
rest
      | Bool
otherwise = Set a -> Set a -> [a] -> [a]
go (a -> Set a -> Set a
forall a. Ord a => a -> Set a -> Set a
Set.insert a
item Set a
seen) Set a
dups [a]
rest

-- | Lock a job's row with the given lock suffix. Returns the ids it locked.
lockJobRow :: (MonadArbiter m) => Text -> JobRead payload -> m [Int64]
lockJobRow :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> JobRead payload -> m [Int64]
lockJobRow Text
suffix JobRead payload
job = do
  schema <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  executeQuery . rawRows (col "id" CInt8) $
    "SELECT id FROM "
      <> jobQueueTable schema (queueName job)
      <> " WHERE id = "
      <> T.pack (show (primaryKey job))
      <> " FOR UPDATE"
      <> suffix

-- | Parameterized concurrency test suite. These tests need a connection pool of
-- at least 10 connections.
-- Build the schema with 'Arbiter.Test.Setup.setupOnce' and empty it before each test with 'Arbiter.Test.Setup.cleanupData'.
concurrencySpec
  :: forall payload m env
   . ( JobPayload payload
     , KnownSymbol (TableForPayload payload (RegistryOf m))
     , MonadArbiter m
     )
  => (Text -> payload)
  -- ^ Constructor for a simple test message payload
  -> (forall a. env -> m a -> IO a)
  -- ^ Runner for monad actions
  -> SpecWith env
concurrencySpec :: forall payload (m :: * -> *) env.
(JobPayload payload,
 KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> payload) -> (forall a. env -> m a -> IO a) -> SpecWith env
concurrencySpec Text -> payload
mkMessage forall a. env -> m a -> IO a
runM = do
  String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Heartbeat Visibility Across Connections" (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
"setVisibilityTimeout updates are immediately visible to concurrent workers" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      -- Insert a job
      IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ())
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ()
forall a b. (a -> b) -> a -> b
$ env
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a. env -> m a -> IO a
runM env
env (m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
 -> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)))
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a b. (a -> b) -> a -> b
$ JobWrite payload
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
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
"heartbeat-visibility-test") (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"Heartbeat"))

      -- Worker A claims the job with 5 second visibility
      claimed1 <- env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (Int
-> NominalDiffTime
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
1 NominalDiffTime
5) :: IO [JobRead payload]
      length claimed1 `shouldBe` 1
      let job = [Job payload Int64 Text UTCTime PayloadKeys]
-> Job payload Int64 Text UTCTime PayloadKeys
forall a. HasCallStack => [a] -> a
head [Job payload Int64 Text UTCTime PayloadKeys]
claimed1

      -- Worker A extends visibility to 60 seconds, as a heartbeat.
      rowsUpdated <- runM env (HL.setVisibilityTimeout 60 job)
      rowsUpdated `shouldBe` 1

      -- Worker B claims on a different pool connection and gets nothing.
      claimed2 <- runM env (HL.claimNextVisibleJobs 1 5) :: IO [JobRead payload]

      length claimed2 `shouldBe` 0

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"heartbeat prevents reclaim after original visibility timeout expires" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      -- Insert a job
      IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ())
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ()
forall a b. (a -> b) -> a -> b
$ env
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a. env -> m a -> IO a
runM env
env (m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
 -> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)))
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a b. (a -> b) -> a -> b
$ JobWrite payload
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
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
"heartbeat-timeout-test") (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"Timeout"))

      -- Claim with a 2 second visibility.
      claimed1 <- env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (Int
-> NominalDiffTime
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
1 NominalDiffTime
2) :: IO [JobRead payload]
      length claimed1 `shouldBe` 1
      let job = [Job payload Int64 Text UTCTime PayloadKeys]
-> Job payload Int64 Text UTCTime PayloadKeys
forall a. HasCallStack => [a] -> a
head [Job payload Int64 Text UTCTime PayloadKeys]
claimed1

      -- Wait 1 second, then extend visibility by 10 more seconds
      threadDelay beforeExtendMicros
      void $ runM env (HL.setVisibilityTimeout 10 job)

      -- Wait 2 more seconds, past the original timeout.
      threadDelay pastOriginalLeaseMicros

      -- The extended visibility blocks a claim.
      claimed2 <- runM env (HL.claimNextVisibleJobs 1 2) :: IO [JobRead payload]
      length claimed2 `shouldBe` 0

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"visibility timeout of 0 makes job immediately claimable" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      -- Insert and claim job
      IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ())
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ()
forall a b. (a -> b) -> a -> b
$ env
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a. env -> m a -> IO a
runM env
env (m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
 -> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)))
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a b. (a -> b) -> a -> b
$ JobWrite payload
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
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
"visibility-zero-test") (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"Zero"))
      claimed1 <- env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (Int
-> NominalDiffTime
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
1 NominalDiffTime
60) :: IO [JobRead payload]
      length claimed1 `shouldBe` 1
      let job = [Job payload Int64 Text UTCTime PayloadKeys]
-> Job payload Int64 Text UTCTime PayloadKeys
forall a. HasCallStack => [a] -> a
head [Job payload Int64 Text UTCTime PayloadKeys]
claimed1

      -- Set visibility to 0.
      void $ runM env (HL.setVisibilityTimeout 0 job)

      -- The job is claimable at once.
      claimed2 <- runM env (HL.claimNextVisibleJobs 1 60) :: IO [JobRead payload]
      length claimed2 `shouldBe` 1

  String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Concurrent Job Claims" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
    let claimsPastLocked :: env -> [Job payload Int64 Text UTCTime PayloadKeys] -> IO ()
claimsPastLocked env
env [Job payload Int64 Text UTCTime PayloadKeys]
jobs = do
          let ([Job payload Int64 Text UTCTime PayloadKeys]
locked, [Job payload Int64 Text UTCTime PayloadKeys]
free) = Int
-> [Job payload Int64 Text UTCTime PayloadKeys]
-> ([Job payload Int64 Text UTCTime PayloadKeys],
    [Job payload Int64 Text UTCTime PayloadKeys])
forall a. Int -> [a] -> ([a], [a])
splitAt Int
30 [Job payload Int64 Text UTCTime PayloadKeys]
jobs
          held <- IO (MVar ())
forall a. IO (MVar a)
newEmptyMVar
          release <- newEmptyMVar
          withAsync
            ( runM env $ withDbTransaction $ do
                for_ locked (lockJobRow "")
                liftIO $ putMVar held () >> takeMVar release
            )
            $ \Async ()
_ -> do
              MVar () -> IO ()
forall a. MVar a -> IO a
takeMVar MVar ()
held
              claimed <- env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (Int
-> NominalDiffTime
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
1 NominalDiffTime
60) :: IO [JobRead payload]
              putMVar release ()
              map primaryKey claimed `shouldBe` map primaryKey free

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"claims past many locked ready ungrouped rows at capacity one" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      inserted <- env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (m [Job payload Int64 Text UTCTime PayloadKeys]
 -> IO [Job payload Int64 Text UTCTime PayloadKeys])
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload]
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch (Int -> JobWrite payload -> [JobWrite payload]
forall a. Int -> a -> [a]
replicate Int
31 (payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"ready")))
      claimsPastLocked env inserted

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"claims past many locked due ungrouped rows at capacity one" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      now <- IO UTCTime
getCurrentTime
      let dueJob Int
i = Maybe UTCTime -> JobWrite payload -> JobWrite payload
forall payload.
Maybe UTCTime -> JobWrite payload -> JobWrite payload
setNotVisibleUntil (UTCTime -> Maybe UTCTime
forall a. a -> Maybe a
Just (NominalDiffTime -> UTCTime -> UTCTime
addUTCTime (Int -> NominalDiffTime
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int
i Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
3600 :: Int)) UTCTime
now)) (payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"due"))
      inserted <- runM env $ HL.insertJobsBatch (map dueJob [1 .. 31])
      claimsPastLocked env inserted

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"leaves ungrouped rows it does not claim unlocked" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      [grouped, ungrouped] <-
        env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (m [Job payload Int64 Text UTCTime PayloadKeys]
 -> IO [Job payload Int64 Text UTCTime PayloadKeys])
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a b. (a -> b) -> a -> b
$
          [JobWrite payload]
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch
            [ Int32 -> JobWrite payload -> JobWrite payload
forall payload. Int32 -> JobWrite payload -> JobWrite payload
setPriority Int32
0 (Text -> payload -> JobWrite payload
forall payload. Text -> payload -> JobWrite payload
defaultGroupedJob Text
"first" (Text -> payload
mkMessage Text
"grouped"))
            , Int32 -> JobWrite payload -> JobWrite payload
forall payload. Int32 -> JobWrite payload -> JobWrite payload
setPriority Int32
5 (payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"ungrouped"))
            ]
      held <- newEmptyMVar
      release <- newEmptyMVar
      withAsync
        ( runM env $ withDbTransaction $ do
            claimed <- HL.claimNextVisibleJobs 1 60
            liftIO $ putMVar held (map primaryKey (claimed :: [JobRead payload])) >> takeMVar release
        )
        $ \Async ()
_ -> do
          claimed <- MVar [Int64] -> IO [Int64]
forall a. MVar a -> IO a
takeMVar MVar [Int64]
held
          lockable <- runM env $ withDbTransaction (lockJobRow " SKIP LOCKED" ungrouped)
          putMVar release ()
          claimed `shouldBe` [primaryKey grouped]
          lockable `shouldBe` [primaryKey ungrouped]

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"claims healthy work beyond exhausted group heads" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
        (IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ())
-> IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ()
forall a b. (a -> b) -> a -> b
$ env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env
        (m [Job payload Int64 Text UTCTime PayloadKeys]
 -> IO [Job payload Int64 Text UTCTime PayloadKeys])
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload]
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch
          [ 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) (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ Text -> payload -> JobWrite payload
forall payload. Text -> payload -> JobWrite payload
defaultGroupedJob (Text
"exhausted-" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show Int
i)) (Text -> payload
mkMessage Text
"old") | Int
i <- [Int
1 .. Int
12 :: Int]
          ]
      exhausted <- env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (Int
-> NominalDiffTime
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
12 NominalDiffTime
60) :: IO [JobRead payload]
      for_ exhausted $ \Job payload Int64 Text UTCTime PayloadKeys
job -> IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (NominalDiffTime
-> Job payload Int64 Text UTCTime PayloadKeys -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
NominalDiffTime -> JobRead payload -> m Int64
HL.setVisibilityTimeout NominalDiffTime
0 Job payload Int64 Text UTCTime PayloadKeys
job)
      Just healthy <- runM env $ HL.insertJob (defaultGroupedJob "healthy" (mkMessage "healthy"))
      claimed <- runM env (HL.claimNextVisibleJobs 1 60) :: IO [JobRead payload]
      map primaryKey claimed `shouldBe` [primaryKey healthy]

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"claims healthy work beyond exhausted due group heads" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
        (IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ())
-> IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ()
forall a b. (a -> b) -> a -> b
$ env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env
        (m [Job payload Int64 Text UTCTime PayloadKeys]
 -> IO [Job payload Int64 Text UTCTime PayloadKeys])
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload]
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch
          [ 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) (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ Text -> payload -> JobWrite payload
forall payload. Text -> payload -> JobWrite payload
defaultGroupedJob (Text
"exhausted-due-" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show Int
i)) (Text -> payload
mkMessage Text
"old")
          | Int
i <- [Int
1 .. Int
12 :: Int]
          ]
      exhausted <- env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (Int
-> NominalDiffTime
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
12 NominalDiffTime
0.2) :: IO [JobRead payload]
      length exhausted `shouldBe` 12
      threadDelay pastShortLeaseMicros
      Just healthy <- runM env $ HL.insertJob (defaultGroupedJob "healthy-due" (mkMessage "healthy"))
      claimed <- runM env (HL.claimNextVisibleJobs 1 60) :: IO [JobRead payload]
      map primaryKey claimed `shouldBe` [primaryKey healthy]

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"claims healthy work beyond force-cancelled heads whose lease lapsed" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
        (IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ())
-> IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ()
forall a b. (a -> b) -> a -> b
$ env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env
        (m [Job payload Int64 Text UTCTime PayloadKeys]
 -> IO [Job payload Int64 Text UTCTime PayloadKeys])
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload]
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch
          [Text -> payload -> JobWrite payload
forall payload. Text -> payload -> JobWrite payload
defaultGroupedJob (Text
"flagged-due-" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show Int
i)) (Text -> payload
mkMessage Text
"old") | Int
i <- [Int
1 .. Int
12 :: Int]]
      flagged <- env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (Int
-> NominalDiffTime
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
12 NominalDiffTime
1) :: IO [JobRead payload]
      length flagged `shouldBe` 12
      runM env $ withDbTransaction $ for_ flagged $ \Job payload Int64 Text UTCTime PayloadKeys
job -> forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
HL.forceCancelJob @payload (Job payload Int64 Text UTCTime PayloadKeys -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey Job payload Int64 Text UTCTime PayloadKeys
job)
      threadDelay pastOneSecondLeaseMicros
      Just healthy <- runM env $ HL.insertJob (defaultGroupedJob "healthy-flagged" (mkMessage "healthy"))
      claimed <- runM env (HL.claimNextVisibleJobs 1 60) :: IO [JobRead payload]
      map primaryKey claimed `shouldBe` [primaryKey healthy]
      live <- runM env (HL.listJobs @payload 20 0)
      length live `shouldBe` 13

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"concurrent workers claim disjoint ungrouped jobs" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      -- Insert 6 ungrouped jobs, remembering their ids.
      inserted <- env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (m [Job payload Int64 Text UTCTime PayloadKeys]
 -> IO [Job payload Int64 Text UTCTime PayloadKeys])
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload]
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch (Int -> JobWrite payload -> [JobWrite payload]
forall a. Int -> a -> [a]
replicate Int
6 (JobWrite payload -> [JobWrite payload])
-> JobWrite payload -> [JobWrite payload]
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"Concurrent"))
      let insertedIds = [Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList ((Job payload Int64 Text UTCTime PayloadKeys -> Int64)
-> [Job payload Int64 Text UTCTime PayloadKeys] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map Job payload Int64 Text UTCTime PayloadKeys -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey [Job payload Int64 Text UTCTime PayloadKeys]
inserted)

      [claimsA, claimsB, claimsC] <-
        mapConcurrently
          (\Int
_ -> env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (Int
-> NominalDiffTime
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
2 NominalDiffTime
60) :: IO [JobRead payload])
          [1 :: Int, 2, 3]

      let raced = (Job payload Int64 Text UTCTime PayloadKeys -> Int64)
-> [Job payload Int64 Text UTCTime PayloadKeys] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map Job payload Int64 Text UTCTime PayloadKeys -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey ([Job payload Int64 Text UTCTime PayloadKeys]
claimsA [Job payload Int64 Text UTCTime PayloadKeys]
-> [Job payload Int64 Text UTCTime PayloadKeys]
-> [Job payload Int64 Text UTCTime PayloadKeys]
forall a. Semigroup a => a -> a -> a
<> [Job payload Int64 Text UTCTime PayloadKeys]
claimsB [Job payload Int64 Text UTCTime PayloadKeys]
-> [Job payload Int64 Text UTCTime PayloadKeys]
-> [Job payload Int64 Text UTCTime PayloadKeys]
forall a. Semigroup a => a -> a -> a
<> [Job payload Int64 Text UTCTime PayloadKeys]
claimsC)
      -- No job claimed twice in the racy pass.
      findDuplicates raced `shouldBe` []
      -- Drain the jobs the race missed. A follow-up claim surfaces the unclaimed ones.
      drainedRef <- newIORef ([] :: [Int64])
      drainAll (runM env (HL.claimNextVisibleJobs 6 60) :: IO [JobRead payload]) $ \Job payload Int64 Text UTCTime PayloadKeys
job ->
        IORef [Int64] -> ([Int64] -> ([Int64], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Int64]
drainedRef (\[Int64]
acc -> (Job payload Int64 Text UTCTime PayloadKeys -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey Job payload Int64 Text UTCTime PayloadKeys
job Int64 -> [Int64] -> [Int64]
forall a. a -> [a] -> [a]
: [Int64]
acc, ()))
      drained <- readIORef drainedRef
      -- Every inserted job was claimed exactly once across the racing workers and the drain.
      let allClaimed = [Int64]
raced [Int64] -> [Int64] -> [Int64]
forall a. Semigroup a => a -> a -> a
<> [Int64]
drained
      findDuplicates allClaimed `shouldBe` []
      Set.fromList allClaimed `shouldBe` insertedIds

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"concurrent workers respect per-group ordering for grouped jobs" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      -- Each round seeds a group and races 10 claimers with 2 inserters on the
      -- same group. Per-group ordering allows at most 1 claim per round.
      let numIterations :: Int
numIterations = Int
500
          numWorkers :: Int
numWorkers = Int
10
          seedPerRound :: Int
seedPerRound = Int
5
          insertersPerRound :: Int
insertersPerRound = Int
2
      violationRef <- Int -> IO (IORef Int)
forall a. a -> IO (IORef a)
newIORef (Int
0 :: Int)
      insertedRef <- newIORef (0 :: Int)
      processedRef <- newIORef (Set.empty :: Set.Set Int64)

      for_ [(1 :: Int) .. numIterations] $ \Int
_ -> do
        -- Insert 5 jobs in one group
        IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
          (IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ())
-> IO [Job payload Int64 Text UTCTime PayloadKeys] -> IO ()
forall a b. (a -> b) -> a -> b
$ env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env
          (m [Job payload Int64 Text UTCTime PayloadKeys]
 -> IO [Job payload Int64 Text UTCTime PayloadKeys])
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload]
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch (Int -> JobWrite payload -> [JobWrite payload]
forall a. Int -> a -> [a]
replicate Int
seedPerRound (JobWrite payload -> [JobWrite payload])
-> JobWrite payload -> [JobWrite payload]
forall a b. (a -> b) -> a -> b
$ 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
"hol-stress") (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"Grouped"))
        IORef Int -> (Int -> (Int, ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef Int
insertedRef (\Int
count -> (Int
count Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
seedPerRound Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
insertersPerRound, ()))

        -- Race the claimers and inserters on the same group.
        results <-
          (IO [Job payload Int64 Text UTCTime PayloadKeys]
 -> IO [Job payload Int64 Text UTCTime PayloadKeys])
-> [IO [Job payload Int64 Text UTCTime PayloadKeys]]
-> IO [[Job payload Int64 Text UTCTime PayloadKeys]]
forall (m :: * -> *) (t :: * -> *) a b.
(MonadUnliftIO m, Traversable t) =>
(a -> m b) -> t a -> m (t b)
mapConcurrently
            IO [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. a -> a
id
            ([IO [Job payload Int64 Text UTCTime PayloadKeys]]
 -> IO [[Job payload Int64 Text UTCTime PayloadKeys]])
-> [IO [Job payload Int64 Text UTCTime PayloadKeys]]
-> IO [[Job payload Int64 Text UTCTime PayloadKeys]]
forall a b. (a -> b) -> a -> b
$ (Int -> IO [Job payload Int64 Text UTCTime PayloadKeys])
-> [Int] -> [IO [Job payload Int64 Text UTCTime PayloadKeys]]
forall a b. (a -> b) -> [a] -> [b]
map
              (\Int
_ -> env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (Int
-> NominalDiffTime
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
1 NominalDiffTime
60) :: IO [JobRead payload])
              [(Int
1 :: Int) .. Int
numWorkers]
              [IO [Job payload Int64 Text UTCTime PayloadKeys]]
-> [IO [Job payload Int64 Text UTCTime PayloadKeys]]
-> [IO [Job payload Int64 Text UTCTime PayloadKeys]]
forall a. Semigroup a => a -> a -> a
<> Int
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
-> [IO [Job payload Int64 Text UTCTime PayloadKeys]]
forall a. Int -> a -> [a]
replicate
                Int
insertersPerRound
                (IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (env
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a. env -> m a -> IO a
runM env
env (m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
 -> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)))
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a b. (a -> b) -> a -> b
$ JobWrite payload
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall payload (m :: * -> *).
QueueOperation m payload =>
JobWrite payload -> m (Maybe (JobRead payload))
HL.insertJob (JobWrite payload
 -> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys)))
-> JobWrite payload
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a b. (a -> b) -> a -> b
$ 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
"hol-stress") (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"concurrent")) IO ()
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [])

        -- At most 1 claim per group. A round with no claim is normal when the
        -- insert trigger holds the groups row lock.
        let totalClaimed = [Int] -> Int
forall a. Num a => [a] -> a
forall (t :: * -> *) a. (Foldable t, Num a) => t a -> a
sum (([Job payload Int64 Text UTCTime PayloadKeys] -> Int)
-> [[Job payload Int64 Text UTCTime PayloadKeys]] -> [Int]
forall a b. (a -> b) -> [a] -> [b]
map [Job payload Int64 Text UTCTime PayloadKeys] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [[Job payload Int64 Text UTCTime PayloadKeys]]
results)
        when (totalClaimed > 1) $
          bumpRef violationRef

        -- Ack the racy claims, then drain the rest, recording every processed id.
        let record Job payload Int64 Text UTCTime PayloadKeys
job = IORef (Set Int64) -> (Set Int64 -> (Set Int64, ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef (Set Int64)
processedRef (\Set Int64
acc -> (Int64 -> Set Int64 -> Set Int64
forall a. Ord a => a -> Set a -> Set a
Set.insert (Job payload Int64 Text UTCTime PayloadKeys -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey Job payload Int64 Text UTCTime PayloadKeys
job) Set Int64
acc, ()))
        for_ (concat results) $ \Job payload Int64 Text UTCTime PayloadKeys
job -> Job payload Int64 Text UTCTime PayloadKeys -> IO ()
record Job payload Int64 Text UTCTime PayloadKeys
job IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (Job payload Int64 Text UTCTime PayloadKeys -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob Job payload Int64 Text UTCTime PayloadKeys
job))
        drainAll (runM env (HL.claimNextVisibleJobs 100 60) :: IO [JobRead payload]) $
          \Job payload Int64 Text UTCTime PayloadKeys
job -> Job payload Int64 Text UTCTime PayloadKeys -> IO ()
record Job payload Int64 Text UTCTime PayloadKeys
job IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (Job payload Int64 Text UTCTime PayloadKeys -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob Job payload Int64 Text UTCTime PayloadKeys
job))

      violations <- readIORef violationRef
      violations `shouldBe` 0
      -- Every job inserted across all rounds was claimed and acked exactly once.
      inserted <- readIORef insertedRef
      processed <- readIORef processedRef
      Set.size processed `shouldBe` inserted

    String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"FOR UPDATE SKIP LOCKED prevents concurrent claims of same job" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
      -- Insert a single job
      IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ())
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)) -> IO ()
forall a b. (a -> b) -> a -> b
$ env
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a. env -> m a -> IO a
runM env
env (m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
 -> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys)))
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
-> IO (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
forall a b. (a -> b) -> a -> b
$ JobWrite payload
-> m (Maybe (Job payload Int64 Text UTCTime PayloadKeys))
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
"skip-locked-test") (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"Single"))

      -- Three workers race to claim the same job.
      [claimsA, claimsB, claimsC] <-
        (Int -> IO [Job payload Int64 Text UTCTime PayloadKeys])
-> [Int] -> IO [[Job payload Int64 Text UTCTime PayloadKeys]]
forall (m :: * -> *) (t :: * -> *) a b.
(MonadUnliftIO m, Traversable t) =>
(a -> m b) -> t a -> m (t b)
mapConcurrently
          (\Int
_ -> env
-> m [Job payload Int64 Text UTCTime PayloadKeys]
-> IO [Job payload Int64 Text UTCTime PayloadKeys]
forall a. env -> m a -> IO a
runM env
env (Int
-> NominalDiffTime
-> m [Job payload Int64 Text UTCTime PayloadKeys]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
2 NominalDiffTime
60) :: IO [JobRead payload])
          [Int
1 :: Int, Int
2, Int
3]

      -- Exactly one worker claims it.
      let totalClaimed = [Job payload Int64 Text UTCTime PayloadKeys] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [Job payload Int64 Text UTCTime PayloadKeys]
claimsA Int -> Int -> Int
forall a. Num a => a -> a -> a
+ [Job payload Int64 Text UTCTime PayloadKeys] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [Job payload Int64 Text UTCTime PayloadKeys]
claimsB Int -> Int -> Int
forall a. Num a => a -> a -> a
+ [Job payload Int64 Text UTCTime PayloadKeys] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [Job payload Int64 Text UTCTime PayloadKeys]
claimsC
      totalClaimed `shouldBe` 1

-- | High-contention tests for concurrency defects. Many workers, large job
-- counts, and tight timing windows.
-- Build the schema with 'Arbiter.Test.Setup.setupOnce' and empty it before each test with 'Arbiter.Test.Setup.cleanupData'.
raceConditionSpec
  :: forall payload m env
   . ( JobPayload payload
     , KnownSymbol (TableForPayload payload (RegistryOf m))
     , MonadArbiter m
     )
  => (Text -> payload)
  -- ^ Constructor for a simple test message payload
  -> (forall a. env -> m a -> IO a)
  -- ^ Runner for monad actions
  -> SpecWith env
raceConditionSpec :: forall payload (m :: * -> *) env.
(JobPayload payload,
 KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> payload) -> (forall a. env -> m a -> IO a) -> SpecWith env
raceConditionSpec Text -> payload
mkMessage forall a. env -> m a -> IO a
runM = do
  String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Race Condition Stress Tests" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"High-Contention Claim Stress" (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
"no duplicate claims under extreme contention (ungrouped)" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let numJobs :: Int
numJobs = Int
500
            numWorkers :: Int
numWorkers = Int
40

        IO [JobRead payload] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO [JobRead payload] -> IO ()) -> IO [JobRead payload] -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload] -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch (Int -> JobWrite payload -> [JobWrite payload]
forall a. Int -> a -> [a]
replicate Int
numJobs (JobWrite payload -> [JobWrite payload])
-> JobWrite payload -> [JobWrite payload]
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"stress"))

        claimedRef <- [Int64] -> IO (IORef [Int64])
forall a. a -> IO (IORef a)
newIORef ([] :: [Int64])

        replicateConcurrently_ numWorkers $
          claimAckLoop (runM env (HL.claimNextVisibleJobs 3 60) :: IO [JobRead payload]) 20 $ \JobRead payload
job -> do
            IORef [Int64] -> ([Int64] -> ([Int64], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Int64]
claimedRef (\[Int64]
acc -> (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]
acc, ()))
            IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)

        allClaimed <- readIORef claimedRef
        findDuplicates allClaimed `shouldBe` []
        length allClaimed `shouldBe` numJobs

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"no duplicate claims under extreme contention (many groups)" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let numGroups :: Int
numGroups = Int
50
            jobsPerGroup :: Int
jobsPerGroup = Int
10
            numWorkers :: Int
numWorkers = Int
40

        [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
t a -> (a -> f b) -> f ()
for_ [Int
1 .. Int
numGroups] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \(Int
groupIndex :: Int) -> do
          let groupName :: Text
groupName = Text
"stress-group-" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show Int
groupIndex)
          IO [JobRead payload] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
            (IO [JobRead payload] -> IO ()) -> IO [JobRead payload] -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env
            (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload] -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch
              (Int -> JobWrite payload -> [JobWrite payload]
forall a. Int -> a -> [a]
replicate Int
jobsPerGroup (JobWrite payload -> [JobWrite payload])
-> JobWrite payload -> [JobWrite payload]
forall a b. (a -> b) -> a -> b
$ 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
groupName) (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"grouped"))

        claimedRef <- [Int64] -> IO (IORef [Int64])
forall a. a -> IO (IORef a)
newIORef ([] :: [Int64])

        replicateConcurrently_ numWorkers $
          claimAckLoop (runM env (HL.claimNextVisibleJobs 5 60) :: IO [JobRead payload]) 20 $ \JobRead payload
job -> do
            IORef [Int64] -> ([Int64] -> ([Int64], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Int64]
claimedRef (\[Int64]
acc -> (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]
acc, ()))
            IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)

        allClaimed <- readIORef claimedRef
        let totalJobs = Int
numGroups Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
jobsPerGroup
        findDuplicates allClaimed `shouldBe` []
        length allClaimed `shouldBe` totalJobs

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Insert + Claim Race" (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
"no lost jobs under concurrent insert/claim pressure" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let numRounds :: Int
numRounds = Int
100
            jobsPerRound :: Int
jobsPerRound = Int
20
            numClaimers :: Int
numClaimers = Int
10

        insertedRef <- Set Int64 -> IO (IORef (Set Int64))
forall a. a -> IO (IORef a)
newIORef Set Int64
forall a. Set a
Set.empty
        claimedRef <- newIORef Set.empty

        -- Run multiple rounds of concurrent insert + claim
        replicateM_ numRounds $ do
          _ <-
            mapConcurrently
              id
              ( [ do
                    -- Inserter
                    jobs <- runM env $ HL.insertJobsBatch (replicate jobsPerRound $ defaultJob (mkMessage "race"))
                    let ids = [Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList ([Int64] -> Set Int64) -> [Int64] -> Set Int64
forall a b. (a -> b) -> a -> b
$ (JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey [JobRead payload]
jobs
                    atomicModifyIORef' insertedRef (\Set Int64
acc -> (Set Int64 -> Set Int64 -> Set Int64
forall a. Ord a => Set a -> Set a -> Set a
Set.union Set Int64
acc Set Int64
ids, ()))
                ]
                  <> replicate
                    numClaimers
                    ( do
                        -- Claimer
                        claimed <- runM env (HL.claimNextVisibleJobs 10 60) :: IO [JobRead payload]
                        let ids = [Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList ([Int64] -> Set Int64) -> [Int64] -> Set Int64
forall a b. (a -> b) -> a -> b
$ (JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey [JobRead payload]
claimed
                        atomicModifyIORef' claimedRef (\Set Int64
acc -> (Set Int64 -> Set Int64 -> Set Int64
forall a. Ord a => Set a -> Set a -> Set a
Set.union Set Int64
acc Set Int64
ids, ()))
                        for_ claimed $ \JobRead payload
job -> env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)
                    )
              )
          pure ()

        -- Drain remaining
        drainAll (runM env (HL.claimNextVisibleJobs 100 60) :: IO [JobRead payload]) $ \JobRead payload
job -> do
          IORef (Set Int64) -> (Set Int64 -> (Set Int64, ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef (Set Int64)
claimedRef (\Set Int64
acc -> (Int64 -> Set Int64 -> Set Int64
forall a. Ord a => a -> Set a -> Set a
Set.insert (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job) Set Int64
acc, ()))
          IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)

        inserted <- readIORef insertedRef
        claimed <- readIORef claimedRef

        -- No job is lost.
        Set.size inserted `shouldBe` (numRounds * jobsPerRound)
        claimed `shouldBe` inserted

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Visibility Timeout Races" (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
"acks every job under aggressive visibility expiration" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let numJobs :: Int
numJobs = Int
50
            numWorkers :: Int
numWorkers = Int
20
            visibilityMs :: Int
visibilityMs = Int
100 -- 100ms visibility
        IO [JobRead payload] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO [JobRead payload] -> IO ()) -> IO [JobRead payload] -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload] -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch (Int -> JobWrite payload -> [JobWrite payload]
forall a. Int -> a -> [a]
replicate Int
numJobs (JobWrite payload -> [JobWrite payload])
-> JobWrite payload -> [JobWrite payload]
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"timeout-race"))

        -- Track which jobs were successfully acked
        ackedRef <- Set Int64 -> IO (IORef (Set Int64))
forall a. a -> IO (IORef a)
newIORef Set Int64
forall a. Set a
Set.empty

        replicateConcurrently_ numWorkers $
          -- A 50 empty-streak limit keeps workers retrying while jobs are in flight.
          claimAckLoop (runM env (HL.claimNextVisibleJobs 2 (fromIntegral visibilityMs / 1000)) :: IO [JobRead payload]) 50 $
            \JobRead payload
job -> do
              Int -> IO ()
threadDelay (Int
visibilityMs Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
500) -- Half the visibility time
              rows <- env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)
              when (rows > 0) $
                atomicModifyIORef' ackedRef (\Set Int64
acc -> (Int64 -> Set Int64 -> Set Int64
forall a. Ord a => a -> Set a -> Set a
Set.insert (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job) Set Int64
acc, ()))

        acked <- readIORef ackedRef
        -- Every job is acked.
        Set.size acked `shouldBe` numJobs

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"exactly one ack succeeds per job under reclaim pressure" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let numAttempts :: Int
numAttempts = Int
50
            numWorkersPerAttempt :: Int
numWorkersPerAttempt = Int
5

        -- Insert a fresh job per iteration.
        successCounts <- Int -> IO Int -> IO [Int]
forall (m :: * -> *) a. Applicative m => Int -> m a -> m [a]
replicateM Int
numAttempts (IO Int -> IO [Int]) -> IO Int -> IO [Int]
forall a b. (a -> b) -> a -> b
$ do
          IO (Maybe (JobRead payload)) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Maybe (JobRead payload)) -> IO ())
-> IO (Maybe (JobRead payload)) -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m (Maybe (JobRead payload)) -> IO (Maybe (JobRead payload))
forall a. env -> m a -> IO a
runM env
env (m (Maybe (JobRead payload)) -> IO (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload)) -> IO (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ JobWrite payload -> m (Maybe (JobRead payload))
forall payload (m :: * -> *).
QueueOperation m payload =>
JobWrite payload -> m (Maybe (JobRead payload))
HL.insertJob (payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"single-race"))

          results <-
            (IO Int -> IO Int) -> [IO Int] -> IO [Int]
forall (m :: * -> *) (t :: * -> *) a b.
(MonadUnliftIO m, Traversable t) =>
(a -> m b) -> t a -> m (t b)
mapConcurrently
              IO Int -> IO Int
forall a. a -> a
id
              ([IO Int] -> IO [Int]) -> [IO Int] -> IO [Int]
forall a b. (a -> b) -> a -> b
$ Int -> IO Int -> [IO Int]
forall a. Int -> a -> [a]
replicate Int
numWorkersPerAttempt
              (IO Int -> [IO Int]) -> IO Int -> [IO Int]
forall a b. (a -> b) -> a -> b
$ do
                claimed <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
1 NominalDiffTime
0.1) :: IO [JobRead payload]
                case claimed of
                  [] -> Int -> IO Int
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int
0
                  (JobRead payload
job : [JobRead payload]
_) -> do
                    Int -> IO ()
threadDelay Int
holdBeforeAckMicros
                    rows <- env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)
                    pure (fromIntegral rows :: Int)

          pure (sum results)

        -- Each iteration has exactly 1 successful ack.
        successCounts `shouldBe` replicate numAttempts 1

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Batched Mode Stress" (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
"no duplicate claims in batched mode under heavy contention" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let numGroups :: Int
numGroups = Int
30
            jobsPerGroup :: Int
jobsPerGroup = Int
20
            numWorkers :: Int
numWorkers = Int
25

        [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
t a -> (a -> f b) -> f ()
for_ [Int
1 .. Int
numGroups] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \(Int
groupIndex :: Int) -> do
          let groupName :: Text
groupName = Text
"batch-stress-" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show Int
groupIndex)
          IO [JobRead payload] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
            (IO [JobRead payload] -> IO ()) -> IO [JobRead payload] -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env
            (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload] -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch
              (Int -> JobWrite payload -> [JobWrite payload]
forall a. Int -> a -> [a]
replicate Int
jobsPerGroup (JobWrite payload -> [JobWrite payload])
-> JobWrite payload -> [JobWrite payload]
forall a b. (a -> b) -> a -> b
$ 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
groupName) (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"batched"))

        claimedRef <- [Int64] -> IO (IORef [Int64])
forall a. a -> IO (IORef a)
newIORef ([] :: [Int64])

        replicateConcurrently_ numWorkers $
          claimAckLoop (concatMap NE.toList <$> runM env (HL.claimNextVisibleJobsBatched 5 4 60) :: IO [JobRead payload]) 20 $ \JobRead payload
job -> do
            IORef [Int64] -> ([Int64] -> ([Int64], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Int64]
claimedRef (\[Int64]
acc -> (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]
acc, ()))
            IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)

        allClaimed <- readIORef claimedRef
        let totalJobs = Int
numGroups Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
jobsPerGroup
        findDuplicates allClaimed `shouldBe` []
        length allClaimed `shouldBe` totalJobs

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"batched claims from same group are contiguous (no interleaving)" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let numGroups :: Int
numGroups = Int
5
            jobsPerGroup :: Int
jobsPerGroup = Int
50
            numWorkers :: Int
numWorkers = Int
20
            numRounds :: Int
numRounds = Int
30

        gapCount <- Int -> IO (IORef Int)
forall a. a -> IO (IORef a)
newIORef (Int
0 :: Int)

        replicateM_ numRounds $ do
          for_ [1 .. numGroups] $ \(Int
groupIndex :: Int) ->
            IO [JobRead payload] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
              (IO [JobRead payload] -> IO ()) -> IO [JobRead payload] -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env
              (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload] -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch
                (Int -> JobWrite payload -> [JobWrite payload]
forall a. Int -> a -> [a]
replicate Int
jobsPerGroup (JobWrite payload -> [JobWrite payload])
-> JobWrite payload -> [JobWrite payload]
forall a b. (a -> b) -> a -> b
$ 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
"split-" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show Int
groupIndex))) (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"split"))

          resultsRef <- newIORef ([] :: [(Int, [(Maybe Text, Int64)])])
          _ <-
            mapConcurrently
              ( \Int
workerId -> do
                  claimed <- (NonEmpty (JobRead payload) -> [JobRead payload])
-> [NonEmpty (JobRead payload)] -> [JobRead payload]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap NonEmpty (JobRead payload) -> [JobRead payload]
forall a. NonEmpty a -> [a]
NE.toList ([NonEmpty (JobRead payload)] -> [JobRead payload])
-> IO [NonEmpty (JobRead payload)] -> IO [JobRead payload]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> env
-> m [NonEmpty (JobRead payload)]
-> IO [NonEmpty (JobRead payload)]
forall a. env -> m a -> IO a
runM env
env (Int -> Int -> NominalDiffTime -> m [NonEmpty (JobRead payload)]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> Int -> NominalDiffTime -> m [NonEmpty (JobRead payload)]
HL.claimNextVisibleJobsBatched Int
10 Int
3 NominalDiffTime
60) :: IO [JobRead payload]
                  atomicModifyIORef' resultsRef (\[(Int, [(Maybe Text, Int64)])]
acc -> ((Int
workerId, (JobRead payload -> (Maybe Text, Int64))
-> [JobRead payload] -> [(Maybe Text, Int64)]
forall a b. (a -> b) -> [a] -> [b]
map (\JobRead payload
job -> (JobRead payload -> Maybe Text
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> Maybe Text
groupKey JobRead payload
job, JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job)) [JobRead payload]
claimed) (Int, [(Maybe Text, Int64)])
-> [(Int, [(Maybe Text, Int64)])] -> [(Int, [(Maybe Text, Int64)])]
forall a. a -> [a] -> [a]
: [(Int, [(Maybe Text, Int64)])]
acc, ()))
                  for_ claimed $ \JobRead payload
job -> env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)
              )
              [1 .. numWorkers]

          -- Per-worker ids in each group are contiguous. A gap means another
          -- worker interleaved claims from the same group.
          results <- readIORef resultsRef
          let perGroup Text
wantedGroup =
                [ [Int64] -> [Int64]
forall a. Ord a => [a] -> [a]
sort [Int64]
ids
                | (Int
_, [(Maybe Text, Int64)]
jobs) <- [(Int, [(Maybe Text, Int64)])]
results
                , let ids :: [Int64]
ids = [Int64
jobId | (Just Text
groupName, Int64
jobId) <- [(Maybe Text, Int64)]
jobs, Text
groupName Text -> Text -> Bool
forall a. Eq a => a -> a -> Bool
== Text
wantedGroup]
                , Bool -> Bool
not ([Int64] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [Int64]
ids)
                ]
              allGroupKeys = Set Text -> [Text]
forall a. Set a -> [a]
Set.toList ([Text] -> Set Text
forall a. Ord a => [a] -> Set a
Set.fromList [Text
groupName | (Int
_, [(Maybe Text, Int64)]
jobs) <- [(Int, [(Maybe Text, Int64)])]
results, (Just Text
groupName, Int64
_) <- [(Maybe Text, Int64)]
jobs])
              hasGaps [a]
ids = ((a, a) -> Bool) -> [(a, a)] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (\(a
prev, a
next) -> a
next a -> a -> a
forall a. Num a => a -> a -> a
- a
prev a -> a -> Bool
forall a. Eq a => a -> a -> Bool
/= a
1) ([(a, a)] -> Bool) -> [(a, a)] -> Bool
forall a b. (a -> b) -> a -> b
$ [a] -> [a] -> [(a, a)]
forall a b. [a] -> [b] -> [(a, b)]
zip [a]
ids (Int -> [a] -> [a]
forall a. Int -> [a] -> [a]
drop Int
1 [a]
ids)
              gappedGroups =
                [ Text
groupName
                | Text
groupName <- [Text]
allGroupKeys
                , ([Int64] -> Bool) -> [[Int64]] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any [Int64] -> Bool
forall {a}. (Eq a, Num a) => [a] -> Bool
hasGaps (Text -> [[Int64]]
perGroup Text
groupName)
                ]

          when (not (null gappedGroups)) $
            atomicModifyIORef' gapCount (\Int
count -> (Int
count Int -> Int -> Int
forall a. Num a => a -> a -> a
+ [Text] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [Text]
gappedGroups, ()))

          -- Drain remaining
          drainAll (concatMap NE.toList <$> runM env (HL.claimNextVisibleJobsBatched 100 20 60) :: IO [JobRead payload]) $
            \JobRead payload
job -> IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)

        readIORef gapCount >>= (`shouldBe` 0)

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"batched and single mode don't interfere" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- Mix of batched and single mode workers on same table
        let numGroups :: Int
numGroups = Int
20
            jobsPerGroup :: Int
jobsPerGroup = Int
15
            numBatchedWorkers :: Int
numBatchedWorkers = Int
10
            numSingleWorkers :: Int
numSingleWorkers = Int
10

        [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
t a -> (a -> f b) -> f ()
for_ [Int
1 .. Int
numGroups] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \(Int
groupIndex :: Int) -> do
          let groupName :: Text
groupName = Text
"mixed-mode-" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show Int
groupIndex)
          IO [JobRead payload] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
            (IO [JobRead payload] -> IO ()) -> IO [JobRead payload] -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env
            (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ [JobWrite payload] -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
HL.insertJobsBatch
              (Int -> JobWrite payload -> [JobWrite payload]
forall a. Int -> a -> [a]
replicate Int
jobsPerGroup (JobWrite payload -> [JobWrite payload])
-> JobWrite payload -> [JobWrite payload]
forall a b. (a -> b) -> a -> b
$ 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
groupName) (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"mixed"))

        claimedRef <- [Int64] -> IO (IORef [Int64])
forall a. a -> IO (IORef a)
newIORef ([] :: [Int64])

        let onJob JobRead payload
job = do
              IORef [Int64] -> ([Int64] -> ([Int64], ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [Int64]
claimedRef (\[Int64]
acc -> (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]
acc, ()))
              IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)
            batchedWorker =
              IO [JobRead payload] -> Int -> (JobRead payload -> IO ()) -> IO ()
forall a. IO [a] -> Int -> (a -> IO ()) -> IO ()
claimAckLoop ((NonEmpty (JobRead payload) -> [JobRead payload])
-> [NonEmpty (JobRead payload)] -> [JobRead payload]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap NonEmpty (JobRead payload) -> [JobRead payload]
forall a. NonEmpty a -> [a]
NE.toList ([NonEmpty (JobRead payload)] -> [JobRead payload])
-> IO [NonEmpty (JobRead payload)] -> IO [JobRead payload]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> env
-> m [NonEmpty (JobRead payload)]
-> IO [NonEmpty (JobRead payload)]
forall a. env -> m a -> IO a
runM env
env (Int -> Int -> NominalDiffTime -> m [NonEmpty (JobRead payload)]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> Int -> NominalDiffTime -> m [NonEmpty (JobRead payload)]
HL.claimNextVisibleJobsBatched Int
3 Int
3 NominalDiffTime
60) :: IO [JobRead payload]) Int
50 JobRead payload -> IO ()
onJob
            singleWorker =
              IO [JobRead payload] -> Int -> (JobRead payload -> IO ()) -> IO ()
forall a. IO [a] -> Int -> (a -> IO ()) -> IO ()
claimAckLoop (env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
2 NominalDiffTime
60) :: IO [JobRead payload]) Int
50 JobRead payload -> IO ()
onJob

        _ <-
          mapConcurrently
            id
            (replicate numBatchedWorkers batchedWorker <> replicate numSingleWorkers singleWorker)

        allClaimed <- readIORef claimedRef
        let totalJobs = Int
numGroups Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
jobsPerGroup
        findDuplicates allClaimed `shouldBe` []
        length allClaimed `shouldBe` totalJobs

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Ack Race Conditions" (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
"concurrent ack attempts on same job - exactly one succeeds" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let numIterations :: Int
numIterations = Int
100
            numRacingWorkers :: Int
numRacingWorkers = Int
10

        successCounts <- [Int] -> (Int -> IO Int) -> IO [Int]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
t a -> (a -> f b) -> f (t b)
for [Int
1 .. Int
numIterations] ((Int -> IO Int) -> IO [Int]) -> (Int -> IO Int) -> IO [Int]
forall a b. (a -> b) -> a -> b
$ \(Int
_ :: Int) -> do
          -- Insert a job
          IO (Maybe (JobRead payload)) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Maybe (JobRead payload)) -> IO ())
-> IO (Maybe (JobRead payload)) -> IO ()
forall a b. (a -> b) -> a -> b
$ env -> m (Maybe (JobRead payload)) -> IO (Maybe (JobRead payload))
forall a. env -> m a -> IO a
runM env
env (m (Maybe (JobRead payload)) -> IO (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload)) -> IO (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ JobWrite payload -> m (Maybe (JobRead payload))
forall payload (m :: * -> *).
QueueOperation m payload =>
JobWrite payload -> m (Maybe (JobRead payload))
HL.insertJob (payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkMessage Text
"ack-race"))

          -- Claim it
          claimed <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
1 NominalDiffTime
60) :: IO [JobRead payload]
          case claimed of
            [] -> Int -> IO Int
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int
0
            (JobRead payload
job : [JobRead payload]
_) -> do
              -- Many workers try to ack the same job concurrently
              results <-
                (IO Int -> IO Int) -> [IO Int] -> IO [Int]
forall (m :: * -> *) (t :: * -> *) a b.
(MonadUnliftIO m, Traversable t) =>
(a -> m b) -> t a -> m (t b)
mapConcurrently
                  IO Int -> IO Int
forall a. a -> a
id
                  ([IO Int] -> IO [Int]) -> [IO Int] -> IO [Int]
forall a b. (a -> b) -> a -> b
$ Int -> IO Int -> [IO Int]
forall a. Int -> a -> [a]
replicate Int
numRacingWorkers
                  (IO Int -> [IO Int]) -> IO Int -> [IO Int]
forall a b. (a -> b) -> a -> b
$ do
                    rows <- env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (JobRead payload -> m Int64
forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
HL.ackJob JobRead payload
job)
                    pure (fromIntegral rows :: Int)
              pure (sum results)

        -- Each iteration has exactly 1 successful ack.
        successCounts `shouldBe` replicate numIterations 1