shibuya-core-0.8.0.0: test/Shibuya/Runner/SupervisedSpec.hs
{-# LANGUAGE OverloadedStrings #-}
module Shibuya.Runner.SupervisedSpec (spec) where
import Control.Concurrent.NQE.Supervisor (Strategy (..))
import Control.Concurrent.STM (atomically, check, readTVar, readTVarIO)
import Control.Monad (forM, forM_, replicateM)
import Data.IORef (IORef, atomicModifyIORef', atomicWriteIORef, modifyIORef', newIORef, readIORef)
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 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.TestHarness (finalizedExactlyOnce)
import Shibuya.Core.Ack (AckDecision (..), HaltReason (..), RetryDelay (..))
import Shibuya.Core.AckHandle (AckHandle (..))
import Shibuya.Core.Ingested (Ingested, Message (..), mkIngested)
import Shibuya.Core.Metrics
( InFlightInfo (..),
ProcessorId (..),
ProcessorMetrics (..),
ProcessorState (..),
StreamStats (..),
sampleMetrics,
)
import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), mkEnvelope)
import Shibuya.Handler (Handler)
import Shibuya.Internal.Runner.Master
( getAllMetrics,
getAllMetricsIO,
getProcessorMetricsIO,
startMaster,
stopMaster,
)
import Shibuya.Internal.Runner.Supervised
( SupervisedProcessor (..),
getMetrics,
isDone,
runSupervised,
runWithMetrics,
)
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 "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
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 "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} -- 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}
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}
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 ()
}
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