shibuya-core 0.10.0.1 → 0.11.0.0

raw patch · 24 files changed

+2308/−292 lines, 24 filesdep ~aesonPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: aeson

API changes (from Hackage documentation)

Files

• CHANGELOG.md +59/−2
• shibuya-core.cabal +7/−2
• src/Shibuya.hs +26/−0
• src/Shibuya/App.hs +87/−16
• src/Shibuya/Core/Error.hs +4/−0
• src/Shibuya/Core/Metrics.hs +302/−7
• src/Shibuya/Core/Types.hs +24/−1
• src/Shibuya/Internal/App.hs +4/−1
• src/Shibuya/Internal/Runner/BatchProcessor.hs +142/−90
• src/Shibuya/Internal/Runner/Ingester.hs +95/−15
• src/Shibuya/Internal/Runner/Master.hs +136/−16
• src/Shibuya/Internal/Runner/Pause.hs +54/−0
• src/Shibuya/Internal/Runner/Supervised.hs +224/−117
• src/Shibuya/Telemetry/Effect.hs +16/−6
• src/Shibuya/Telemetry/Propagation.hs +12/−7
• test/Main.hs +6/−0
• test/Shibuya/App/BatchSpec.hs +38/−2
• test/Shibuya/Core/MetricsSpec.hs +187/−0
• test/Shibuya/Runner/PauseSpec.hs +212/−0
• test/Shibuya/Runner/SupervisedSpec.hs +201/−1
• test/Shibuya/Telemetry/EffectSpec.hs +23/−0
• test/Shibuya/Telemetry/LifecycleSpec.hs +417/−0
• test/Shibuya/Telemetry/PropagationSpec.hs +28/−6
• test/Shibuya/Telemetry/SemanticSpec.hs +4/−3

@@ -1,5 +1,62 @@ # Changelog +## 0.11.0.0 — 2026-10-25++### Breaking Changes++- `AppConfig` gains `metricsDetail`; `ProcessorMetrics` gains optional `latency`,+  `oldestInFlight`, and `progress` fields. `MetricsHandle` gains detail state.+  Direct record construction must supply the new fields.+- `ProcessorState` gains `Paused`; `ConfigError` gains+  `InvalidMaxTrackedPartitions`. Exhaustive matches must handle these constructors.+- The exposed `SupervisedProcessor` and `AppHandle` records gain `pause` and+  `shutdownTracer`, respectively; direct construction must supply them.++### New Features++- Add processor intake pause/resume APIs with atomic admission control and refusal+  during draining. Already admitted polls and inbox work may finish while paused.+- Add opt-in detailed metrics with bounded synchronized in-flight slots, latency+  histograms, oldest in-flight inspection, and acknowledged cursor progress bounded+  per partition. Basic accounting remains the default.+- Add lossless tagged `Cursor` JSON and `MessageId` JSON instances, detail-aware+  handle and runner constructors, and read-only slot-miss diagnostics.+- Trace source-stream lifetimes and once-only adapter shutdowns as Internal spans+  carrying `shibuya.processor.id`, including exception and cancellation outcomes.+  Capture the startup tracer to preserve the existing stop API.++### Bug Fixes++- Reduce serial intake startup migration by waiting for input before processor+  setup and yielding the producer before it publishes messages. Preserve bounded+  admission, source ownership and cancellation using ordinary Haskell operations.+- Reduce intake overhead by checking queue emptiness and receiving in one STM+  transaction, preserving bounded admission and completion/halt wakeups.+- Preserve ingestion cancellation when tracing callbacks swallow asynchronous+  exceptions. Keep source effects on their owning thread and isolate SDK callbacks+  on an owned worker, preventing source entry after cancellation during span start.+- Remove per-message zip state from pause admission and avoid unreachable pause+  checks in blocking standalone helpers, preserving supervised intake control.+- Preserve shutdown deadlines and caller cancellation when tracing callbacks catch+  asynchronous exceptions; prevent cancelled shutdowns from starting more adapter+  cleanup.+- Omit invalid contexts from W3C propagation instead of emitting unusable headers.++### Other Changes++- The final matched comparison against 0.10.0.1 passed all 62 existing benchmarks+  at the 10% timing threshold with package-default RTS settings. The 100-message+  standalone counter measured 41.32 to 38.16 microseconds (-7.7%) after the startup+  handoff repair. These measurements were taken on a shared workstation.+- The same comparison observed higher allocation for serial delayed I/O+  (471,316 to 570,252 bytes per invocation, +21.0%) and ahead-5+  (2,132,353 to 2,171,847 bytes, +1.85%). Their timing changes were +2.9% and+  -2.7%, respectively; the allocation increases remain unexplained.+- Share the private disabled compatibility tracer across invocations and document+  its non-recording, zero-ID callback sentinel.+- Specialize basic workers without per-message detail closures; document bracketed+  application acquisition and cleanup in the OpenTelemetry guide.+ ## 0.10.0.1 — 2026-10-05  ### Other Changes@@ -310,7 +367,7 @@   `messaging.system` default — without opening a second span. The   per-record `Shibuya.Adapter.Kafka.Tracing.traced` wrapper that   previously bolted these on can now be removed; see the audit-  document `docs/plans/9-otel-audit-findings.md` (Finding F1, P0)+  document `docs/plans/9-audit-and-improve-opentelemetry-api.md` (Finding F1, P0)   for the full motivation. - `Shibuya.Telemetry.Propagation.currentTraceHeaders ::   (Tracing :> es, IOE :> es) => Eff es (Maybe TraceHeaders)` looks@@ -320,7 +377,7 @@   is disabled or there is no active span. The intended call sites   are adapter-side DLQ writes (so the failing-consumer's trace   links to the resulting DLQ message) and ad-hoc producer paths.-  See the audit document `docs/plans/9-otel-audit-findings.md`+  See the audit document `docs/plans/9-audit-and-improve-opentelemetry-api.md`   (Findings F3 and F5).  ## 0.4.0.0 — 2026-04-29

@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-core-version: 0.10.0.1+version: 0.11.0.0 synopsis: Supervised queue processing framework for Haskell description:   A supervised queue processing framework inspired by Broadway (Elixir).@@ -39,8 +39,10 @@     Shibuya.Internal.Runner.BatchProcessor     Shibuya.Internal.Runner.Batcher     Shibuya.Internal.Runner.Finalize+    Shibuya.Internal.Runner.Ingester     Shibuya.Internal.Runner.KeyedScheduler     Shibuya.Internal.Runner.Master+    Shibuya.Internal.Runner.Pause     Shibuya.Internal.Runner.Supervised     Shibuya.Policy     Shibuya.Stream@@ -51,7 +53,6 @@    other-modules:     Shibuya.Internal.Runner.Halt-    Shibuya.Internal.Runner.Ingester     Shibuya.Prelude    default-extensions:@@ -120,6 +121,7 @@     Shibuya.Batch.TestHarness     Shibuya.BatchSpec     Shibuya.Core.AckSpec+    Shibuya.Core.MetricsSpec     Shibuya.Core.RetrySpec     Shibuya.Core.TypesSpec     Shibuya.PolicySpec@@ -127,13 +129,16 @@     Shibuya.Runner.BatchProcessorSpec     Shibuya.Runner.BatcherSpec     Shibuya.Runner.PartitionOrderingSpec+    Shibuya.Runner.PauseSpec     Shibuya.Runner.SupervisedSpec     Shibuya.RunnerSpec     Shibuya.Telemetry.EffectSpec+    Shibuya.Telemetry.LifecycleSpec     Shibuya.Telemetry.PropagationSpec     Shibuya.Telemetry.SemanticSpec    build-depends:+    aeson,     QuickCheck ^>=2.15,     base ^>=4.21.0.0,     bytestring,

