shibuya-core 0.11.0.0 → 0.12.0.0

raw patch · 10 files changed

+513/−123 lines, 10 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +33/−0
• shibuya-core.cabal +1/−1
• src/Shibuya.hs +2/−1
• src/Shibuya/Core/AckHandle.hs +2/−1
• src/Shibuya/Core/Ingested.hs +20/−2
• src/Shibuya/Internal/Runner/BatchProcessor.hs +38/−3
• src/Shibuya/Internal/Runner/Supervised.hs +108/−99
• test/Shibuya/Runner/BatchProcessorSpec.hs +115/−12
• test/Shibuya/Runner/SupervisedSpec.hs +168/−3
• test/Shibuya/Telemetry/SemanticSpec.hs +26/−1

@@ -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

@@ -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).

@@ -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 (..),

@@ -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 (..),   )

@@ -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     }

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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