shibuya-core-0.12.0.0: test/Shibuya/Runner/SupervisedSpec.hs
{-# LANGUAGE OverloadedStrings #-}
module Shibuya.Runner.SupervisedSpec (spec) where
import Control.Concurrent (myThreadId)
import Control.Concurrent.NQE.Supervisor (Strategy (..))
import Control.Concurrent.STM (atomically, check, newEmptyTMVarIO, readTVar, readTVarIO, takeTMVar, tryPutTMVar)
import Control.Exception (MaskingState (..), getMaskingState)
import Control.Monad (forM, forM_, replicateM, replicateM_, when)
import Data.IORef (IORef, atomicModifyIORef', atomicWriteIORef, modifyIORef', newIORef, readIORef)
import Data.List (sort)
import Data.List.NonEmpty qualified as NE
import Data.Map.Strict qualified as Map
import Data.Text qualified as Text
import Data.Time (UTCTime (..), fromGregorian)
import Effectful (Eff, IOE, liftIO, runEff, (:>))
import Effectful.State.Static.Local qualified as State
import Shibuya (ProcessorHalt (..))
import Shibuya.Adapter (Adapter (..))
import Shibuya.Adapter.Mock
( getTrackedDecisions,
mkTrackedIngested,
newTrackingAck,
trackedListAdapter,
)
import Shibuya.App
( AppConfig (..),
ShutdownConfig (..),
defaultAppConfig,
mkProcessor,
runApp,
stopAppGracefully,
)
import Shibuya.Batch (BatchConfig (..), BatchKey (..), ackAll, ackAllOk, defaultBatchConfig)
import Shibuya.Batch.TestHarness (finalizedExactlyOnce)
import Shibuya.Core.Ack
( AckDecision (..),
DeadLetterCode,
DeadLetterReason (..),
HaltReason (..),
RetryDelay (..),
mkDeadLetterCode,
)
import Shibuya.Core.AckHandle (AckHandle (..))
import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), Message (..), mkIngested)
import Shibuya.Core.Metrics
( InFlightInfo (..),
ProcessorId (..),
ProcessorMetrics (..),
ProcessorState (..),
StreamStats (..),
sampleMetrics,
)
import Shibuya.Core.Metrics qualified as Metrics
import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), mkEnvelope)
import Shibuya.Handler (Handler)
import Shibuya.Internal.Runner.Master
( ProcessorLifecycle (..),
getAllMetrics,
getAllMetricsIO,
getLifecycleSnapshot,
getProcessorMetricsIO,
startMaster,
stopMaster,
)
import Shibuya.Internal.Runner.Supervised
( SupervisedProcessor (..),
getMetrics,
isDone,
runSupervised,
runSupervisedBatchWith,
runSupervisedWith,
runWithMetrics,
runWithMetricsBatch,
)
import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))
import Shibuya.Telemetry.Effect (runTracingNoop)
import Streamly.Data.Stream qualified as Stream
import Test.Hspec
import UnliftIO (try)
import UnliftIO qualified as UIO
import UnliftIO.Concurrent (threadDelay)
spec :: Spec
spec = do
describe "delivery status" $ do
it "treats deliveries built by mkIngested as current" $ do
status <- runEff $ (mkIngested (createTestEnvelope 1) (AckHandle (const (pure ())))).deliveryStatus
status `shouldBe` DeliveryCurrent
forM_ [Metrics.MetricsBasic, Metrics.MetricsDetailed Metrics.defaultDetailOptions] $ \detail -> do
it ("skips the handler and finalizer for a superseded delivery: " <> show detail) $ do
seen <- newIORef []
(decisions, metrics) <- runEff $ runTracingNoop $ do
tracking <- newTrackingAck
let values =
[ let item = mkTrackedIngested tracking ((createTestEnvelope n) {cursor = Just (CursorInt (fromIntegral n))})
in item {deliveryStatus = pure (if n == 2 then DeliverySuperseded else DeliveryCurrent)}
| n <- [1 .. 3]
]
handler message = liftIO (modifyIORef' seen (<> [message.envelope.messageId])) >> pure AckOk
master <- startMaster IgnoreAll
proc <- runSupervisedWith detail master 10 (ProcessorId "status") Unordered Serial (testAdapter values) handler
liftIO $ atomically $ readTVar proc.done >>= check
metrics <- liftIO $ sampleMetrics proc.metrics
liftIO $ Metrics.slotMisses proc.metrics `shouldReturn` 0
decisions <- getTrackedDecisions tracking
stopMaster master
pure (decisions, metrics)
readIORef seen `shouldReturn` [MessageId "msg-1", MessageId "msg-3"]
map fst (reverse decisions) `shouldBe` [MessageId "msg-1", MessageId "msg-3"]
metrics.stats.received `shouldBe` 3
metrics.stats.processed `shouldBe` 2
metrics.stats.failed `shouldBe` 0
metrics.oldestInFlight `shouldBe` Nothing
case detail of
Metrics.MetricsBasic -> metrics.latency `shouldBe` Nothing
_ -> do
fmap (.count) metrics.latency `shouldBe` Just 2
(metrics.progress >>= (.latest)) `shouldBe` Just (CursorInt 3)
it "asks for status at handler time, not at ingestion" $ do
queued <- UIO.newEmptyMVar
current <- newIORef True
seen <- newIORef []
result <- UIO.timeout 5000000 $ runEff $ runTracingNoop $ do
tracking <- newTrackingAck
let values =
[ let item = mkTrackedIngested tracking (createTestEnvelope n)
in if n == 2 then item {deliveryStatus = liftIO $ (\x -> if x then DeliveryCurrent else DeliverySuperseded) <$> readIORef current} else item
| n <- [1 .. 3]
]
source =
Stream.unfoldrM
( \case
[] -> liftIO (UIO.putMVar queued ()) >> pure Nothing
x : xs -> pure (Just (x, xs))
)
values
adapter = (testAdapter values) {source}
handler message = do
liftIO $ modifyIORef' seen (<> [message.envelope.messageId])
when (message.envelope.messageId == MessageId "msg-1") $ liftIO $ do
UIO.takeMVar queued
atomicWriteIORef current False
pure AckOk
_ <- runWithMetrics 10 (ProcessorId "status-time") adapter handler
getTrackedDecisions tracking
fmap (map fst . reverse) result `shouldBe` Just [MessageId "msg-1", MessageId "msg-3"]
readIORef seen `shouldReturn` [MessageId "msg-1", MessageId "msg-3"]
forM_ [Ahead 2, Async 2] $ \concurrency ->
forM_ [False, True] $ \batching ->
forM_ [Metrics.MetricsBasic, Metrics.MetricsDetailed Metrics.defaultDetailOptions] $ \detail ->
it ("checks queued delivery status after finalization with concurrent keys: " <> show (concurrency, batching, detail)) $ do
otherStarted <- UIO.newEmptyMVar
firstFinalized <- UIO.newEmptyMVar
current <- newIORef True
sampled <- newIORef []
handled <- newIORef []
finalized <- newIORef []
result <- UIO.timeout 5_000_000 $
runEff $
runTracingNoop $
UIO.bracket (startMaster IgnoreAll) stopMaster $ \master -> do
let values =
[ ( mkIngested ((createTestEnvelope n) {partition = Just (if n == 3 then "b" else "a")}) $
AckHandle $ \decision -> liftIO $ do
atomicModifyIORef' finalized (\xs -> ((n, decision) : xs, ()))
when (n == 1) $ do
atomicWriteIORef current False
UIO.putMVar firstFinalized ()
)
{ deliveryStatus = liftIO $ do
atomicModifyIORef' sampled (\xs -> (n : xs, ()))
fresh <- readIORef current
pure $ if n == 2 && not fresh then DeliverySuperseded else DeliveryCurrent
}
| n <- [1 .. 3]
]
handler message = do
let mid = message.envelope.messageId
liftIO $ atomicModifyIORef' handled (\xs -> (mid : xs, ()))
-- The second key cannot start until the scheduler has read
-- the queued successor on key a. Both handlers must overlap:
-- a waits for b to start, and b waits for a to finalize.
if mid == MessageId "msg-1"
then liftIO (UIO.takeMVar otherStarted) >> pure (AckRetry (RetryDelay 0))
else do
when (mid == MessageId "msg-3") $ liftIO $ do
UIO.putMVar otherStarted ()
UIO.takeMVar firstFinalized
pure AckOk
config = defaultBatchConfig {batchSize = 1, batchKey = \env -> BatchKey (maybe "" id env.partition)}
proc <-
if batching
then runSupervisedBatchWith detail master 10 (ProcessorId "concurrent-status") concurrency config (testAdapter values) (\_ messages -> ackAll <$> handler (NE.head messages))
else runSupervisedWith detail master 10 (ProcessorId "concurrent-status") PartitionedInOrder concurrency (testAdapter values) handler
liftIO $ atomically $ readTVar proc.done >>= check
liftIO $ Metrics.slotMisses proc.metrics `shouldReturn` 0
liftIO $ sampleMetrics proc.metrics
case result of
Nothing -> expectationFailure "concurrent delivery admission did not finish"
Just metrics -> do
metrics.stats.received `shouldBe` 3
metrics.stats.processed `shouldBe` 2
metrics.stats.failed `shouldBe` 0
metrics.oldestInFlight `shouldBe` Nothing
metrics.state `shouldBe` Idle
case detail of
Metrics.MetricsBasic -> metrics.latency `shouldBe` Nothing
_ -> fmap (.count) metrics.latency `shouldBe` Just 2
sort <$> readIORef sampled `shouldReturn` [1, 2, 3]
sort <$> readIORef handled `shouldReturn` [MessageId "msg-1", MessageId "msg-3"]
decisions <- readIORef finalized
sort (map fst decisions) `shouldBe` [1, 3]
lookup 1 decisions `shouldBe` Just (AckRetry (RetryDelay 0))
lookup 3 decisions `shouldBe` Just AckOk
forM_ [False, True] $ \batching ->
it ("reports throwing status as infrastructure failure without handler or finalizer: batch=" <> show batching) $ do
seen <- newIORef (0 :: Int)
finalized <- newIORef (0 :: Int)
result <- UIO.timeout 5000000 $ runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
let pid = ProcessorId "throwing-status"
item =
(mkIngested (createTestEnvelope 1) (AckHandle (\_ -> liftIO $ modifyIORef' finalized (+ 1))))
{ deliveryStatus = liftIO $ UIO.throwIO (userError "status infrastructure failure")
}
adapter = testAdapter [item]
proc <-
if batching
then runSupervisedBatchWith Metrics.MetricsBasic master 10 pid Serial defaultBatchConfig {batchSize = 1} adapter (\_ _ -> liftIO (modifyIORef' seen (+ 1)) >> pure ackAllOk)
else runSupervised master 10 pid Unordered Serial adapter (\_ -> liftIO (modifyIORef' seen (+ 1)) >> pure AckOk)
liftIO $ atomically $ readTVar proc.done >>= check
lifecycle <- getLifecycleSnapshot master
stopMaster master
pure (Map.lookup pid lifecycle)
case result of
Just (Just (LifecycleFailed message Nothing)) -> message `shouldSatisfy` Text.isInfixOf "status infrastructure failure"
other -> expectationFailure (show other)
readIORef seen `shouldReturn` 0
readIORef finalized `shouldReturn` 0
describe "Shibuya.Internal.Runner.Master" $ do
it "starts and stops cleanly" $ do
result <- runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
stopMaster master
pure ()
result `shouldBe` ()
it "returns empty metrics initially" $ do
metrics <- runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
m <- getAllMetrics master
stopMaster master
pure m
metrics `shouldSatisfy` null
it "metrics queries return after stopMaster" $ do
master <- runEff $ runTracingNoop $ startMaster IgnoreAll
runEff $ runTracingNoop $ stopMaster master
allMetrics <- UIO.timeout 1_000_000 (getAllMetricsIO master)
processorMetrics <- UIO.timeout 1_000_000 (getProcessorMetricsIO master (ProcessorId "x"))
allMetrics `shouldSatisfy` maybe False null
processorMetrics `shouldBe` Just Nothing
describe "Shibuya.Internal.Runner.Supervised" $ do
forM_ [False, True] $ \batching ->
it ("isolates intake state while preserving serial handler state for " <> if batching then "batches" else "messages") $ do
owner <- myThreadId
observed <- newIORef []
finalState <- runEff $ State.execState (100 :: Int) $ runTracingNoop $ do
let source =
Stream.unfoldrM
( \n ->
if n == (3 :: Int)
then pure Nothing
else do
liftIO $ myThreadId >>= (`shouldNotBe` owner)
value <- State.get @Int
State.modify @Int (+ 1)
pure $ Just (mkIngested (mkEnvelope (MessageId (Text.pack (show n))) value) (AckHandle (const $ pure ())), n + 1)
)
0
adapter = Adapter {adapterName = "test:state", source = source, shutdown = pure ()}
observe message = do
value <- State.get @Int
State.modify @Int (+ 10)
liftIO $ do
myThreadId >>= (`shouldBe` owner)
modifyIORef' observed (<> [(message.envelope.payload, value)])
if batching
then runWithMetricsBatch 1 (ProcessorId "state") Serial defaultBatchConfig {batchSize = 1} adapter (\_ messages -> mapM_ observe messages >> pure ackAllOk)
else runWithMetrics 1 (ProcessorId "state") adapter (\message -> observe message >> pure AckOk)
finalState `shouldBe` 130
readIORef observed `shouldReturn` [(100, 100), (101, 110), (102, 120)]
it "records only successful acknowledgement and dead-letter cursors" $ do
forM_ [AckOk, AckDeadLetter MaxRetriesExceeded, AckRetry (RetryDelay 0), AckHalt (HaltFatal "stop")] $ \decision -> do
final <- cursorOutcome decision False
let expected = case decision of
AckOk -> Just (CursorInt 5)
AckDeadLetter _ -> Just (CursorInt 5)
_ -> Nothing
(final >>= (.latest)) `shouldBe` expected
cursorOutcome AckOk True `shouldReturn` Nothing
it "preserves source cursor order per partition in detailed ordered processing" $ do
final <- runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
tracking <- newTrackingAck
let adapter = trackedListAdapter tracking [((mkEnvelope (MessageId (Text.pack (show n))) n) {cursor = Just (CursorInt n), partition = Just (if odd n then "a" else "b")}) | n <- [1 .. 20]]
proc <- runSupervisedWith (Metrics.MetricsDetailed Metrics.defaultDetailOptions) master 100 (ProcessorId "cursor-ordered") PartitionedInOrder (Async 4) adapter (\_ -> pure AckOk)
liftIO $ atomically $ readTVar proc.done >>= check
m <- getMetrics proc
stopMaster master
pure m.progress
fmap (.partitions) final `shouldBe` Just (Map.fromList [("a", CursorInt 19), ("b", CursorInt 20)])
it "reports growing oldest age and clears it after finalization" $ do
started <- UIO.newEmptyMVar
release <- UIO.newEmptyMVar
runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
messages <- createTestMessages 1
sp <-
runSupervisedWith
(Metrics.MetricsDetailed Metrics.defaultDetailOptions)
master
10
(ProcessorId "detailed-age")
Unordered
Serial
(testAdapter messages)
(\_ -> liftIO (UIO.putMVar started () >> UIO.takeMVar release) >> pure AckOk)
liftIO (UIO.takeMVar started)
first <- liftIO $ sampleMetrics sp.metrics
liftIO $ threadDelay 20_000
second <- liftIO $ sampleMetrics sp.metrics
liftIO $ do
fmap (.messageId) second.oldestInFlight `shouldBe` Just (MessageId "msg-1")
fmap (.ageSeconds) second.oldestInFlight `shouldSatisfy` (> fmap (.ageSeconds) first.oldestInFlight)
UIO.putMVar release ()
UIO.timeout 1_000_000 (atomically $ readTVar sp.done >>= check) `shouldReturn` Just ()
completed <- liftIO $ sampleMetrics sp.metrics
liftIO $ do
completed.oldestInFlight `shouldBe` Nothing
fmap (.count) completed.latency `shouldBe` Just 1
fmap (.sumSeconds) completed.latency `shouldSatisfy` (>= Just 0.02)
Metrics.slotMisses sp.metrics `shouldReturn` 0
stopMaster master
it "identifies the oldest of four concurrent attempts as slots are released" $ do
starts <- replicateM 4 UIO.newEmptyMVar
releases <- replicateM 4 UIO.newEmptyMVar
runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
messages <- createTestMessages 4
let source =
Stream.mapM
( \(i, msg) -> do
when (i > 0) $ UIO.readMVar (starts !! (i - 1))
pure msg
)
(Stream.fromList (zip [0 ..] messages))
adapter = (testAdapter messages) {source}
handler msg = do
let i = read (drop 8 msg.envelope.payload) - 1
liftIO $ UIO.putMVar (starts !! i) () >> UIO.takeMVar (releases !! i)
pure AckOk
sp <-
runSupervisedWith
(Metrics.MetricsDetailed Metrics.defaultDetailOptions)
master
10
(ProcessorId "detailed-four")
Unordered
(Async 4)
adapter
handler
liftIO $ mapM_ UIO.readMVar starts
liftIO $ sampleMetrics sp.metrics >>= \m -> fmap (.messageId) m.oldestInFlight `shouldBe` Just (MessageId "msg-1")
liftIO $ do
UIO.putMVar (releases !! 3) ()
UIO.putMVar (releases !! 2) ()
UIO.putMVar (releases !! 0) ()
let awaitSecond = do
m <- sampleMetrics sp.metrics
if fmap (.messageId) m.oldestInFlight == Just (MessageId "msg-2") then pure () else threadDelay 1000 >> awaitSecond
UIO.timeout 1_000_000 awaitSecond `shouldReturn` Just ()
UIO.putMVar (releases !! 1) ()
UIO.timeout 1_000_000 (atomically $ readTVar sp.done >>= check) `shouldReturn` Just ()
Metrics.slotMisses sp.metrics `shouldReturn` 0
stopMaster master
it "counts completed finalization failures including retry delays" $ do
runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
let message =
mkIngested
(mkEnvelope (MessageId "failed-detail") ())
(AckHandle (\_ -> liftIO $ ioError (userError "failed finalizer")))
adapter = Adapter "failed-detail" (Stream.fromList [message]) (pure ())
sp <-
runSupervisedWith
(Metrics.MetricsDetailed Metrics.defaultDetailOptions)
master
1
(ProcessorId "failed-detail")
Unordered
Serial
adapter
(\_ -> pure AckOk)
liftIO $ UIO.timeout 1_000_000 (atomically $ readTVar sp.done >>= check) `shouldReturn` Just ()
m <- liftIO $ sampleMetrics sp.metrics
liftIO $ do
m.oldestInFlight `shouldBe` Nothing
fmap (.count) m.latency `shouldBe` Just 1
fmap (.sumSeconds) m.latency `shouldSatisfy` (>= Just 0.31)
stopMaster master
it "abandons detail when cancellation interrupts a handler or finalizer" $ do
forM_ [False, True] $ \inFinalizer -> replicateM_ 100 $ do
entered <- UIO.newEmptyMVar
blocker <- UIO.newEmptyMVar
runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
let waitHere = liftIO (UIO.putMVar entered () >> UIO.takeMVar blocker)
message = mkIngested (mkEnvelope (MessageId "cancel") ()) (AckHandle (\_ -> when inFinalizer waitHere))
adapter = Adapter "cancel-detail" (Stream.fromList [message]) (pure ())
sp <-
runSupervisedWith
(Metrics.MetricsDetailed Metrics.defaultDetailOptions)
master
1
(ProcessorId "cancel-detail")
Unordered
Serial
adapter
(\_ -> when (not inFinalizer) waitHere >> pure AckOk)
liftIO $ UIO.takeMVar entered
stopMaster master
liftIO $ do
m <- sampleMetrics sp.metrics
m.oldestInFlight `shouldBe` Nothing
m.latency `shouldBe` Nothing
it "restores interruptibility inside a supervised child" $ do
observed <- newEmptyTMVarIO
result <-
UIO.timeout 1_000_000 $
runEff $
runTracingNoop $ do
messages <- createTestMessages 1
let handler _ = do
state <- liftIO getMaskingState
liftIO $ atomically $ tryPutTMVar observed state >> pure ()
pure AckOk
master <- startMaster IgnoreAll
_ <- runSupervised master 1 (ProcessorId "masking-state") Unordered Serial (testAdapter messages) handler
state <- liftIO $ atomically $ takeTMVar observed
stopMaster master
pure state
result `shouldBe` Just Unmasked
describe "runWithMetrics" $ do
it "processes messages and tracks metrics" $ do
(finalMetrics, processedMsgs) <- runEff $ runTracingNoop $ do
-- Track processed messages
processedRef <- liftIO $ newIORef ([] :: [String])
-- Create test messages
messages <- createTestMessages 3
-- Create adapter and handler
let adapter = testAdapter messages
handler = testHandler processedRef
procId = ProcessorId "test-processor"
-- Run with metrics
sp <- runWithMetrics 10 procId adapter handler
-- Get results
metrics <- getMetrics sp
msgs <- liftIO $ readIORef processedRef
pure (metrics, msgs)
-- Verify metrics
finalMetrics.stats.received `shouldBe` 3
finalMetrics.stats.processed `shouldBe` 3
finalMetrics.stats.failed `shouldBe` 0
finalMetrics.state `shouldBe` Idle
-- Verify all messages were processed
length processedMsgs `shouldBe` 3
it "tracks failed messages" $ do
finalMetrics <- runEff $ runTracingNoop $ do
messages <- createTestMessages 3
let adapter = testAdapter messages
handler = failingHandler
procId = ProcessorId "failing-processor"
sp <- runWithMetrics 10 procId adapter handler
getMetrics sp
-- All should be failed (handler throws)
finalMetrics.stats.failed `shouldBe` 3
it "marks processor as done when stream exhausted" $ do
done <- runEff $ runTracingNoop $ do
messages <- createTestMessages 2
let adapter = testAdapter messages
handler = alwaysAckOk
procId = ProcessorId "quick-processor"
sp <- runWithMetrics 10 procId adapter handler
isDone sp
done `shouldBe` True
it "completes when the stream is longer than the inbox" $ do
processedRef <- newIORef (0 :: Int)
result <-
UIO.timeout 10_000_000 $
runEff $
runTracingNoop $ do
messages <- createTestMessages 100
let handler _ = do
liftIO $ modifyIORef' processedRef (+ 1)
pure AckOk
adapter = testAdapter messages
procId = ProcessorId "longer-than-inbox"
sp <- runWithMetrics 5 procId adapter handler
metrics <- getMetrics sp
processed <- liftIO $ readIORef processedRef
pure (metrics, processed)
case result of
Nothing -> expectationFailure "runWithMetrics timed out"
Just (finalMetrics, processed) -> do
processed `shouldBe` 100
finalMetrics.stats.received `shouldBe` 100
finalMetrics.stats.processed `shouldBe` 100
describe "AckHalt behavior" $ do
it "stops processing when handler returns AckHalt" $ do
processedCount <- do
processedRef <- newIORef (0 :: Int)
let handler _ = do
count <- liftIO $ readIORef processedRef
liftIO $ modifyIORef' processedRef (+ 1)
if count >= 2
then pure $ AckHalt (HaltFatal "stopping after 3")
else pure AckOk
-- Try to run and catch ProcessorHalt
result <- try $ runEff $ runTracingNoop $ do
messages <- createTestMessages 10
let adapter = testAdapter messages
procId = ProcessorId "halting-processor"
runWithMetrics 10 procId adapter handler
-- Verify ProcessorHalt was thrown
case result of
Left (ProcessorHalt reason) ->
reason `shouldBe` HaltFatal "stopping after 3"
Right _ ->
expectationFailure "Expected ProcessorHalt exception"
readIORef processedRef
-- Should have processed exactly 3 messages before halt
processedCount `shouldBe` 3
it "updates metrics to halted state on AckHalt" $ do
finalMetrics <- runEff $ runTracingNoop $ do
messages <- createTestMessages 5
let handler _ = pure $ AckHalt (HaltFatal "immediate halt")
adapter = testAdapter messages
master <- startMaster IgnoreAll
-- runSupervised catches ProcessorHalt, so we can check metrics
sp <- runSupervised master 10 (ProcessorId "halt-metrics") Unordered Serial adapter handler
-- Wait for processor to halt
liftIO $ threadDelay 100000 -- 100ms
m <- liftIO $ sampleMetrics sp.metrics
stopMaster master
pure m
case finalMetrics.state of
Failed msg _ -> msg `shouldSatisfy` (Text.isInfixOf "immediate halt")
other -> expectationFailure $ "Expected Failed state, got: " ++ show other
it "halt in one supervised processor doesn't affect others" $ do
(countA, countB) <- runEff $ runTracingNoop $ do
countARef <- liftIO $ newIORef (0 :: Int)
countBRef <- liftIO $ newIORef (0 :: Int)
messagesA <- createTestMessages 10
messagesB <- createTestMessages 10
-- Handler A halts after 2 messages
let handlerA _ = do
count <- liftIO $ readIORef countARef
liftIO $ modifyIORef' countARef (+ 1)
if count >= 1
then pure $ AckHalt (HaltFatal "A stopping")
else pure AckOk
-- Handler B processes all messages
let handlerB _ = do
liftIO $ modifyIORef' countBRef (+ 1)
pure AckOk
let adapterA = testAdapter messagesA
adapterB = testAdapter messagesB
master <- startMaster IgnoreAll
_spA <- runSupervised master 10 (ProcessorId "A") Unordered Serial adapterA handlerA
_spB <- runSupervised master 10 (ProcessorId "B") Unordered Serial adapterB handlerB
-- Wait for both to complete
liftIO $ threadDelay 500000 -- 500ms
cA <- liftIO $ readIORef countARef
cB <- liftIO $ readIORef countBRef
stopMaster master
pure (cA, cB)
-- A should have stopped after 2 messages
countA `shouldBe` 2
-- B should have processed all 10
countB `shouldBe` 10
describe "Concurrency modes" $ do
describe "Ahead mode" $ do
it "processes all messages successfully" $ do
countRef <- newIORef (0 :: Int)
finalCount <- runEff $ runTracingNoop $ do
messages <- createTestMessages 5
let handler _ = do
liftIO $ atomicModifyIORef' countRef (\n -> (n + 1, ()))
liftIO $ threadDelay 10000 -- 10ms
pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
_ <- runSupervised master 10 (ProcessorId "ahead") Unordered (Ahead 3) adapter handler
-- Wait for completion
liftIO $ threadDelay 300000 -- 300ms
stopMaster master
liftIO $ readIORef countRef
finalCount `shouldBe` 5
it "respects maxBuffer limit" $ do
-- This tests that we don't start too many concurrent tasks
maxInFlightRef <- newIORef (0 :: Int)
currentInFlightRef <- newIORef (0 :: Int)
_ <- runEff $ runTracingNoop $ do
messages <- createTestMessages 10
let handler _ = do
-- Atomically increment current in-flight and get new value
cur <-
liftIO $
atomicModifyIORef' currentInFlightRef (\n -> let n' = n + 1 in (n', n'))
-- Update max
liftIO $ atomicModifyIORef' maxInFlightRef (\m -> (max m cur, ()))
-- Simulate some work
liftIO $ threadDelay 50000 -- 50ms
-- Decrement in-flight
liftIO $ atomicModifyIORef' currentInFlightRef (\c -> (c - 1, ()))
pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
-- Use Ahead with max 3 concurrent
_sp <- runSupervised master 10 (ProcessorId "ahead-limit") Unordered (Ahead 3) adapter handler
liftIO $ threadDelay 1000000 -- 1s to let it complete
stopMaster master
maxInFlight <- readIORef maxInFlightRef
-- Max in-flight should not exceed buffer size + 1 (buffer + currently processing)
maxInFlight `shouldSatisfy` (<= 4)
describe "Async mode" $ do
it "processes all messages concurrently" $ do
countRef <- newIORef (0 :: Int)
finalCount <- runEff $ runTracingNoop $ do
messages <- createTestMessages 5
let handler _ = do
liftIO $ atomicModifyIORef' countRef (\n -> (n + 1, ()))
liftIO $ threadDelay 10000 -- 10ms
pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
_ <- runSupervised master 10 (ProcessorId "async") Unordered (Async 5) adapter handler
-- Wait for completion
liftIO $ threadDelay 300000 -- 300ms
stopMaster master
liftIO $ readIORef countRef
finalCount `shouldBe` 5
it "respects concurrency limit" $ do
maxInFlightRef <- newIORef (0 :: Int)
currentInFlightRef <- newIORef (0 :: Int)
_ <- runEff $ runTracingNoop $ do
messages <- createTestMessages 10
let handler _ = do
cur <-
liftIO $
atomicModifyIORef' currentInFlightRef (\n -> let n' = n + 1 in (n', n'))
liftIO $ atomicModifyIORef' maxInFlightRef (\m -> (max m cur, ()))
liftIO $ threadDelay 50000 -- 50ms
liftIO $ atomicModifyIORef' currentInFlightRef (\c -> (c - 1, ()))
pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
_sp <- runSupervised master 10 (ProcessorId "async-limit") Unordered (Async 3) adapter handler
liftIO $ threadDelay 1000000 -- 1s
stopMaster master
maxInFlight <- readIORef maxInFlightRef
maxInFlight `shouldSatisfy` (<= 4)
describe "Halt with concurrency" $ do
it "waits for in-flight to complete before halting" $ do
processedRef <- newIORef ([] :: [Int])
_ <- runEff $ runTracingNoop $ do
messages <- createTestMessages 10
let handler ingested = do
let msgNum = case ingested.envelope.cursor of
Just (CursorInt n) -> n
_ -> 0
-- Simulate work
liftIO $ threadDelay 50000 -- 50ms
liftIO $ atomicModifyIORef' processedRef (\xs -> (msgNum : xs, ()))
-- Halt on message 3
if msgNum == 3
then pure $ AckHalt (HaltFatal "halt on 3")
else pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
_sp <- runSupervised master 10 (ProcessorId "halt-concurrent") Unordered (Async 3) adapter handler
liftIO $ threadDelay 500000 -- 500ms
stopMaster master
processed <- readIORef processedRef
-- With Async 3, messages 1, 2, 3 start together
-- When 3 halts, 1 and 2 should complete
length processed `shouldSatisfy` (>= 3)
it "stops reading new messages after halt decision" $ do
processedRef <- newIORef (0 :: Int)
_ <- runEff $ runTracingNoop $ do
messages <- createTestMessages 20
let handler _ = do
count <-
liftIO $
atomicModifyIORef' processedRef (\n -> let n' = n + 1 in (n', n'))
liftIO $ threadDelay 20000 -- 20ms
-- Halt after processing 5 messages
if count >= 5
then pure $ AckHalt (HaltFatal "halt at 5")
else pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
_sp <- runSupervised master 10 (ProcessorId "halt-stop-read") Unordered (Async 3) adapter handler
liftIO $ threadDelay 500000 -- 500ms
stopMaster master
processed <- readIORef processedRef
-- Should process around 5-8 messages (5 before halt + a few in-flight)
-- but definitely not all 20
processed `shouldSatisfy` (< 15)
describe "Error handling" $ do
it "handler exception doesn't affect other in-flight handlers" $ do
resultsRef <- newIORef ([] :: [Either String Int])
_ <- runEff $ runTracingNoop $ do
messages <- createTestMessages 5
let handler ingested = do
let msgNum = case ingested.envelope.cursor of
Just (CursorInt n) -> n
_ -> 0
-- Message 2 throws an exception
if msgNum == 2
then do
liftIO $ atomicModifyIORef' resultsRef (\xs -> (Left "error on 2" : xs, ()))
error "Intentional failure on message 2"
else do
liftIO $ threadDelay 30000 -- 30ms
liftIO $ atomicModifyIORef' resultsRef (\xs -> (Right msgNum : xs, ()))
pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
_ <- runSupervised master 10 (ProcessorId "error-handling") Unordered (Async 3) adapter handler
liftIO $ threadDelay 500000 -- 500ms
stopMaster master
results <- readIORef resultsRef
-- Message 2 should have errored
let errors = [e | Left e <- results]
length errors `shouldBe` 1
-- Other messages (1, 3, 4, 5) should have succeeded
let successes = [n | Right n <- results]
length successes `shouldSatisfy` (>= 3)
it "finalizes throwing handlers with immediate retry" $ do
(metrics, tracked) <- runEff $ runTracingNoop $ do
tracking <- newTrackingAck
let envelopes = createTestEnvelopes 5
adapter = trackedListAdapter tracking envelopes
throwingIds = [MessageId "msg-2", MessageId "msg-4"]
handler ingested =
if ingested.envelope.messageId `elem` throwingIds
then error "handler failed for conservation test"
else pure AckOk
sp <- runWithMetrics 10 (ProcessorId "handler-exception-finalize") adapter handler
decisions <- getTrackedDecisions tracking
finalMetrics <- getMetrics sp
pure (finalMetrics, decisions)
let expected =
Map.fromList
[ (MessageId "msg-1", AckOk),
(MessageId "msg-2", AckRetry (RetryDelay 0)),
(MessageId "msg-3", AckOk),
(MessageId "msg-4", AckRetry (RetryDelay 0)),
(MessageId "msg-5", AckOk)
]
finalizedExactlyOnce tracked expected `shouldBe` Right ()
metrics.stats.processed `shouldBe` 3
metrics.stats.failed `shouldBe` 2
it "preserves an application failure through finalization" $ do
let code = validDeadLetterCode "keiro.router.selection.recipient_overflow"
reason =
ApplicationFailure
code
"selected 101 recipients; configured limit is 100"
decision = AckDeadLetter reason
(metrics, tracked) <- runEff $ runTracingNoop $ do
tracking <- newTrackingAck
let adapter = trackedListAdapter tracking [createTestEnvelope 1]
handler _ = pure decision
sp <- runWithMetrics 1 (ProcessorId "application-failure-finalize") adapter handler
decisions <- getTrackedDecisions tracking
finalMetrics <- getMetrics sp
pure (finalMetrics, decisions)
finalizedExactlyOnce
tracked
(Map.singleton (MessageId "msg-1") decision)
`shouldBe` Right ()
metrics.stats.processed `shouldBe` 0
metrics.stats.failed `shouldBe` 1
it "retries transient finalizer failures on the single-message path" $ do
(attempts, tracked) <- runEff $ runTracingNoop $ do
attemptsRef <- liftIO $ newIORef (0 :: Int)
decisionsRef <- liftIO $ newIORef ([] :: [(MessageId, AckDecision)])
let env = createTestEnvelope 1
ingested =
mkIngested
env
( AckHandle $ \decision ->
liftIO $ do
attempt <-
atomicModifyIORef'
attemptsRef
(\n -> let n' = n + 1 in (n', n'))
if attempt <= 2
then ioError (userError "transient finalizer failure")
else modifyIORef' decisionsRef ((env.messageId, decision) :)
)
adapter = testAdapter [ingested]
_sp <- runWithMetrics 10 (ProcessorId "transient-finalizer") adapter alwaysAckOk
attemptCount <- liftIO $ readIORef attemptsRef
decisions <- liftIO $ readIORef decisionsRef
pure (attemptCount, decisions)
attempts `shouldBe` 3
finalizedExactlyOnce tracked (Map.singleton (MessageId "msg-1") AckOk) `shouldBe` Right ()
it "halts loudly when finalizer retry is exhausted" $ do
(metrics, done, tracked) <- runEff $ runTracingNoop $ do
tracking <- newTrackingAck
let env1 = createTestEnvelope 1
env2 = createTestEnvelope 2
failing =
mkIngested env1 (AckHandle $ \_ -> liftIO $ ioError (userError "permanent finalizer failure"))
trackedSecond = mkTrackedIngested tracking env2
adapter = testAdapter [failing, trackedSecond]
master <- startMaster IgnoreAll
sp <- runSupervised master 10 (ProcessorId "permanent-finalizer") Unordered (Async 2) adapter alwaysAckOk
liftIO $ threadDelay 700000
finalMetrics <- liftIO $ sampleMetrics sp.metrics
doneState <- liftIO $ readTVarIO sp.done
decisions <- getTrackedDecisions tracking
stopMaster master
pure (finalMetrics, doneState, decisions)
done `shouldBe` True
case metrics.state of
Failed msg _ -> msg `shouldSatisfy` Text.isInfixOf "msg-1"
other -> expectationFailure $ "Expected Failed state naming msg-1, got: " ++ show other
finalizedExactlyOnce tracked (Map.singleton (MessageId "msg-2") AckOk) `shouldBe` Right ()
describe "Metrics tracking" $ do
it "tracks in-flight count during concurrent processing" $ do
maxInFlightObserved <- newIORef (0 :: Int)
_ <- runEff $ runTracingNoop $ do
messages <- createTestMessages 5
let handler _ = do
liftIO $ threadDelay 100000 -- 100ms
pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
sp <- runSupervised master 10 (ProcessorId "metrics-test") Unordered (Async 3) adapter handler
-- Check metrics while processing
liftIO $ do
threadDelay 50000 -- 50ms - should be in the middle of processing
metrics <- sampleMetrics sp.metrics
case metrics.state of
Processing info _ _ -> modifyIORef' maxInFlightObserved (max info.inFlight)
_ -> pure ()
liftIO $ threadDelay 600000 -- 600ms to complete
stopMaster master
maxObserved <- readIORef maxInFlightObserved
-- Should have observed at least 2 concurrent handlers
maxObserved `shouldSatisfy` (>= 2)
it "reports correct maxConcurrency in metrics" $ do
observedMaxConc <- runEff $ runTracingNoop $ do
messages <- createTestMessages 3
let handler _ = do
liftIO $ threadDelay 50000 -- 50ms
pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
sp <- runSupervised master 10 (ProcessorId "max-conc") Unordered (Ahead 7) adapter handler
-- Check metrics while processing
result <- liftIO $ do
threadDelay 25000 -- 25ms
metrics <- sampleMetrics sp.metrics
case metrics.state of
Processing info _ _ -> pure $ Just info.maxConcurrency
_ -> pure Nothing
liftIO $ threadDelay 300000
stopMaster master
pure result
observedMaxConc `shouldBe` Just 7
describe "Ahead mode ordering" $ do
it "completes handlers and emits results in input order" $ do
-- Track the order in which handlers START and COMPLETE
startOrderRef <- newIORef ([] :: [Int])
completeOrderRef <- newIORef ([] :: [Int])
_ <- runEff $ runTracingNoop $ do
messages <- createTestMessages 5
let handler ingested = do
let msgNum = case ingested.envelope.cursor of
Just (CursorInt n) -> n
_ -> 0
-- Record start
liftIO $ atomicModifyIORef' startOrderRef (\xs -> (msgNum : xs, ()))
-- Variable delay: message 1 is slowest, 5 is fastest
liftIO $ threadDelay ((6 - msgNum) * 20000)
-- Record completion
liftIO $ atomicModifyIORef' completeOrderRef (\xs -> (msgNum : xs, ()))
pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
_ <- runSupervised master 10 (ProcessorId "ahead-order") Unordered (Ahead 5) adapter handler
liftIO $ threadDelay 500000 -- 500ms
stopMaster master
-- With Ahead mode, all 5 should complete
completeOrder <- readIORef completeOrderRef
length completeOrder `shouldBe` 5
-- The key property: all messages were processed
-- (OrderingPolicy of side effects may vary, but stream output is ordered)
describe "Robustness" $ do
describe "Adapter source exceptions" $ do
it "drains already-ingested messages, marks Failed, and propagates the ingester exception under IgnoreGraceful" $ do
processedRef <- newIORef (0 :: Int)
spRef <- newIORef (Nothing :: Maybe SupervisedProcessor)
result <-
UIO.withAsync
( runEff $ runTracingNoop $ do
-- Create an adapter where source throws after 3 messages
let failingSource = Stream.unfoldrM step (0 :: Int)
where
step n
| n < 3 = do
msg <- liftIO $ createSingleMessage (n + 1)
pure $ Just (msg, n + 1)
| n == 3 = error "Network failure mid-stream"
| otherwise = pure Nothing
let adapter =
Adapter
{ adapterName = "test:failing-source",
source = failingSource,
shutdown = pure ()
}
let handler _ = do
liftIO $ modifyIORef' processedRef (+ 1)
pure AckOk
master <- startMaster IgnoreGraceful
sp <- runSupervised master 10 (ProcessorId "failing-adapter") Unordered Serial adapter handler
liftIO $ atomicWriteIORef spRef (Just sp)
-- Wait for the linked ingester failure to reach this thread.
liftIO $ threadDelay 200000 -- 200ms
stopMaster master
)
UIO.waitCatch
case result of
Left err ->
Text.pack (show err) `shouldSatisfy` Text.isInfixOf "Network failure mid-stream"
Right () ->
expectationFailure "Expected the ingester exception to propagate"
-- Should have processed the 3 good messages before the error
processed <- readIORef processedRef
processed `shouldBe` 3
mSp <- readIORef spRef
case mSp of
Nothing -> expectationFailure "Expected supervised processor handle"
Just sp -> do
done <- readTVarIO sp.done
done `shouldBe` True
metrics <- sampleMetrics sp.metrics
case metrics.state of
Failed msg _ -> msg `shouldSatisfy` Text.isInfixOf "Network failure mid-stream"
other -> expectationFailure $ "Expected Failed state, got: " ++ show other
it "drains already-ingested messages and marks Failed without propagating under IgnoreAll" $ do
processedRef <- newIORef (0 :: Int)
spRef <- newIORef (Nothing :: Maybe SupervisedProcessor)
result <-
UIO.withAsync
( runEff $ runTracingNoop $ do
let failingSource = Stream.unfoldrM step (0 :: Int)
where
step n
| n < 3 = do
msg <- liftIO $ createSingleMessage (n + 1)
pure $ Just (msg, n + 1)
| n == 3 = error "Network failure mid-stream"
| otherwise = pure Nothing
let adapter =
Adapter
{ adapterName = "test:failing-source",
source = failingSource,
shutdown = pure ()
}
let handler _ = do
liftIO $ modifyIORef' processedRef (+ 1)
pure AckOk
master <- startMaster IgnoreAll
sp <- runSupervised master 10 (ProcessorId "failing-adapter-ignore") Unordered Serial adapter handler
liftIO $ atomicWriteIORef spRef (Just sp)
liftIO $ threadDelay 200000 -- 200ms
stopMaster master
)
UIO.waitCatch
case result of
Left err ->
expectationFailure $ "Expected IgnoreAll to isolate the ingester exception, got: " ++ show err
Right () ->
pure ()
processed <- readIORef processedRef
processed `shouldBe` 3
mSp <- readIORef spRef
case mSp of
Nothing -> expectationFailure "Expected supervised processor handle"
Just sp -> do
done <- readTVarIO sp.done
done `shouldBe` True
metrics <- sampleMetrics sp.metrics
case metrics.state of
Failed msg _ -> msg `shouldSatisfy` Text.isInfixOf "Network failure mid-stream"
other -> expectationFailure $ "Expected Failed state, got: " ++ show other
it "never drops an ingester failure (repeated)" $ do
results <- forM [(1 :: Int) .. 25] $ \i ->
runEff $ runTracingNoop $ do
let failingSource = Stream.unfoldrM (\() -> error "boom immediately") ()
adapter =
Adapter
{ adapterName = "test:failing-immediate",
source = failingSource,
shutdown = pure ()
}
handler _ = pure AckOk
master <- startMaster IgnoreAll
sp <- runSupervised master 10 (ProcessorId $ "failing-immediate-" <> Text.pack (show i)) Unordered Serial adapter handler
doneResult <-
liftIO $
UIO.timeout 2_000_000 $
atomically $ do
done <- readTVar sp.done
check done
metrics <- liftIO $ sampleMetrics sp.metrics
stopMaster master
pure (i, doneResult, metrics)
forM_ results $ \(i, doneResult, metrics) -> do
doneResult `shouldBe` Just ()
case metrics.state of
Failed msg _ -> msg `shouldSatisfy` Text.isInfixOf "boom immediately"
other -> expectationFailure $ "Iteration " ++ show i ++ " expected Failed state, got: " ++ show other
describe "Rapid start/stop cycles" $ do
it "handles rapid start/stop without resource leaks" $ do
-- Run 50 rapid cycles
results <- replicateM 50 $ runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
messages <- createTestMessages 5
let adapter = testAdapter messages
handler _ = pure AckOk
sp <- runSupervised master 10 (ProcessorId "rapid") Unordered Serial adapter handler
-- Minimal wait
liftIO $ threadDelay 10000 -- 10ms
stopMaster master
-- Check it stopped cleanly
isDone sp
-- All should complete (no hangs, no crashes)
length results `shouldBe` 50
it "handles concurrent start/stop cycles" $ do
-- Run 20 concurrent cycles
countRef <- newIORef (0 :: Int)
_ <- runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
-- Start 10 processors rapidly
_sps <- forM [(1 :: Int) .. 10] $ \i -> do
messages <- createTestMessages 3
let adapter = testAdapter messages
handler _ = do
liftIO $ atomicModifyIORef' countRef (\n -> (n + 1, ()))
liftIO $ threadDelay 5000 -- 5ms
pure AckOk
runSupervised master 10 (ProcessorId $ "concurrent-" <> Text.pack (show i)) Unordered Serial adapter handler
-- Wait briefly then stop
liftIO $ threadDelay 100000 -- 100ms
stopMaster master
-- Should have processed messages from multiple processors
count <- readIORef countRef
count `shouldSatisfy` (> 0)
describe "IgnoreGraceful supervision strategy (StopAllOnFailure)" $ do
it "stops all processors when adapter source fails with IgnoreGraceful strategy" $ do
countARef <- newIORef (0 :: Int)
countBRef <- newIORef (0 :: Int)
result <-
UIO.withAsync
( runEff $ runTracingNoop $ do
messagesB <- createTestMessages 20
-- Adapter A throws after 3 messages (ingester crash, not handler)
let failingSourceA = Stream.unfoldrM step (0 :: Int)
where
step n
| n < 3 = do
msg <- liftIO $ createSingleMessage (n + 1)
pure $ Just (msg, n + 1)
| otherwise = error "Adapter A source failed!"
let adapterA =
Adapter
{ adapterName = "test:failing-A",
source = failingSourceA,
shutdown = pure ()
}
-- Handler A counts messages
let handlerA _ = do
liftIO $ modifyIORef' countARef (+ 1)
liftIO $ threadDelay 10000 -- 10ms
pure AckOk
-- Handler B processes slowly
let handlerB _ = do
liftIO $ modifyIORef' countBRef (+ 1)
liftIO $ threadDelay 30000 -- 30ms
pure AckOk
let adapterB = testAdapter messagesB
-- Use the NQE strategy backing StopAllOnFailure.
master <- startMaster IgnoreGraceful
_spA <- runSupervised master 10 (ProcessorId "killall-A") Unordered Serial adapterA handlerA
_spB <- runSupervised master 10 (ProcessorId "killall-B") Unordered Serial adapterB handlerB
-- Wait for A to fail and trigger KillAll
liftIO $ threadDelay 500000 -- 500ms
stopMaster master
)
UIO.waitCatch
case result of
Left err ->
Text.pack (show err) `shouldSatisfy` Text.isInfixOf "Adapter A source failed"
Right () ->
expectationFailure "Expected adapter A source failure to propagate"
-- A should have processed messages before adapter failed
countA <- readIORef countARef
countA `shouldSatisfy` (<= 3)
-- B should have been killed (not processed all 20)
countB <- readIORef countBRef
countB `shouldSatisfy` (< 20)
describe "Multiple concurrent halts" $ do
it "handles multiple handlers returning AckHalt simultaneously" $ do
haltCountRef <- newIORef (0 :: Int)
processedRef <- newIORef (0 :: Int)
_ <- runEff $ runTracingNoop $ do
messages <- createTestMessages 10
let handler ingested = do
let msgNum = case ingested.envelope.cursor of
Just (CursorInt n) -> n
_ -> 0
liftIO $ atomicModifyIORef' processedRef (\n -> (n + 1, ()))
liftIO $ threadDelay 30000 -- 30ms
-- Messages 2, 3, 4 all return halt
if msgNum `elem` [2, 3, 4]
then do
liftIO $ atomicModifyIORef' haltCountRef (\n -> (n + 1, ()))
pure $ AckHalt (HaltFatal $ "halt from " <> Text.pack (show msgNum))
else pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
_ <- runSupervised master 10 (ProcessorId "multi-halt") Unordered (Async 5) adapter handler
liftIO $ threadDelay 500000 -- 500ms
stopMaster master
-- Multiple halts should have been recorded
haltCount <- readIORef haltCountRef
haltCount `shouldSatisfy` (>= 1) -- At least one halt processed
-- Should not have processed all messages
processed <- readIORef processedRef
processed `shouldSatisfy` (< 10)
describe "Stability under load" $ do
it "processes many messages without issues" $ do
processedRef <- newIORef (0 :: Int)
finalCount <- runEff $ runTracingNoop $ do
-- 500 messages
messages <- createTestMessages 500
let handler _ = do
liftIO $ atomicModifyIORef' processedRef (\n -> (n + 1, ()))
pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
_sp <- runSupervised master 100 (ProcessorId "load-test") Unordered (Async 10) adapter handler
-- Wait for completion
liftIO $ threadDelay 2000000 -- 2s
stopMaster master
liftIO $ readIORef processedRef
finalCount `shouldBe` 500
it "handles mixed success and failure under load" $ do
successRef <- newIORef (0 :: Int)
failRef <- newIORef (0 :: Int)
_ <- runEff $ runTracingNoop $ do
messages <- createTestMessages 200
let handler ingested = do
let msgNum = case ingested.envelope.cursor of
Just (CursorInt n) -> n
_ -> 0
-- Every 7th message fails
if msgNum `mod` 7 == 0
then do
liftIO $ atomicModifyIORef' failRef (\n -> (n + 1, ()))
error "Intentional failure"
else do
liftIO $ atomicModifyIORef' successRef (\n -> (n + 1, ()))
pure AckOk
master <- startMaster IgnoreAll
let adapter = testAdapter messages
_ <- runSupervised master 50 (ProcessorId "mixed-load") Unordered (Async 5) adapter handler
liftIO $ threadDelay 1000000 -- 1s
stopMaster master
success <- readIORef successRef
failures <- readIORef failRef
-- Should have processed many messages
success `shouldSatisfy` (> 150)
-- Should have recorded failures
failures `shouldSatisfy` (> 20)
describe "Graceful Shutdown" $ do
describe "stopAppGracefully" $ do
it "drains in-flight messages before stopping" $ do
processedRef <- newIORef (0 :: Int)
drained <- runEff $ runTracingNoop $ do
messages <- createTestMessages 10
let handler _ = do
liftIO $ do
threadDelay 50000 -- 50ms per message
modifyIORef' processedRef (+ 1)
pure AckOk
let adapter = testAdapter messages
processor = mkProcessor adapter handler
result <- runApp defaultAppConfig {inboxSize = 10} [(ProcessorId "drain-test", processor)]
case result of
Left _ -> pure False
Right appHandle -> do
-- Give some time for processing to start
liftIO $ threadDelay 100000 -- 100ms
-- Graceful shutdown with generous timeout
let config = ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6} -- 5 seconds
stopAppGracefully config appHandle
drained `shouldBe` True
-- All messages should have been processed
processed <- readIORef processedRef
processed `shouldBe` 10
it "respects timeout when processors are slow" $ do
processedRef <- newIORef (0 :: Int)
drained <- runEff $ runTracingNoop $ do
messages <- createTestMessages 20
let handler _ = do
liftIO $ do
threadDelay 500000 -- 500ms per message - very slow
modifyIORef' processedRef (+ 1)
pure AckOk
let adapter = testAdapter messages
processor = mkProcessor adapter handler
result <- runApp defaultAppConfig {inboxSize = 5} [(ProcessorId "timeout-test", processor)]
case result of
Left _ -> pure True -- Treat error as "drained" for simplicity
Right appHandle -> do
-- Start immediately
liftIO $ threadDelay 100000 -- 100ms
-- Very short timeout (0.3 seconds)
let config = ShutdownConfig {drainTimeout = 0.3, totalShutdownTimeout = 1.3}
stopAppGracefully config appHandle
-- Should timeout (not all drained)
drained `shouldBe` False
-- Should have processed only some messages
processed <- readIORef processedRef
processed `shouldSatisfy` (< 20)
it "returns True when all processors finish quickly" $ do
drained <- runEff $ runTracingNoop $ do
messages <- createTestMessages 5
let handler _ = pure AckOk -- Instant processing
let adapter = testAdapter messages
processor = mkProcessor adapter handler
result <- runApp defaultAppConfig {inboxSize = 10} [(ProcessorId "quick-test", processor)]
case result of
Left _ -> pure False
Right appHandle -> do
-- Give time to complete
liftIO $ threadDelay 200000 -- 200ms
let config = ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}
stopAppGracefully config appHandle
drained `shouldBe` True
-- Test helpers
testTime :: UTCTime
testTime = UTCTime (fromGregorian 2026 1 1) 0
createTestEnvelope :: Int -> Envelope String
createTestEnvelope i =
(mkEnvelope (MessageId $ "msg-" <> Text.pack (show i)) ("message-" <> show i))
{ cursor = Just (CursorInt i),
enqueuedAt = Just testTime
}
createTestEnvelopes :: Int -> [Envelope String]
createTestEnvelopes n = map createTestEnvelope [1 .. n]
createTestMessages :: (IOE :> es) => Int -> Eff es [Ingested es String]
createTestMessages n = mapM createMessage [1 .. n]
where
createMessage i = do
let env = createTestEnvelope i
ackHandle = AckHandle $ \_ -> pure ()
pure $ mkIngested env ackHandle
-- | Create a single message (for use in streaming contexts)
-- Polymorphic over effect stack for use with tracing.
createSingleMessage :: (IOE :> es) => Int -> IO (Ingested es String)
createSingleMessage i = do
let env = createTestEnvelope i
ackHandle = AckHandle $ \_ -> pure ()
pure $ mkIngested env ackHandle
testAdapter :: [Ingested es String] -> Adapter es String
testAdapter messages =
Adapter
{ adapterName = "test:supervised",
source = Stream.fromList messages,
shutdown = pure ()
}
validDeadLetterCode :: Text.Text -> DeadLetterCode
validDeadLetterCode code =
case mkDeadLetterCode code of
Left err -> error $ "invalid test fixture: " <> show err
Right valid -> valid
testHandler :: (IOE :> es) => IORef [String] -> Handler es String
testHandler ref ingested = do
liftIO $ modifyIORef' ref (ingested.envelope.payload :)
pure AckOk
failingHandler :: Handler es msg
failingHandler _ = error "Intentional failure"
alwaysAckOk :: Handler es msg
alwaysAckOk _ = pure AckOk
cursorOutcome :: AckDecision -> Bool -> IO (Maybe Metrics.ProgressSnapshot)
cursorOutcome decision failFinalizer = runEff $ runTracingNoop $ do
master <- startMaster IgnoreAll
let message =
mkIngested
((mkEnvelope (MessageId "cursor") ()) {cursor = Just (CursorInt 5), partition = Just "p"})
(AckHandle (\_ -> when failFinalizer (liftIO $ UIO.throwIO (userError "finalization failed"))))
adapter = Adapter "cursor" (Stream.fromList [message]) (pure ())
proc <- runSupervisedWith (Metrics.MetricsDetailed Metrics.defaultDetailOptions) master 10 (ProcessorId "cursor") Unordered Serial adapter (\_ -> pure decision)
liftIO $ atomically $ readTVar proc.done >>= check
m <- getMetrics proc
stopMaster master
pure m.progress