@@ -13,6 +13,11 @@     getAppMetrics,     getAppMaster,     waitApp,+    pauseProcessor,+    resumeProcessor,+    isProcessorPaused,+    ControlOutcome (..),+    PauseState (..),     stopApp,     stopAppGracefully,     ShutdownConfig (..),@@ -92,6 +97,14 @@     BatchStats (..),     InFlightInfo (..),     MetricsMap,+    MetricsDetail (..),+    DetailOptions (..),+    defaultDetailOptions,+    defaultMetricsDetail,+    LatencySummary (..),+    LatencyBucket (..),+    OldestInFlight (..),+    ProgressSnapshot (..),      -- * Tracing     Tracing,@@ -105,7 +118,9 @@   ( AppConfig (..),     AppError (..),     AppHandle,+    ControlOutcome (..),     Master,+    PauseState (..),     QueueProcessor (..),     ShutdownConfig (..),     SupervisionStrategy (..),@@ -113,8 +128,11 @@     defaultShutdownConfig,     getAppMaster,     getAppMetrics,+    isProcessorPaused,     mkBatchProcessor,     mkProcessor,+    pauseProcessor,+    resumeProcessor,     runApp,     stopApp,     stopAppGracefully,@@ -139,12 +157,20 @@ import Shibuya.Core.Lease (Lease (..)) import Shibuya.Core.Metrics   ( BatchStats (..),+    DetailOptions (..),     InFlightInfo (..),+    LatencyBucket (..),+    LatencySummary (..),+    MetricsDetail (..),     MetricsMap,+    OldestInFlight (..),     ProcessorId (..),     ProcessorMetrics (..),     ProcessorState (..),+    ProgressSnapshot (..),     StreamStats (..),+    defaultDetailOptions,+    defaultMetricsDetail,   ) import Shibuya.Core.Retry import Shibuya.Core.Types (Attempt (..), Cursor (..), Envelope (..), Headers, MessageId (..), TraceHeaders, mkEnvelope)

@@ -15,6 +15,11 @@     stopApp,     stopAppGracefully,     waitApp,+    pauseProcessor,+    resumeProcessor,+    isProcessorPaused,+    ControlOutcome (..),+    PauseState (..),      -- * Shutdown Configuration     ShutdownConfig (..),@@ -40,6 +45,7 @@   ) where +import Control.Concurrent.Async qualified as Async import Control.Concurrent.NQE.Supervisor qualified as NQE import Control.Concurrent.STM   ( STM,@@ -52,8 +58,11 @@     readTMVar,     readTVar,     registerDelay,+    retry,+    throwSTM,     writeTVar,   )+import Control.Exception qualified as IOException import Control.Monad (forM, forM_, void) import Data.Bifunctor (first) import Data.Foldable (traverse_)@@ -65,6 +74,9 @@ import Effectful.Exception qualified as Exception import GHC.Generics (Generic) import Numeric.Natural (Natural)+import OpenTelemetry.Context qualified as TraceContext+import OpenTelemetry.Context.ThreadLocal qualified as TraceContext+import OpenTelemetry.Trace.Core qualified as OTel import Shibuya.Adapter (Adapter (..)) import Shibuya.Batch import Shibuya.Core.Error (ConfigError (..), HandlerError (..), PolicyError (..), RuntimeError (..))@@ -73,6 +85,7 @@     ProcessorId (..),     ProcessorMetrics (..),   )+import Shibuya.Core.Metrics qualified as Metrics import Shibuya.Internal.App   ( AppHandle (..),     OwnershipFailure (..),@@ -82,24 +95,28 @@     mkProcessor,   ) import Shibuya.Internal.Runner.Master-  ( Master,+  ( ControlOutcome (..),+    Master,+    PauseState (..),+    drainAndResume,     getAllMetrics,     getAllMetricsIO,     getProcessorMetrics,     getProcessorMetricsIO,     markMasterDraining,     markMasterRunning,-    markProcessorDraining,     startMaster,     stopMaster,   )+import Shibuya.Internal.Runner.Master qualified as Master import Shibuya.Internal.Runner.Supervised   ( SupervisedProcessor (..),-    runSupervised,-    runSupervisedBatch,+    runSupervisedBatchWith,+    runSupervisedWith,   ) import Shibuya.Policy (Concurrency (..), OrderingPolicy (..), validatePolicy)-import Shibuya.Telemetry.Effect (Tracing)+import Shibuya.Telemetry.Effect (Tracing, getTracer, isTracingEnabled)+import Shibuya.Telemetry.Semantic (attrShibuyaProcessorId, internalSpanArgs) import UnliftIO (SomeException, displayException) import UnliftIO qualified as UIO @@ -174,13 +191,15 @@   { -- | How processor failures affect siblings.     strategy :: !SupervisionStrategy,     -- | Bounded-inbox capacity per processor (backpressure). Must be >= 1.-    inboxSize :: !Int+    inboxSize :: !Int,+    -- | Optional latency and in-flight detail; defaults to basic accounting.+    metricsDetail :: !Metrics.MetricsDetail   }   deriving stock (Eq, Show, Generic)  -- | 'IgnoreFailures' with an inbox of 100. defaultAppConfig :: AppConfig-defaultAppConfig = AppConfig {strategy = IgnoreFailures, inboxSize = 100}+defaultAppConfig = AppConfig {strategy = IgnoreFailures, inboxSize = 100, metricsDetail = Metrics.defaultMetricsDetail}  -- | Run queue processors concurrently under NQE supervision. --@@ -213,8 +232,10 @@           (startMaster nqeStrategy)           stopMaster           ( \master -> do-              processors <- spawnProcessors master (fromIntegral config.inboxSize) namedProcessors+              processors <- spawnProcessors config.metricsDetail master (fromIntegral config.inboxSize) namedProcessors               markMasterRunning master+              enabled <- isTracingEnabled+              shutdownTracer <- if enabled then Just <$> getTracer else pure Nothing               shutdownStarted <- liftIO $ newTVarIO False               shutdownResult <- liftIO newEmptyTMVarIO               pure@@ -222,7 +243,8 @@                   { master = master,                     processors = Map.fromList processors,                     shutdownStarted,-                    shutdownResult+                    shutdownResult,+                    shutdownTracer                   }           )       case startupResult of@@ -253,6 +275,9 @@ validateAppConfig :: AppConfig -> Either AppError () validateAppConfig config   | config.inboxSize < 1 = Left $ AppConfigInvalid $ InvalidInboxSize config.inboxSize+  | Metrics.MetricsDetailed options <- config.metricsDetail,+    options.maxTrackedPartitions < 0 =+      Left $ AppConfigInvalid $ InvalidMaxTrackedPartitions options.maxTrackedPartitions   | otherwise = Right ()  -- | Reject duplicate processor identifiers before the master or any adapter is@@ -293,19 +318,21 @@ -- | Spawn all processors under supervision. spawnProcessors ::   (IOE :> es, Tracing :> es) =>+  Metrics.MetricsDetail ->   Master ->   Natural ->   [(ProcessorId, QueueProcessor es)] ->   Eff es [(ProcessorId, (SupervisedProcessor, QueueProcessor es))]-spawnProcessors master inboxSize = traverse spawnOne+spawnProcessors detail master inboxSize = traverse spawnOne   where     spawnOne (procId, qp) = case qp of       QueueProcessor {adapter, handler, ordering, concurrency} -> do-        sp <- runSupervised master inboxSize procId ordering concurrency adapter handler+        sp <- runSupervisedWith detail master inboxSize procId ordering concurrency adapter handler         pure (procId, (sp, qp))       BatchingProcessor {adapter, batchHandler, batchConfig, concurrency} -> do         sp <--          runSupervisedBatch+          runSupervisedBatchWith+            detail             master             inboxSize             procId@@ -380,14 +407,13 @@      shutdownAndDrain = do       markMasterDraining appHandle.master-      forM_ (Map.keys appHandle.processors) $-        markProcessorDraining appHandle.master+      drainAndResume appHandle.master        -- Catch only synchronous adapter failures. External cancellation and the       -- total timeout must abort this phase so the master is force-stopped.       shutdownResults <--        forM (Map.elems appHandle.processors) $ \processor ->-          Exception.trySync (shutdownAdapter processor)+        forM (Map.toList appHandle.processors) $ \(procId, processor) ->+          Exception.trySync (traceShutdown procId (shutdownAdapter processor))        case [failure | Left failure <- shutdownResults] of         firstFailure : _ -> Exception.throwIO firstFailure@@ -395,6 +421,39 @@           let drainTimeoutMicros = nominalToMicros config.drainTimeout           liftIO $ waitForDrainWithTimeout drainTimeoutMicros (Map.elems appHandle.processors) +    traceShutdown procId action = case appHandle.shutdownTracer of+      Nothing -> action+      Just tracer -> withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->+        UIO.mask $ \restore -> do+          context <- TraceContext.getContext+          ready <- newEmptyTMVarIO+          finished <- newEmptyTMVarIO+          -- SDK hooks catch SomeException, including cancellation. Keep them+          -- on a separate owned thread; only the caller runs adapter effects.+          -- Even if span creation swallows cancellation, it can only observe+          -- the published outcome, never start adapter cleanup afterwards.+          let trace = UIO.bracket (TraceContext.attachContext context) TraceContext.detachContext $ \_ ->+                OTel.inSpan' tracer (procId.unProcessorId <> " shutdown") internalSpanArgs $ \sp -> do+                  OTel.addAttribute sp attrShibuyaProcessorId procId.unProcessorId+                  atomically $ putTMVar ready sp+                  result <- atomically $ readTMVar finished+                  either IOException.throwIO pure result+          UIO.withAsync trace $ \worker -> do+            result <- IOException.try @SomeException $ restore $ do+              sp <-+                atomically $+                  readTMVar ready `orElse` (Async.waitCatchSTM worker >>= either throwSTM (const retry))+              UIO.bracket (TraceContext.attachContext $ TraceContext.insertSpan sp context) TraceContext.detachContext $ \_ ->+                runInIO action+            atomically $ putTMVar finished result+            case result of+              -- Do not wait for export after the caller has been cancelled.+              -- withAsync still cancels and joins the tracing worker.+              Left failure | Exception.isAsyncException failure -> IOException.throwIO failure+              _ -> do+                traced <- restore $ UIO.waitCatch worker+                either IOException.throwIO (const $ either IOException.throwIO pure traced) result+     shutdownAdapter (_, qp) = case qp of       QueueProcessor {adapter} -> adapter.shutdown       BatchingProcessor {adapter} -> adapter.shutdown@@ -431,3 +490,15 @@   -- Block until all done TVars are True   forM_ (Map.elems appHandle.processors) $ \(sp, _) ->     readTVar sp.done >>= check++-- | Pause source admissions; already admitted polls and inbox work may drain.+pauseProcessor :: (IOE :> es) => AppHandle es -> ProcessorId -> Eff es ControlOutcome+pauseProcessor app = Master.pauseProcessor app.master++-- | Resume intake. Successful state is captured by the control transaction.+resumeProcessor :: (IOE :> es) => AppHandle es -> ProcessorId -> Eff es ControlOutcome+resumeProcessor app = Master.resumeProcessor app.master++-- | Nothing for unknown, terminal, draining or uncontrollable processors.+isProcessorPaused :: (IOE :> es) => AppHandle es -> ProcessorId -> Eff es (Maybe Bool)+isProcessorPaused app = Master.isProcessorPaused app.master

@@ -67,6 +67,8 @@     InvalidInboxSize !Int   | -- | Processor identifiers must be unique within one application.     DuplicateProcessorId !ProcessorId+  | -- | Detail partition capacity must be nonnegative.+    InvalidMaxTrackedPartitions !Int   deriving stock (Eq, Show, Generic)  -- | Convert configuration error to text for display.@@ -75,3 +77,5 @@   "inboxSize must be >= 1, got " <> Text.pack (show n) configErrorToText (DuplicateProcessorId (ProcessorId pid)) =   "processor IDs must be unique, duplicate: " <> pid+configErrorToText (InvalidMaxTrackedPartitions n) =+  "maxTrackedPartitions must be nonnegative, got " <> Text.pack (show n)

@@ -1,5 +1,10 @@ -- | Metrics and state tracking for processors. -- Provides introspection into what's happening in the system.+-- Detailed accounting is opt-in through 'MetricsDetailed': a bounded synchronized+-- slot per concurrent attempt measures handler plus finalization duration and+-- publishes oldest identity/start/age. Interrupted attempts are abandoned; basic+-- handles retain counters without detail clocks or ownership brackets. Detailed+-- progress records successful acknowledgements with a bounded partition snapshot. module Shibuya.Core.Metrics   ( -- * Processor State     ProcessorState (..),@@ -28,9 +33,32 @@     emptyProcessorMetrics,     HotCounters (..),     MetricsHandle (..),+    metricsDetailLevel,+    MetricsDetail (..),+    DetailOptions (..),+    defaultDetailOptions,+    defaultMetricsDetail,+    LatencyBucket (..),+    LatencySummary (..),+    OldestInFlight (..),+    ProgressSnapshot (..),+    recordAcknowledgedCursor,+    latencyBucketBoundsSeconds,+    latencyBucketLabels,+    InFlightSlots,+    claimInFlightSlot,+    releaseInFlightSlot,+    abandonInFlightSlot,+    slotMisses,+    newMetricsHandleWith,+    newMetricsHandleWithClockAndDetail,     newMetricsHandle,     newMetricsHandleWithClock,     sampleMetrics,+    markPaused,+    markResumed,+    markPausedSTM,+    markResumedSTM,     incrementReceived,     beginProcessing,     AckDecisionMetric (..),@@ -49,19 +77,26 @@   ) where -import Control.Concurrent.STM (TVar, atomically, modifyTVar', newTVarIO, readTVarIO)+import Control.Concurrent.MVar (MVar, modifyMVar, modifyMVar_, newMVar, readMVar)+import Control.Concurrent.STM (STM, TVar, atomically, modifyTVar', newTVarIO, readTVarIO)+import Control.Exception (mask_) import Control.Monad (void, when) import Data.Aeson (FromJSON (..), FromJSONKey (..), ToJSON (..), ToJSONKey (..), object, withObject, (.:), (.:?)) import Data.Aeson qualified as Aeson import Data.Atomics.Counter (AtomicCounter, incrCounter, readCounter) import Data.Atomics.Counter qualified as Counter-import Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef, writeIORef)+import Data.IORef (IORef, atomicModifyIORef', atomicWriteIORef, newIORef, readIORef, writeIORef)+import Data.IntMap.Strict qualified as IntMap+import Data.List (minimumBy) import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map import Data.Maybe (fromMaybe)+import Data.Ord (comparing) import Data.Text qualified as Text import Data.Time.Clock (addUTCTime) import Data.Word (Word64) import GHC.Clock (getMonotonicTimeNSec)+import Shibuya.Core.Types (Cursor, MessageId) import Shibuya.Prelude  -- | Processor identifier.@@ -101,6 +136,8 @@     Idle   | -- | Currently processing (in-flight info, sampled burst start, sampled last progress)     Processing !InFlightInfo !UTCTime !UTCTime+  | -- | Intake paused (live in-flight info, pause time, sampled last progress)+    Paused !InFlightInfo !UTCTime !UTCTime   | -- | Failed with error (error message, timestamp)     Failed !Text !UTCTime   | -- | Processor has been stopped@@ -117,6 +154,14 @@         "lastActivity" Aeson..= lastActivity,         "lastProgress" Aeson..= lastProgress       ]+  toJSON (Paused info pausedAt lastProgress) =+    object+      [ "status" Aeson..= ("paused" :: Text),+        "inFlight" Aeson..= info.inFlight,+        "maxConcurrency" Aeson..= info.maxConcurrency,+        "pausedAt" Aeson..= pausedAt,+        "lastProgress" Aeson..= lastProgress+      ]   toJSON (Failed err timestamp) =     object       [ "status" Aeson..= ("failed" :: Text),@@ -136,6 +181,7 @@         lastActivity <- v .: "lastActivity"         lastProgress <- v .:? "lastProgress"         pure $ Processing (InFlightInfo inFlightCount maxConc) lastActivity (fromMaybe lastActivity lastProgress)+      "paused" -> Paused <$> (InFlightInfo <$> v .: "inFlight" <*> v .: "maxConcurrency") <*> v .: "pausedAt" <*> v .: "lastProgress"       "failed" -> Failed <$> v .: "error" <*> v .: "timestamp"       "stopped" -> pure Stopped       other -> fail $ "Unknown processor state: " <> Text.unpack other@@ -182,6 +228,176 @@ emptyBatchStats :: BatchStats emptyBatchStats = BatchStats 0 0 0 0 0 0 +-- | Basic accounting is the default. Detailed timing includes handler execution+-- and all finalization retries, once per message attempt or emitted batch.+data MetricsDetail = MetricsBasic | MetricsDetailed !DetailOptions+  deriving stock (Eq, Show, Generic)++-- | Bounds for optional detail. Zero partitions reserves latest-only progress.+data DetailOptions = DetailOptions {maxTrackedPartitions :: !Int}+  deriving stock (Eq, Show, Generic)++defaultDetailOptions :: DetailOptions+defaultDetailOptions = DetailOptions 256++defaultMetricsDetail :: MetricsDetail+defaultMetricsDetail = MetricsBasic++data LatencyBucket = LatencyBucket {le :: !Double, count :: !Integer}+  deriving stock (Eq, Show, Generic)+  deriving anyclass (ToJSON, FromJSON)++data LatencySummary = LatencySummary+  {count :: !Integer, sumSeconds :: !Double, buckets :: ![LatencyBucket]}+  deriving stock (Eq, Show, Generic)+  deriving anyclass (ToJSON, FromJSON)++data OldestInFlight = OldestInFlight+  {messageId :: !MessageId, startedAt :: !UTCTime, ageSeconds :: !Double}+  deriving stock (Eq, Show, Generic)+  deriving anyclass (ToJSON, FromJSON)++-- | Last successful finalization recorded, with bounded partition detail.+-- Cursors are opaque; concurrent completion order need not be source order.+data ProgressSnapshot = ProgressSnapshot+  { latest :: !(Maybe Cursor),+    partitions :: !(Map Text Cursor),+    truncated :: !Bool+  }+  deriving stock (Eq, Show, Generic)++instance ToJSON ProgressSnapshot where+  toJSON = Aeson.genericToJSON omitNothingOptions++instance FromJSON ProgressSnapshot where+  parseJSON = Aeson.genericParseJSON omitNothingOptions++data ProgressTable = ProgressTable+  {snapshot :: !(TVar ProgressSnapshot), maxPartitions :: !Int}++-- | Record only after successful AckOk or AckDeadLetter finalization.+-- Missing cursors and basic handles do not enter STM.+recordAcknowledgedCursor :: MetricsHandle -> Maybe Text -> Maybe Cursor -> IO ()+recordAcknowledgedCursor handle partition cursor = case (handle.progressTable, cursor) of+  (Just table, Just position) -> atomically $ modifyTVar' table.snapshot $ \old ->+    let (partitions, truncated) = case partition of+          Nothing -> (old.partitions, old.truncated)+          Just key+            | Map.member key old.partitions || Map.size old.partitions < table.maxPartitions ->+                (Map.insert key position old.partitions, old.truncated)+            | otherwise -> (old.partitions, True)+     in ProgressSnapshot (Just position) partitions truncated+  _ -> pure ()++latencyBucketBoundsSeconds :: [Double]+latencyBucketBoundsSeconds = [0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60]++latencyBucketBoundsNs :: [Word64]+latencyBucketBoundsNs = [100_000, 250_000, 500_000, 1_000_000, 2_500_000, 5_000_000, 10_000_000, 25_000_000, 50_000_000, 100_000_000, 250_000_000, 500_000_000, 1_000_000_000, 2_500_000_000, 5_000_000_000, 10_000_000_000, 30_000_000_000, 60_000_000_000]++latencyBucketLabels :: [Text]+latencyBucketLabels = ["0.0001", "0.00025", "0.0005", "0.001", "0.0025", "0.005", "0.01", "0.025", "0.05", "0.1", "0.25", "0.5", "1", "2.5", "5", "10", "30", "60", "+Inf"]++-- Each slot is also a histogram shard. Its lock is never held across user code.+data DetailSlot = DetailSlot+  { start :: !(Maybe (MessageId, Word64)),+    observations :: !Integer,+    totalNs :: !Integer,+    bins :: !(IntMap.IntMap Integer)+  }++data InFlightSlots = InFlightSlots+  {cells :: ![MVar DetailSlot], misses :: !(IORef Integer)}++-- | Diagnostic exhaustion count. Supported runners must never miss a slot.+slotMisses :: MetricsHandle -> IO Integer+slotMisses handle = maybe (pure 0) (readIORef . (.misses)) handle.slots++-- | Claim a bounded slot, publishing identity and start under the same lock.+-- Callers must mask ownership transfer until they install exception cleanup.+claimInFlightSlot :: MetricsHandle -> MessageId -> IO Int+claimInFlightSlot handle messageId = case handle.slots of+  Nothing -> pure (-1)+  Just table -> mask_ $ do+    now <- handle.progressClock+    let claim _ [] = do+          atomicModifyIORef' table.misses (\n -> (n + 1, ()))+          pure (-1)+        claim index (cell : rest) = do+          owned <- modifyMVar cell $ \slot -> case slot.start of+            Nothing -> pure (slot {start = Just (messageId, now)}, True)+            Just _ -> pure (slot, False)+          if owned then pure index else claim (index + 1) rest+    claim 0 table.cells+{-# INLINE claimInFlightSlot #-}++-- | Complete one attempt, including finalization. Interrupted attempts instead+-- use 'abandonInFlightSlot' and contribute no completed observation.+releaseInFlightSlot :: MetricsHandle -> Int -> IO ()+releaseInFlightSlot handle index = withSlot handle index $ \cell -> mask_ $ do+  now <- handle.progressClock+  modifyMVar_ cell $ \slot -> case slot.start of+    Nothing -> pure slot+    Just (_, began) -> do+      let duration = if now >= began then now - began else 0+          bucketIndex = length (takeWhile (< duration) latencyBucketBoundsNs)+      pure $!+        DetailSlot+          Nothing+          (slot.observations + 1)+          (slot.totalNs + toInteger duration)+          (IntMap.insertWith (+) bucketIndex 1 slot.bins)++abandonInFlightSlot :: MetricsHandle -> Int -> IO ()+abandonInFlightSlot handle index = withSlot handle index $ \cell ->+  modifyMVar_ cell $ \slot -> pure $! slot {start = Nothing}++withSlot :: MetricsHandle -> Int -> (MVar DetailSlot -> IO ()) -> IO ()+withSlot handle index action+  | index < 0 = pure ()+  | otherwise = case handle.slots of+      Nothing -> pure ()+      Just table -> case drop index table.cells of+        cell : _ -> action cell+        [] -> pure ()++-- Copy each strict shard under its lock, format only after releasing it.+-- Sampling across slots is best effort; identity and timestamp always agree.+sampleDetail :: MetricsHandle -> IO (Maybe LatencySummary, Maybe OldestInFlight)+sampleDetail handle = case handle.slots of+  Nothing -> pure (Nothing, Nothing)+  Just table -> do+    copies <- mapM readMVar table.cells+    let observations = sum (map (.observations) copies)+        totalNs = sum (map (.totalNs) copies)+        bins = IntMap.unionsWith (+) (map (.bins) copies)+        cumulative = drop 1 (scanl (+) 0 [IntMap.findWithDefault 0 i bins | i <- [0 .. 17]])+        summary =+          if observations == 0+            then Nothing+            else+              Just+                ( LatencySummary+                    observations+                    (fromInteger totalNs / 1_000_000_000)+                    (zipWith LatencyBucket latencyBucketBoundsSeconds cumulative)+                )+        starts = [start | DetailSlot {start = Just start} <- copies]+    oldest <- case starts of+      [] -> pure Nothing+      _ -> do+        now <- handle.progressClock+        let (messageId, began) = minimumBy (comparing snd) starts+            elapsed = if now >= began then now - began else 0+        pure $+          Just+            ( OldestInFlight+                messageId+                (monotonicToUTC handle began)+                (fromIntegral elapsed / 1_000_000_000)+            )+    pure (summary, oldest)+ -- | Combined processor metrics. data ProcessorMetrics = ProcessorMetrics   { -- | Current state@@ -191,11 +407,22 @@     -- | Batch statistics     batch :: !BatchStats,     -- | When the processor started-    startedAt :: !UTCTime+    startedAt :: !UTCTime,+    latency :: !(Maybe LatencySummary),+    oldestInFlight :: !(Maybe OldestInFlight),+    progress :: !(Maybe ProgressSnapshot)   }   deriving stock (Eq, Show, Generic)-  deriving anyclass (ToJSON, FromJSON) +instance ToJSON ProcessorMetrics where+  toJSON = Aeson.genericToJSON omitNothingOptions++instance FromJSON ProcessorMetrics where+  parseJSON = Aeson.genericParseJSON omitNothingOptions++omitNothingOptions :: Aeson.Options+omitNothingOptions = Aeson.defaultOptions {Aeson.omitNothingFields = True}+ -- | Empty processor metrics. emptyProcessorMetrics :: UTCTime -> ProcessorMetrics emptyProcessorMetrics now =@@ -203,7 +430,10 @@     { state = Idle,       stats = emptyStreamStats,       batch = emptyBatchStats,-      startedAt = now+      startedAt = now,+      latency = Nothing,+      oldestInFlight = Nothing,+      progress = Nothing     }  -- | Map of processor IDs to their metrics.@@ -232,9 +462,19 @@     progressClock :: !(IO Word64),     lastObservedProgress :: !(IORef (Int, Int, Int)),     stateActiveRef :: !(IORef Bool),+    detail :: !MetricsDetail,+    slots :: !(Maybe InFlightSlots),+    progressTable :: !(Maybe ProgressTable),     cold :: !(TVar ProcessorMetrics)   } +-- | Read the configured level once, outside the processing loop. Keep this+-- accessor opaque: exposing the record match lets GHC carry the individual+-- counter references through each basic worker closure instead of one handle.+{-# OPAQUE metricsDetailLevel #-}+metricsDetailLevel :: MetricsHandle -> MetricsDetail+metricsDetailLevel handle = handle.detail+ newMetricsHandle :: UTCTime -> IO MetricsHandle newMetricsHandle = newMetricsHandleWithClock getMonotonicTimeNSec @@ -242,8 +482,31 @@ -- clock is intended for deterministic tests; production callers should use -- 'newMetricsHandle'. newMetricsHandleWithClock :: IO Word64 -> UTCTime -> IO MetricsHandle-newMetricsHandleWithClock clock now = do+newMetricsHandleWithClock clock = newMetricsHandleWithClockAndDetail clock MetricsBasic 1++newMetricsHandleWith :: MetricsDetail -> Int -> UTCTime -> IO MetricsHandle+newMetricsHandleWith = newMetricsHandleWithClockAndDetail getMonotonicTimeNSec++newMetricsHandleWithClockAndDetail :: IO Word64 -> MetricsDetail -> Int -> UTCTime -> IO MetricsHandle+newMetricsHandleWithClockAndDetail clock detail maxConcurrency now = do+  when (maxConcurrency < 1) $ ioError (userError "metrics concurrency must be positive")+  case detail of+    MetricsDetailed options+      | options.maxTrackedPartitions < 0 ->+          ioError (userError "maxTrackedPartitions must be nonnegative")+    _ -> pure ()+  slots <- case detail of+    MetricsBasic -> pure Nothing+    MetricsDetailed _ -> do+      cells <- mapM (const (newMVar (DetailSlot Nothing 0 0 IntMap.empty))) [1 .. maxConcurrency]+      misses <- newIORef 0+      pure (Just (InFlightSlots cells misses))   progressOriginNs <- clock+  progressTable <- case detail of+    MetricsBasic -> pure Nothing+    MetricsDetailed options -> do+      snapshot <- newTVarIO (ProgressSnapshot Nothing Map.empty False)+      pure (Just (ProgressTable snapshot options.maxTrackedPartitions))   received <- Counter.newCounter 0   processed <- Counter.newCounter 0   failed <- Counter.newCounter 0@@ -271,6 +534,9 @@         progressClock = clock,         lastObservedProgress = lastObservedProgress,         stateActiveRef = stateActiveRef,+        detail = detail,+        slots = slots,+        progressTable = progressTable,         cold = cold       } @@ -295,9 +561,38 @@       sampledState = case coldSnapshot.state of         Failed err timestamp -> Failed err timestamp         Stopped -> Stopped+        Paused _ since _ -> Paused (InFlightInfo inFlight maxConcurrency) since lastProgressAt         _ | inFlight > 0 -> Processing (InFlightInfo inFlight maxConcurrency) burstStartedAt lastProgressAt         _ -> Idle-  pure coldSnapshot {state = sampledState, stats = sampledStats}+  (latency, oldestInFlight) <- sampleDetail handle+  progress <- case handle.progressTable of+    Nothing -> pure Nothing+    Just table -> do+      snapshot <- readTVarIO table.snapshot+      pure $ case snapshot.latest of+        Nothing -> Nothing+        Just _ -> Just snapshot+  pure coldSnapshot {state = sampledState, stats = sampledStats, latency, oldestInFlight, progress}++-- | Publish pause intent without resetting the last-progress clock.+markPaused :: MetricsHandle -> UTCTime -> IO ()+markPaused handle = atomically . markPausedSTM handle++markPausedSTM :: MetricsHandle -> UTCTime -> STM ()+markPausedSTM handle since = modifyTVar' handle.cold $ \m -> case m.state of+  Failed {} -> m+  Stopped -> m+  Paused {} -> m+  _ -> m {state = Paused (InFlightInfo 0 0) since since}++-- | Clear only a pause; terminal and failure observations retain precedence.+markResumed :: MetricsHandle -> IO ()+markResumed = atomically . markResumedSTM++markResumedSTM :: MetricsHandle -> STM ()+markResumedSTM handle = modifyTVar' handle.cold $ \m -> case m.state of+  Paused {} -> m {state = Idle}+  _ -> m  incrementReceived :: MetricsHandle -> IO () incrementReceived handle =

@@ -24,17 +24,20 @@ where  import Control.DeepSeq (NFData (..))+import Data.Aeson (FromJSON (..), ToJSON (..), object, withObject, (.:), (.=)) import Data.ByteString (ByteString) import Data.HashMap.Strict (HashMap) import Data.String (IsString)+import Data.Text qualified as Text import OpenTelemetry.Attributes (Attribute (..)) import Shibuya.Prelude+import Text.Read (readMaybe)  -- | Stable identity for idempotency & observability. -- Every message has a unique identifier. newtype MessageId = MessageId {unMessageId :: Text}   deriving stock (Eq, Ord, Show, Generic)-  deriving newtype (IsString)+  deriving newtype (IsString, ToJSON, FromJSON)   deriving anyclass (NFData)  -- | Optional cursor / offset / global position.@@ -44,6 +47,26 @@   | CursorText !Text   deriving stock (Eq, Ord, Show, Generic)   deriving anyclass (NFData)++-- | Decimal text preserves integer identity in browser clients above 2^53.+instance ToJSON Cursor where+  toJSON (CursorInt n) = object ["kind" .= ("integer" :: Text), "value" .= Text.pack (show n)]+  toJSON (CursorText t) = object ["kind" .= ("text" :: Text), "value" .= t]++instance FromJSON Cursor where+  parseJSON = withObject "Cursor" $ \v -> do+    kind <- v .: "kind"+    value <- v .: "value"+    case (kind :: Text) of+      "text" -> pure (CursorText value)+      "integer" -> case readMaybe (Text.unpack value) :: Maybe Integer of+        Just n+          | Text.pack (show n) == value,+            n >= toInteger (minBound :: Int),+            n <= toInteger (maxBound :: Int) ->+              pure (CursorInt (fromInteger n))+        _ -> fail "Cursor integer must be canonical decimal text in the Int range"+      _ -> fail "Unknown cursor kind"  -- | Zero-indexed delivery attempt count. -- 0 means first delivery; 1 means first retry; and so on.

@@ -15,6 +15,7 @@ import Data.Map.Strict (Map) import Effectful (Eff, IOE, (:>)) import Effectful.Exception qualified as Exception+import OpenTelemetry.Trace.Core (Tracer) import Shibuya.Adapter (Adapter (..)) import Shibuya.Batch (BatchConfig, BatchHandler) import Shibuya.Core.Metrics (ProcessorId (..))@@ -103,5 +104,7 @@     -- | Coordinates repeated and concurrent graceful-stop calls. The first     -- caller performs shutdown; every caller observes the same terminal result.     shutdownStarted :: !(TVar Bool),-    shutdownResult :: !(TMVar (Either SomeException Bool))+    shutdownResult :: !(TMVar (Either SomeException Bool)),+    -- | Tracer selected at startup, retaining the IOE-only stop API.+    shutdownTracer :: !(Maybe Tracer)   }

@@ -33,6 +33,7 @@ where  import Control.Applicative ((<|>))+import Control.Monad (when) import Data.Foldable (for_, traverse_) import Data.HashMap.Strict qualified as HashMap import Data.List.NonEmpty (NonEmpty)@@ -41,6 +42,7 @@ import Data.Maybe (fromMaybe, isJust, listToMaybe) import Data.Text qualified as Text import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), liftIO, withEffToIO, (:>))+import Effectful.Exception qualified as Exception import GHC.IO (unsafeUnmask) import OpenTelemetry.Attributes (toAttribute) import OpenTelemetry.Trace.Core qualified as OTel@@ -69,6 +71,7 @@     recordBatchOutcomeMetrics,     sampleMetrics,   )+import Shibuya.Core.Metrics qualified as Metrics import Shibuya.Core.Types (Envelope (..)) import Shibuya.Internal.Runner.Finalize (finalizeWithRetry) import Shibuya.Internal.Runner.Halt@@ -130,7 +133,22 @@   BatchHandler es msg ->   (BatchInfo, NonEmpty (Ingested es msg)) ->   Eff es ()-processOneBatch metricsHandle procId maxConc stopSignal exitPublisher handler (info, batch) = do+processOneBatch metricsHandle = case Metrics.metricsDetailLevel metricsHandle of+  Metrics.MetricsBasic -> processOneBatchWithDetail False metricsHandle+  Metrics.MetricsDetailed _ -> processOneBatchWithDetail True metricsHandle++processOneBatchWithDetail ::+  (IOE :> es, Tracing :> es) =>+  Bool ->+  MetricsHandle ->+  ProcessorId ->+  Int ->+  ProcessorSignal ->+  ProcessorExitPublisher ->+  BatchHandler es msg ->+  (BatchInfo, NonEmpty (Ingested es msg)) ->+  Eff es ()+processOneBatchWithDetail 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).@@ -153,106 +171,122 @@               ]       addAttributes traceSpan frameworkAttrs -      -- Increment in-flight (a batch counts as one in-flight unit) and report it.-      currentInflight <- liftIO $ beginProcessing metricsHandle maxConc-      addAttribute traceSpan attrShibuyaInflightCount currentInflight-      addAttribute traceSpan attrShibuyaInflightMax maxConc+      let process currentInflight = do+            addAttribute traceSpan attrShibuyaInflightCount currentInflight+            addAttribute traceSpan attrShibuyaInflightMax maxConc -      addEvent traceSpan (mkEvent eventBatchStarted [])+            addEvent traceSpan (mkEvent eventBatchStarted []) -      terminal <- liftIO $ readProcessorExit stopSignal+            terminal <- liftIO $ readProcessorExit stopSignal -      -- Run the handler under exception isolation. On any exception, record it-      -- on the span and substitute the whole-batch retry default.-      (handlerResult, skippedAfterHalt) <--        case terminal of-          Just _ -> pure (Left (), True)-          Nothing -> do-            result <--              catchAny-                (Right <$> handler info (toMessage <$> batch))-                ( \ex -> do-                    recordException traceSpan ex-                    pure (Left ())-                )-            pure (result, False)-      let (resolvedAck, handlerThrew) = case handlerResult of-            Right a -> (a, False)-            Left () -> (ackAll (AckRetry (RetryDelay 0)), True)+            -- Run the handler under exception isolation. On any exception, record it+            -- on the span and substitute the whole-batch retry default.+            (handlerResult, skippedAfterHalt) <-+              case terminal of+                Just _ -> pure (Left (), True)+                Nothing -> do+                  result <-+                    catchAny+                      (Right <$> handler info (toMessage <$> batch))+                      ( \ex -> do+                          recordException traceSpan ex+                          pure (Left ())+                      )+                  pure (result, False)+            let (resolvedAck, handlerThrew) = case handlerResult of+                  Right a -> (a, False)+                  Left () -> (ackAll (AckRetry (RetryDelay 0)), True) -      -- RELIABLE FINALIZATION: iterate OUR OWN retained list, never the-      -- handler's output. For each retained message, choose its decision once-      -- via findWithDefault, then call the idempotent adapter finalizer with-      -- bounded retry. Do not let one adapter failure prevent attempts for the-      -- rest of the batch.-      results <--        mapM-          ( \ingested -> do-              let explicitDecision = Map.lookup ingested.envelope.messageId resolvedAck.decisions-                  d = fromMaybe resolvedAck.fallback explicitDecision-              finalResult <- finalizeWithRetry traceSpan ingested d-              pure (ingested.envelope.messageId, isJust explicitDecision, d, finalResult)-          )-          (NE.toList batch)-      let decisions = [d | (_, _, d, _) <- results]-          finalizeFailures = [(mid, ex) | (mid, _, _, Left ex) <- results]+            -- RELIABLE FINALIZATION: iterate OUR OWN retained list, never the+            -- handler's output. For each retained message, choose its decision once+            -- via findWithDefault, then call the idempotent adapter finalizer with+            -- bounded retry. Do not let one adapter failure prevent attempts for the+            -- rest of the batch.+            results <-+              mapM+                ( \ingested -> do+                    let explicitDecision = Map.lookup ingested.envelope.messageId resolvedAck.decisions+                        d = fromMaybe resolvedAck.fallback explicitDecision+                    finalResult <- finalizeWithRetry traceSpan ingested d+                    let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor+                    when detailed $ case (finalResult, d) of+                      (Right (), AckOk) -> recordCursor+                      (Right (), AckDeadLetter _) -> recordCursor+                      _ -> pure ()+                    pure (ingested.envelope.messageId, isJust explicitDecision, d, finalResult)+                )+                (NE.toList batch)+            let decisions = [d | (_, _, d, _) <- results]+                finalizeFailures = [(mid, ex) | (mid, _, _, Left ex) <- results] -      -- Compute halt and partial-failure signals from the resolved decisions.-      let finalizationFailure =-            case finalizeFailures of-              [] -> Nothing-              failed@((firstMessageId, _) : _) ->-                Just $-                  ProcessorFailed-                    ( "batch finalization failed for message ids: "-                        <> Text.intercalate ", " [tshow mid | (mid, _) <- failed]-                    )-                    (Just firstMessageId)-          firstHalt = listToMaybe [r | AckHalt r <- decisions]-          requestedExit = finalizationFailure <|> (ProcessorHalted <$> firstHalt)-          overrideFailures =-            [ ()-            | (_, explicitlyNamed, d, _) <- results,-              explicitlyNamed,-              isFailing d-            ]-          partialInc = not handlerThrew && not (null overrideFailures)+            -- Compute halt and partial-failure signals from the resolved decisions.+            let finalizationFailure =+                  case finalizeFailures of+                    [] -> Nothing+                    failed@((firstMessageId, _) : _) ->+                      Just $+                        ProcessorFailed+                          ( "batch finalization failed for message ids: "+                              <> Text.intercalate ", " [tshow mid | (mid, _) <- failed]+                          )+                          (Just firstMessageId)+                firstHalt = listToMaybe [r | AckHalt r <- decisions]+                requestedExit = finalizationFailure <|> (ProcessorHalted <$> firstHalt)+                overrideFailures =+                  [ ()+                  | (_, explicitlyNamed, d, _) <- results,+                    explicitlyNamed,+                    isFailing d+                  ]+                partialInc = not handlerThrew && not (null overrideFailures) -      -- Span status: error on halt or exception, otherwise Ok.-      addEvent traceSpan (mkEvent eventBatchCompleted [])-      case requestedExit of-        Just processorExit -> setStatus traceSpan (OTel.Error (processorExitText processorExit))-        Nothing ->-          if skippedAfterHalt-            then setStatus traceSpan (OTel.Error "batch skipped after halt")-            else-              if handlerThrew-                then setStatus traceSpan (OTel.Error "batch handler exception")-                else setStatus traceSpan OTel.Ok+            -- Span status: error on halt or exception, otherwise Ok.+            addEvent traceSpan (mkEvent eventBatchCompleted [])+            case requestedExit of+              Just processorExit -> setStatus traceSpan (OTel.Error (processorExitText processorExit))+              Nothing ->+                if skippedAfterHalt+                  then setStatus traceSpan (OTel.Error "batch skipped after halt")+                  else+                    if handlerThrew+                      then setStatus traceSpan (OTel.Error "batch handler exception")+                      else setStatus traceSpan OTel.Ok -      traverse_ (recordException traceSpan . snd) finalizeFailures+            traverse_ (recordException traceSpan . snd) finalizeFailures -      -- Record metrics: decrement in-flight, add per-message counter deltas,-      -- advance batch counters, set Failed state on halt or exhausted-      -- finalization retry.-      liftIO $-        recordBatchOutcomeMetrics-          metricsHandle-          (triggerMetric info.trigger)-          info.size-          handlerThrew-          partialInc-          (decisionMetric <$> decisions)-          (processorExitText <$> requestedExit)+            -- Record metrics: decrement in-flight, add per-message counter deltas,+            -- advance batch counters, set Failed state on halt or exhausted+            -- finalization retry.+            liftIO $+              recordBatchOutcomeMetrics+                metricsHandle+                (triggerMetric info.trigger)+                info.size+                handlerThrew+                partialInc+                (decisionMetric <$> decisions)+                (processorExitText <$> requestedExit) -      -- Halt: set the shared flag; do NOT throw (let the stream drain).-      for_ requestedExit $ \processorExit ->-        liftIO $ requestProcessorExit exitPublisher processorExit+            -- Halt: set the shared flag; do NOT throw (let the stream drain).+            for_ requestedExit $ \processorExit ->+              liftIO $ requestProcessorExit exitPublisher processorExit+      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 firstMsg.envelope.messageId+          restore (process currentInflight)+            `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)+          liftIO (Metrics.releaseInFlightSlot metricsHandle slot)+            `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)   where     isFailing :: AckDecision -> Bool     isFailing (AckDeadLetter _) = True     isFailing (AckRetry _) = True     isFailing _ = False+{-# INLINE processOneBatchWithDetail #-}  decisionMetric :: AckDecision -> AckDecisionMetric decisionMetric AckOk = CountProcessed@@ -296,7 +330,25 @@   ProcessorSignal ->   ProcessorExitPublisher ->   Eff es ()-processBatchesUntilDrained metricsHandle procId concurrency handler batchesStream stopSignal exitPublisher = do+processBatchesUntilDrained metricsHandle procId concurrency handler batchesStream stopSignal exitPublisher =+  -- Specialize the complete loop once; keep mode selection and ownership+  -- closures out of the basic batch hot path.+  case Metrics.metricsDetailLevel metricsHandle of+    Metrics.MetricsBasic -> processBatchesUntilDrainedWithDetail False metricsHandle procId concurrency handler batchesStream stopSignal exitPublisher+    Metrics.MetricsDetailed _ -> processBatchesUntilDrainedWithDetail True metricsHandle procId concurrency handler batchesStream stopSignal exitPublisher++processBatchesUntilDrainedWithDetail ::+  (IOE :> es, Tracing :> es) =>+  Bool ->+  MetricsHandle ->+  ProcessorId ->+  Concurrency ->+  BatchHandler es msg ->+  Stream IO (BatchInfo, NonEmpty (Ingested es msg)) ->+  ProcessorSignal ->+  ProcessorExitPublisher ->+  Eff es ()+processBatchesUntilDrainedWithDetail detailed metricsHandle procId concurrency handler batchesStream stopSignal exitPublisher = do   let maxConc = case concurrency of         Serial -> 1         Ahead n -> n@@ -306,8 +358,7 @@     -- The supervisor keeps framework coordination masked; only the owned batch     -- action is unmasked so user code and finalizers remain cancellable.     let runBatchAction batch =-          runInIO $-            processOneBatch metricsHandle procId maxConc stopSignal exitPublisher handler batch+          runInIO $ processOneBatchWithDetail detailed metricsHandle procId maxConc stopSignal exitPublisher handler batch         batchAction = unsafeUnmask . runBatchAction         pendingLimit = max 2 (2 * max 1 maxConc)     case concurrency of@@ -319,6 +370,7 @@         runKeyedScheduler (max 1 n) pendingLimit (Just . fstBatchKey) batchAction batchesStream       Async n ->         runKeyedScheduler (max 1 n) pendingLimit (Just . fstBatchKey) batchAction batchesStream+{-# INLINE processBatchesUntilDrainedWithDetail #-}  fstBatchKey :: (BatchInfo, NonEmpty (Ingested es msg)) -> BatchKey fstBatchKey (info, _) = info.batchKey

@@ -7,16 +7,29 @@ module Shibuya.Internal.Runner.Ingester   ( runIngester,     runIngesterWithMetrics,+    runIngesterWithMetricsUnpaused,   ) where +import Control.Concurrent.Async qualified as Async import Control.Concurrent.NQE.Process (Inbox, inboxToMailbox, send)-import Effectful (Eff, IOE, liftIO, (:>))+import Control.Concurrent.STM (atomically, newEmptyTMVarIO, orElse, putTMVar, readTMVar, retry, throwSTM)+import Control.Exception qualified as IOException+import Effectful (Eff, IOE, UnliftStrategy (..), liftIO, withEffToIO, (:>))+import Effectful.Dispatch.Static (unEff, unsafeEff)+import Effectful.Exception qualified as Exception+import OpenTelemetry.Context qualified as TraceContext+import OpenTelemetry.Context.ThreadLocal qualified as TraceContext+import OpenTelemetry.Trace.Core qualified as OTel import Shibuya.Core.Ingested (Ingested)-import Shibuya.Core.Metrics (MetricsHandle, incrementReceived)+import Shibuya.Core.Metrics (MetricsHandle, ProcessorId (..), incrementReceived)+import Shibuya.Internal.Runner.Pause (PauseHandle, waitWhilePaused)+import Shibuya.Telemetry.Effect (Tracing, getTracer, isTracingEnabled)+import Shibuya.Telemetry.Semantic (attrShibuyaProcessorId, ingestSpanName, internalSpanArgs) import Streamly.Data.Fold qualified as Fold import Streamly.Data.Stream (Stream) import Streamly.Data.Stream qualified as Stream+import UnliftIO qualified as UIO  -- | Run the ingester: reads from stream, sends to inbox. -- Sends each element to the inbox with backpressure - if inbox is full, blocks.@@ -40,23 +53,90 @@ -- | Run the ingester with metrics tracking. -- Increments the received count for each message sent to inbox. runIngesterWithMetrics ::-  (IOE :> es) =>+  (IOE :> es, Tracing :> es) =>+  ProcessorId ->   -- | Metrics handle (for updating received count)   MetricsHandle ->+  PauseHandle ->   -- | Source stream from adapter   Stream (Eff es) (Ingested es msg) ->   -- | Target inbox (bounded for backpressure)   Inbox (Ingested es msg) ->   Eff es ()-runIngesterWithMetrics metricsHandle source inbox = do-  let mailbox = inboxToMailbox inbox-  Stream.fold Fold.drain $-    Stream.mapM-      ( \msg -> do-          -- Increment received count-          liftIO $ incrementReceived metricsHandle-          -- Send to inbox (blocks if full - backpressure)-          liftIO $ send msg mailbox-          pure msg-      )-      source+runIngesterWithMetrics procId metricsHandle pauseHandle source inbox =+  traceIngestion procId $+    -- The fold and admission check use IO directly, avoiding an Eff action+    -- allocation for each step. Safe use of unsafeEff: IOE is required, and+    -- only this sequential fold calls unEff, on the owning ingester thread.+    -- The fold neither forks nor exports this environment or callback.+    unsafeEff $ \environment -> do+      let mailbox = inboxToMailbox inbox+      Stream.fold Fold.drain $+        Stream.mapM+          ( \msg -> do+              incrementReceived metricsHandle+              send msg mailbox+              pure msg+          )+          (Stream.morphInner (\step -> waitWhilePaused pauseHandle >> unEff step environment) source)++-- | Metrics and tracing for an intake that has no live pause capability.+runIngesterWithMetricsUnpaused ::+  (IOE :> es, Tracing :> es) =>+  ProcessorId ->+  MetricsHandle ->+  Stream (Eff es) (Ingested es msg) ->+  Inbox (Ingested es msg) ->+  Eff es ()+runIngesterWithMetricsUnpaused procId metricsHandle source inbox =+  traceIngestion procId ingest+  where+    ingest = do+      let mailbox = inboxToMailbox inbox+      Stream.fold Fold.drain $+        Stream.mapM+          ( \msg -> do+              liftIO $ incrementReceived metricsHandle+              liftIO $ send msg mailbox+              pure msg+          )+          source++traceIngestion ::+  (IOE :> es, Tracing :> es) =>+  ProcessorId ->+  Eff es () ->+  Eff es ()+traceIngestion procId action = do+  enabled <- isTracingEnabled+  if not enabled+    then action+    else do+      tracer <- getTracer+      -- Only the owner unlifts source effects. The SDK worker is entirely IO,+      -- so a sequential unlift preserves the source's thread and local state.+      withEffToIO SeqUnlift $ \runInIO -> UIO.mask $ \restore -> do+        context <- TraceContext.getContext+        ready <- newEmptyTMVarIO+        finished <- newEmptyTMVarIO+        -- SDK callbacks catch cancellation. Isolate them from source ownership:+        -- after cancellation the worker can only observe the published outcome,+        -- never enter the source even if onStart consumes its cancellation.+        let trace = UIO.bracket (TraceContext.attachContext context) TraceContext.detachContext $ \_ ->+              OTel.inSpan' tracer ingestSpanName internalSpanArgs $ \sp -> do+                OTel.addAttribute sp attrShibuyaProcessorId procId.unProcessorId+                atomically $ putTMVar ready sp+                result <- atomically $ readTMVar finished+                either IOException.throwIO pure result+        UIO.withAsync trace $ \worker -> do+          result <- IOException.try @IOException.SomeException $ restore $ do+            sp <- atomically $ readTMVar ready `orElse` (Async.waitCatchSTM worker >>= either throwSTM (const retry))+            UIO.bracket (TraceContext.attachContext $ TraceContext.insertSpan sp context) TraceContext.detachContext $ \_ ->+              runInIO action+          atomically $ putTMVar finished result+          case result of+            -- withAsync still cancels and joins the worker on this path.+            Left failure | Exception.isAsyncException failure -> IOException.throwIO failure+            _ -> do+              traced <- restore $ UIO.waitCatch worker+              either IOException.throwIO (const $ either IOException.throwIO pure traced) result

@@ -33,6 +33,17 @@      -- * Processor Management     registerProcessor,+    registerProcessorWithPause,+    ProcessorEntry (..),+    PauseState (..),+    ControlOutcome (..),+    pauseProcessorIO,+    resumeProcessorIO,+    isProcessorPausedIO,+    pauseProcessor,+    resumeProcessor,+    isProcessorPaused,+    drainAndResume,     unregisterProcessor,     markMasterRunning,     markMasterDraining,@@ -49,13 +60,17 @@ import Control.Concurrent.NQE.Supervisor (Strategy (..), Supervisor) import Control.Concurrent.NQE.Supervisor qualified as Supervisor import Control.Concurrent.STM-  ( TVar,+  ( STM,+    TVar,     atomically,     modifyTVar',     newTVarIO,     readTVar,+    retry,+    writeTVar,   ) import Control.Exception qualified as Exception+import Control.Monad (forM_, void, when) import Data.IORef (IORef, atomicModifyIORef', atomicWriteIORef, newIORef, readIORef) import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map@@ -67,7 +82,9 @@     ProcessorMetrics,     sampleMetrics,   )+import Shibuya.Core.Metrics qualified as Metrics import Shibuya.Core.Types (MessageId)+import Shibuya.Internal.Runner.Pause import Shibuya.Prelude import UnliftIO (cancel) @@ -91,7 +108,8 @@   deriving (Generic)  data MasterRegistry = MasterRegistry-  { liveMetrics :: !(Map ProcessorId MetricsHandle),+  { liveMetrics :: !(Map ProcessorId ProcessorEntry),+    draining :: !Bool,     lifecycles :: !LifecycleSnapshot   }   deriving (Generic)@@ -148,7 +166,7 @@   -- to 'acquireOwned', which then owns cleanup; there is no interruptible gap   -- that needs an extra exception frame here.   let sup = Process supAsync mailbox-  registryVar <- newTVarIO $ MasterRegistry Map.empty Map.empty+  registryVar <- newTVarIO $ MasterRegistry Map.empty False Map.empty   phaseRef <- newIORef MasterStarting   let propagate = case strategy of         KillAll -> True@@ -160,9 +178,11 @@ -- | Stop the master and all child processors. -- Cancels the supervisor, which cancels all children via NQE's stopAll. stopMaster :: (IOE :> es) => Master -> Eff es ()-stopMaster master = liftIO $ do-  atomicWriteIORef master.state.phaseRef MasterStopped-  cancel (getProcessAsync master.state.supervisor)+stopMaster master = do+  drainAndResume master+  liftIO $ do+    atomicWriteIORef master.state.phaseRef MasterStopped+    cancel (getProcessAsync master.state.supervisor)  -- | Get metrics for all processors. getAllMetrics :: (IOE :> es) => Master -> Eff es MetricsMap@@ -172,7 +192,7 @@ getAllMetricsIO :: Master -> IO MetricsMap getAllMetricsIO master = do   registry <- atomically $ readTVar master.state.registry-  traverse sampleMetrics registry.liveMetrics+  traverse (sampleMetrics . (.metrics)) registry.liveMetrics  -- | Get metrics for a specific processor. getProcessorMetrics :: (IOE :> es) => Master -> ProcessorId -> Eff es (Maybe ProcessorMetrics)@@ -182,7 +202,7 @@ getProcessorMetricsIO :: Master -> ProcessorId -> IO (Maybe ProcessorMetrics) getProcessorMetricsIO master pid = do   registry <- atomically $ readTVar master.state.registry-  traverse sampleMetrics (Map.lookup pid registry.liveMetrics)+  traverse (sampleMetrics . (.metrics)) (Map.lookup pid registry.liveMetrics)  -- | Read the master lifecycle phase. getMasterPhase :: (IOE :> es) => Master -> Eff es MasterPhase@@ -203,15 +223,22 @@ -- | Register a processor with the master. -- The processor should call this with its metrics handle. registerProcessor :: (IOE :> es) => Master -> ProcessorId -> MetricsHandle -> Eff es ()-registerProcessor master pid metricsHandle =-  liftIO $-    atomically $-      modifyTVar' master.state.registry $ \registry ->-        registry-          { liveMetrics = Map.insert pid metricsHandle registry.liveMetrics,-            lifecycles = Map.insert pid LifecycleRunning registry.lifecycles-          }+registerProcessor master pid metricsHandle = registerProcessorWithPause master pid metricsHandle Nothing +-- | Register real intake control; the legacy wrapper has no control capability.+registerProcessorWithPause :: (IOE :> es) => Master -> ProcessorId -> MetricsHandle -> Maybe PauseHandle -> Eff es ()+registerProcessorWithPause master pid metricsHandle pauseHandle = liftIO $ atomically $ do+  registry <- readTVar master.state.registry+  when registry.draining $ do+    forM_ pauseHandle $ \h -> void (requestResumeSTM h)+    Metrics.markResumedSTM metricsHandle+  writeTVar+    master.state.registry+    registry+      { liveMetrics = Map.insert pid (ProcessorEntry metricsHandle pauseHandle) registry.liveMetrics,+        lifecycles = Map.insert pid (if registry.draining then LifecycleDraining else LifecycleRunning) registry.lifecycles+      }+ -- | Unregister a processor from the master. unregisterProcessor :: (IOE :> es) => Master -> ProcessorId -> Eff es () unregisterProcessor master pid =@@ -270,3 +297,96 @@       registry         { lifecycles = Map.insert pid (LifecycleFailed failure messageId) registry.lifecycles         }++-- | Live observation and optional intake control have the same registry owner.+data ProcessorEntry = ProcessorEntry {metrics :: !MetricsHandle, pause :: !(Maybe PauseHandle)}++data PauseState = IntakeRunning | IntakePaused !UTCTime+  deriving stock (Eq, Show, Generic)++-- | Outcome and successful intake state captured by the same STM transaction.+data ControlOutcome+  = ControlApplied !PauseState+  | ControlAlreadyInState !PauseState+  | ControlNotFound+  | ControlNotControllable+  | ControlTerminal+  | ControlDraining+  deriving stock (Eq, Show, Generic)++controllable :: Master -> ProcessorId -> STM (Either ControlOutcome ProcessorEntry)+controllable master pid = do+  registry <- readTVar master.state.registry+  pure $ case Map.lookup pid registry.lifecycles of+    Nothing -> Left ControlNotFound+    Just LifecycleStopped -> Left ControlTerminal+    Just LifecycleFailed {} -> Left ControlTerminal+    Just LifecycleDraining -> Left ControlDraining+    _ | registry.draining -> Left ControlDraining+    _ -> case Map.lookup pid registry.liveMetrics of+      Just entry | Just _ <- entry.pause -> Right entry+      _ -> Left ControlNotControllable++-- | Pause intake atomically with lifecycle validation and visible intent.+pauseProcessorIO :: Master -> ProcessorId -> IO ControlOutcome+pauseProcessorIO master pid = do+  now <- getCurrentTime+  atomically $+    controllable master pid >>= \case+      Left outcome -> pure outcome+      Right entry -> case entry.pause of+        Nothing -> pure ControlNotControllable+        Just handle -> do+          changed <- requestPauseSTM handle now+          pausedAt <- readTVar handle.pauseState+          case pausedAt of+            Nothing -> retry -- cannot occur in this transaction+            Just since -> do+              Metrics.markPausedSTM entry.metrics since+              pure $ (if changed then ControlApplied else ControlAlreadyInState) (IntakePaused since)++-- | Resume only live non-draining processors; terminal state is retained.+resumeProcessorIO :: Master -> ProcessorId -> IO ControlOutcome+resumeProcessorIO master pid =+  atomically $+    controllable master pid >>= \case+      Left outcome -> pure outcome+      Right entry -> case entry.pause of+        Nothing -> pure ControlNotControllable+        Just handle -> do+          changed <- requestResumeSTM handle+          Metrics.markResumedSTM entry.metrics+          pure $ (if changed then ControlApplied else ControlAlreadyInState) IntakeRunning++isProcessorPausedIO :: Master -> ProcessorId -> IO (Maybe Bool)+isProcessorPausedIO master pid =+  atomically $+    controllable master pid >>= \case+      Left _ -> pure Nothing+      Right entry -> case entry.pause of+        Nothing -> pure Nothing+        Just handle -> Just . maybe False (const True) <$> readTVar handle.pauseState++pauseProcessor :: (IOE :> es) => Master -> ProcessorId -> Eff es ControlOutcome+pauseProcessor master = liftIO . pauseProcessorIO master++resumeProcessor :: (IOE :> es) => Master -> ProcessorId -> Eff es ControlOutcome+resumeProcessor master = liftIO . resumeProcessorIO master++isProcessorPaused :: (IOE :> es) => Master -> ProcessorId -> Eff es (Maybe Bool)+isProcessorPaused master = liftIO . isProcessorPausedIO master++-- | Shutdown owns admission from this transaction onward. No later control can+-- close a gate again while adapters are shutting down.+drainAndResume :: (IOE :> es) => Master -> Eff es ()+drainAndResume master = liftIO $ atomically $ do+  registry <- readTVar master.state.registry+  forM_ (Map.elems registry.liveMetrics) $ \entry -> do+    forM_ entry.pause $ \handle -> void (requestResumeSTM handle)+    Metrics.markResumedSTM entry.metrics+  writeTVar+    master.state.registry+    registry+      { draining = True,+        lifecycles = Map.map (\case LifecycleRunning -> LifecycleDraining; other -> other) registry.lifecycles+      }

@@ -0,0 +1,54 @@+-- | __Internal module.__ Exposed for the test suite and benchmarks only.+-- No PVP guarantees: anything here may change or disappear in any release.+-- Application authors should import "Shibuya" instead.+module Shibuya.Internal.Runner.Pause+  ( PauseHandle (..),+    newPauseHandle,+    requestPause,+    requestResume,+    requestPauseSTM,+    requestResumeSTM,+    isPaused,+    waitWhilePaused,+  )+where++import Control.Concurrent.STM+import Data.Maybe (isJust)+import Data.Time.Clock (UTCTime, getCurrentTime)++-- | One cell publishes intake intent and the original pause timestamp together.+newtype PauseHandle = PauseHandle {pauseState :: TVar (Maybe UTCTime)}++newPauseHandle :: IO PauseHandle+newPauseHandle = PauseHandle <$> newTVarIO Nothing++requestPause :: PauseHandle -> IO Bool+requestPause handle = getCurrentTime >>= atomically . requestPauseSTM handle++requestPauseSTM :: PauseHandle -> UTCTime -> STM Bool+requestPauseSTM handle now =+  readTVar handle.pauseState >>= \case+    Just _ -> pure False+    Nothing -> writeTVar handle.pauseState (Just now) >> pure True++requestResume :: PauseHandle -> IO Bool+requestResume = atomically . requestResumeSTM++requestResumeSTM :: PauseHandle -> STM Bool+requestResumeSTM handle = do+  previous <- readTVar handle.pauseState+  writeTVar handle.pauseState Nothing+  pure (isJust previous)++isPaused :: PauseHandle -> IO Bool+isPaused handle = isJust <$> readTVarIO handle.pauseState++-- | Admission check only. An already admitted poll cannot be revoked.+-- The running path reads once; a paused wait is interruptible by cancellation.+waitWhilePaused :: PauseHandle -> IO ()+waitWhilePaused handle = do+  paused <- isPaused handle+  if paused+    then atomically $ readTVar handle.pauseState >>= maybe (pure ()) (const retry)+    else pure ()

@@ -13,6 +13,8 @@   ( -- * Running with Supervision     runSupervised,     runSupervisedBatch,+    runSupervisedWith,+    runSupervisedBatchWith,      -- * Standalone (without Master)     runWithMetrics,@@ -28,6 +30,7 @@   ) where +import Control.Concurrent (yield) import Control.Concurrent.NQE.Process (Inbox, mailboxEmptySTM, newBoundedInbox, receiveSTM) import Control.Concurrent.NQE.Supervisor (addChild) import Control.Concurrent.STM@@ -35,7 +38,6 @@     atomically,     modifyTVar',     newTVarIO,-    orElse,     readTVar,     readTVarIO,     retry,@@ -94,16 +96,17 @@     requestProcessorExit,     throwProcessorExit,   )-import Shibuya.Internal.Runner.Ingester (runIngesterWithMetrics)+import Shibuya.Internal.Runner.Ingester (runIngesterWithMetrics, runIngesterWithMetricsUnpaused) import Shibuya.Internal.Runner.KeyedScheduler (runKeyedScheduler) import Shibuya.Internal.Runner.Master   ( Master (..),     MasterState (..),     markProcessorFailedIO,     markProcessorStoppedIO,-    registerProcessor,+    registerProcessorWithPause,     unregisterProcessor,   )+import Shibuya.Internal.Runner.Pause (PauseHandle, newPauseHandle) import Shibuya.Policy (Concurrency (..), OrderingPolicy (..)) import Shibuya.Prelude import Shibuya.Telemetry.Effect@@ -150,7 +153,8 @@     -- | Whether processing is complete     done :: !(TVar Bool),     -- | The async handle if running under supervision-    child :: !(Maybe (Async ()))+    child :: !(Maybe (Async ())),+    pause :: !(Maybe PauseHandle)   }  -- | Get current metrics for the processor.@@ -190,15 +194,35 @@   -- | Message handler   Handler es msg ->   Eff es SupervisedProcessor-runSupervised master inboxSize procId ordering concurrency adapter handler = Exception.mask_ $ do+runSupervised = runSupervisedWith Metrics.defaultMetricsDetail++runSupervisedWith ::+  (IOE :> es, Tracing :> es) =>+  Metrics.MetricsDetail ->+  Master ->+  -- | Inbox size (for backpressure)+  Natural ->+  -- | Processor identifier+  ProcessorId ->+  -- | OrderingPolicy policy+  OrderingPolicy ->+  -- | Concurrency mode+  Concurrency ->+  -- | Queue adapter+  Adapter es msg ->+  -- | Message handler+  Handler es msg ->+  Eff es SupervisedProcessor+runSupervisedWith detail master inboxSize procId ordering concurrency adapter handler = Exception.mask_ $ do   now <- liftIO getCurrentTime    -- Initialize state-  metricsHandle <- liftIO $ newMetricsHandle now+  metricsHandle <- liftIO $ Metrics.newMetricsHandleWith detail (concurrencyBound concurrency) now+  pauseHandle <- liftIO newPauseHandle   doneVar <- liftIO $ newTVarIO False    -- Register with Master-  registerProcessor master procId metricsHandle+  registerProcessorWithPause master procId metricsHandle (Just pauseHandle)    -- Add as supervised child using NQE's Supervisor   -- ConcUnlift Persistent allows the runInIO function to be used in the async child@@ -209,7 +233,7 @@   supervisedChild <- withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->     let processorAction =           runInIO $-            runIngesterAndProcessor metricsHandle procId inboxSize ordering concurrency adapter handler+            runIngesterAndProcessor metricsHandle (Just pauseHandle) procId inboxSize ordering concurrency adapter handler         unregisterAction = runInIO $ unregisterProcessor master procId      in addChild master.state.supervisor $           superviseProcessorLifecycleIO master procId processorAction@@ -226,6 +250,7 @@       { metrics = metricsHandle,         processorId = procId,         done = doneVar,+        pause = Just pauseHandle,         child = Just supervisedChild       } @@ -234,6 +259,7 @@ -- This blocks until the stream is exhausted and all messages are processed -- using serial processing with the same concurrent ingester/drainer shape as -- the supervised runner.+-- No live control handle escapes this blocking call, so intake is unpaused. runWithMetrics ::   (IOE :> es, Tracing :> es) =>   -- | Inbox size (for backpressure)@@ -252,7 +278,7 @@   metricsHandle <- liftIO $ newMetricsHandle now   doneVar <- liftIO $ newTVarIO False -  runIngesterAndProcessor metricsHandle procId inboxSize Unordered Serial adapter handler+  runIngesterAndProcessor metricsHandle Nothing procId inboxSize Unordered Serial adapter handler     `finally` liftIO (atomically (writeTVar doneVar True))    pure@@ -260,6 +286,7 @@       { metrics = metricsHandle,         processorId = procId,         done = doneVar,+        pause = Nothing,         child = Nothing       } @@ -269,6 +296,7 @@ runIngesterAndProcessor ::   (IOE :> es, Tracing :> es) =>   MetricsHandle ->+  Maybe PauseHandle ->   ProcessorId ->   Natural ->   OrderingPolicy ->@@ -276,7 +304,7 @@   Adapter es msg ->   Handler es msg ->   Eff es ()-runIngesterAndProcessor metricsHandle procId inboxSize ordering concurrency adapter handler = do+runIngesterAndProcessor metricsHandle pauseHandle procId inboxSize ordering concurrency adapter handler = do   -- Create bounded inbox (this is where inboxSize is used for backpressure)   inbox <- liftIO $ newBoundedInbox inboxSize @@ -289,13 +317,39 @@     -- Use finally to ensure streamDoneVar is always set, even if ingester fails     -- 'unsafeUnmask' is scoped inside the surrounding withAsync/finally owner;     -- cancellation therefore reaches adapter code without bypassing cleanup.-    let ingesterWithSignal =-          unsafeUnmask (runInIO (runIngesterWithMetrics metricsHandle adapter.source inbox))+    let ingest = case pauseHandle of+          Nothing -> runIngesterWithMetricsUnpaused procId metricsHandle adapter.source inbox+          Just pause -> runIngesterWithMetrics procId metricsHandle pause adapter.source inbox+        ingesterWithSignal =+          ( do+              -- Honor the startup scheduling hint before publishing messages.+              -- A later yield can move the newly awakened serial consumer to+              -- another capability while intake is still running.+              case concurrency of+                Serial -> yield+                _ -> pure ()+              unsafeUnmask (runInIO ingest)+          )             `finally` atomically (writeTVar streamDoneVar True)      UIO.withAsync ingesterWithSignal $ \ingesterAsync -> do+      -- Block before allocating processor state. Paired with the producer's+      -- startup yield, this avoids leaving both threads runnable at handoff.+      -- Observe readiness without removing a message or changing admission.+      case concurrency of+        Serial -> unsafeUnmask $ atomically $ do+          empty <- mailboxEmptySTM inbox+          when empty $ do+            done <- readTVar streamDoneVar+            when (not done) retry+        _ -> pure ()       -- Processor: process messages, exit when stream done and inbox empty-      runInIO $ processUntilDrained metricsHandle procId ordering concurrency handler inbox streamDoneVar+      -- Saturated literal arguments plus INLINE on the loop and worker erase+      -- the detail ownership action from the basic specialization. Selecting a+      -- partially applied worker retains a per-message closure.+      case Metrics.metricsDetailLevel metricsHandle of+        Metrics.MetricsBasic -> runInIO $ processUntilDrained False metricsHandle procId ordering concurrency handler inbox streamDoneVar+        Metrics.MetricsDetailed _ -> runInIO $ processUntilDrained True metricsHandle procId ordering concurrency handler inbox streamDoneVar       UIO.waitCatch ingesterAsync >>= \case         Left ingesterErr -> do           now <- getCurrentTime@@ -327,19 +381,40 @@   -- | Batch handler   BatchHandler es msg ->   Eff es SupervisedProcessor-runSupervisedBatch master inboxSize procId concurrency batchConfig adapter batchHandler = Exception.mask_ $ do+runSupervisedBatch = runSupervisedBatchWith Metrics.defaultMetricsDetail++runSupervisedBatchWith ::+  (IOE :> es, Tracing :> es) =>+  Metrics.MetricsDetail ->+  Master ->+  -- | Inbox size (for backpressure)+  Natural ->+  -- | Processor identifier+  ProcessorId ->+  -- | Concurrency mode (bounds how many BATCHES run at once)+  Concurrency ->+  -- | Batch configuration+  BatchConfig es msg ->+  -- | Queue adapter+  Adapter es msg ->+  -- | Batch handler+  BatchHandler es msg ->+  Eff es SupervisedProcessor+runSupervisedBatchWith detail master inboxSize procId concurrency batchConfig adapter batchHandler = Exception.mask_ $ do   now <- liftIO getCurrentTime -  metricsHandle <- liftIO $ newMetricsHandle now+  metricsHandle <- liftIO $ Metrics.newMetricsHandleWith detail (concurrencyBound concurrency) now+  pauseHandle <- liftIO newPauseHandle   doneVar <- liftIO $ newTVarIO False -  registerProcessor master procId metricsHandle+  registerProcessorWithPause master procId metricsHandle (Just pauseHandle)    supervisedChild <- withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->     let processorAction =           runInIO $             runIngesterAndProcessorBatch               metricsHandle+              (Just pauseHandle)               procId               inboxSize               concurrency@@ -361,6 +436,7 @@       { metrics = metricsHandle,         processorId = procId,         done = doneVar,+        pause = Just pauseHandle,         child = Just supervisedChild       } @@ -390,6 +466,7 @@ -- | Run a batching processor with metrics but without Master supervision. -- Blocks until the adapter stream is exhausted and every accumulated batch has -- been processed (including the end-of-input flush). Useful for tests.+-- No live control handle escapes this blocking call, so intake is unpaused. runWithMetricsBatch ::   (IOE :> es, Tracing :> es) =>   Natural ->@@ -407,6 +484,7 @@    runIngesterAndProcessorBatch     metricsHandle+    Nothing     procId     inboxSize     concurrency@@ -420,6 +498,7 @@       { metrics = metricsHandle,         processorId = procId,         done = doneVar,+        pause = Nothing,         child = Nothing       } @@ -444,6 +523,7 @@ runIngesterAndProcessorBatch ::   (IOE :> es, Tracing :> es) =>   MetricsHandle ->+  Maybe PauseHandle ->   ProcessorId ->   Natural ->   Concurrency ->@@ -451,15 +531,18 @@   Adapter es msg ->   BatchHandler es msg ->   Eff es ()-runIngesterAndProcessorBatch metricsHandle procId inboxSize concurrency batchConfig adapter batchHandler = do+runIngesterAndProcessorBatch metricsHandle pauseHandle procId inboxSize concurrency batchConfig adapter batchHandler = do   inbox <- liftIO $ newBoundedInbox inboxSize   streamDoneVar <- liftIO $ newTVarIO False   stopSignal <- liftIO newProcessorSignal   let exitPublisher = newProcessorExitPublisherWithWake stopSignal streamDoneVar    withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do-    let ingesterWithSignal =-          unsafeUnmask (runInIO (runIngesterWithMetrics metricsHandle adapter.source inbox))+    let ingest = case pauseHandle of+          Nothing -> runIngesterWithMetricsUnpaused procId metricsHandle adapter.source inbox+          Just pause -> runIngesterWithMetrics procId metricsHandle pause adapter.source inbox+        ingesterWithSignal =+          unsafeUnmask (runInIO ingest)             `finally` atomically (writeTVar streamDoneVar True)      UIO.withAsync ingesterWithSignal $ \ingesterAsync -> do@@ -511,22 +594,23 @@         Just _ -> pure Nothing         Nothing -> do           result <--            atomically $-              (Just <$> receiveSTM inbox)-                `orElse` do-                  -- Source completion and terminal publication share this-                  -- wake cell. A populated inbox therefore keeps the original-                  -- receive-only hot branch, while either terminal event wakes-                  -- an empty intake wait.+            atomically $ do+              -- Inspect and receive in the same transaction. Avoid an orElse+              -- child transaction for every message; only an empty inbox needs+              -- the source-completion/terminal wake cell in its read set.+              empty <- mailboxEmptySTM inbox+              if empty+                then do                   done <- readTVar streamDoneVar-                  empty <- mailboxEmptySTM inbox-                  if done && empty then pure Nothing else retry+                  if done then pure Nothing else retry+                else Just <$> receiveSTM inbox           pure $ fmap (,()) result  -- | Process messages from inbox until stream is done and inbox is empty. -- Supports Serial, Ahead, and Async concurrency modes. processUntilDrained ::   (IOE :> es, Tracing :> es) =>+  Bool ->   MetricsHandle ->   ProcessorId ->   OrderingPolicy ->@@ -535,7 +619,7 @@   Inbox (Ingested es msg) ->   TVar Bool ->   Eff es ()-processUntilDrained metricsHandle procId ordering concurrency handler inbox streamDoneVar = do+processUntilDrained detailed metricsHandle procId ordering concurrency handler inbox streamDoneVar = do   stopSignal <- liftIO newProcessorSignal   let exitPublisher = newProcessorExitPublisherWithWake stopSignal streamDoneVar @@ -555,8 +639,8 @@   withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do     let inboxStream = inboxToStream inbox streamDoneVar stopSignal         runProcessAction ingested =-          runInIO $-            processOne metricsHandle spanName constantFrameworkAttrs maxConc exitPublisher handler ingested+          runInIO $ processOne detailed metricsHandle spanName constantFrameworkAttrs maxConc exitPublisher handler ingested+         -- Restore normal interruptibility for handlers and finalizers while the         -- concurrent Streamly scheduler retains its inherited mask.         processAction = unsafeUnmask . runProcessAction@@ -598,6 +682,7 @@     -- After draining, check if we halted     maybeExit <- readProcessorExit stopSignal     traverse_ throwProcessorExit maybeExit+{-# INLINE processUntilDrained #-}  handlerStartedEvent :: OTel.NewEvent handlerStartedEvent = mkEvent eventHandlerStarted []@@ -607,6 +692,7 @@ -- Thread-safe for concurrent execution. processOne ::   (IOE :> es, Tracing :> es) =>+  Bool ->   MetricsHandle ->   Text ->   HashMap.HashMap Text Attribute ->@@ -615,7 +701,7 @@   Handler es msg ->   Ingested es msg ->   Eff es ()-processOne metricsHandle spanName constantFrameworkAttrs maxConc exitPublisher handler ingested = do+processOne detailed metricsHandle spanName constantFrameworkAttrs maxConc exitPublisher handler ingested = do   -- Extract parent context from message headers for distributed tracing   let parentCtx = ingested.envelope.traceContext >>= extractTraceContext @@ -637,100 +723,115 @@           mergedAttrs = HashMap.union ingested.envelope.attributes frameworkAttrs       addAttributes traceSpan mergedAttrs -      -- Increment in-flight and add inflight attributes.-      currentInflight <- liftIO $ beginProcessing metricsHandle maxConc+      let process currentInflight = do+            addAttribute traceSpan attrShibuyaInflightCount currentInflight+            addAttribute traceSpan attrShibuyaInflightMax maxConc -      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 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-              )+            when (isLeft handlerResult) $+              addEvent traceSpan $+                mkEvent+                  eventAckDecision+                  [(attrShibuyaAckDecision, OTel.toAttribute ("ack_retry" :: Text))] -      when (isLeft handlerResult) $-        addEvent traceSpan $-          mkEvent-            eventAckDecision-            [(attrShibuyaAckDecision, OTel.toAttribute ("ack_retry" :: Text))]+            finalizeResult <- finalizeWithRetry traceSpan ingested decision -      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 -      -- 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+                  -- 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 -            -- 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 $-        case finalizeResult of-          Left _ -> finishFinalizationFailure metricsHandle (finalizationFailureText msgIdText)-          Right () -> finishProcessing metricsHandle (metricForResult result)+            -- Decrement in-flight and update stats.+            liftIO $+              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 ()+            -- 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@@ -749,6 +850,7 @@     showHaltReason :: HaltReason -> Text     showHaltReason (HaltOrderedStream t) = "halt_ordered_stream: " <> t     showHaltReason (HaltFatal t) = "halt_fatal: " <> t+{-# INLINE processOne #-}  metricForResult :: Either HandlerError AckDecision -> Either Text AckDecisionMetric metricForResult (Left err) = Left (handlerErrorToText err)@@ -759,3 +861,8 @@   where     haltReasonText (HaltOrderedStream t) = t     haltReasonText (HaltFatal t) = t++concurrencyBound :: Concurrency -> Int+concurrencyBound Serial = 1+concurrencyBound (Ahead n) = n+concurrencyBound (Async n) = n

@@ -57,6 +57,7 @@ import OpenTelemetry.Trace.Core qualified as OTel import OpenTelemetry.Trace.Id qualified as OTel.Id import OpenTelemetry.Trace.TraceState qualified as OTel.TraceState+import System.IO.Unsafe (unsafePerformIO)  -------------------------------------------------------------------------------- -- Effect Definition@@ -95,16 +96,20 @@ runTracing t = evalStaticRep (TracingRep t True)  -- | Run with tracing disabled (near-zero overhead).--- All tracing operations become no-ops.+-- All tracing operations become no-ops. The compatibility tracer returned by+-- 'getTracer' is shared across invocations, independent of the global provider. runTracingNoop ::   (IOE :> es) =>   Eff (Tracing : es) a ->   Eff es a-runTracingNoop action = do-  -- Create a minimal no-op tracer-  noopProvider <- liftIO $ OTel.createTracerProvider [] OTel.emptyTracerProviderOptions-  let noopTracer = OTel.makeTracer noopProvider instrumentationLib OTel.tracerOptions-  evalStaticRep (TracingRep noopTracer False) action+runTracingNoop = evalStaticRep (TracingRep noopTracer False)++-- One private provider without processors or background resources. Sharing it+-- preserves the concrete getTracer API without repeated provider construction.+noopTracer :: OTel.Tracer+noopTracer = unsafePerformIO $ do+  provider <- OTel.createTracerProvider [] OTel.emptyTracerProviderOptions+  pure $ OTel.makeTracer provider instrumentationLib OTel.tracerOptions   where     instrumentationLib =       OTel.InstrumentationLibrary@@ -113,6 +118,7 @@           OTel.librarySchemaUrl = "",           OTel.libraryAttributes = Attr.emptyAttributes         }+{-# NOINLINE noopTracer #-}  -------------------------------------------------------------------------------- -- Span Operations@@ -142,6 +148,10 @@  -- | Create a span with access to the Span handle. -- Use this when you need to add attributes or events to the span.+-- When tracing is disabled, the callback receives a shared non-recording span+-- with invalid all-zero IDs. It represents absence, cannot be correlated, and+-- must not be used as a parent. Shibuya's injection helper returns no headers+-- for this sentinel. Use 'isTracingEnabled' to branch on presence explicitly. -- -- Example: --

@@ -17,7 +17,7 @@ import OpenTelemetry.Context qualified as Ctx import OpenTelemetry.Context.ThreadLocal qualified as Ctx import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C-import OpenTelemetry.Trace.Core (Span, SpanContext)+import OpenTelemetry.Trace.Core (Span, SpanContext, getSpanContext, isValid) import Shibuya.Core.Types (TraceHeaders) import Shibuya.Telemetry.Effect (Tracing, isTracingEnabled) @@ -40,6 +40,8 @@  -- | Inject current span's context into headers for propagation. -- Use this when producing messages that should carry trace context.+-- Returns an empty list for an invalid context, including the disabled+-- @withSpan'@ sentinel; zero IDs must never be propagated as W3C headers. -- -- Example: --@@ -49,11 +51,12 @@ -- @ injectTraceContext :: Span -> IO TraceHeaders injectTraceContext otelSpan = do-  (traceparent, tracestate) <- W3C.encodeSpanContext otelSpan-  pure-    [ ("traceparent", traceparent),-      ("tracestate", tracestate)-    ]+  context <- getSpanContext otelSpan+  if not (isValid context)+    then pure []+    else do+      (traceparent, tracestate) <- W3C.encodeSpanContext otelSpan+      pure [("traceparent", traceparent), ("tracestate", tracestate)]  -- | Look up the currently-active OTel span and encode its context as -- W3C trace headers, ready to attach to an outgoing message.@@ -87,4 +90,6 @@       ctx <- Ctx.getContext       case Ctx.lookupSpan ctx of         Nothing -> pure Nothing-        Just sp -> Just <$> injectTraceContext sp+        Just sp -> do+          headers <- injectTraceContext sp+          pure $ if null headers then Nothing else Just headers

@@ -7,6 +7,7 @@ import Shibuya.Batch.ReliabilitySpec qualified import Shibuya.BatchSpec qualified import Shibuya.Core.AckSpec qualified+import Shibuya.Core.MetricsSpec qualified import Shibuya.Core.RetrySpec qualified import Shibuya.Core.TypesSpec qualified import Shibuya.PolicySpec qualified@@ -14,15 +15,19 @@ import Shibuya.Runner.BatchProcessorSpec qualified import Shibuya.Runner.BatcherSpec qualified import Shibuya.Runner.PartitionOrderingSpec qualified+import Shibuya.Runner.PauseSpec qualified as Pause import Shibuya.Runner.SupervisedSpec qualified import Shibuya.RunnerSpec qualified import Shibuya.Telemetry.EffectSpec qualified+import Shibuya.Telemetry.LifecycleSpec qualified import Shibuya.Telemetry.PropagationSpec qualified import Shibuya.Telemetry.SemanticSpec qualified import Test.Hspec  main :: IO () main = hspec $ do+  Pause.spec+  Shibuya.Core.MetricsSpec.spec   Shibuya.App.BatchSpec.spec   Shibuya.App.LifecycleSpec.spec   Shibuya.Batch.ReliabilitySpec.spec@@ -37,6 +42,7 @@   Shibuya.Runner.BatchProcessorSpec.spec   Shibuya.Runner.PartitionOrderingSpec.spec   Shibuya.Runner.SupervisedSpec.spec+  Shibuya.Telemetry.LifecycleSpec.spec   Shibuya.Telemetry.EffectSpec.spec   Shibuya.Telemetry.PropagationSpec.spec   Shibuya.Telemetry.SemanticSpec.spec

@@ -1,18 +1,21 @@ module Shibuya.App.BatchSpec (spec) where -import Control.Concurrent (threadDelay)+import Control.Concurrent (newEmptyMVar, putMVar, takeMVar, threadDelay) import Control.Concurrent.STM (TVar, atomically, check, newTVarIO, readTVar, writeTVar) import Data.IORef (IORef, modifyIORef', newIORef, readIORef) import Data.List (nub)+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.Adapter (Adapter (..)) import Shibuya.Adapter.Mock (TrackingAck (..), newTrackingAck, trackingAckHandle) import Shibuya.App-  ( AppError (..),+  ( AppConfig (..),+    AppError (..),     QueueProcessor (..),     defaultAppConfig,+    getAppMetrics,     mkBatchProcessor,     runApp,     stopApp,@@ -28,6 +31,7 @@   ) import Shibuya.Core.Ingested (Ingested, mkIngested) import Shibuya.Core.Metrics (ProcessorId (..))+import Shibuya.Core.Metrics qualified as Metrics import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), mkEnvelope) import Shibuya.Policy (Concurrency (..), OrderingPolicy (..)) import Shibuya.Telemetry.Effect (runTracingNoop)@@ -37,6 +41,37 @@ spec :: Spec spec = describe "Shibuya.App.Batch" $ do   describe "runApp with a BatchingProcessor" $ do+    it "counts detailed duration per emitted batch through AppConfig" $ do+      started <- newEmptyMVar+      proceed <- newEmptyMVar+      runEff $ runTracingNoop $ do+        tracking <- newTrackingAck+        messages <- createTrackedMessages tracking 5+        let handler _ _ = liftIO (putMVar started () >> takeMVar proceed) >> pure ackAllOk+            config = defaultBatchConfig {batchSize = 2, batchTimeout = 60}+            processor = mkBatchProcessor (listAdapter' messages) handler config+        runApp+          defaultAppConfig {metricsDetail = Metrics.MetricsDetailed Metrics.defaultDetailOptions}+          [(ProcessorId "batch-detail", processor)]+          >>= \case+            Left err -> liftIO $ expectationFailure (show err)+            Right app -> do+              liftIO $ takeMVar started >> putMVar proceed () >> takeMVar started+              metrics <- getAppMetrics app+              let m = metrics Map.! ProcessorId "batch-detail"+              liftIO $ do+                fmap (.count) m.latency `shouldBe` Just 1+                m.batch.batchesEmitted `shouldBe` 1+                m.batch.batchedMessages `shouldBe` 2+                fmap (.latest) m.progress `shouldBe` Just (Just (CursorInt 2))+                fmap (.partitions) m.progress `shouldBe` Just (Map.fromList [("p-0", CursorInt 2), ("p-1", CursorInt 1)])+                putMVar proceed () >> takeMVar started+              metrics2 <- getAppMetrics app+              liftIO $ fmap (.count) (metrics2 Map.! ProcessorId "batch-detail").latency `shouldBe` Just 2+              liftIO $ putMVar proceed ()+              waitApp app+              stopApp app+     -- Milestone 1: a bad batch config is rejected by runApp with AppBatchConfigError.     it "rejects a batch config with size 0 (AppBatchConfigError)" $ do       result <- runEff $ runTracingNoop $ do@@ -136,6 +171,7 @@           env =             (mkEnvelope msgId ("message-" <> show i))               { cursor = Just (CursorInt i),+                partition = Just (if even i then "p-0" else "p-1"),                 enqueuedAt = Just testTime               }       pure $ mkIngested env (trackingAckHandle tracking msgId)

@@ -0,0 +1,187 @@+module Shibuya.Core.MetricsSpec (spec) where++import Control.Concurrent (forkIO, newEmptyMVar, putMVar, takeMVar)+import Control.Monad (forM_, replicateM, replicateM_)+import Data.Aeson (eitherDecode, encode)+import Data.IORef+import Data.Map.Strict qualified as Map+import Data.Maybe (fromJust)+import Data.Text qualified as Text+import Data.Time (UTCTime (..), fromGregorian)+import Data.Word (Word64)+import Shibuya.Core.Metrics+import Shibuya.Core.Types (Cursor (..), MessageId (..))+import Test.Hspec+import UnliftIO qualified as UIO++spec :: Spec+spec = describe "Detailed metrics" $ do+  it "accepts timestamp zero, bounds claims, and counts slot misses" $ do+    (_, h) <- handle 2+    sampleMetrics h >>= \m -> m.latency `shouldBe` Nothing+    a <- claimInFlightSlot h (MessageId "a")+    b <- claimInFlightSlot h (MessageId "b")+    a `shouldNotBe` b+    claimInFlightSlot h (MessageId "c") `shouldReturn` (-1)+    slotMisses h `shouldReturn` 1+    sampleMetrics h >>= \m -> fmap (.messageId) m.oldestInFlight `shouldBe` Just (MessageId "a")+    abandonInFlightSlot h a+    abandonInFlightSlot h b+    sampleMetrics h >>= \m -> m.oldestInFlight `shouldBe` Nothing++  it "records cumulative buckets and nanosecond sums on release and reuse" $ do+    (clock, h) <- handle 1+    replicateM_ 100 $ do+      start <- readIORef clock+      slot <- claimInFlightSlot h (MessageId "a")+      writeIORef clock (start + 3_000_000)+      releaseInFlightSlot h slot+    m <- sampleMetrics h+    let summary = fromJust m.latency+    summary.count `shouldBe` 100+    summary.sumSeconds `shouldBe` 0.3+    map (.count) summary.buckets `shouldBe` replicate 5 0 ++ replicate 13 100+    m.oldestInFlight `shouldBe` Nothing++  it "reports oldest identity, start and growing age together" $ do+    (clock, h) <- handle 2+    a <- claimInFlightSlot h (MessageId "a")+    writeIORef clock 1_000_000_000+    _ <- claimInFlightSlot h (MessageId "b")+    writeIORef clock 2_000_000_000+    sampleMetrics h >>= \m -> m.oldestInFlight `shouldBe` Just (OldestInFlight (MessageId "a") epoch 2)+    releaseInFlightSlot h a+    sampleMetrics h >>= \m -> fmap (.messageId) m.oldestInFlight `shouldBe` Just (MessageId "b")++  it "counts observations beyond the finite bounds without wrapping the sum" $ do+    (clock, h) <- handle 1+    slot <- claimInFlightSlot h (MessageId "long")+    writeIORef clock 70_000_000_000+    releaseInFlightSlot h slot+    m <- sampleMetrics h+    let summary = fromJust m.latency+    summary.count `shouldBe` 1+    summary.sumSeconds `shouldBe` 70+    map (.count) summary.buckets `shouldBe` replicate 18 0++  it "never reads the detail clock on the basic hooks" $ do+    calls <- newIORef (0 :: Int)+    h <- newMetricsHandleWithClock (modifyIORef' calls (+ 1) >> pure 0) epoch+    initialCalls <- readIORef calls+    claimInFlightSlot h (MessageId "a") `shouldReturn` (-1)+    releaseInFlightSlot h (-1)+    abandonInFlightSlot h (-1)+    m <- sampleMetrics h+    m.latency `shouldBe` Nothing+    m.oldestInFlight `shouldBe` Nothing+    readIORef calls `shouldReturn` initialCalls++  it "aggregates lifetime nanoseconds beyond a Word64 across occupied slots" $ do+    (clock, h) <- handle 2+    a <- claimInFlightSlot h (MessageId "a")+    b <- claimInFlightSlot h (MessageId "b")+    writeIORef clock maxBound+    releaseInFlightSlot h a+    releaseInFlightSlot h b+    m <- sampleMetrics h+    let summary = fromJust m.latency+    summary.count `shouldBe` 2+    summary.sumSeconds `shouldBe` (fromInteger (2 * toInteger (maxBound :: Word64)) / 1e9)+    map (.count) summary.buckets `shouldBe` replicate 18 0+    m.oldestInFlight `shouldBe` Nothing++  it "publishes complete slots and reuses bounded storage at 1/4/64/256 concurrency" $ do+    forM_ [1, 4, 64, 256] $ \capacity -> do+      (_, h) <- handle capacity+      replicateM_ 100 $ do+        gate <- newEmptyMVar+        results <- replicateM capacity newEmptyMVar+        mapM_ (\out -> forkIO (takeMVar gate >> claimInFlightSlot h (MessageId "same") >>= putMVar out)) results+        replicateM_ capacity (putMVar gate ())+        claimed <- mapM takeMVar results+        length (filter (>= 0) claimed) `shouldBe` capacity+        slotMisses h `shouldReturn` 0+        sampleMetrics h >>= \m -> fmap (.messageId) m.oldestInFlight `shouldBe` Just (MessageId "same")+        mapM_ (abandonInFlightSlot h) claimed+        sampleMetrics h >>= \m -> m.oldestInFlight `shouldBe` Nothing++  it "samples coherently at gated claim and release clock boundaries" $ replicateM_ 100 $ do+    calls <- newIORef (0 :: Int)+    claiming <- newEmptyMVar+    publish <- newEmptyMVar+    releasing <- newEmptyMVar+    complete <- newEmptyMVar+    let clock = do+          n <- atomicModifyIORef' calls (\n -> (n + 1, n))+          case n of+            0 -> pure 0+            1 -> putMVar claiming () >> takeMVar publish >> pure 0+            2 -> putMVar releasing () >> takeMVar complete >> pure 3_000_000+            _ -> pure 3_000_000+    h <- newMetricsHandleWithClockAndDetail clock (MetricsDetailed defaultDetailOptions) 1 epoch+    claim <- UIO.async $ claimInFlightSlot h (MessageId "gated")+    takeMVar claiming+    sampleMetrics h >>= \m -> m.oldestInFlight `shouldBe` Nothing+    putMVar publish ()+    slot <- UIO.wait claim+    release <- UIO.async $ releaseInFlightSlot h slot+    takeMVar releasing+    m <- sampleMetrics h+    fmap (.messageId) m.oldestInFlight `shouldBe` Just (MessageId "gated")+    m.latency `shouldBe` Nothing+    putMVar complete ()+    UIO.wait release+    final <- sampleMetrics h+    final.oldestInFlight `shouldBe` Nothing+    fmap (.count) final.latency `shouldBe` Just 1++  it "rejects invalid detail options and concurrency before reading the clock" $ do+    let noClock = error "clock acquired before validation"+    newMetricsHandleWithClockAndDetail noClock (MetricsDetailed (DetailOptions (-1))) 1 epoch `shouldThrow` anyIOException+    newMetricsHandleWithClockAndDetail noClock (MetricsDetailed defaultDetailOptions) 0 epoch `shouldThrow` anyIOException++  describe "acknowledged cursor progress" $ do+    it "omits missing cursors and ignores basic accounting" $ do+      basic <- newMetricsHandle epoch+      (_, detailed) <- handle 1+      recordAcknowledgedCursor basic (Just "a") (Just (CursorInt 7))+      recordAcknowledgedCursor detailed (Just "a") Nothing+      mapM_ (\h -> sampleMetrics h >>= \m -> m.progress `shouldBe` Nothing) [basic, detailed]++    it "records latest completion, replaces known keys and keeps truncation sticky" $ do+      h <- newMetricsHandleWith (MetricsDetailed (DetailOptions 2)) 1 epoch+      recordAcknowledgedCursor h Nothing (Just (CursorInt 7))+      sampleMetrics h >>= \m -> m.progress `shouldBe` Just (ProgressSnapshot (Just (CursorInt 7)) Map.empty False)+      forM_ [("a", 8), ("b", 9), ("c", 10), ("a", 3)] $ \(key, n) -> recordAcknowledgedCursor h (Just key) (Just (CursorInt n))+      sampleMetrics h >>= \m -> m.progress `shouldBe` Just (ProgressSnapshot (Just (CursorInt 3)) (Map.fromList [("a", CursorInt 3), ("b", CursorInt 9)]) True)+      recordAcknowledgedCursor h (Just "a") Nothing+      sampleMetrics h >>= \m -> fmap (.latest) m.progress `shouldBe` Just (Just (CursorInt 3))++    it "atomically caps concurrent partitions at capacity zero, one and two" $ do+      forM_ [0, 1, 2] $ \capacity -> do+        h <- newMetricsHandleWith (MetricsDetailed (DetailOptions capacity)) 1 epoch+        UIO.mapConcurrently_ (\n -> recordAcknowledgedCursor h (Just (Text.pack (show n))) (Just (CursorInt n))) [1 .. 64]+        m <- sampleMetrics h+        let snapshot = fromJust m.progress+        Map.size snapshot.partitions `shouldBe` capacity+        snapshot.truncated `shouldBe` True+        snapshot.latest `shouldSatisfy` maybe False (const True)++    it "round trips lossless tagged cursor identities and rejects noncanonical integers" $ do+      forM_ [CursorInt minBound, CursorInt maxBound, CursorInt 9007199254740993, CursorText "9007199254740993", CursorText ""] $ \cursor ->+        eitherDecode (encode cursor) `shouldBe` Right cursor+      forM_ ["01", "-0", "+1", "1.5", " 1", "9223372036854775808"] $ \value ->+        (eitherDecode ("{\"kind\":\"integer\",\"value\":\"" <> value <> "\"}") :: Either String Cursor) `shouldSatisfy` either (const True) (const False)+      (eitherDecode "{\"kind\":\"integer\",\"value\":1}" :: Either String Cursor) `shouldSatisfy` either (const True) (const False)+      (eitherDecode "{\"kind\":\"unknown\",\"value\":\"1\"}" :: Either String Cursor) `shouldSatisfy` either (const True) (const False)+      forM_ [ProgressSnapshot Nothing Map.empty False, ProgressSnapshot (Just (CursorText "a")) Map.empty True] $ \snapshot ->+        eitherDecode (encode snapshot) `shouldBe` Right snapshot++epoch :: UTCTime+epoch = UTCTime (fromGregorian 2026 1 1) 0++handle :: Int -> IO (IORef Word64, MetricsHandle)+handle n = do+  clock <- newIORef 0+  h <- newMetricsHandleWithClockAndDetail (readIORef clock) (MetricsDetailed defaultDetailOptions) n epoch+  pure (clock, h)

@@ -0,0 +1,212 @@+module Shibuya.Runner.PauseSpec (spec) where++import Control.Concurrent (myThreadId, yield)+import Control.Concurrent.NQE.Process (newBoundedInbox)+import Control.Concurrent.NQE.Supervisor (Strategy (IgnoreAll))+import Control.Concurrent.STM+import Control.Monad (forM_, replicateM_, void)+import Data.Map.Strict qualified as Map+import Data.Time.Clock (getCurrentTime)+import Effectful (liftIO, runEff)+import Effectful.State.Static.Local qualified as State+import Shibuya qualified as Public+import Shibuya.Adapter (Adapter (..))+import Shibuya.Batch (BatchConfig (..), ackAllOk, defaultBatchConfig)+import Shibuya.Core.Ack (AckDecision (..), HaltReason (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (mkIngested)+import Shibuya.Core.Metrics+import Shibuya.Core.Types (MessageId (..), mkEnvelope)+import Shibuya.Internal.Runner.Ingester (runIngesterWithMetrics)+import Shibuya.Internal.Runner.Master+import Shibuya.Internal.Runner.Pause+import Shibuya.Internal.Runner.Supervised (SupervisedProcessor (..), runSupervised)+import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))+import Shibuya.Telemetry.Effect (runTracingNoop)+import Streamly.Data.Stream qualified as Stream+import Test.Hspec+import UnliftIO qualified as UIO++spec :: Spec+spec = describe "pause and resume" $ do+  it "preserves source-local effects and counts only yielded messages across skipped steps" $ do+    owner <- myThreadId+    now <- getCurrentTime+    h <- newMetricsHandle now+    pause <- newPauseHandle+    inbox <- newBoundedInbox 10+    updates <- runEff $ State.execState (0 :: Int) $ runTracingNoop $ do+      let numbers =+            Stream.unfoldrM+              ( \n -> do+                  liftIO $ myThreadId >>= (`shouldBe` owner)+                  State.modify @Int (+ 1)+                  pure $ if n < 6 then Just (n, n + 1) else Nothing+              )+              (0 :: Int)+          source =+            fmap+              (\n -> mkIngested (mkEnvelope (MessageId "source-local") n) (AckHandle (\_ -> pure ())))+              (Stream.filter even numbers)+      runIngesterWithMetrics (ProcessorId "source-local") h pause source inbox+    updates `shouldBe` 7+    snapshot <- sampleMetrics h+    snapshot.stats.received `shouldBe` 3++  it "requests nothing further when paused before source admission" $ do+    now <- getCurrentTime+    h <- newMetricsHandle now+    pause <- newPauseHandle+    requestPause pause `shouldReturn` True+    polled <- newEmptyTMVarIO+    inbox <- newBoundedInbox 10+    let action =+          runEff $+            runTracingNoop $+              runIngesterWithMetrics+                (ProcessorId "pause-ingester")+                h+                pause+                (Stream.repeatM (liftIO (atomically $ putTMVar polled ()) >> pure (mkIngested (mkEnvelope (MessageId "a") ()) (AckHandle (\_ -> pure ())))))+                inbox+    UIO.withAsync action $ \_ -> do+      UIO.timeout 100_000 (atomically $ takeTMVar polled) `shouldReturn` Nothing+      requestResume pause `shouldReturn` True+      UIO.timeout 1_000_000 (atomically $ takeTMVar polled) `shouldReturn` Just ()++  it "allows an admitted poll and in-flight work to drain, isolates siblings, and resumes before shutdown" $ do+    forM_ [False, True] $ \batching -> do+      polls <- newTQueueIO+      releases <- newTQueueIO+      starts <- newTQueueIO+      finishes <- newTVarIO (0 :: Int)+      stop <- newTVarIO False+      handlerGate <- newEmptyTMVarIO+      sibling <- newEmptyTMVarIO+      runEff $ runTracingNoop $ do+        let source =+              Stream.unfoldrM+                ( \n -> do+                    liftIO $ atomically $ writeTQueue polls n+                    next <-+                      liftIO $+                        atomically $+                          (readTVar stop >>= check >> pure Nothing) `orElse` (Just <$> readTQueue releases)+                    pure $ fmap (\() -> (mkIngested (mkEnvelope (MessageId "live") ()) (AckHandle (\_ -> liftIO $ atomically $ modifyTVar' finishes (+ 1))), n + 1)) next+                )+                (1 :: Int)+            adapter = Adapter "live" source (liftIO $ atomically $ writeTVar stop True)+            work = liftIO ((atomically (writeTQueue starts ()) >> atomically (takeTMVar handlerGate))) >> pure AckOk+            processor =+              if batching+                then Public.mkBatchProcessor adapter (\_ _ -> work >> pure ackAllOk) defaultBatchConfig {batchSize = 1}+                else Public.mkProcessor adapter (\_ -> work)+            siblingAdapter = Adapter "sibling" (Stream.fromList [mkIngested (mkEnvelope (MessageId "sibling") ()) (AckHandle (\_ -> pure ()))]) (pure ())+        result <- Public.runApp Public.defaultAppConfig [(ProcessorId "live", processor), (ProcessorId "sibling", Public.mkProcessor siblingAdapter (\_ -> liftIO (atomically $ putTMVar sibling ()) >> pure AckOk))]+        app <- either (liftIO . ioError . userError . show) pure result+        liftIO $ atomically (readTQueue polls) `shouldReturn` 1+        first <- Public.pauseProcessor app (ProcessorId "live")+        liftIO $ first `shouldSatisfy` isAppliedPause+        second <- Public.pauseProcessor app (ProcessorId "live")+        liftIO $ case (first, second) of+          (ControlApplied state, ControlAlreadyInState same) -> same `shouldBe` state+          other -> expectationFailure (show other)+        liftIO $ atomically $ writeTQueue releases ()+        liftIO $ bounded (atomically $ readTQueue starts) >> bounded (atomically $ takeTMVar sibling)+        during <- Public.getAppMetrics app+        liftIO $ (during Map.! ProcessorId "live").state `shouldSatisfy` pausedWith 1+        liftIO $ atomically $ putTMVar handlerGate ()+        liftIO $ bounded $ atomically $ readTVar finishes >>= check . (== 1)+        -- Completion includes the metrics decrement, which follows finalization.+        liftIO $+          bounded $+            let await = do+                  m <- getProcessorMetricsIO (Public.getAppMaster app) (ProcessorId "live")+                  if maybe False (pausedWith 0 . (.state)) m then pure () else yield >> await+             in await+        liftIO $ UIO.timeout 100_000 (atomically $ readTQueue polls) `shouldReturn` Nothing+        Public.resumeProcessor app (ProcessorId "live") >>= liftIO . (`shouldBe` ControlApplied IntakeRunning)+        Public.resumeProcessor app (ProcessorId "live") >>= liftIO . (`shouldBe` ControlAlreadyInState IntakeRunning)+        liftIO $ atomically (readTQueue polls) `shouldReturn` 2+        _ <- Public.pauseProcessor app (ProcessorId "live")+        drained <- Public.stopAppGracefully Public.defaultShutdownConfig app+        liftIO $ drained `shouldBe` True+        liftIO $ readTVarIO stop `shouldReturn` True++  it "reports unknown, uncontrollable, terminal and draining outcomes atomically" $ do+    master <- runEff $ startMaster IgnoreAll+    UIO.finally+      ( do+          now <- getCurrentTime+          h <- newMetricsHandle now+          pause <- newPauseHandle+          pauseProcessorIO master (ProcessorId "unknown") `shouldReturn` ControlNotFound+          runEff $ registerProcessor master (ProcessorId "fixture") h+          pauseProcessorIO master (ProcessorId "fixture") `shouldReturn` ControlNotControllable+          runEff $ registerProcessorWithPause master (ProcessorId "live") h (Just pause)+          replicateM_ 20 $ do+            UIO.mapConcurrently_ (\n -> void $ if even n then pauseProcessorIO master (ProcessorId "live") else resumeProcessorIO master (ProcessorId "live")) [1 .. 20 :: Int]+            intended <- isPaused pause+            sampled <- sampleMetrics h+            isJustPause sampled.state `shouldBe` intended+          _ <- UIO.concurrently (runEff $ drainAndResume master) (pauseProcessorIO master (ProcessorId "live"))+          isPaused pause `shouldReturn` False+          pauseProcessorIO master (ProcessorId "live") `shouldReturn` ControlDraining+          isProcessorPausedIO master (ProcessorId "live") `shouldReturn` Nothing+          markProcessorStoppedIO master (ProcessorId "live")+          resumeProcessorIO master (ProcessorId "live") `shouldReturn` ControlTerminal+          runEff $ unregisterProcessor master (ProcessorId "live")+          pauseProcessorIO master (ProcessorId "live") `shouldReturn` ControlTerminal+      )+      (runEff $ stopMaster master)++  it "keeps terminal halt and finalization failure above pause intent" $ do+    forM_ [False, True] $ \failure -> do+      started <- newEmptyTMVarIO+      release <- newEmptyTMVarIO+      master <- runEff $ startMaster IgnoreAll+      UIO.finally+        ( runEff $ runTracingNoop $ do+            let message = mkIngested (mkEnvelope (MessageId "terminal") ()) (AckHandle (\_ -> if failure then liftIO (ioError (userError "failure")) else pure ()))+                adapter = Adapter "terminal" (Stream.fromList [message]) (pure ())+            proc <-+              runSupervised+                master+                10+                (ProcessorId "terminal")+                Unordered+                Serial+                adapter+                ( \_ -> do+                    liftIO $ (atomically (putTMVar started ()) >> atomically (takeTMVar release))+                    pure $ if failure then AckOk else AckHalt (HaltFatal "stop")+                )+            liftIO $ bounded (atomically $ takeTMVar started)+            _ <- liftIO $ pauseProcessorIO master (ProcessorId "terminal")+            liftIO $ atomically $ putTMVar release ()+            liftIO $ bounded (atomically $ readTVar proc.done >>= check)+            liftIO $ resumeProcessorIO master (ProcessorId "terminal") `shouldReturn` ControlTerminal+            lifecycle <- liftIO $ getLifecycleSnapshotIO master+            liftIO $+              if failure+                then (lifecycle Map.! ProcessorId "terminal") `shouldSatisfy` (\case LifecycleFailed {} -> True; _ -> False)+                else (lifecycle Map.! ProcessorId "terminal") `shouldBe` LifecycleStopped+            m <- liftIO $ sampleMetrics proc.metrics+            liftIO $ m.state `shouldSatisfy` (\case Failed {} -> True; _ -> False)+        )+        (runEff $ stopMaster master)++isAppliedPause :: ControlOutcome -> Bool+isAppliedPause (ControlApplied (IntakePaused _)) = True+isAppliedPause _ = False++isJustPause :: ProcessorState -> Bool+isJustPause Paused {} = True+isJustPause _ = False++pausedWith :: Int -> ProcessorState -> Bool+pausedWith n (Paused info _ _) = info.inFlight == n+pausedWith _ _ = False++bounded :: IO a -> IO a+bounded action = UIO.timeout 2_000_000 action >>= maybe (expectationFailure "timed out" >> error "timeout") pure

@@ -2,15 +2,17 @@  module Shibuya.Runner.SupervisedSpec (spec) where +import Control.Concurrent (myThreadId) import Control.Concurrent.NQE.Supervisor (Strategy (..)) import Control.Concurrent.STM (atomically, check, newEmptyTMVarIO, readTVar, readTVarIO, takeTMVar, tryPutTMVar) import Control.Exception (MaskingState (..), getMaskingState)-import Control.Monad (forM, forM_, replicateM)+import Control.Monad (forM, forM_, replicateM, replicateM_, when) 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 Effectful.State.Static.Local qualified as State import Shibuya (ProcessorHalt (..)) import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Mock@@ -27,6 +29,7 @@     runApp,     stopAppGracefully,   )+import Shibuya.Batch (BatchConfig (..), ackAllOk, defaultBatchConfig) import Shibuya.Batch.TestHarness (finalizedExactlyOnce) import Shibuya.Core.Ack   ( AckDecision (..),@@ -46,6 +49,7 @@     StreamStats (..),     sampleMetrics,   )+import Shibuya.Core.Metrics qualified as Metrics import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), mkEnvelope) import Shibuya.Handler (Handler) import Shibuya.Internal.Runner.Master@@ -60,7 +64,9 @@     getMetrics,     isDone,     runSupervised,+    runSupervisedWith,     runWithMetrics,+    runWithMetricsBatch,   ) import Shibuya.Policy (Concurrency (..), OrderingPolicy (..)) import Shibuya.Telemetry.Effect (runTracingNoop)@@ -99,6 +105,186 @@       processorMetrics `shouldBe` Just Nothing    describe "Shibuya.Internal.Runner.Supervised" $ do+    forM_ [False, True] $ \batching ->+      it ("isolates intake state while preserving serial handler state for " <> if batching then "batches" else "messages") $ do+        owner <- myThreadId+        observed <- newIORef []+        finalState <- runEff $ State.execState (100 :: Int) $ runTracingNoop $ do+          let source =+                Stream.unfoldrM+                  ( \n ->+                      if n == (3 :: Int)+                        then pure Nothing+                        else do+                          liftIO $ myThreadId >>= (`shouldNotBe` owner)+                          value <- State.get @Int+                          State.modify @Int (+ 1)+                          pure $ Just (mkIngested (mkEnvelope (MessageId (Text.pack (show n))) value) (AckHandle (const $ pure ())), n + 1)+                  )+                  0+              adapter = Adapter {adapterName = "test:state", source = source, shutdown = pure ()}+              observe message = do+                value <- State.get @Int+                State.modify @Int (+ 10)+                liftIO $ do+                  myThreadId >>= (`shouldBe` owner)+                  modifyIORef' observed (<> [(message.envelope.payload, value)])+          if batching+            then runWithMetricsBatch 1 (ProcessorId "state") Serial defaultBatchConfig {batchSize = 1} adapter (\_ messages -> mapM_ observe messages >> pure ackAllOk)+            else runWithMetrics 1 (ProcessorId "state") adapter (\message -> observe message >> pure AckOk)+        finalState `shouldBe` 130+        readIORef observed `shouldReturn` [(100, 100), (101, 110), (102, 120)]++    it "records only successful acknowledgement and dead-letter cursors" $ do+      forM_ [AckOk, AckDeadLetter MaxRetriesExceeded, AckRetry (RetryDelay 0), AckHalt (HaltFatal "stop")] $ \decision -> do+        final <- cursorOutcome decision False+        let expected = case decision of+              AckOk -> Just (CursorInt 5)+              AckDeadLetter _ -> Just (CursorInt 5)+              _ -> Nothing+        (final >>= (.latest)) `shouldBe` expected+      cursorOutcome AckOk True `shouldReturn` Nothing++    it "preserves source cursor order per partition in detailed ordered processing" $ do+      final <- runEff $ runTracingNoop $ do+        master <- startMaster IgnoreAll+        tracking <- newTrackingAck+        let adapter = trackedListAdapter tracking [((mkEnvelope (MessageId (Text.pack (show n))) n) {cursor = Just (CursorInt n), partition = Just (if odd n then "a" else "b")}) | n <- [1 .. 20]]+        proc <- runSupervisedWith (Metrics.MetricsDetailed Metrics.defaultDetailOptions) master 100 (ProcessorId "cursor-ordered") PartitionedInOrder (Async 4) adapter (\_ -> pure AckOk)+        liftIO $ atomically $ readTVar proc.done >>= check+        m <- getMetrics proc+        stopMaster master+        pure m.progress+      fmap (.partitions) final `shouldBe` Just (Map.fromList [("a", CursorInt 19), ("b", CursorInt 20)])++    it "reports growing oldest age and clears it after finalization" $ do+      started <- UIO.newEmptyMVar+      release <- UIO.newEmptyMVar+      runEff $ runTracingNoop $ do+        master <- startMaster IgnoreAll+        messages <- createTestMessages 1+        sp <-+          runSupervisedWith+            (Metrics.MetricsDetailed Metrics.defaultDetailOptions)+            master+            10+            (ProcessorId "detailed-age")+            Unordered+            Serial+            (testAdapter messages)+            (\_ -> liftIO (UIO.putMVar started () >> UIO.takeMVar release) >> pure AckOk)+        liftIO (UIO.takeMVar started)+        first <- liftIO $ sampleMetrics sp.metrics+        liftIO $ threadDelay 20_000+        second <- liftIO $ sampleMetrics sp.metrics+        liftIO $ do+          fmap (.messageId) second.oldestInFlight `shouldBe` Just (MessageId "msg-1")+          fmap (.ageSeconds) second.oldestInFlight `shouldSatisfy` (> fmap (.ageSeconds) first.oldestInFlight)+          UIO.putMVar release ()+          UIO.timeout 1_000_000 (atomically $ readTVar sp.done >>= check) `shouldReturn` Just ()+        completed <- liftIO $ sampleMetrics sp.metrics+        liftIO $ do+          completed.oldestInFlight `shouldBe` Nothing+          fmap (.count) completed.latency `shouldBe` Just 1+          fmap (.sumSeconds) completed.latency `shouldSatisfy` (>= Just 0.02)+          Metrics.slotMisses sp.metrics `shouldReturn` 0+        stopMaster master++    it "identifies the oldest of four concurrent attempts as slots are released" $ do+      starts <- replicateM 4 UIO.newEmptyMVar+      releases <- replicateM 4 UIO.newEmptyMVar+      runEff $ runTracingNoop $ do+        master <- startMaster IgnoreAll+        messages <- createTestMessages 4+        let source =+              Stream.mapM+                ( \(i, msg) -> do+                    when (i > 0) $ UIO.readMVar (starts !! (i - 1))+                    pure msg+                )+                (Stream.fromList (zip [0 ..] messages))+            adapter = (testAdapter messages) {source}+            handler msg = do+              let i = read (drop 8 msg.envelope.payload) - 1+              liftIO $ UIO.putMVar (starts !! i) () >> UIO.takeMVar (releases !! i)+              pure AckOk+        sp <-+          runSupervisedWith+            (Metrics.MetricsDetailed Metrics.defaultDetailOptions)+            master+            10+            (ProcessorId "detailed-four")+            Unordered+            (Async 4)+            adapter+            handler+        liftIO $ mapM_ UIO.readMVar starts+        liftIO $ sampleMetrics sp.metrics >>= \m -> fmap (.messageId) m.oldestInFlight `shouldBe` Just (MessageId "msg-1")+        liftIO $ do+          UIO.putMVar (releases !! 3) ()+          UIO.putMVar (releases !! 2) ()+          UIO.putMVar (releases !! 0) ()+          let awaitSecond = do+                m <- sampleMetrics sp.metrics+                if fmap (.messageId) m.oldestInFlight == Just (MessageId "msg-2") then pure () else threadDelay 1000 >> awaitSecond+          UIO.timeout 1_000_000 awaitSecond `shouldReturn` Just ()+          UIO.putMVar (releases !! 1) ()+          UIO.timeout 1_000_000 (atomically $ readTVar sp.done >>= check) `shouldReturn` Just ()+          Metrics.slotMisses sp.metrics `shouldReturn` 0+        stopMaster master++    it "counts completed finalization failures including retry delays" $ do+      runEff $ runTracingNoop $ do+        master <- startMaster IgnoreAll+        let message =+              mkIngested+                (mkEnvelope (MessageId "failed-detail") ())+                (AckHandle (\_ -> liftIO $ ioError (userError "failed finalizer")))+            adapter = Adapter "failed-detail" (Stream.fromList [message]) (pure ())+        sp <-+          runSupervisedWith+            (Metrics.MetricsDetailed Metrics.defaultDetailOptions)+            master+            1+            (ProcessorId "failed-detail")+            Unordered+            Serial+            adapter+            (\_ -> pure AckOk)+        liftIO $ UIO.timeout 1_000_000 (atomically $ readTVar sp.done >>= check) `shouldReturn` Just ()+        m <- liftIO $ sampleMetrics sp.metrics+        liftIO $ do+          m.oldestInFlight `shouldBe` Nothing+          fmap (.count) m.latency `shouldBe` Just 1+          fmap (.sumSeconds) m.latency `shouldSatisfy` (>= Just 0.31)+        stopMaster master++    it "abandons detail when cancellation interrupts a handler or finalizer" $ do+      forM_ [False, True] $ \inFinalizer -> replicateM_ 100 $ do+        entered <- UIO.newEmptyMVar+        blocker <- UIO.newEmptyMVar+        runEff $ runTracingNoop $ do+          master <- startMaster IgnoreAll+          let waitHere = liftIO (UIO.putMVar entered () >> UIO.takeMVar blocker)+              message = mkIngested (mkEnvelope (MessageId "cancel") ()) (AckHandle (\_ -> when inFinalizer waitHere))+              adapter = Adapter "cancel-detail" (Stream.fromList [message]) (pure ())+          sp <-+            runSupervisedWith+              (Metrics.MetricsDetailed Metrics.defaultDetailOptions)+              master+              1+              (ProcessorId "cancel-detail")+              Unordered+              Serial+              adapter+              (\_ -> when (not inFinalizer) waitHere >> pure AckOk)+          liftIO $ UIO.takeMVar entered+          stopMaster master+          liftIO $ do+            m <- sampleMetrics sp.metrics+            m.oldestInFlight `shouldBe` Nothing+            m.latency `shouldBe` Nothing+     it "restores interruptibility inside a supervised child" $ do       observed <- newEmptyTMVarIO       result <-@@ -1190,3 +1376,17 @@  alwaysAckOk :: Handler es msg alwaysAckOk _ = pure AckOk++cursorOutcome :: AckDecision -> Bool -> IO (Maybe Metrics.ProgressSnapshot)+cursorOutcome decision failFinalizer = runEff $ runTracingNoop $ do+  master <- startMaster IgnoreAll+  let message =+        mkIngested+          ((mkEnvelope (MessageId "cursor") ()) {cursor = Just (CursorInt 5), partition = Just "p"})+          (AckHandle (\_ -> when failFinalizer (liftIO $ UIO.throwIO (userError "finalization failed"))))+      adapter = Adapter "cursor" (Stream.fromList [message]) (pure ())+  proc <- runSupervisedWith (Metrics.MetricsDetailed Metrics.defaultDetailOptions) master 10 (ProcessorId "cursor") Unordered Serial adapter (\_ -> pure decision)+  liftIO $ atomically $ readTVar proc.done >>= check+  m <- getMetrics proc+  stopMaster master+  pure m.progress

@@ -7,6 +7,7 @@ import Data.Text (Text) import Effectful (liftIO, runEff) import OpenTelemetry.Trace.Core qualified as OTel+import OpenTelemetry.Trace.Id (Base (..), spanIdBaseEncodedText, traceIdBaseEncodedText) import Shibuya.Telemetry.Effect   ( addAttribute,     addEvent,@@ -17,7 +18,9 @@     withSpan,     withSpan',   )+import Shibuya.Telemetry.Propagation (injectTraceContext) import Shibuya.Telemetry.Semantic (consumerSpanArgs, mkEvent)+import System.Mem.StableName (makeStableName) import Test.Hspec  spec :: Spec@@ -54,6 +57,26 @@         _tracer <- getTracer         pure ("got tracer" :: String)       result `shouldBe` "got tracer"++    it "shares the compatibility provider across independent invocations" $ do+      first <- runEff $ runTracingNoop getTracer+      second <- runEff $ runTracingNoop getTracer+      let firstProvider = OTel.getTracerTracerProvider first+          secondProvider = OTel.getTracerTracerProvider second+      firstProvider `seq` secondProvider `seq` pure ()+      a <- makeStableName firstProvider+      b <- makeStableName secondProvider+      (a == b) `shouldBe` True+      OTel.inSpan' second "direct noop tracer" OTel.defaultSpanArguments OTel.isRecording+        `shouldReturn` False++    it "provides an invalid non-recording sentinel that cannot be propagated" $ do+      runEff $ runTracingNoop $ withSpan' "disabled" consumerSpanArgs $ \sp -> liftIO $ do+        OTel.isRecording sp `shouldReturn` False+        ctx <- OTel.getSpanContext sp+        traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "00000000000000000000000000000000"+        spanIdBaseEncodedText Base16 (OTel.spanId ctx) `shouldBe` "0000000000000000"+        injectTraceContext sp `shouldReturn` []      it "handles exceptions in actions correctly" $ do       ref <- newIORef (0 :: Int)

@@ -0,0 +1,417 @@+module Shibuya.Telemetry.LifecycleSpec (spec) where++import Control.Concurrent (myThreadId, newEmptyMVar, putMVar, readMVar, takeMVar, tryPutMVar)+import Control.Concurrent.NQE.Process (newBoundedInbox)+import Control.Concurrent.STM (atomically, retry)+import Control.Exception (SomeException, bracket, displayException, finally, try)+import Control.Monad (forM_, unless, void, when)+import Data.Foldable (toList, traverse_)+import Data.HashMap.Strict qualified as HashMap+import Data.IORef (modifyIORef', newIORef, readIORef)+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Time (getCurrentTime)+import Effectful (Eff, IOE, liftIO, runEff, (:>))+import Effectful.Exception qualified as Exception+import Effectful.State.Static.Local qualified as State+import OpenTelemetry.Attributes (emptyAttributes, getAttributeMap, toAttribute)+import OpenTelemetry.Context qualified as TraceContext+import OpenTelemetry.Context.ThreadLocal qualified as TraceContext+import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)+import OpenTelemetry.Processor.Span (SpanProcessor (..))+import OpenTelemetry.Trace.Core qualified as OTel+import OpenTelemetry.Trace.Id.Generator (IdGenerator (..))+import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)+import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter.Mock (listAdapter)+import Shibuya.App+import Shibuya.Core.Ack (AckDecision (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (mkIngested)+import Shibuya.Core.Metrics (newMetricsHandle)+import Shibuya.Core.Types (MessageId (..), mkEnvelope)+import Shibuya.Internal.Runner.Ingester (runIngesterWithMetrics, runIngesterWithMetricsUnpaused)+import Shibuya.Internal.Runner.Master (ProcessorLifecycle (..), getLifecycleSnapshot)+import Shibuya.Internal.Runner.Pause (newPauseHandle, requestPause, requestResume)+import Shibuya.Telemetry.Effect (Tracing, runTracing, runTracingNoop, withSpan')+import Streamly.Data.Stream qualified as Stream+import Test.Hspec+import UnliftIO qualified as UIO++spec :: Spec+spec = describe "Shibuya.Telemetry.Lifecycle" $ do+  forM_ [False, True] $ \batching ->+    it ("emits one ingestion lifetime and once-only shutdown for " <> if batching then "batch" else "single") $ do+      calls <- newIORef (0 :: Int)+      (_, spans) <- withExporter $ \tracer -> runEff $ runTracing tracer $ do+        let messages = [mkIngested (mkEnvelope (MessageId "one") ()) (AckHandle (const $ pure ()))]+            adapter = (listAdapter messages) {shutdown = liftIO $ modifyIORef' calls (+ 1)}+            processor =+              if batching+                then mkBatchProcessor adapter (\_ _ -> pure ackAllOk) defaultBatchConfig+                else mkProcessor adapter (const $ pure AckOk)+        app <- start processor+        Exception.finally (waitApp app) (stopApp app)+        stopApp app+      readIORef calls `shouldReturn` 1+      expectSpan "shibuya.ingest" spans >>= expectIdentity+      expectSpan "lifecycle shutdown" spans >>= expectIdentity++  forM_ [False, True] $ \batching ->+    it ("does not start ingestion after onStart consumes cancellation for " <> if batching then "batch" else "single") $ do+      (_, _) <- withBlockedIngestion False $ \tracer gates -> do+        app <- runEff $ runTracing tracer $ do+          let adapter = (listAdapter []) {source = Stream.repeatM (liftIO gates.sourceWait >> Exception.throwIO (userError "test source released"))}+              processor =+                if batching+                  then mkBatchProcessor adapter (\_ _ -> pure ackAllOk) defaultBatchConfig+                  else mkProcessor adapter (const $ pure AckOk)+          start processor+        finally+          ( do+              gates.awaitHook+              UIO.withAsync+                (runEff $ runTracing tracer $ stopAppGracefully defaultShutdownConfig {drainTimeout = 0.01, totalShutdownTimeout = 0.1} app)+                $ \stopper ->+                  finally+                    ( do+                        UIO.timeout 1_000_000 (UIO.wait stopper) `shouldReturn` Just False+                        gates.sourceEntries `shouldReturn` 0+                    )+                    gates.release+          )+          (gates.release >> void (try @SomeException $ runEff (runTracing tracer $ stopApp app)))+      pure ()++  forM_ [False, True] $ \onEnd ->+    it ("retains direct ingester cancellation during " <> if onEnd then "onEnd" else "onStart") $ do+      (_, _) <- withBlockedIngestion onEnd $ \tracer gates -> do+        metrics <- getCurrentTime >>= newMetricsHandle+        inbox <- newBoundedInbox 1+        let source = if onEnd then Stream.fromList [] else Stream.repeatM (liftIO gates.sourceWait >> Exception.throwIO (userError "test source released"))+        UIO.withAsync (runEff $ runTracing tracer $ runIngesterWithMetricsUnpaused (ProcessorId "lifecycle") metrics source inbox) $ \ingester ->+          finally+            ( do+                gates.awaitHook+                UIO.withAsync (UIO.cancel ingester) $ \canceller ->+                  finally+                    ( do+                        UIO.timeout 1_000_000 (UIO.wait canceller) `shouldReturn` Just ()+                        UIO.waitCatch ingester >>= \case+                          Left failure -> Exception.isAsyncException failure `shouldBe` True+                          Right () -> expectationFailure "ingestion swallowed cancellation"+                        gates.sourceEntries `shouldReturn` 0+                    )+                    gates.release+            )+            gates.release+      pure ()++  forM_ [False, True] $ \pauseAware ->+    it ("preserves source thread, local state and context with " <> if pauseAware then "pause admission" else "unpaused admission") $ do+      owner <- myThreadId+      metrics <- getCurrentTime >>= newMetricsHandle+      pause <- newPauseHandle+      inbox <- newBoundedInbox 10+      seen <- newIORef []+      ((updates, parentContext), spans) <- withExporter $ \tracer -> do+        (parentContext, updates) <- runEff $+          State.runState (0 :: Int) $+            runTracing tracer $+              withSpan' "source caller" OTel.defaultSpanArguments $ \parent -> do+                let source =+                      Stream.unfoldrM+                        ( \n -> do+                            liftIO $ myThreadId >>= (`shouldBe` owner)+                            State.modify @Int (+ 1)+                            liftIO $ do+                              current <- TraceContext.lookupSpan <$> TraceContext.getContext+                              traverse_ (\sp -> OTel.getSpanContext sp >>= \ctx -> modifyIORef' seen (ctx :)) current+                            pure $ if n < 2 then Just (mkIngested (mkEnvelope (MessageId "one") ()) (AckHandle (const $ pure ())), n + 1) else Nothing+                        )+                        (0 :: Int)+                if pauseAware+                  then runIngesterWithMetrics (ProcessorId "lifecycle") metrics pause source inbox+                  else runIngesterWithMetricsUnpaused (ProcessorId "lifecycle") metrics source inbox+                parentContext <- liftIO $ OTel.getSpanContext parent+                liftIO $ do+                  restored <- TraceContext.lookupSpan <$> TraceContext.getContext+                  case restored of+                    Nothing -> expectationFailure "source caller context was not restored"+                    Just sp -> OTel.getSpanContext sp >>= \ctx -> OTel.spanId ctx `shouldBe` OTel.spanId parentContext+                pure parentContext+        pure (updates, parentContext)+      updates `shouldBe` 3+      ingestion <- expectSpan "shibuya.ingest" spans+      expectIdentity ingestion+      let ingestionContext = OTel.spanContext ingestion+      observed <- readIORef seen+      map OTel.spanId observed `shouldBe` replicate 3 (OTel.spanId ingestionContext)+      case OTel.spanParent ingestion of+        Nothing -> expectationFailure "ingestion lost its caller parent"+        Just sp -> OTel.getSpanContext sp >>= \ctx -> OTel.spanId ctx `shouldBe` OTel.spanId parentContext++  forM_ [False, True] $ \paused ->+    it ("retains cancellation of a " <> if paused then "paused source" else "running source") $ do+      entered <- newEmptyMVar+      gate <- newEmptyMVar+      calls <- newIORef (0 :: Int)+      pause <- newPauseHandle+      metrics <- getCurrentTime >>= newMetricsHandle+      inbox <- newBoundedInbox 1+      let release = void (tryPutMVar gate ()) >> void (requestResume pause)+          source = Stream.repeatM $ liftIO $ do+            modifyIORef' calls (+ 1)+            when paused $ void (requestPause pause)+            putMVar entered ()+            unless paused $ readMVar gate+            pure (mkIngested (mkEnvelope (MessageId "one") ()) (AckHandle (const $ pure ())))+      (_, _) <- withExporter $ \tracer ->+        UIO.withAsync (runEff $ runTracing tracer $ runIngesterWithMetrics (ProcessorId "lifecycle") metrics pause source inbox) $ \ingester ->+          finally+            ( do+                takeMVar entered+                -- For the paused case, no further source admission is allowed.+                when paused $ UIO.timeout 20_000 (takeMVar entered) `shouldReturn` Nothing+                UIO.withAsync (UIO.cancel ingester) $ \canceller ->+                  finally+                    ( do+                        UIO.timeout 1_000_000 (UIO.wait canceller) `shouldReturn` Just ()+                        UIO.waitCatch ingester >>= \case+                          Left failure -> Exception.isAsyncException failure `shouldBe` True+                          Right () -> expectationFailure "source cancellation disappeared"+                        readIORef calls `shouldReturn` 1+                    )+                    release+            )+            release+      pure ()++  it "preserves the direct source exception when onEnd also fails" $ do+    metrics <- getCurrentTime >>= newMetricsHandle+    inbox <- newBoundedInbox 1+    let failEnd processor = processor {spanProcessorOnEnd = \_ -> ioError (userError "export failed")}+        source = Stream.repeatM (Exception.throwIO (userError "original source failure"))+    (result, _) <- withExporterUsing failEnd $ \tracer ->+      try @SomeException $+        runEff $+          runTracing tracer $+            runIngesterWithMetricsUnpaused (ProcessorId "lifecycle") metrics source inbox+    case result of+      Left failure -> displayException failure `shouldBe` "user error (original source failure)"+      Right () -> expectationFailure "source exception disappeared"++  it "records source failure and retains the terminal lifecycle" $ do+    (state, spans) <- withExporter $ \tracer -> runEff $ runTracing tracer $ do+      let adapter = (listAdapter []) {source = Stream.repeatM (Exception.throwIO $ userError "source failed")}+      app <- start $ mkProcessor adapter (const $ pure AckOk)+      Exception.finally+        (waitApp app >> getLifecycleSnapshot (getAppMaster app))+        (stopApp app)+    case Map.lookup (ProcessorId "lifecycle") state of+      Just (LifecycleFailed message _) -> message `shouldSatisfy` (Text.isInfixOf "source failed")+      _ -> expectationFailure $ "expected source failure, got " <> show state+    expectSpan "shibuya.ingest" spans >>= expectError++  it "records a shutdown exception, preserves it, and caches it" $ do+    calls <- newIORef (0 :: Int)+    (results, spans) <- withExporter $ \tracer -> runEff $ runTracing tracer $ do+      let adapter = (listAdapter []) {shutdown = liftIO (modifyIORef' calls (+ 1)) >> Exception.throwIO (userError "shutdown failed")}+      app <- start $ mkProcessor adapter (const $ pure AckOk)+      waitApp app+      first <- Exception.try @SomeException $ stopApp app+      second <- Exception.try @SomeException $ stopApp app+      pure [first, second]+    forM_ results $ \result -> case result of+      Left failure -> displayException failure `shouldContain` "shutdown failed"+      Right () -> expectationFailure "shutdown exception disappeared"+    readIORef calls `shouldReturn` 1+    expectSpan "lifecycle shutdown" spans >>= expectError++  forM_ [False, True] $ \onEnd -> do+    let hookName = if onEnd then "onEnd" else "onStart"+    it ("retains the shutdown deadline when " <> hookName <> " swallows cancellation") $ do+      calls <- newIORef (0 :: Int)+      ((first, second), _) <- withExporterUsing (blockingShutdownHook onEnd (pure ())) $ \tracer -> runEff $ runTracing tracer $ do+        let adapter = (listAdapter []) {shutdown = liftIO $ modifyIORef' calls (+ 1)}+        app <- start $ mkProcessor adapter (const $ pure AckOk)+        waitApp app+        first <- stopAppGracefully defaultShutdownConfig {totalShutdownTimeout = 0.05} app+        second <- stopAppGracefully defaultShutdownConfig app+        pure (first, second)+      (first, second) `shouldBe` (False, False)+      readIORef calls `shouldReturn` if onEnd then 1 else 0++    it ("retains caller cancellation when " <> hookName <> " swallows it") $ do+      entered <- newEmptyMVar+      calls <- newIORef (0 :: Int)+      ((first, second), _) <- withExporterUsing (blockingShutdownHook onEnd (putMVar entered ())) $ \tracer -> do+        app <- runEff $ runTracing tracer $ do+          let adapter = (listAdapter []) {shutdown = liftIO $ modifyIORef' calls (+ 1)}+          app <- start $ mkProcessor adapter (const $ pure AckOk)+          waitApp app+          pure app+        first <- UIO.withAsync (runEff $ runTracing tracer $ stopApp app) $ \stopper -> do+          takeMVar entered+          UIO.cancel stopper+          UIO.waitCatch stopper+        second <- try @SomeException $ runEff $ runTracing tracer $ stopApp app+        pure (first, second)+      forM_ [first, second] $ \case+        Left failure -> Exception.isAsyncException failure `shouldBe` True+        Right () -> expectationFailure "shutdown swallowed caller cancellation"+      readIORef calls `shouldReturn` if onEnd then 1 else 0++  it "records shutdown deadline cancellation without changing the False result" $ do+    ((first, second), spans) <- withExporter $ \tracer -> runEff $ runTracing tracer $ do+      let adapter = (listAdapter []) {shutdown = liftIO $ atomically retry}+      app <- start $ mkProcessor adapter (const $ pure AckOk)+      waitApp app+      first <- stopAppGracefully defaultShutdownConfig {totalShutdownTimeout = 0.05} app+      second <- stopAppGracefully defaultShutdownConfig app+      pure (first, second)+    (first, second) `shouldBe` (False, False)+    expectSpan "lifecycle shutdown" spans >>= expectError++  it "exports cancellation of the ingester when the master is force-stopped" $ do+    started <- newEmptyMVar+    (drained, spans) <- withExporter $ \tracer -> runEff $ runTracing tracer $ do+      let adapter =+            (listAdapter [])+              { source = Stream.repeatM (liftIO (putMVar started () >> atomically retry))+              }+      app <- start $ mkProcessor adapter (const $ pure AckOk)+      liftIO $ takeMVar started+      stopAppGracefully defaultShutdownConfig {drainTimeout = 0.05, totalShutdownTimeout = 1} app+    drained `shouldBe` False+    expectSpan "shibuya.ingest" spans >>= expectError++  it "uses the startup tracer even if stop runs under a different runner" $ do+    (_, spans) <- withExporter $ \tracer -> do+      app <- runEff $ runTracing tracer $ do+        app <- start $ mkProcessor (listAdapter []) (const $ pure AckOk)+        waitApp app+        pure app+      runEff $ runTracingNoop $ stopApp app+    expectSpan "lifecycle shutdown" spans >>= expectIdentity++  it "retains adapter shutdown updates to caller-local effects" $ do+    (updates, _) <- withExporter $ \tracer -> runEff $ State.execState (0 :: Int) $ runTracing tracer $ do+      let adapter = (listAdapter []) {shutdown = State.modify @Int (+ 1)}+      app <- start $ mkProcessor adapter (const $ pure AckOk)+      waitApp app+      stopApp app+    updates `shouldBe` 1++  it "parents shutdown to the caller's active span" $ do+    (parentContext, spans) <- withExporter $ \tracer -> runEff $ runTracing tracer $ do+      app <- start $ mkProcessor (listAdapter []) (const $ pure AckOk)+      waitApp app+      withSpan' "stop caller" OTel.defaultSpanArguments $ \parent -> do+        stopApp app+        liftIO $ OTel.getSpanContext parent+    shutdownSpan <- expectSpan "lifecycle shutdown" spans+    case OTel.spanParent shutdownSpan of+      Nothing -> expectationFailure "shutdown lost its caller's parent context"+      Just parent -> do+        actual <- OTel.getSpanContext parent+        OTel.traceId actual `shouldBe` OTel.traceId parentContext+        OTel.spanId actual `shouldBe` OTel.spanId parentContext++  it "exports no lifecycle spans with tracing disabled" $ do+    (_, spans) <- withExporter $ \_ -> runEff $ runTracingNoop $ do+      app <- start $ mkProcessor (listAdapter []) (const $ pure AckOk)+      waitApp app+      stopApp app+    length spans `shouldBe` 0++start :: (IOE :> es, Tracing :> es) => QueueProcessor es -> Eff es (AppHandle es)+start processor =+  runApp defaultAppConfig [(ProcessorId "lifecycle", processor)] >>= \case+    Left err -> liftIO $ expectationFailure (show err) >> error "unreachable"+    Right app -> pure app++withExporter :: (OTel.Tracer -> IO a) -> IO (a, [OTel.ImmutableSpan])+withExporter = withExporterUsing id++-- Block only the shutdown span so startup and ingestion remain independent.+blockingShutdownHook :: Bool -> IO () -> SpanProcessor -> SpanProcessor+blockingShutdownHook onEnd entered processor =+  if onEnd+    then processor {spanProcessorOnEnd = \sp -> block sp >> spanProcessorOnEnd processor sp}+    else processor {spanProcessorOnStart = \sp ctx -> block sp >> spanProcessorOnStart processor sp ctx}+  where+    block sp = do+      hot <- readIORef (OTel.spanHot sp)+      when (OTel.hotName hot == "lifecycle shutdown") $ entered >> atomically retry++-- Release gates before joining test-owned threads, even on the broken runtime.+-- Keeping these gates reachable also prevents a collected STM wait from faking+-- successful cancellation via BlockedIndefinitelyOnSTM.+data IngestionGates = IngestionGates+  { awaitHook :: IO (),+    sourceWait :: IO (),+    sourceEntries :: IO Int,+    release :: IO ()+  }++withBlockedIngestion :: Bool -> (OTel.Tracer -> IngestionGates -> IO a) -> IO (a, [OTel.ImmutableSpan])+withBlockedIngestion onEnd action = do+  entered <- newEmptyMVar+  hookGate <- newEmptyMVar+  sourceGate <- newEmptyMVar+  calls <- newIORef (0 :: Int)+  let release = void (tryPutMVar hookGate ()) >> void (tryPutMVar sourceGate ())+      gates =+        IngestionGates+          { awaitHook = takeMVar entered,+            sourceWait = modifyIORef' calls (+ 1) >> readMVar sourceGate,+            sourceEntries = readIORef calls,+            release = release+          }+      block sp = do+        hot <- readIORef (OTel.spanHot sp)+        when (OTel.hotName hot == "shibuya.ingest") $ putMVar entered () >> readMVar hookGate+      configure processor =+        if onEnd+          then processor {spanProcessorOnEnd = \sp -> block sp >> spanProcessorOnEnd processor sp}+          else processor {spanProcessorOnStart = \sp ctx -> block sp >> spanProcessorOnStart processor sp ctx}+  withExporterUsing configure $ \tracer -> finally (action tracer gates) release++withExporterUsing :: (SpanProcessor -> SpanProcessor) -> (OTel.Tracer -> IO a) -> IO (a, [OTel.ImmutableSpan])+withExporterUsing configure action = do+  (processor, spansRef) <- inMemoryListExporter+  result <- bracket+    (OTel.createTracerProvider [configure processor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator})+    (\p -> void $ OTel.shutdownTracerProvider p (Just 5_000_000))+    $ \provider -> do+      let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "lifecycle-test" "" "" emptyAttributes) OTel.tracerOptions+      -- Bound every test, including waits, so a lifecycle regression cannot hang.+      UIO.timeout 5_000_000 (action tracer) >>= \case+        Nothing -> expectationFailure "lifecycle test timed out" >> error "unreachable"+        Just value -> pure value+  spans <- readIORef spansRef+  pure (result, spans)++expectSpan :: Text -> [OTel.ImmutableSpan] -> IO OTel.ImmutableSpan+expectSpan name spans = do+  named <- mapM (\sp -> do hot <- readIORef (OTel.spanHot sp); pure (OTel.hotName hot, sp)) spans+  case [sp | (n, sp) <- named, n == name] of+    [sp] -> pure sp+    matches -> expectationFailure ("expected one " <> show name <> " span; got " <> show (length matches)) >> error "unreachable"++expectIdentity :: OTel.ImmutableSpan -> Expectation+expectIdentity sp = do+  OTel.spanKind sp `shouldBe` OTel.Internal+  hot <- readIORef (OTel.spanHot sp)+  HashMap.lookup "shibuya.processor.id" (getAttributeMap (OTel.hotAttributes hot))+    `shouldBe` Just (toAttribute ("lifecycle" :: Text))++expectError :: OTel.ImmutableSpan -> Expectation+expectError sp = do+  expectIdentity sp+  hot <- readIORef (OTel.spanHot sp)+  OTel.hotStatus hot `shouldSatisfy` (\case OTel.Error _ -> True; _ -> False)+  map OTel.eventName (toList $ appendOnlyBoundedCollectionValues $ OTel.hotEvents hot)+    `shouldContain` ["exception"]

@@ -3,9 +3,13 @@ -- | Tests for W3C Trace Context propagation. module Shibuya.Telemetry.PropagationSpec (spec) where +import Control.Exception (bracket) import Data.ByteString (ByteString)+import Data.IORef (readIORef) import Effectful (runEff) import OpenTelemetry.Attributes (emptyAttributes)+import OpenTelemetry.Context qualified as Ctx+import OpenTelemetry.Context.ThreadLocal qualified as Ctx import OpenTelemetry.Exporter.InMemory (inMemoryListExporter) import OpenTelemetry.Trace.Core   ( InstrumentationLibrary (..),@@ -19,7 +23,8 @@ import OpenTelemetry.Trace.Core qualified as OTel import OpenTelemetry.Trace.Id (Base (..)) import OpenTelemetry.Trace.Id qualified as OTel.Id-import Shibuya.Telemetry.Effect (runTracing, runTracingNoop, withSpan)+import OpenTelemetry.Trace.Id.Generator (IdGenerator (..))+import Shibuya.Telemetry.Effect (runTracing, runTracingNoop, withSpan, withSpan') import Shibuya.Telemetry.Propagation (currentTraceHeaders, extractTraceContext) import Test.Hspec @@ -129,6 +134,19 @@       result <- runEff $ runTracingNoop currentTraceHeaders       result `shouldBe` Nothing +    it "returns Nothing for an invalid active context even when enabled" $ do+      sentinel <- runEff $ runTracingNoop $ withSpan' "disabled" defaultSpanArguments pure+      bracket+        (createTracerProvider [] emptyTracerProviderOptions)+        (\p -> shutdownTracerProvider p (Just 5_000_000))+        $ \provider -> do+          let tracer = makeTracer provider (InstrumentationLibrary "test" "" "" emptyAttributes) tracerOptions+          bracket+            (Ctx.attachContext $ Ctx.insertSpan sentinel Ctx.empty)+            Ctx.detachContext+            $ \_ -> do+              runEff (runTracing tracer currentTraceHeaders) `shouldReturn` Nothing+     it "returns Nothing outside any active span" $ do       provider <- createTracerProvider [] emptyTracerProviderOptions       let tracer =@@ -147,8 +165,8 @@       result `shouldBe` Nothing      it "returns headers carrying the active span's traceparent" $ do-      (processor, _spansRef) <- inMemoryListExporter-      provider <- createTracerProvider [processor] emptyTracerProviderOptions+      (processor, spansRef) <- inMemoryListExporter+      provider <- createTracerProvider [processor] emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator}       let tracer =             makeTracer               provider@@ -164,6 +182,10 @@       _ <- shutdownTracerProvider provider (Just 5_000_000)       case result of         Nothing -> expectationFailure "expected currentTraceHeaders to return Just"-        Just hs -> case lookup "traceparent" hs of-          Nothing -> expectationFailure "expected traceparent header to be present"-          Just _ -> pure ()+        Just hs -> do+          spans <- readIORef spansRef+          case (extractTraceContext hs, spans) of+            (Just ctx, [exported]) -> do+              OTel.traceId ctx `shouldBe` OTel.traceId (OTel.spanContext exported)+              OTel.spanId ctx `shouldBe` OTel.spanId (OTel.spanContext exported)+            _ -> expectationFailure "expected valid headers and exactly one exported span"

@@ -29,6 +29,7 @@   ( Event (..),     ImmutableSpan (..),     InstrumentationLibrary (..),+    SpanKind (..),     SpanStatus (..),     createTracerProvider,     emptyTracerProviderOptions,@@ -79,7 +80,7 @@       pure ()      _ <- shutdownTracerProvider provider (Just 5_000_000)-    spans <- readIORef spansRef+    spans <- filter ((== Consumer) . spanKind) <$> readIORef spansRef     case spans of       [s] -> do         hot <- readIORef (spanHot s)@@ -130,7 +131,7 @@       pure ()      _ <- shutdownTracerProvider provider (Just 5_000_000)-    spans <- readIORef spansRef+    spans <- filter ((== Consumer) . spanKind) <$> readIORef spansRef     case spans of       [s] -> do         hot <- readIORef (spanHot s)@@ -165,7 +166,7 @@       pure ()      _ <- shutdownTracerProvider provider (Just 5_000_000)-    spans <- readIORef spansRef+    spans <- filter ((== Consumer) . spanKind) <$> readIORef spansRef     case spans of       [s] -> do         hot <- readIORef (spanHot s)