diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,38 @@ # Changelog +## 0.12.0.0 — 2026-10-26 + +### Breaking Changes + +- `Ingested` gains `deliveryStatus`. Direct record constructors must add + `deliveryStatus = pure DeliveryCurrent`; positional construction requires the + extra field. `mkIngested` continues to build current deliveries. + +### New Features + +- Export `DeliveryStatus` so adapters can supersede buffered deliveries before + handler admission. Sample status once; skip handlers, finalizers and processing + outcomes for superseded deliveries. Filter batches in order, update their + retained size, and skip processing when no current members remain. +- Require delivery-status actions to return promptly without side effects or + synchronous exceptions. Unexpected exceptions propagate as infrastructure + failures; cancellation preserves existing ownership behavior. + +### Other Changes + +- Add concurrent-key regression coverage for Ahead and Async runners, single and + batch handlers, and basic and detailed metrics. Verify queued successors are + checked after finalization and superseded deliveries skip handlers and finalizers. +- Accept this candidate's measured allocation increase of approximately 8 bytes + per constructed record and 24 bytes per processed message in the existing + framework workloads. All 62 timing comparisons passed the 10% gate; this + candidate-specific acceptance leaves the general allocation policy unchanged. +- The release comparison against 0.11.0.0 measured 10,000-message processing at + 3.519 to 3.741 ms (+6.31%), with allocation increasing from 22,517,226 to + 22,756,034 bytes. Construction allocation increased from 15,017,773 to + 15,097,727 bytes for 10,000 records. Measurements used package-default RTS + settings on a shared workstation. + ## 0.11.0.0 — 2026-10-25 ### Breaking Changes diff --git a/shibuya-core.cabal b/shibuya-core.cabal --- a/shibuya-core.cabal +++ b/shibuya-core.cabal @@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-core -version: 0.11.0.0 +version: 0.12.0.0 synopsis: Supervised queue processing framework for Haskell description: A supervised queue processing framework inspired by Broadway (Elixir). diff --git a/src/Shibuya.hs b/src/Shibuya.hs --- a/src/Shibuya.hs +++ b/src/Shibuya.hs @@ -79,6 +79,7 @@ AckHandle (..), Lease (..), Ingested (..), + DeliveryStatus (..), mkIngested, toMessage, @@ -153,7 +154,7 @@ ) import Shibuya.Core.AckHandle (AckHandle (..)) import Shibuya.Core.Error (ConfigError (..), HandlerError (..), PolicyError (..), RuntimeError (..)) -import Shibuya.Core.Ingested (Ingested (..), Message (..), mkIngested, toMessage) +import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), Message (..), mkIngested, toMessage) import Shibuya.Core.Lease (Lease (..)) import Shibuya.Core.Metrics ( BatchStats (..), diff --git a/src/Shibuya/Core/AckHandle.hs b/src/Shibuya/Core/AckHandle.hs --- a/src/Shibuya/Core/AckHandle.hs +++ b/src/Shibuya/Core/AckHandle.hs @@ -6,7 +6,8 @@ -- failure. Adapters must therefore make finalization idempotent or -- phase-tracked: re-running a partially completed finalization with the same -- decision must be safe. The framework never intentionally finalizes one --- delivery with two different decisions. +-- delivery with two different decisions. A delivery whose admission status is +-- DeliverySuperseded receives no finalizer call. module Shibuya.Core.AckHandle ( AckHandle (..), ) diff --git a/src/Shibuya/Core/Ingested.hs b/src/Shibuya/Core/Ingested.hs --- a/src/Shibuya/Core/Ingested.hs +++ b/src/Shibuya/Core/Ingested.hs @@ -3,16 +3,28 @@ -- Exactly one thing flows through the system. module Shibuya.Core.Ingested ( Ingested (..), + DeliveryStatus (..), Message (..), toMessage, mkIngested, ) where +import Effectful (Eff) + import Shibuya.Core.AckHandle (AckHandle) import Shibuya.Core.Lease (Lease) import Shibuya.Core.Types (Envelope) +-- | Adapter admission snapshot; a current answer does not reserve the delivery +-- against a subsequent revocation or concurrent retry. +data DeliveryStatus + = -- | Invoke the handler and finalize normally. + DeliveryCurrent + | -- | Redelivery is arranged or the attempt is fenced. Skip handler and finalizer. + DeliverySuperseded + deriving stock (Eq, Show) + -- | Framework-side message with the adapter-provided ack finalizer. -- Adapters construct this; application handlers receive @Message@ instead. data Ingested es msg = Ingested @@ -21,7 +33,12 @@ -- | Handle for acknowledging the message ack :: !(AckHandle es), -- | Optional lease for visibility timeout extension - lease :: !(Maybe (Lease es)) + lease :: !(Maybe (Lease es)), + -- | Read once at processing admission (once per member on the batch path). + -- Must return promptly, without side effects or synchronous exceptions; + -- asynchronous cancellation propagates. Superseded deliveries own no metrics + -- slot and receive no handler or finalizer call. + deliveryStatus :: !(Eff es DeliveryStatus) } -- | The read-only view a handler receives: envelope plus optional lease, @@ -47,5 +64,6 @@ Ingested { envelope = envelope, ack = ack, - lease = Nothing + lease = Nothing, + deliveryStatus = pure DeliveryCurrent } diff --git a/src/Shibuya/Internal/Runner/BatchProcessor.hs b/src/Shibuya/Internal/Runner/BatchProcessor.hs --- a/src/Shibuya/Internal/Runner/BatchProcessor.hs +++ b/src/Shibuya/Internal/Runner/BatchProcessor.hs @@ -59,7 +59,7 @@ HaltReason (..), RetryDelay (..), ) -import Shibuya.Core.Ingested (Ingested (..), toMessage) +import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), toMessage) import Shibuya.Core.Metrics ( AckDecisionMetric (..), BatchTriggerMetric (..), @@ -148,7 +148,42 @@ BatchHandler es msg -> (BatchInfo, NonEmpty (Ingested es msg)) -> Eff es () -processOneBatchWithDetail detailed metricsHandle procId maxConc stopSignal exitPublisher handler (info, batch) = do +processOneBatchWithDetail detailed metricsHandle procId maxConc stopSignal exitPublisher handler original@(info, batch) = + scan 0 (NE.toList batch) + where + -- All-current batches retain the original pair without constructing a list. + scan _ [] = run original + scan count (member : rest) = do + status <- member.deliveryStatus + case status of + DeliveryCurrent -> scan (count + 1) rest + DeliverySuperseded -> do + suffix <- retain rest + case NE.nonEmpty (take count (NE.toList batch) <> suffix) of + Nothing -> pure () + Just current -> run (info {size = NE.length current}, current) + retain [] = pure [] + retain (member : rest) = do + status <- member.deliveryStatus + suffix <- retain rest + pure $ case status of + DeliveryCurrent -> member : suffix + DeliverySuperseded -> suffix + run = processCurrentBatchWithDetail detailed metricsHandle procId maxConc stopSignal exitPublisher handler +{-# INLINE processOneBatchWithDetail #-} + +processCurrentBatchWithDetail :: + (IOE :> es, Tracing :> es) => + Bool -> + MetricsHandle -> + ProcessorId -> + Int -> + ProcessorSignal -> + ProcessorExitPublisher -> + BatchHandler es msg -> + (BatchInfo, NonEmpty (Ingested es msg)) -> + Eff es () +processCurrentBatchWithDetail detailed metricsHandle procId maxConc stopSignal exitPublisher handler (info, batch) = do -- Use the first message's trace context as the batch span's parent. A batch -- may span several traces; picking the first is a pragmatic single parent -- (full fan-in links are a later refinement). @@ -286,7 +321,7 @@ isFailing (AckDeadLetter _) = True isFailing (AckRetry _) = True isFailing _ = False -{-# INLINE processOneBatchWithDetail #-} +{-# INLINE processCurrentBatchWithDetail #-} decisionMetric :: AckDecision -> AckDecisionMetric decisionMetric AckOk = CountProcessed diff --git a/src/Shibuya/Internal/Runner/Supervised.hs b/src/Shibuya/Internal/Runner/Supervised.hs --- a/src/Shibuya/Internal/Runner/Supervised.hs +++ b/src/Shibuya/Internal/Runner/Supervised.hs @@ -65,7 +65,7 @@ renderDeadLetterReason, ) import Shibuya.Core.Error (HandlerError (..), handlerErrorToText) -import Shibuya.Core.Ingested (Ingested (..), toMessage) +import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), toMessage) import Shibuya.Core.Metrics ( AckDecisionMetric (..), MetricsHandle (..), @@ -723,115 +723,124 @@ mergedAttrs = HashMap.union ingested.envelope.attributes frameworkAttrs addAttributes traceSpan mergedAttrs - let process currentInflight = do - addAttribute traceSpan attrShibuyaInflightCount currentInflight - addAttribute traceSpan attrShibuyaInflightMax maxConc + status <- ingested.deliveryStatus + case status of + DeliverySuperseded -> do + addEvent traceSpan $ + mkEvent + eventAckDecision + [(attrShibuyaAckDecision, OTel.toAttribute ("superseded" :: Text))] + setStatus traceSpan OTel.Ok + DeliveryCurrent -> do + let process currentInflight = do + addAttribute traceSpan attrShibuyaInflightCount currentInflight + addAttribute traceSpan attrShibuyaInflightMax maxConc - -- Record handler start event - addEvent traceSpan handlerStartedEvent + -- Record handler start event + addEvent traceSpan handlerStartedEvent - -- Call handler and finalizer separately. A handler exception is - -- substituted with immediate retry so the adapter always observes a - -- finalization decision for an ingested message. - -- - -- This is a *separate* per-message 'catchAny' from the one inside - -- 'finalizeWithRetry' below. 0.7.1.0 used a single combined catch around - -- handler+finalize, which skipped finalization when the handler threw; - -- splitting them is what makes "always finalize" hold. The extra - -- per-message exception frame costs ~126 bytes/message (measured, EP-31 - -- S3: docs/plans/31-…-shared-per-message-…-allocation-regression.md) and is - -- accepted as the price of the always-finalize guarantee — do not merge the - -- two catches back together to reclaim it. - handlerResult <- - catchAny - (Right <$> handler (toMessage ingested)) - ( \ex -> do - recordException traceSpan ex - pure (Left ex) - ) - let (decision, result) = case handlerResult of - Right d -> (d, Right d) - Left ex -> - ( AckRetry (RetryDelay 0), - Left $ HandlerException $ Text.pack $ show ex + -- Call handler and finalizer separately. A handler exception is + -- substituted with immediate retry so an admitted delivery always receives a + -- finalization decision. Superseded deliveries are excluded. + -- + -- This is a *separate* per-message 'catchAny' from the one inside + -- 'finalizeWithRetry' below. 0.7.1.0 used a single combined catch around + -- handler+finalize, which skipped finalization when the handler threw; + -- splitting them is what makes "always finalize admitted deliveries" hold. The extra + -- per-message exception frame costs ~126 bytes/message (measured, EP-31 + -- S3: docs/plans/31-…-shared-per-message-…-allocation-regression.md) and is + -- accepted as the price of the always-finalize guarantee — do not merge the + -- two catches back together to reclaim it. + handlerResult <- + catchAny + (Right <$> handler (toMessage ingested)) + ( \ex -> do + recordException traceSpan ex + pure (Left ex) ) - - when (isLeft handlerResult) $ - addEvent traceSpan $ - mkEvent - eventAckDecision - [(attrShibuyaAckDecision, OTel.toAttribute ("ack_retry" :: Text))] - - finalizeResult <- finalizeWithRetry traceSpan ingested decision - - -- Record completion event and set status - case finalizeResult of - Left _ -> do - addEvent traceSpan $ - mkEvent - eventAckDecision - [(attrShibuyaAckDecision, OTel.toAttribute ("finalization_failed" :: Text))] - setStatus traceSpan $ OTel.Error $ finalizationFailureText msgIdText - Right () -> case result of - Right decision' -> do - let decisionText = showAckDecision decision' - addEvent traceSpan $ - mkEvent - eventHandlerCompleted - [(attrShibuyaAckDecision, OTel.toAttribute decisionText)] - addAttribute traceSpan attrShibuyaAckDecision decisionText + let (decision, result) = case handlerResult of + Right d -> (d, Right d) + Left ex -> + ( AckRetry (RetryDelay 0), + Left $ HandlerException $ Text.pack $ show ex + ) - -- Set span status based on decision - case decision' of - AckOk -> setStatus traceSpan OTel.Ok - AckRetry _ -> setStatus traceSpan OTel.Ok - AckDeadLetter reason -> do - addAttribute - traceSpan - attrShibuyaDeadLetterReasonCode - (deadLetterCodeText (deadLetterReasonCode reason)) - setStatus traceSpan $ OTel.Error $ renderDeadLetterReason reason - AckHalt reason -> - setStatus traceSpan $ OTel.Error $ showHaltReason reason - Left err -> do + when (isLeft handlerResult) $ addEvent traceSpan $ mkEvent eventAckDecision - [(attrShibuyaAckDecision, OTel.toAttribute ("error" :: Text))] - setStatus traceSpan $ OTel.Error $ handlerErrorToText err + [(attrShibuyaAckDecision, OTel.toAttribute ("ack_retry" :: Text))] - let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor - when detailed $ case (finalizeResult, result) of - (Right (), Right AckOk) -> recordCursor - (Right (), Right (AckDeadLetter _)) -> recordCursor - _ -> pure () + finalizeResult <- finalizeWithRetry traceSpan ingested decision - -- Decrement in-flight and update stats. - liftIO $ - case finalizeResult of - Left _ -> finishFinalizationFailure metricsHandle (finalizationFailureText msgIdText) - Right () -> finishProcessing metricsHandle (metricForResult result) + -- Record completion event and set status + case finalizeResult of + Left _ -> do + addEvent traceSpan $ + mkEvent + eventAckDecision + [(attrShibuyaAckDecision, OTel.toAttribute ("finalization_failed" :: Text))] + setStatus traceSpan $ OTel.Error $ finalizationFailureText msgIdText + Right () -> case result of + Right decision' -> do + let decisionText = showAckDecision decision' + addEvent traceSpan $ + mkEvent + eventHandlerCompleted + [(attrShibuyaAckDecision, OTel.toAttribute decisionText)] + addAttribute traceSpan attrShibuyaAckDecision decisionText - -- Handle halt (set flag, don't throw - let stream drain) - case finalizeResult of - Left _ -> + -- Set span status based on decision + case decision' of + AckOk -> setStatus traceSpan OTel.Ok + AckRetry _ -> setStatus traceSpan OTel.Ok + AckDeadLetter reason -> do + addAttribute + traceSpan + attrShibuyaDeadLetterReasonCode + (deadLetterCodeText (deadLetterReasonCode reason)) + setStatus traceSpan $ OTel.Error $ renderDeadLetterReason reason + AckHalt reason -> + setStatus traceSpan $ OTel.Error $ showHaltReason reason + Left err -> do + addEvent traceSpan $ + mkEvent + eventAckDecision + [(attrShibuyaAckDecision, OTel.toAttribute ("error" :: Text))] + setStatus traceSpan $ OTel.Error $ handlerErrorToText err + + let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor + when detailed $ case (finalizeResult, result) of + (Right (), Right AckOk) -> recordCursor + (Right (), Right (AckDeadLetter _)) -> recordCursor + _ -> pure () + + -- Decrement in-flight and update stats. liftIO $ - requestProcessorExit exitPublisher (ProcessorFailed (finalizationFailureText msgIdText) (Just messageId)) - Right () -> case result of - Right (AckHalt reason) -> - liftIO $ requestProcessorExit exitPublisher (ProcessorHalted reason) - _ -> pure () - if not detailed - then do - currentInflight <- liftIO $ beginProcessing metricsHandle maxConc - process currentInflight - else Exception.mask $ \restore -> do - currentInflight <- liftIO $ beginProcessing metricsHandle maxConc - slot <- liftIO $ Metrics.claimInFlightSlot metricsHandle messageId - restore (process currentInflight) - `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot) - liftIO (Metrics.releaseInFlightSlot metricsHandle slot) - `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot) + case finalizeResult of + Left _ -> finishFinalizationFailure metricsHandle (finalizationFailureText msgIdText) + Right () -> finishProcessing metricsHandle (metricForResult result) + + -- Handle halt (set flag, don't throw - let stream drain) + case finalizeResult of + Left _ -> + liftIO $ + requestProcessorExit exitPublisher (ProcessorFailed (finalizationFailureText msgIdText) (Just messageId)) + Right () -> case result of + Right (AckHalt reason) -> + liftIO $ requestProcessorExit exitPublisher (ProcessorHalted reason) + _ -> pure () + if not detailed + then do + currentInflight <- liftIO $ beginProcessing metricsHandle maxConc + process currentInflight + else Exception.mask $ \restore -> do + currentInflight <- liftIO $ beginProcessing metricsHandle maxConc + slot <- liftIO $ Metrics.claimInFlightSlot metricsHandle messageId + restore (process currentInflight) + `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot) + liftIO (Metrics.releaseInFlightSlot metricsHandle slot) + `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot) where isLeft :: Either a b -> Bool isLeft (Left _) = True diff --git a/test/Shibuya/Runner/BatchProcessorSpec.hs b/test/Shibuya/Runner/BatchProcessorSpec.hs --- a/test/Shibuya/Runner/BatchProcessorSpec.hs +++ b/test/Shibuya/Runner/BatchProcessorSpec.hs @@ -1,13 +1,8 @@ module Shibuya.Runner.BatchProcessorSpec (spec) where -import Control.Concurrent.STM - ( atomically, - modifyTVar', - newTVarIO, - readTVar, - readTVarIO, - writeTVar, - ) +import Control.Concurrent.NQE.Supervisor (Strategy (..)) +import Control.Concurrent.STM (atomically, check, modifyTVar', newTVarIO, readTVar, readTVarIO, writeTVar) +import Control.Monad (forM_) import Data.IORef (IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef) import Data.List (sort) import Data.List.NonEmpty (NonEmpty) @@ -18,6 +13,7 @@ import Data.Time (UTCTime (..), fromGregorian) import Effectful (Eff, IOE, liftIO, runEff, (:>)) import Shibuya (ProcessorFailure (..), ProcessorHalt (..)) +import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Mock ( TrackingAck (..), getTrackedDecisions, @@ -25,11 +21,14 @@ trackingAckHandle, ) import Shibuya.Batch - ( BatchInfo (..), + ( BatchConfig (..), + BatchHandler, + BatchInfo (..), BatchKey (..), BatchTrigger (..), ackAllOk, ackExcept, + defaultBatchConfig, defaultBatchKey, withFallback, ) @@ -42,23 +41,113 @@ mkDeadLetterCode, ) import Shibuya.Core.AckHandle (AckHandle (..)) -import Shibuya.Core.Ingested (Ingested, Message (..), mkIngested) +import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), Message (..), mkIngested) import Shibuya.Core.Metrics ( BatchStats (..), ProcessorId (..), ProcessorMetrics (..), StreamStats (..), ) -import Shibuya.Core.Types (Envelope (..), MessageId (..), mkEnvelope) +import Shibuya.Core.Metrics qualified as Metrics +import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), mkEnvelope) import Shibuya.Internal.Runner.BatchProcessor (runBatchesWithMetrics) +import Shibuya.Internal.Runner.Master (startMaster, stopMaster) +import Shibuya.Internal.Runner.Supervised (SupervisedProcessor (..), runSupervisedBatchWith) import Shibuya.Policy (Concurrency (..)) -import Shibuya.Telemetry.Effect (runTracingNoop) +import Shibuya.Telemetry.Effect (Tracing, runTracingNoop) +import Streamly.Data.Stream qualified as Stream import Test.Hspec import UnliftIO (throwIO, try) import UnliftIO.Concurrent (threadDelay) spec :: Spec spec = describe "Shibuya.Internal.Runner.BatchProcessor" $ do + describe "delivery status" $ do + forM_ [Metrics.MetricsBasic, Metrics.MetricsDetailed Metrics.defaultDetailOptions] $ \detail -> do + forM_ [[1, 3, 5], [1 .. 5], []] $ \retained -> + it ("samples every status once and retains current members " <> show retained <> " / " <> show detail) $ do + calls <- newIORef [] + handled <- newIORef [] + (tracked, metrics) <- runEff $ runTracingNoop $ do + tracking <- newTrackingAck + (info, batch) <- buildBatch tracking 5 TriggerSize + let members = + NE.zipWith + ( \n item -> + item + { envelope = item.envelope {cursor = Just (CursorInt (fromIntegral n))}, + deliveryStatus = liftIO $ do + previous <- atomicModifyIORef' calls (\xs -> (xs <> [n], xs)) + -- A superseded status flips on a second sample; two reads + -- would wrongly re-admit it and fail both assertions. + pure $ if n `elem` retained || n `elem` previous then DeliveryCurrent else DeliverySuperseded + } + ) + (NE.fromList [1 .. 5 :: Int]) + batch + handler actual messages = do + liftIO $ modifyIORef' handled (<> [(actual, map (.envelope.messageId) (NE.toList messages))]) + pure ackAllOk + m <- runStatusBatches detail handler [(info, members)] + t <- getTrackedDecisions tracking + pure (t, m) + readIORef calls `shouldReturn` [1 .. 5] + map fst (reverse tracked) `shouldBe` [MessageId ("msg-" <> Text.pack (show n)) | n <- retained] + observed <- readIORef handled + case retained of + [] -> do + observed `shouldBe` [] + metrics.batch.batchesEmitted `shouldBe` 0 + _ -> case observed of + [(actual, ids)] -> do + actual.size `shouldBe` length retained + actual.batchKey `shouldBe` defaultBatchKey + actual.trigger `shouldBe` TriggerSize + actual.partition `shouldBe` Nothing + ids `shouldBe` map fst (reverse tracked) + _ -> expectationFailure (show observed) + metrics.stats.processed `shouldBe` length retained + metrics.stats.failed `shouldBe` 0 + metrics.oldestInFlight `shouldBe` Nothing + case detail of + Metrics.MetricsBasic -> metrics.latency `shouldBe` Nothing + _ -> do + fmap (.count) metrics.latency `shouldBe` (if null retained then Nothing else Just 1) + (metrics.progress >>= (.latest)) `shouldBe` (if null retained then Nothing else Just (CursorInt (fromIntegral (last retained)))) + + it "checks a later batch after the previous batch finalizes" $ do + current <- newIORef True + handled <- newIORef [] + finalized <- newIORef [] + metrics <- runEff $ runTracingNoop $ do + tracking <- newTrackingAck + (info, first) <- buildBatch tracking 1 TriggerSize + let a = (NE.head first) {ack = AckHandle (\_ -> liftIO $ atomicModifyIORef' current (const (False, ())))} + b = + (NE.head first) + { deliveryStatus = liftIO $ (\x -> if x then DeliveryCurrent else DeliverySuperseded) <$> readIORef current, + ack = AckHandle (\_ -> liftIO $ modifyIORef' finalized (<> [2 :: Int])) + } + handler _ _ = liftIO (modifyIORef' handled (<> [1 :: Int])) >> pure ackAllOk + runBatchesWithMetrics (ProcessorId "status-later") Serial handler [(info, NE.singleton a), (info, NE.singleton b)] + readIORef handled `shouldReturn` [1] + readIORef finalized `shouldReturn` [] + metrics.batch.batchesEmitted `shouldBe` 1 + + it "lets throwing status escape as infrastructure failure" $ do + handlerCalls <- newIORef (0 :: Int) + finalized <- newIORef (0 :: Int) + result <- try @IO @IOError $ runEff $ runTracingNoop $ do + tracking <- newTrackingAck + (info, batch) <- buildBatch tracking 1 TriggerSize + let item = (NE.head batch) {deliveryStatus = liftIO $ ioError (userError "batch status infrastructure failure"), ack = AckHandle (\_ -> liftIO $ modifyIORef' finalized (+ 1))} + runBatchesWithMetrics (ProcessorId "status-throw") Serial (\_ _ -> liftIO (modifyIORef' handlerCalls (+ 1)) >> pure ackAllOk) [(info, NE.singleton item)] + case result of + Left _ -> pure () + Right _ -> expectationFailure "expected infrastructure failure" + readIORef handlerCalls `shouldReturn` 0 + readIORef finalized `shouldReturn` 0 + describe "decision resolution and finalization (M1)" $ do it "finalizes each of 5 messages successfully with per-message decisions" $ do (tracked, metrics) <- runEff $ runTracingNoop $ do @@ -358,3 +447,17 @@ case mkDeadLetterCode code of Left err -> error $ "invalid test fixture: " <> show err Right valid -> valid + +-- Exercise the actual supervised batch runner in both metrics modes. +runStatusBatches :: (IOE :> es, Tracing :> es) => Metrics.MetricsDetail -> BatchHandler es msg -> [(BatchInfo, NonEmpty (Ingested es msg))] -> Eff es ProcessorMetrics +runStatusBatches detail handler batches = do + master <- startMaster IgnoreAll + let values = concatMap (NE.toList . snd) batches + adapter = Adapter "status-batch" (Stream.fromList values) (pure ()) + config = defaultBatchConfig {batchSize = length values} + proc <- runSupervisedBatchWith detail master 10 (ProcessorId "status-batch") Serial config adapter handler + liftIO $ atomically $ readTVar proc.done >>= check + metrics <- liftIO $ Metrics.sampleMetrics proc.metrics + liftIO $ Metrics.slotMisses proc.metrics `shouldReturn` 0 + stopMaster master + pure metrics diff --git a/test/Shibuya/Runner/SupervisedSpec.hs b/test/Shibuya/Runner/SupervisedSpec.hs --- a/test/Shibuya/Runner/SupervisedSpec.hs +++ b/test/Shibuya/Runner/SupervisedSpec.hs @@ -8,6 +8,8 @@ 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) @@ -29,7 +31,7 @@ runApp, stopAppGracefully, ) -import Shibuya.Batch (BatchConfig (..), ackAllOk, defaultBatchConfig) +import Shibuya.Batch (BatchConfig (..), BatchKey (..), ackAll, ackAllOk, defaultBatchConfig) import Shibuya.Batch.TestHarness (finalizedExactlyOnce) import Shibuya.Core.Ack ( AckDecision (..), @@ -40,7 +42,7 @@ mkDeadLetterCode, ) import Shibuya.Core.AckHandle (AckHandle (..)) -import Shibuya.Core.Ingested (Ingested, Message (..), mkIngested) +import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), Message (..), mkIngested) import Shibuya.Core.Metrics ( InFlightInfo (..), ProcessorId (..), @@ -53,8 +55,10 @@ import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), mkEnvelope) import Shibuya.Handler (Handler) import Shibuya.Internal.Runner.Master - ( getAllMetrics, + ( ProcessorLifecycle (..), + getAllMetrics, getAllMetricsIO, + getLifecycleSnapshot, getProcessorMetricsIO, startMaster, stopMaster, @@ -64,6 +68,7 @@ getMetrics, isDone, runSupervised, + runSupervisedBatchWith, runSupervisedWith, runWithMetrics, runWithMetricsBatch, @@ -78,6 +83,166 @@ 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 diff --git a/test/Shibuya/Telemetry/SemanticSpec.hs b/test/Shibuya/Telemetry/SemanticSpec.hs --- a/test/Shibuya/Telemetry/SemanticSpec.hs +++ b/test/Shibuya/Telemetry/SemanticSpec.hs @@ -50,7 +50,7 @@ mkDeadLetterCode, ) import Shibuya.Core.AckHandle (AckHandle (..)) -import Shibuya.Core.Ingested (mkIngested) +import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), mkIngested) import Shibuya.Core.Metrics (ProcessorId (..)) import Shibuya.Core.Types (Envelope (..), MessageId (..), mkEnvelope) import Shibuya.Internal.Runner.Supervised (runWithMetrics) @@ -63,6 +63,31 @@ it "keeps the application dead-letter reason code wire key stable" $ do attrShibuyaDeadLetterReasonCode `shouldBe` "shibuya.dead_letter.reason.code" + + it "records superseded without starting a handler" $ do + (processor, spansRef) <- inMemoryListExporter + provider <- createTracerProvider [processor] emptyTracerProviderOptions + let tracer = mkTestTracer provider + runEff $ runTracing tracer $ do + let item = + (mkIngested (mkEnvelope (MessageId "superseded") ("hello" :: Text)) (AckHandle (\_ -> error "finalizer must not run"))) + { deliveryStatus = pure DeliverySuperseded + } + _ <- runWithMetrics 4 (ProcessorId "skip-proc") (listAdapter [item]) (\_ -> error "handler must not run") + pure () + _ <- shutdownTracerProvider provider (Just 5_000_000) + spans <- filter ((== Consumer) . spanKind) <$> readIORef spansRef + case spans of + [processingSpan] -> do + hot <- readIORef (spanHot processingSpan) + hotName hot `shouldBe` "skip-proc process" + let events = toList (appendOnlyBoundedCollectionValues (hotEvents hot)) + map eventName events `shouldBe` ["shibuya.ack.decision"] + hotStatus hot `shouldBe` Ok + case events of + [event] -> getAttributeMap (eventAttributes event) `shouldHaveTextAttribute` ("shibuya.ack.decision", "superseded") + _ -> expectationFailure "expected acknowledgement event" + _ -> expectationFailure ("expected processing span, got " <> show (length spans)) it "emits a process span with conventions-aligned attributes and events" $ do (processor, spansRef) <- inMemoryListExporter