packages feed

shibuya-core (empty) → 0.1.0.0

raw patch · 36 files changed

+4272/−0 lines, 36 files

Files

+ CHANGELOG.md view
@@ -0,0 +1,29 @@+# Changelog++## 0.1.0.0 — 2026-02-24++Initial release.++### New Features++- Multi-queue processing with `runApp` and `QueueProcessor` API+- Backpressure via bounded inbox+- AckHalt support to stop processing on halt decision+- Serial, Ahead, and Async concurrent processing modes+- Policy validation enforcement (StrictInOrder requires Serial)+- Graceful shutdown with configurable drain timeout+- NQE-based supervision via Master/Supervisor+- OpenTelemetry tracing integration with W3C trace context propagation+- Mock adapter for testing++### Bug Fixes++- Fix race conditions and incomplete cleanup in runner+- Fix data races in concurrent test handlers using atomicModifyIORef'++### Other Changes++- Consolidate error handling with unified error types+- Define own SupervisionStrategy type to decouple from NQE+- Use registerDelay for cleaner timeout handling+- Replace polling with STM blocking in waitApp
+ shibuya-core.cabal view
@@ -0,0 +1,134 @@+cabal-version: 3.14+name: shibuya-core+version: 0.1.0.0+synopsis: Supervised queue processing framework for Haskell+description:+  A supervised queue processing framework inspired by Broadway (Elixir).+  Provides unified queue abstraction, NQE-based supervision, backpressure+  via bounded inbox, explicit ack semantics, and OpenTelemetry tracing.++author: Nadeem Bitar+maintainer: nadeem@gmail.com+license: MIT+build-type: Simple+category: Concurrency+extra-doc-files: CHANGELOG.md++common warnings+  ghc-options: -Wall++library+  import: warnings+  exposed-modules:+    Shibuya.Adapter+    Shibuya.Adapter.Mock+    Shibuya.App+    Shibuya.Core+    Shibuya.Core.Ack+    Shibuya.Core.AckHandle+    Shibuya.Core.Error+    Shibuya.Core.Ingested+    Shibuya.Core.Lease+    Shibuya.Core.Types+    Shibuya.Handler+    Shibuya.Policy+    Shibuya.Prelude+    Shibuya.Runner.Master+    Shibuya.Runner.Metrics+    Shibuya.Runner.Supervised+    Shibuya.Stream+    Shibuya.Telemetry+    Shibuya.Telemetry.Config+    Shibuya.Telemetry.Effect+    Shibuya.Telemetry.Propagation+    Shibuya.Telemetry.Semantic++  other-modules:+    Shibuya.Runner.Halt+    Shibuya.Runner.Ingester+    Shibuya.Runner.Processor+    Shibuya.Runner.Serial++  default-extensions:+    DeriveAnyClass+    DerivingStrategies+    DuplicateRecordFields+    LambdaCase+    NoFieldSelectors+    OverloadedLabels+    OverloadedRecordDot+    OverloadedStrings+    QuasiQuotes++  build-depends:+    aeson ^>=2.2,+    base ^>=4.21.0.0,+    bytestring ^>=0.12.2.0,+    containers ^>=0.7,+    effectful ^>=2.6.1.0,+    effectful-core ^>=2.6.1.0,+    generic-lens ^>=2.3.0.0,+    hs-opentelemetry-api ^>=0.3,+    hs-opentelemetry-propagator-w3c ^>=0.1,+    lens ^>=5.3.5,+    nqe ^>=0.6,+    stm ^>=2.5,+    streamly ^>=0.11,+    streamly-core ^>=0.3,+    text ^>=2.1.3,+    time ^>=1.14,+    unliftio ^>=0.2,+    unordered-containers ^>=0.2,+    uuid ^>=1.3,+    vector ^>=0.13,++  hs-source-dirs: src+  default-language: GHC2024++test-suite shibuya-core-test+  import: warnings+  default-language: GHC2024+  type: exitcode-stdio-1.0+  hs-source-dirs: test+  main-is: Main.hs+  default-extensions:+    DeriveAnyClass+    DerivingStrategies+    DuplicateRecordFields+    LambdaCase+    NoFieldSelectors+    OverloadedLabels+    OverloadedRecordDot+    OverloadedStrings+    QuasiQuotes++  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N++  other-modules:+    Shibuya.Core.AckSpec+    Shibuya.Core.TypesSpec+    Shibuya.PolicySpec+    Shibuya.Runner.SupervisedSpec+    Shibuya.RunnerSpec+    Shibuya.Telemetry.EffectSpec+    Shibuya.Telemetry.PropagationSpec++  build-depends:+    QuickCheck ^>=2.15,+    base ^>=4.21.0.0,+    bytestring,+    effectful,+    hs-opentelemetry-api,+    hspec ^>=2.11,+    memory,+    nqe,+    shibuya-core,+    stm,+    streamly,+    streamly-core,+    text,+    time,+    unliftio,
+ src/Shibuya/Adapter.hs view
@@ -0,0 +1,23 @@+-- | Adapter API - library-author facing.+-- Adapters bridge external systems to the framework.+-- Adapter owns queue semantics; runner never touches offsets directly.+module Shibuya.Adapter+  ( Adapter (..),+  )+where++import Data.Text (Text)+import Effectful (Eff)+import Shibuya.Core.Ingested (Ingested)+import Streamly.Data.Stream (Stream)++-- | Queue adapter interface.+-- Provides a stream of ingested messages and shutdown capability.+data Adapter es msg = Adapter+  { -- | Name for logging/observability+    adapterName :: !Text,+    -- | Stream of leased messages+    source :: Stream (Eff es) (Ingested es msg),+    -- | Stop polling, release resources+    shutdown :: Eff es ()+  }
+ src/Shibuya/Adapter/Mock.hs view
@@ -0,0 +1,56 @@+-- | Mock adapter for testing.+-- Provides adapters that produce messages from in-memory sources.+module Shibuya.Adapter.Mock+  ( -- * Mock Adapters+    listAdapter,++    -- * Test Helpers+    TrackingAck (..),+    newTrackingAck,+    trackingAckHandle,+    getTrackedDecisions,+  )+where++import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Effectful (Eff, IOE, liftIO, (:>))+import Shibuya.Adapter (Adapter (..))+import Shibuya.Core.Ack (AckDecision)+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (Ingested)+import Shibuya.Core.Types (MessageId (..))+import Streamly.Data.Stream qualified as Stream++-- | Create an adapter from a list of ingested messages.+-- Useful for testing handlers with predetermined input.+listAdapter :: (IOE :> es) => [Ingested es msg] -> Adapter es msg+listAdapter msgs =+  Adapter+    { adapterName = "mock:list",+      source = Stream.fromList msgs,+      shutdown = pure ()+    }++-- | Tracking state for ack decisions.+data TrackingAck = TrackingAck+  { trackedDecisions :: IORef [(MessageId, AckDecision)]+  }++-- | Create an AckHandle that tracks all decisions made.+-- Useful for testing that handlers make correct ack decisions.+trackingAckHandle ::+  (IOE :> es) =>+  TrackingAck ->+  MessageId ->+  AckHandle es+trackingAckHandle tracking msgId =+  AckHandle $ \decision ->+    liftIO $ modifyIORef' tracking.trackedDecisions ((msgId, decision) :)++-- | Create a new TrackingAck.+newTrackingAck :: (IOE :> es) => Eff es TrackingAck+newTrackingAck = liftIO $ TrackingAck <$> newIORef []++-- | Get all tracked decisions.+getTrackedDecisions :: (IOE :> es) => TrackingAck -> Eff es [(MessageId, AckDecision)]+getTrackedDecisions tracking = liftIO $ readIORef tracking.trackedDecisions
+ src/Shibuya/App.hs view
@@ -0,0 +1,285 @@+-- | Application entry point for running Shibuya queue processors.+module Shibuya.App+  ( -- * Running Processors+    runApp,+    QueueProcessor (..),+    mkProcessor,+    AppHandle (..),++    -- * AppHandle Operations+    getAppMetrics,+    getAppMaster,+    stopApp,+    stopAppGracefully,+    waitApp,++    -- * Shutdown Configuration+    ShutdownConfig (..),+    defaultShutdownConfig,++    -- * Supervision Strategy+    SupervisionStrategy (..),++    -- * Errors+    AppError (..),++    -- * Re-exports+    ProcessorId (..),+    ProcessorMetrics (..),+  )+where++import Control.Concurrent.NQE.Supervisor qualified as NQE+import Control.Concurrent.STM (STM, atomically, check, orElse, readTVar, registerDelay)+import Control.Monad (forM_, void)+import Data.Foldable (traverse_)+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.Text qualified as Text+import Data.Time.Clock (NominalDiffTime)+import Effectful (Eff, IOE, liftIO, (:>))+import GHC.Generics (Generic)+import Numeric.Natural (Natural)+import Shibuya.Adapter (Adapter (..))+import Shibuya.Core.Error (HandlerError (..), PolicyError (..), RuntimeError (..))+import Shibuya.Handler (Handler)+import Shibuya.Policy (Concurrency (..), Ordering (..), validatePolicy)+import Shibuya.Runner.Master+  ( Master,+    getAllMetrics,+    startMaster,+    stopMaster,+  )+import Shibuya.Runner.Metrics+  ( MetricsMap,+    ProcessorId (..),+    ProcessorMetrics (..),+  )+import Shibuya.Runner.Supervised+  ( SupervisedProcessor (..),+    runSupervised,+  )+import Shibuya.Telemetry.Effect (Tracing)+import UnliftIO (SomeException, catch, displayException)+import Prelude hiding (Ordering)++--------------------------------------------------------------------------------+-- Supervision Strategy+--------------------------------------------------------------------------------++-- | Supervision strategy for processor failures.+--+-- This is Shibuya's own type that maps to NQE's supervision strategies,+-- decoupling users from the NQE library.+data SupervisionStrategy+  = -- | Ignore all child exits, keep running.+    -- Failed processors are marked as Failed in metrics but don't affect others.+    IgnoreFailures+  | -- | Stop all processors if any fails.+    -- A single processor failure triggers shutdown of all processors.+    StopAllOnFailure+  deriving stock (Eq, Show, Generic)++-- | Convert Shibuya's strategy type to NQE's internal type.+toNQEStrategy :: SupervisionStrategy -> NQE.Strategy+toNQEStrategy = \case+  IgnoreFailures -> NQE.IgnoreAll+  StopAllOnFailure -> NQE.KillAll++--------------------------------------------------------------------------------+-- Shutdown Configuration+--------------------------------------------------------------------------------++-- | Configuration for graceful shutdown behavior.+data ShutdownConfig = ShutdownConfig+  { -- | Maximum time to wait for in-flight messages to drain.+    -- After this timeout, remaining processors are forcefully stopped.+    -- Default: 30 seconds.+    drainTimeout :: !NominalDiffTime+  }+  deriving stock (Eq, Show, Generic)++-- | Default shutdown configuration with 30 second drain timeout.+defaultShutdownConfig :: ShutdownConfig+defaultShutdownConfig = ShutdownConfig {drainTimeout = 30}++--------------------------------------------------------------------------------+-- Errors+--------------------------------------------------------------------------------++-- | Application errors.+-- Uses structured error types from Shibuya.Core.Error.+data AppError+  = -- | Invalid policy configuration+    AppPolicyError !PolicyError+  | -- | Handler execution error+    AppHandlerError !HandlerError+  | -- | Runtime error+    AppRuntimeError !RuntimeError+  deriving stock (Eq, Show)++-- | A queue processor pairs an adapter with its handler.+-- The message type is existentially hidden, allowing heterogeneous queues.+data QueueProcessor es where+  QueueProcessor ::+    { adapter :: Adapter es msg,+      handler :: Handler es msg,+      ordering :: Ordering,+      concurrency :: Concurrency+    } ->+    QueueProcessor es++-- | Convenience constructor with default policies (Unordered + Serial).+-- Provides backward compatibility with existing code.+mkProcessor :: Adapter es msg -> Handler es msg -> QueueProcessor es+mkProcessor adapter handler = QueueProcessor adapter handler Unordered Serial++-- | Handle for a running multi-queue application.+-- Provides introspection and control over all processors.+data AppHandle es = AppHandle+  { -- | The master coordinator+    master :: !Master,+    -- | Map of processor IDs to their handles+    processors :: !(Map ProcessorId (SupervisedProcessor, QueueProcessor es))+  }++-- | Run queue processors concurrently under NQE supervision.+--+-- Each processor runs independently. Returns immediately with a handle+-- for introspection and control.+--+-- Example:+--+-- @+-- result <- runApp IgnoreFailures 100+--   [ ("orders", QueueProcessor ordersAdapter ordersHandler)+--   , ("events", QueueProcessor eventsAdapter eventsHandler)+--   ]+-- @+runApp ::+  (IOE :> es, Tracing :> es) =>+  -- | Supervision strategy+  SupervisionStrategy ->+  -- | Inbox size for backpressure+  Int ->+  -- | Named processors+  [(ProcessorId, QueueProcessor es)] ->+  Eff es (Either AppError (AppHandle es))+runApp strategy inboxSize namedProcessors =+  -- Validate all policies first+  case validateAllPolicies namedProcessors of+    Left err -> pure $ Left $ AppPolicyError err+    Right () -> do+      let nqeStrategy = toNQEStrategy strategy+      catch+        ( do+            -- Start the master coordinator+            master <- startMaster nqeStrategy++            -- Spawn each processor under supervision+            processors <- spawnProcessors master (fromIntegral inboxSize) namedProcessors++            pure $+              Right+                AppHandle+                  { master = master,+                    processors = Map.fromList processors+                  }+        )+        ( \(e :: SomeException) ->+            pure $ Left $ AppRuntimeError $ SupervisorFailed $ Text.pack $ displayException e+        )++-- | Validate all processor policies before starting.+validateAllPolicies :: [(ProcessorId, QueueProcessor es)] -> Either PolicyError ()+validateAllPolicies = traverse_ validateOne+  where+    validateOne (_, QueueProcessor _ _ ord conc) = validatePolicy ord conc++-- | Spawn all processors under supervision.+spawnProcessors ::+  (IOE :> es, Tracing :> es) =>+  Master ->+  Natural ->+  [(ProcessorId, QueueProcessor es)] ->+  Eff es [(ProcessorId, (SupervisedProcessor, QueueProcessor es))]+spawnProcessors master inboxSize = traverse spawnOne+  where+    spawnOne (procId, qp@(QueueProcessor adapter handler _ordering concurrency)) = do+      sp <- runSupervised master inboxSize procId concurrency adapter handler+      pure (procId, (sp, qp))++--------------------------------------------------------------------------------+-- AppHandle Operations+--------------------------------------------------------------------------------++-- | Get metrics for all processors.+getAppMetrics :: (IOE :> es) => AppHandle es -> Eff es MetricsMap+getAppMetrics appHandle = getAllMetrics appHandle.master++-- | Get the master handle for direct access.+-- This is useful for integrating with the metrics server.+getAppMaster :: AppHandle es -> Master+getAppMaster appHandle = appHandle.master++-- | Gracefully stop all processors with default configuration.+-- Uses 'defaultShutdownConfig' (30 second drain timeout).+-- For custom timeout, use 'stopAppGracefully'.+stopApp :: (IOE :> es) => AppHandle es -> Eff es ()+stopApp = void . stopAppGracefully defaultShutdownConfig++-- | Gracefully stop all processors with configurable drain timeout.+--+-- Shutdown sequence:+-- 1. Signal all adapters to stop producing (close source streams)+-- 2. Wait for processors to drain in-flight messages (with timeout)+-- 3. Force stop any remaining processors after timeout+-- 4. Stop the master coordinator+--+-- Returns whether all processors drained cleanly (True) or were forced (False).+stopAppGracefully :: (IOE :> es) => ShutdownConfig -> AppHandle es -> Eff es Bool+stopAppGracefully config appHandle = do+  -- 1. Signal adapters to stop producing+  mapM_ shutdownAdapter (Map.elems appHandle.processors)++  -- 2. Wait for drain with timeout+  let timeoutMicros = floor (config.drainTimeout * 1_000_000)+  drained <- liftIO $ waitForDrainWithTimeout timeoutMicros (Map.elems appHandle.processors)++  -- 3. Log warning if forced shutdown (caller can check return value)+  -- Note: We don't log here to avoid IO dependencies, caller can log if needed++  -- 4. Stop master (cancels any remaining processors)+  stopMaster appHandle.master++  pure drained+  where+    shutdownAdapter (_, QueueProcessor adapter _ _ _) = adapter.shutdown++-- | Wait for all processors to be done, with timeout.+-- Returns True if all drained cleanly, False if timeout occurred.+-- Note: Requires -threaded RTS for registerDelay to work properly.+waitForDrainWithTimeout :: Int -> [(SupervisedProcessor, a)] -> IO Bool+waitForDrainWithTimeout timeoutMicros processors = do+  -- Create a timeout TVar that becomes True after the deadline+  timeoutVar <- registerDelay timeoutMicros++  -- Wait for either all done or timeout+  atomically $+    (allDone processors >> pure True)+      `orElse` (readTVar timeoutVar >>= check >> pure False)+  where+    allDone :: [(SupervisedProcessor, a)] -> STM ()+    allDone procs = forM_ procs $ \(sp, _) -> readTVar sp.done >>= check++-- | Wait for all processors to complete.+-- For infinite streams, this will block forever.+-- Use 'stopApp' to gracefully terminate.+--+-- Uses STM to block efficiently until all processors are done,+-- rather than polling.+waitApp :: (IOE :> es) => AppHandle es -> Eff es ()+waitApp appHandle = liftIO $ atomically $ do+  -- Block until all done TVars are True+  forM_ (Map.elems appHandle.processors) $ \(sp, _) ->+    readTVar sp.done >>= check
+ src/Shibuya/Core.hs view
@@ -0,0 +1,94 @@+-- | Shibuya Core - Public API+--+-- This module re-exports all public types and functions from the Shibuya framework.+-- Import this module for application development.+--+-- Example:+--+-- @+-- import Shibuya.Core+--+-- main = runEff $ do+--   let processor = QueueProcessor myAdapter myHandler+--   result <- runApp IgnoreFailures 100 [(ProcessorId "main", processor)]+--   case result of+--     Right handle -> waitApp handle+--     Left err -> print err+-- @+module Shibuya.Core+  ( -- * Message Types+    MessageId (..),+    Cursor (..),+    Envelope (..),++    -- * Ack Semantics+    AckDecision (..),+    RetryDelay (..),+    DeadLetterReason (..),+    HaltReason (..),++    -- * Handler Types+    AckHandle (..),+    Lease (..),+    Ingested (..),++    -- * Handler+    Handler,++    -- * Adapter+    Adapter (..),++    -- * Policy+    Ordering (..),+    Concurrency (..),+    validatePolicy,++    -- * App+    runApp,+    AppError (..),+    QueueProcessor (..),+    mkProcessor,+    AppHandle (..),+    waitApp,+    stopApp,+    stopAppGracefully,+    getAppMetrics,++    -- * Shutdown Configuration+    ShutdownConfig (..),+    defaultShutdownConfig,++    -- * Supervision Strategy+    SupervisionStrategy (..),++    -- * Errors+    PolicyError (..),+    HandlerError (..),+    RuntimeError (..),++    -- * Metrics+    ProcessorId (..),+    ProcessorState (..),+    ProcessorMetrics (..),+    StreamStats (..),+    InFlightInfo (..),+    MetricsMap,++    -- * Halt+    ProcessorHalt (..),+  )+where++import Shibuya.Adapter (Adapter (..))+import Shibuya.App (AppError (..), AppHandle (..), QueueProcessor (..), ShutdownConfig (..), SupervisionStrategy (..), defaultShutdownConfig, getAppMetrics, mkProcessor, runApp, stopApp, stopAppGracefully, waitApp)+import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..), RetryDelay (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Error (HandlerError (..), PolicyError (..), RuntimeError (..))+import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Lease (Lease (..))+import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..))+import Shibuya.Handler (Handler)+import Shibuya.Policy (Concurrency (..), Ordering (..), validatePolicy)+import Shibuya.Runner.Halt (ProcessorHalt (..))+import Shibuya.Runner.Metrics (InFlightInfo (..), MetricsMap, ProcessorId (..), ProcessorMetrics (..), ProcessorState (..), StreamStats (..))+import Prelude hiding (Ordering)
+ src/Shibuya/Core/Ack.hs view
@@ -0,0 +1,54 @@+-- | Ack semantics for message processing.+-- Handlers decide meaning, not mechanics.+-- This is explicit to support halt-on-error for ordered streams.+module Shibuya.Core.Ack+  ( -- * Retry+    RetryDelay (..),++    -- * Dead Letter+    DeadLetterReason (..),++    -- * Halt+    HaltReason (..),++    -- * Handler Decision+    AckDecision (..),+  )+where++import Shibuya.Prelude++-- | Delay before retry.+newtype RetryDelay = RetryDelay {unRetryDelay :: NominalDiffTime}+  deriving stock (Eq, Show)++-- | Why a message is being dead-lettered.+data DeadLetterReason+  = -- | Message is permanently unprocessable+    PoisonPill !Text+  | -- | Message payload failed validation/parsing+    InvalidPayload !Text+  | -- | Retry limit exceeded+    MaxRetriesExceeded+  deriving stock (Eq, Show, Generic)++-- | Why processing should halt.+data HaltReason+  = -- | Must stop to preserve ordering guarantees+    HaltOrderedStream !Text+  | -- | Unrecoverable error+    HaltFatal !Text+  deriving stock (Eq, Show, Generic)++-- | Handler outcome (semantic, not mechanical).+-- The handler returns this to express intent; the framework handles the mechanics.+data AckDecision+  = -- | Message processed successfully+    AckOk+  | -- | Retry after delay+    AckRetry !RetryDelay+  | -- | Move to dead letter queue+    AckDeadLetter !DeadLetterReason+  | -- | Stop processing+    AckHalt !HaltReason+  deriving stock (Eq, Show, Generic)
+ src/Shibuya/Core/AckHandle.hs view
@@ -0,0 +1,18 @@+-- | AckHandle for mechanical acknowledgment.+-- This is the Broadway.Acknowledger equivalent, but typed.+-- Must be called exactly once; adapter enforces idempotency.+module Shibuya.Core.AckHandle+  ( AckHandle (..),+  )+where++import Effectful (Eff)+import Shibuya.Core.Ack (AckDecision)++-- | Mechanical ack interface (adapter-provided).+-- The handler's AckDecision is passed to finalize, which performs+-- the actual commit/retry/dead-letter operation.+newtype AckHandle es = AckHandle+  { -- | Finalize the message with the given decision+    finalize :: AckDecision -> Eff es ()+  }
+ src/Shibuya/Core/Error.hs view
@@ -0,0 +1,54 @@+-- | Unified error types for Shibuya.+-- Provides structured errors with consistent representation across the library.+module Shibuya.Core.Error+  ( -- * Policy Errors+    PolicyError (..),+    policyErrorToText,++    -- * Handler Errors+    HandlerError (..),+    handlerErrorToText,++    -- * Runtime Errors+    RuntimeError (..),+    runtimeErrorToText,+  )+where++import Shibuya.Prelude++-- | Policy validation errors.+data PolicyError+  = -- | Invalid combination of ordering and concurrency+    InvalidPolicyCombo !Text+  deriving stock (Eq, Show, Generic)++-- | Convert policy error to text for display.+policyErrorToText :: PolicyError -> Text+policyErrorToText (InvalidPolicyCombo msg) = msg++-- | Handler execution errors.+data HandlerError+  = -- | Handler threw an exception+    HandlerException !Text+  | -- | Handler timed out+    HandlerTimeout+  deriving stock (Eq, Show, Generic)++-- | Convert handler error to text for display.+handlerErrorToText :: HandlerError -> Text+handlerErrorToText (HandlerException msg) = msg+handlerErrorToText HandlerTimeout = "Handler timed out"++-- | Runtime errors during processing.+data RuntimeError+  = -- | Supervisor failed+    SupervisorFailed !Text+  | -- | Inbox overflow (backpressure failure)+    InboxOverflow+  deriving stock (Eq, Show, Generic)++-- | Convert runtime error to text for display.+runtimeErrorToText :: RuntimeError -> Text+runtimeErrorToText (SupervisorFailed msg) = msg+runtimeErrorToText InboxOverflow = "Inbox overflow"
+ src/Shibuya/Core/Ingested.hs view
@@ -0,0 +1,22 @@+-- | Ingested message type - what handlers receive.+-- Combines Broadway.Message + Acknowledger + optional lease.+-- Exactly one thing flows through the system.+module Shibuya.Core.Ingested+  ( Ingested (..),+  )+where++import Shibuya.Core.AckHandle (AckHandle)+import Shibuya.Core.Lease (Lease)+import Shibuya.Core.Types (Envelope)++-- | What handlers receive for processing.+-- Contains the message envelope, ack handle, and optional lease.+data Ingested es msg = Ingested+  { -- | Message metadata and payload+    envelope :: !(Envelope msg),+    -- | Handle for acknowledging the message+    ack :: !(AckHandle es),+    -- | Optional lease for visibility timeout extension+    lease :: !(Maybe (Lease es))+  }
+ src/Shibuya/Core/Lease.hs view
@@ -0,0 +1,20 @@+-- | Lease types for temporary message ownership.+-- Lease exists only to model temporary ownership (SQS visibility timeout, DB locks).+-- Kafka adapters won't use this; SQS/Redis/DB queues will.+module Shibuya.Core.Lease+  ( Lease (..),+  )+where++import Data.Text (Text)+import Data.Time (NominalDiffTime)+import Effectful (Eff)++-- | Optional capability for sources with visibility/ownership.+-- Allows handlers to extend the lease if processing takes longer than expected.+data Lease es = Lease+  { -- | Identifier for the lease+    leaseId :: !Text,+    -- | Extend the lease by the given duration+    leaseExtend :: NominalDiffTime -> Eff es ()+  }
+ src/Shibuya/Core/Types.hs view
@@ -0,0 +1,56 @@+-- | Core types for the Shibuya framework.+-- These types exist everywhere and should be extremely stable.+-- No behavior, no effects, no policy - just identity + payload + metadata.+module Shibuya.Core.Types+  ( -- * Message Identity+    MessageId (..),++    -- * Cursor / Offset+    Cursor (..),++    -- * Message Envelope+    Envelope (..),++    -- * Trace Context+    TraceHeaders,+  )+where++import Data.ByteString (ByteString)+import Data.String (IsString)+import Shibuya.Prelude++-- | 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)++-- | Optional cursor / offset / global position.+-- Used to track position in ordered streams.+data Cursor+  = CursorInt !Int+  | CursorText !Text+  deriving stock (Eq, Ord, Show, Generic)++-- | W3C Trace Context headers for distributed tracing.+-- Contains traceparent and optionally tracestate headers.+type TraceHeaders = [(ByteString, ByteString)]++-- | Normalized message envelope (Broadway.Message equivalent).+-- Contains message metadata plus the payload.+data Envelope msg = Envelope+  { -- | Unique message identifier+    messageId :: !MessageId,+    -- | Optional position/offset+    cursor :: !(Maybe Cursor),+    -- | Optional partition key (for Kafka-style queues)+    partition :: !(Maybe Text),+    -- | When the message was enqueued+    enqueuedAt :: !(Maybe UTCTime),+    -- | W3C trace context headers for distributed tracing+    traceContext :: !(Maybe TraceHeaders),+    -- | The actual message payload+    payload :: !msg+  }+  deriving stock (Eq, Show, Functor, Generic)
+ src/Shibuya/Handler.hs view
@@ -0,0 +1,15 @@+-- | Handler API - this is all application authors need to know.+-- Handlers cannot ack directly, cannot influence concurrency,+-- and express intent only via AckDecision.+module Shibuya.Handler+  ( Handler,+  )+where++import Effectful (Eff)+import Shibuya.Core.Ack (AckDecision)+import Shibuya.Core.Ingested (Ingested)++-- | Handler function type.+-- Takes an ingested message and returns an ack decision.+type Handler es msg = Ingested es msg -> Eff es AckDecision
+ src/Shibuya/Policy.hs view
@@ -0,0 +1,44 @@+-- | Ordering and concurrency policies.+-- Runner policy that maps ordering guarantees to concurrency constraints.+module Shibuya.Policy+  ( -- * Ordering+    Ordering (..),++    -- * Concurrency+    Concurrency (..),++    -- * Validation+    validatePolicy,+  )+where++import Shibuya.Core.Error (PolicyError (..))+import Shibuya.Prelude+import Prelude hiding (Ordering)++-- | Message ordering guarantees.+data Ordering+  = -- | Event-sourced subscriptions - must be Serial+    StrictInOrder+  | -- | Kafka-style - parallel across partitions+    PartitionedInOrder+  | -- | No ordering guarantees+    Unordered+  deriving stock (Eq, Show, Generic)++-- | Concurrency mode.+data Concurrency+  = -- | One message at a time+    Serial+  | -- | Prefetch N, process in order+    Ahead !Int+  | -- | Process N concurrently+    Async !Int+  deriving stock (Eq, Show, Generic)++-- | Validate policy combinations.+-- Invariant: StrictInOrder => Serial+validatePolicy :: Ordering -> Concurrency -> Either PolicyError ()+validatePolicy StrictInOrder (Ahead _) = Left $ InvalidPolicyCombo "StrictInOrder requires Serial concurrency"+validatePolicy StrictInOrder (Async _) = Left $ InvalidPolicyCombo "StrictInOrder requires Serial concurrency"+validatePolicy _ _ = Right ()
+ src/Shibuya/Prelude.hs view
@@ -0,0 +1,30 @@+{-# LANGUAGE PackageImports #-}++module Shibuya.Prelude+  ( -- base+    Generic,+    MonadIO,+    Natural,+    -- text+    Text,+    -- time+    UTCTime,+    Day,+    LocalTime,+    NominalDiffTime,+    getCurrentTime,+    -- lens+    module Control.Lens,+    -- vector+    Vector,+  )+where++import "base" Control.Monad.IO.Class (MonadIO)+import "base" GHC.Generics (Generic)+import "base" Numeric.Natural (Natural)+import "generic-lens" Data.Generics.Labels ()+import "lens" Control.Lens+import "text" Data.Text (Text)+import "time" Data.Time (Day, LocalTime, NominalDiffTime, UTCTime, getCurrentTime)+import "vector" Data.Vector (Vector)
+ src/Shibuya/Runner/Halt.hs view
@@ -0,0 +1,19 @@+-- | Halt exception for processor termination.+-- Thrown when a handler returns AckHalt to stop processing.+module Shibuya.Runner.Halt+  ( ProcessorHalt (..),+  )+where++import Control.Exception (Exception)+import Shibuya.Core.Ack (HaltReason)+import Shibuya.Prelude++-- | Exception thrown when processing should halt.+-- The supervisor catches this to handle graceful shutdown.+data ProcessorHalt = ProcessorHalt+  { reason :: !HaltReason+  }+  deriving stock (Show, Generic)++instance Exception ProcessorHalt
+ src/Shibuya/Runner/Ingester.hs view
@@ -0,0 +1,60 @@+-- | Ingester - reads from adapter stream and sends to inbox.+-- Provides backpressure via bounded inbox.+module Shibuya.Runner.Ingester+  ( runIngester,+    runIngesterWithMetrics,+  )+where++import Control.Concurrent.NQE.Process (Inbox, inboxToMailbox, send)+import Control.Concurrent.STM (TVar, atomically, modifyTVar')+import Effectful (Eff, IOE, liftIO, (:>))+import Shibuya.Core.Ingested (Ingested)+import Shibuya.Runner.Metrics (ProcessorMetrics (..), incReceived)+import Streamly.Data.Fold qualified as Fold+import Streamly.Data.Stream (Stream)+import Streamly.Data.Stream qualified as Stream++-- | Run the ingester: reads from stream, sends to inbox.+-- Sends each element to the inbox with backpressure - if inbox is full, blocks.+-- Completes when the stream completes.+runIngester ::+  (IOE :> es) =>+  -- | Source stream from adapter+  Stream (Eff es) (Ingested es msg) ->+  -- | Target inbox (bounded for backpressure)+  Inbox (Ingested es msg) ->+  Eff es ()+runIngester source inbox = do+  let mailbox = inboxToMailbox inbox+  -- Send each stream element to the mailbox, then drain+  -- This provides backpressure: if mailbox is full (bounded), send blocks+  Stream.fold Fold.drain $+    Stream.mapM+      (\msg -> send msg mailbox >> pure msg)+      source++-- | Run the ingester with metrics tracking.+-- Increments 'received' count for each message sent to inbox.+runIngesterWithMetrics ::+  (IOE :> es) =>+  -- | Metrics TVar (for updating received count)+  TVar ProcessorMetrics ->+  -- | Source stream from adapter+  Stream (Eff es) (Ingested es msg) ->+  -- | Target inbox (bounded for backpressure)+  Inbox (Ingested es msg) ->+  Eff es ()+runIngesterWithMetrics metricsVar source inbox = do+  let mailbox = inboxToMailbox inbox+  Stream.fold Fold.drain $+    Stream.mapM+      ( \msg -> do+          -- Increment received count+          liftIO $ atomically $ modifyTVar' metricsVar $ \m ->+            m {stats = incReceived m.stats}+          -- Send to inbox (blocks if full - backpressure)+          liftIO $ send msg mailbox+          pure msg+      )+      source
+ src/Shibuya/Runner/Master.hs view
@@ -0,0 +1,195 @@+-- | Master process - central coordinator for queue processors.+-- Provides supervision, metrics collection, and control API.+--+-- Architecture:+-- - Master is an NQE Process that handles control messages+-- - Holds a Supervisor for managing child processors+-- - Maintains TVar MetricsMap for O(1) metrics access+-- - Processors register their metrics TVars with the Master+module Shibuya.Runner.Master+  ( -- * Master Handle+    Master (..),+    MasterState (..),++    -- * Control Messages+    MasterMessage (..),++    -- * Starting the Master+    startMaster,+    stopMaster,++    -- * Introspection+    getAllMetrics,+    getAllMetricsIO,+    getProcessorMetrics,+    getProcessorMetricsIO,++    -- * Processor Management+    registerProcessor,+    unregisterProcessor,+  )+where++import Control.Concurrent.NQE.Process+  ( Inbox,+    Listen,+    Process (..),+    newInbox,+    query,+    receive,+  )+import Control.Concurrent.NQE.Supervisor (Strategy (..), Supervisor)+import Control.Concurrent.NQE.Supervisor qualified as Supervisor+import Control.Concurrent.STM+  ( TVar,+    atomically,+    modifyTVar',+    newTVarIO,+    readTVar,+  )+import Control.Monad (forever)+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Effectful (Eff, IOE, liftIO, (:>))+import Shibuya.Prelude+import Shibuya.Runner.Metrics+  ( MetricsMap,+    ProcessorId,+    ProcessorMetrics,+  )+import UnliftIO (Async, async, cancel, link)++-- | Messages for the master process.+data MasterMessage+  = -- | Get metrics for all processors+    GetAllMetrics !(Listen MetricsMap)+  | -- | Get metrics for a specific processor+    GetProcessorMetrics ProcessorId !(Listen (Maybe ProcessorMetrics))+  | -- | Register a processor's metrics TVar+    RegisterProcessor !ProcessorId !(TVar ProcessorMetrics) !(Listen ())+  | -- | Unregister a processor+    UnregisterProcessor !ProcessorId !(Listen ())+  | -- | Shutdown all processors+    Shutdown !(Listen ())++-- | Master state held in TVars.+data MasterState = MasterState+  { -- | Map of processor IDs to their metrics TVars+    metrics :: !(TVar (Map ProcessorId (TVar ProcessorMetrics))),+    -- | The supervisor managing child processors+    supervisor :: !Supervisor+  }+  deriving (Generic)++-- | Master handle - provides access to the master process.+data Master = Master+  { -- | The async handle for the master+    handle :: !(Async ()),+    -- | Direct access to master state+    state :: !MasterState,+    -- | Inbox for sending messages+    inbox :: !(Inbox MasterMessage)+  }+  deriving (Generic)++-- | Start the master process.+-- Returns a handle for interacting with the master.+-- The caller is responsible for calling stopMaster when done.+startMaster :: (IOE :> es) => Strategy -> Eff es Master+startMaster strategy = liftIO $ do+  -- Create supervisor+  sup <- Supervisor.supervisor strategy++  metricsMapVar <- newTVarIO Map.empty+  let masterState = MasterState metricsMapVar sup++  masterInbox <- newInbox++  -- Start master loop+  masterHandle <- async $ masterLoop masterState masterInbox+  link masterHandle++  pure+    Master+      { handle = masterHandle,+        state = masterState,+        inbox = masterInbox+      }++-- | Stop the master and all child processors.+-- Cancels the supervisor first (which cancels all children via NQE's stopAll),+-- then cancels the master message loop.+stopMaster :: (IOE :> es) => Master -> Eff es ()+stopMaster master = liftIO $ do+  -- Cancel the supervisor first - this triggers NQE's stopAll which cancels all children+  cancel (getProcessAsync master.state.supervisor)+  -- Then cancel the master message loop+  cancel (master ^. #handle)++-- | The master process main loop.+masterLoop :: MasterState -> Inbox MasterMessage -> IO ()+masterLoop state inbox = forever $ do+  msg <- receive inbox+  handleMessage state msg++-- | Handle a single master message.+handleMessage :: MasterState -> MasterMessage -> IO ()+handleMessage state msg = case msg of+  GetAllMetrics respond -> do+    -- Read all metrics from their TVars+    metricsMap <- atomically $ do+      tvarsMap <- readTVar state.metrics+      traverse readTVar tvarsMap+    atomically $ respond metricsMap+  GetProcessorMetrics pid respond -> do+    result <- atomically $ do+      tvarsMap <- readTVar state.metrics+      case Map.lookup pid tvarsMap of+        Nothing -> pure Nothing+        Just tvar -> Just <$> readTVar tvar+    atomically $ respond result+  RegisterProcessor pid metricsTVar respond -> do+    atomically $ do+      modifyTVar' state.metrics $ Map.insert pid metricsTVar+    atomically $ respond ()+  UnregisterProcessor pid respond -> do+    atomically $ do+      modifyTVar' state.metrics $ Map.delete pid+    atomically $ respond ()+  Shutdown respond -> do+    -- Clear all processors+    atomically $ modifyTVar' state.metrics (const Map.empty)+    atomically $ respond ()++-- | Get metrics for all processors.+getAllMetrics :: (IOE :> es) => Master -> Eff es MetricsMap+getAllMetrics master =+  liftIO $+    GetAllMetrics `query` master.inbox++-- | Get metrics for all processors (IO version for web servers).+getAllMetricsIO :: Master -> IO MetricsMap+getAllMetricsIO master = GetAllMetrics `query` master.inbox++-- | Get metrics for a specific processor.+getProcessorMetrics :: (IOE :> es) => Master -> ProcessorId -> Eff es (Maybe ProcessorMetrics)+getProcessorMetrics master pid =+  liftIO $+    GetProcessorMetrics pid `query` master.inbox++-- | Get metrics for a specific processor (IO version for web servers).+getProcessorMetricsIO :: Master -> ProcessorId -> IO (Maybe ProcessorMetrics)+getProcessorMetricsIO master pid = GetProcessorMetrics pid `query` master.inbox++-- | Register a processor with the master.+-- The processor should call this with its metrics TVar.+registerProcessor :: (IOE :> es) => Master -> ProcessorId -> TVar ProcessorMetrics -> Eff es ()+registerProcessor master pid metricsTVar =+  liftIO $+    RegisterProcessor pid metricsTVar `query` master.inbox++-- | Unregister a processor from the master.+unregisterProcessor :: (IOE :> es) => Master -> ProcessorId -> Eff es ()+unregisterProcessor master pid =+  liftIO $+    UnregisterProcessor pid `query` master.inbox
+ src/Shibuya/Runner/Metrics.hs view
@@ -0,0 +1,167 @@+-- | Metrics and state tracking for processors.+-- Provides introspection into what's happening in the system.+module Shibuya.Runner.Metrics+  ( -- * Processor State+    ProcessorState (..),+    ProcessorId (..),++    -- * In-Flight Tracking+    InFlightInfo (..),+    emptyInFlightInfo,++    -- * Stream Statistics+    StreamStats (..),+    emptyStreamStats,++    -- * Combined Metrics+    ProcessorMetrics (..),+    emptyProcessorMetrics,++    -- * Metrics Map+    MetricsMap,++    -- * Metrics Updates+    incReceived,+    incDropped,+    incProcessed,+    incFailed,+  )+where++import Data.Aeson (FromJSON (..), FromJSONKey (..), ToJSON (..), ToJSONKey (..), object, withObject, (.:))+import Data.Aeson qualified as Aeson+import Data.Map.Strict (Map)+import Data.Text qualified as Text+import Shibuya.Prelude++-- | Processor identifier.+newtype ProcessorId = ProcessorId {unProcessorId :: Text}+  deriving stock (Eq, Ord, Show, Generic)+  deriving newtype (ToJSON, FromJSON, ToJSONKey, FromJSONKey)++-- | Tracks concurrent in-flight messages.+data InFlightInfo = InFlightInfo+  { -- | Currently processing count+    inFlight :: !Int,+    -- | Configured max concurrency (1 for Serial)+    maxConcurrency :: !Int+  }+  deriving stock (Eq, Show, Generic)++instance ToJSON InFlightInfo where+  toJSON info =+    object+      [ "inFlight" Aeson..= info.inFlight,+        "maxConcurrency" Aeson..= info.maxConcurrency+      ]++instance FromJSON InFlightInfo where+  parseJSON = withObject "InFlightInfo" $ \v ->+    InFlightInfo+      <$> v .: "inFlight"+      <*> v .: "maxConcurrency"++-- | Create empty in-flight info with given max concurrency.+emptyInFlightInfo :: Int -> InFlightInfo+emptyInFlightInfo = InFlightInfo 0++-- | Processor runtime state.+data ProcessorState+  = -- | Waiting for messages+    Idle+  | -- | Currently processing (in-flight info, last activity time)+    Processing !InFlightInfo !UTCTime+  | -- | Failed with error (error message, timestamp)+    Failed !Text !UTCTime+  | -- | Processor has been stopped+    Stopped+  deriving stock (Eq, Show, Generic)++instance ToJSON ProcessorState where+  toJSON Idle = object ["status" Aeson..= ("idle" :: Text)]+  toJSON (Processing info lastActivity) =+    object+      [ "status" Aeson..= ("processing" :: Text),+        "inFlight" Aeson..= info.inFlight,+        "maxConcurrency" Aeson..= info.maxConcurrency,+        "lastActivity" Aeson..= lastActivity+      ]+  toJSON (Failed err timestamp) =+    object+      [ "status" Aeson..= ("failed" :: Text),+        "error" Aeson..= err,+        "timestamp" Aeson..= timestamp+      ]+  toJSON Stopped = object ["status" Aeson..= ("stopped" :: Text)]++instance FromJSON ProcessorState where+  parseJSON = withObject "ProcessorState" $ \v -> do+    status <- v .: "status"+    case status :: Text of+      "idle" -> pure Idle+      "processing" -> do+        inFlightCount <- v .: "inFlight"+        maxConc <- v .: "maxConcurrency"+        lastActivity <- v .: "lastActivity"+        pure $ Processing (InFlightInfo inFlightCount maxConc) lastActivity+      "failed" -> Failed <$> v .: "error" <*> v .: "timestamp"+      "stopped" -> pure Stopped+      other -> fail $ "Unknown processor state: " <> Text.unpack other++-- | Stream statistics.+data StreamStats = StreamStats+  { -- | Messages received from stream+    received :: !Int,+    -- | Messages dropped (backpressure)+    dropped :: !Int,+    -- | Messages successfully processed+    processed :: !Int,+    -- | Messages that failed processing+    failed :: !Int+  }+  deriving stock (Eq, Show, Generic)+  deriving anyclass (ToJSON, FromJSON)++-- | Empty stream stats.+emptyStreamStats :: StreamStats+emptyStreamStats = StreamStats 0 0 0 0++-- | Combined processor metrics.+data ProcessorMetrics = ProcessorMetrics+  { -- | Current state+    state :: !ProcessorState,+    -- | Statistics+    stats :: !StreamStats,+    -- | When the processor started+    startedAt :: !UTCTime+  }+  deriving stock (Eq, Show, Generic)+  deriving anyclass (ToJSON, FromJSON)++-- | Empty processor metrics.+emptyProcessorMetrics :: UTCTime -> ProcessorMetrics+emptyProcessorMetrics now =+  ProcessorMetrics+    { state = Idle,+      stats = emptyStreamStats,+      startedAt = now+    }++-- | Map of processor IDs to their metrics.+type MetricsMap = Map ProcessorId ProcessorMetrics++-- | Increment received count.+incReceived :: StreamStats -> StreamStats+incReceived = #received %~ (+ 1)++-- | Increment dropped count.+incDropped :: StreamStats -> StreamStats+incDropped = #dropped %~ (+ 1)++-- | Increment processed count.+incProcessed :: StreamStats -> StreamStats+incProcessed = #processed %~ (+ 1)++-- | Increment failed count.+incFailed :: StreamStats -> StreamStats+incFailed = #failed %~ (+ 1)
+ src/Shibuya/Runner/Processor.hs view
@@ -0,0 +1,78 @@+-- | Processor - receives from inbox, calls handler, finalizes ack.+-- Runs in a loop until terminated.+module Shibuya.Runner.Processor+  ( runProcessor,+    runProcessorN,+    drainInbox,+  )+where++import Control.Concurrent.NQE.Process (Inbox, mailboxEmpty, receive)+import Control.Monad (forever, unless)+import Effectful (Eff, IOE, (:>))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Handler (Handler)++-- | Run the processor forever: receive message, call handler, finalize ack.+-- This runs until an exception is thrown or the thread is cancelled.+runProcessor ::+  (IOE :> es) =>+  -- | Message handler+  Handler es msg ->+  -- | Source inbox+  Inbox (Ingested es msg) ->+  Eff es ()+runProcessor handler inbox = forever $ processOne handler inbox++-- | Run the processor for exactly N messages.+-- Useful for testing.+runProcessorN ::+  (IOE :> es) =>+  -- | Number of messages to process+  Int ->+  -- | Message handler+  Handler es msg ->+  -- | Source inbox+  Inbox (Ingested es msg) ->+  Eff es ()+runProcessorN n handler inbox = go n+  where+    go 0 = pure ()+    go remaining = do+      processOne handler inbox+      go (remaining - 1)++-- | Process a single message: receive, handle, finalize.+processOne ::+  (IOE :> es) =>+  Handler es msg ->+  Inbox (Ingested es msg) ->+  Eff es ()+processOne handler inbox = do+  -- Receive next message (blocks if inbox empty)+  ingested <- receive inbox++  -- Call the handler to get the ack decision+  decision <- handler ingested++  -- Finalize the ack (commit, retry, dead-letter, or halt)+  -- Note: AckHalt is handled by the supervised runner (Supervised.hs throws ProcessorHalt)+  ingested.ack.finalize decision++-- | Drain all remaining messages from inbox until empty.+-- Used after ingester completes to process buffered messages.+drainInbox ::+  (IOE :> es) =>+  -- | Message handler+  Handler es msg ->+  -- | Source inbox+  Inbox (Ingested es msg) ->+  Eff es ()+drainInbox handler inbox = go+  where+    go = do+      empty <- mailboxEmpty inbox+      unless empty $ do+        processOne handler inbox+        go
+ src/Shibuya/Runner/Serial.hs view
@@ -0,0 +1,44 @@+-- | Serial Runner - processes messages one at a time.+-- Simple sequential runner for ordered processing.+module Shibuya.Runner.Serial+  ( runSerial,+  )+where++import Effectful (Eff, IOE, (:>))+import Numeric.Natural (Natural)+import Shibuya.Adapter (Adapter (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Handler (Handler)+import Streamly.Data.Fold qualified as Fold+import Streamly.Data.Stream qualified as Stream++-- | Run the serial runner.+-- Processes each message from the adapter stream sequentially:+-- 1. Pull message from stream+-- 2. Call handler+-- 3. Finalize ack+-- 4. Repeat until stream is exhausted+--+-- This is the simplest runner - no concurrency, no backpressure buffering.+-- Suitable for ordered, low-throughput workloads.+runSerial ::+  (IOE :> es) =>+  -- | Inbox size (unused in serial mode, for API compatibility)+  Natural ->+  -- | Queue adapter+  Adapter es msg ->+  -- | Message handler+  Handler es msg ->+  Eff es ()+runSerial _inboxSize adapter handler = do+  -- Process each message from the stream sequentially+  Stream.fold Fold.drain $+    Stream.mapM processOne adapter.source+  where+    processOne ingested = do+      -- Call the handler to get the ack decision+      decision <- handler ingested+      -- Finalize the ack+      ingested.ack.finalize decision
+ src/Shibuya/Runner/Supervised.hs view
@@ -0,0 +1,488 @@+-- | Supervised runner - runs processors under NQE supervision with metrics.+-- This is the production runner with introspection and control.+--+-- Architecture:+-- - Each processor runs as a child under the Master's supervisor+-- - Metrics are tracked in a TVar and registered with the Master+-- - Backpressure is provided via bounded inbox between stream and processor+module Shibuya.Runner.Supervised+  ( -- * Running with Supervision+    runSupervised,++    -- * Standalone (without Master)+    runWithMetrics,++    -- * Processor Handle+    SupervisedProcessor (..),++    -- * Introspection+    getMetrics,+    getProcessorState,+    isDone,+  )+where++import Control.Concurrent.NQE.Process (Inbox, mailboxEmpty, mailboxEmptySTM, newBoundedInbox, receive, receiveSTM)+import Control.Concurrent.NQE.Supervisor (addChild)+import Control.Concurrent.STM+  ( TVar,+    atomically,+    modifyTVar',+    newTVarIO,+    orElse,+    readTVar,+    readTVarIO,+    retry,+    writeTVar,+  )+import Control.Monad (unless)+import Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef)+import Data.Text qualified as Text+import Effectful (Eff, IOE, liftIO, withEffToIO, (:>))+import Effectful.Dispatch.Static (unsafeEff_)+import Effectful.Internal.Unlift (Limit (..), Persistence (..), UnliftStrategy (..))+import OpenTelemetry.Trace.Core qualified as OTel+import Shibuya.Adapter (Adapter (..))+import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Error (HandlerError (..), handlerErrorToText)+import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Types (Envelope (..), MessageId (..))+import Shibuya.Handler (Handler)+import Shibuya.Policy (Concurrency (..))+import Shibuya.Prelude+import Shibuya.Runner.Halt (ProcessorHalt (..))+import Shibuya.Runner.Ingester (runIngesterWithMetrics)+import Shibuya.Runner.Master (Master (..), MasterState (..), registerProcessor, unregisterProcessor)+import Shibuya.Runner.Metrics+  ( InFlightInfo (..),+    ProcessorId (..),+    ProcessorMetrics (..),+    ProcessorState (..),+    emptyProcessorMetrics,+    incFailed,+    incProcessed,+  )+import Shibuya.Telemetry.Effect+  ( Tracing,+    addAttribute,+    addEvent,+    recordException,+    setStatus,+    withExtractedContext,+    withSpan',+  )+import Shibuya.Telemetry.Propagation (extractTraceContext)+import Shibuya.Telemetry.Semantic+  ( attrMessagingDestinationPartitionId,+    attrMessagingMessageId,+    attrMessagingSystem,+    attrShibuyaAckDecision,+    attrShibuyaInflightCount,+    attrShibuyaInflightMax,+    consumerSpanArgs,+    eventAckDecision,+    eventHandlerCompleted,+    eventHandlerException,+    eventHandlerStarted,+    mkEvent,+    processMessageSpanName,+  )+import Streamly.Data.Fold qualified as Fold+import Streamly.Data.Stream qualified as Stream+import Streamly.Data.Stream.Prelude qualified as StreamP+import UnliftIO (Async, catch, catchAny, finally, throwIO)+import UnliftIO qualified as UIO++-- | Handle for a supervised processor.+-- Provides introspection into the running processor.+data SupervisedProcessor = SupervisedProcessor+  { -- | Live metrics for this processor+    metrics :: !(TVar ProcessorMetrics),+    -- | The processor's ID+    processorId :: !ProcessorId,+    -- | Whether processing is complete+    done :: !(TVar Bool),+    -- | The async handle if running under supervision+    child :: !(Maybe (Async ()))+  }++-- | Get current metrics for the processor.+getMetrics :: (IOE :> es) => SupervisedProcessor -> Eff es ProcessorMetrics+getMetrics sp = liftIO $ readTVarIO sp.metrics++-- | Get current state of the processor.+getProcessorState :: (IOE :> es) => SupervisedProcessor -> Eff es ProcessorState+getProcessorState sp = liftIO $ (.state) <$> readTVarIO sp.metrics++-- | Check if processing is done.+isDone :: (IOE :> es) => SupervisedProcessor -> Eff es Bool+isDone sp = liftIO $ readTVarIO sp.done++-- | Run a processor under the Master's supervision with metrics tracking.+--+-- Architecture:+-- 1. Creates a bounded inbox (using inboxSize for backpressure)+-- 2. Spawns ingester async (reads from adapter, sends to inbox)+-- 3. Runs processor loop (receives from inbox, calls handler)+-- 4. Registers metrics with Master, unregisters on completion+--+-- Returns immediately with a handle for introspection.+runSupervised ::+  (IOE :> es, Tracing :> es) =>+  Master ->+  -- | Inbox size (for backpressure)+  Natural ->+  -- | Processor identifier+  ProcessorId ->+  -- | Concurrency mode+  Concurrency ->+  -- | Queue adapter+  Adapter es msg ->+  -- | Message handler+  Handler es msg ->+  Eff es SupervisedProcessor+runSupervised master inboxSize procId concurrency adapter handler = do+  now <- liftIO getCurrentTime++  -- Initialize state+  let initialMetrics = emptyProcessorMetrics now+  metricsVar <- liftIO $ newTVarIO initialMetrics+  doneVar <- liftIO $ newTVarIO False++  -- Register with Master+  registerProcessor master procId metricsVar++  -- Add as supervised child using NQE's Supervisor+  -- ConcUnlift Persistent allows the runInIO function to be used in the async child+  supervisedChild <- withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->+    addChild master.state.supervisor $+      runInIO $+        -- Catch ProcessorHalt to prevent propagation via link+        -- (Halt is intentional, not a failure - other processors should continue)+        ( runIngesterAndProcessor metricsVar doneVar inboxSize concurrency adapter handler+            `catch` \(ProcessorHalt _) -> pure () -- Convert halt to graceful exit+        )+          `finally` unregisterProcessor master procId++  -- Link so exceptions propagate to the parent+  unsafeEff_ $ UIO.link supervisedChild++  pure+    SupervisedProcessor+      { metrics = metricsVar,+        processorId = procId,+        done = doneVar,+        child = Just supervisedChild+      }++-- | Run a processor with metrics but without Master supervision.+-- Useful for testing or simple single-processor setups.+-- This blocks until the stream is exhausted and all messages are processed.+runWithMetrics ::+  (IOE :> es, Tracing :> es) =>+  -- | Inbox size (for backpressure)+  Natural ->+  -- | Processor identifier+  ProcessorId ->+  -- | Queue adapter+  Adapter es msg ->+  -- | Message handler+  Handler es msg ->+  Eff es SupervisedProcessor+runWithMetrics inboxSize procId adapter handler = do+  now <- liftIO getCurrentTime++  -- Initialize state+  let initialMetrics = emptyProcessorMetrics now+  metricsVar <- liftIO $ newTVarIO initialMetrics+  doneVar <- liftIO $ newTVarIO False++  -- Create bounded inbox+  inbox <- liftIO $ newBoundedInbox inboxSize++  -- Run ingester to completion (all messages sent to inbox)+  runIngesterWithMetrics metricsVar adapter.source inbox++  -- Drain remaining messages from inbox+  drainInboxWithMetrics metricsVar handler inbox++  -- Mark done+  liftIO $ atomically $ writeTVar doneVar True++  pure+    SupervisedProcessor+      { metrics = metricsVar,+        processorId = procId,+        done = doneVar,+        child = Nothing+      }++-- | Run ingester and processor with a bounded inbox.+-- Ingester reads from adapter stream, processor calls handler.+-- When stream exhausts, processor drains remaining messages and exits.+runIngesterAndProcessor ::+  (IOE :> es, Tracing :> es) =>+  TVar ProcessorMetrics ->+  TVar Bool ->+  Natural ->+  Concurrency ->+  Adapter es msg ->+  Handler es msg ->+  Eff es ()+runIngesterAndProcessor metricsVar doneVar inboxSize concurrency adapter handler = do+  -- Create bounded inbox (this is where inboxSize is used for backpressure)+  inbox <- liftIO $ newBoundedInbox inboxSize++  -- Signal when ingester completes (stream exhausted)+  streamDoneVar <- liftIO $ newTVarIO False++  -- Run ingester async, processor in main thread+  withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do+    -- Ingester: run until stream exhausts, then signal done+    -- Use finally to ensure streamDoneVar is always set, even if ingester fails+    let ingesterWithSignal =+          runInIO (runIngesterWithMetrics metricsVar adapter.source inbox)+            `finally` atomically (writeTVar streamDoneVar True)++    UIO.withAsync ingesterWithSignal $ \_ ->+      -- Processor: process messages, exit when stream done and inbox empty+      runInIO $ processUntilDrained metricsVar concurrency handler inbox streamDoneVar++  -- Mark done when processor exits+  liftIO $ atomically $ writeTVar doneVar True++-- | Convert inbox to a stream for use with streamly.+-- Respects both the stream-done signal and halt flag.+--+-- Uses STM to atomically check done/empty and receive, avoiding a race+-- condition where the processor could block on receive after the stream+-- has completed but before the done flag was checked.+inboxToStream ::+  Inbox (Ingested es msg) ->+  TVar Bool ->+  IORef (Maybe HaltReason) ->+  Stream.Stream IO (Ingested es msg)+inboxToStream inbox streamDoneVar haltRef = Stream.unfoldrM step ()+  where+    step _ = do+      -- Check halt flag first (outside STM since it's an IORef)+      halted <- readIORef haltRef+      case halted of+        Just _ -> pure Nothing -- Stop reading+        Nothing -> do+          -- Atomically either receive a message or detect completion.+          -- This avoids TOCTOU race where we check done/empty separately+          -- and then block on receive after the stream has completed.+          result <-+            atomically $+              -- Try to receive a message+              (Just <$> receiveSTM inbox)+                `orElse`+                -- Or check if we're done (stream exhausted and inbox empty)+                ( do+                    done <- readTVar streamDoneVar+                    empty <- mailboxEmptySTM inbox+                    if done && empty+                      then pure Nothing+                      else retry -- Inbox empty but stream not done, wait for message+                )+          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) =>+  TVar ProcessorMetrics ->+  Concurrency ->+  Handler es msg ->+  Inbox (Ingested es msg) ->+  TVar Bool ->+  Eff es ()+processUntilDrained metricsVar concurrency handler inbox streamDoneVar = do+  haltRef <- liftIO $ newIORef Nothing++  let maxConc = case concurrency of+        Serial -> 1+        Ahead n -> n+        Async n -> n++  withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do+    let inboxStream = inboxToStream inbox streamDoneVar haltRef+        processAction = runInIO . processOne metricsVar maxConc haltRef handler++    case concurrency of+      Serial ->+        Stream.fold Fold.drain $+          Stream.mapM processAction inboxStream+      Ahead n ->+        Stream.fold Fold.drain $+          StreamP.parMapM (StreamP.maxBuffer n . StreamP.ordered True) processAction inboxStream+      Async n ->+        Stream.fold Fold.drain $+          StreamP.parMapM (StreamP.maxBuffer n) processAction inboxStream++    -- After draining, check if we halted+    maybeHalt <- readIORef haltRef+    case maybeHalt of+      Just reason -> throwIO $ ProcessorHalt reason+      Nothing -> pure ()++-- | Drain inbox until empty, with metrics tracking.+-- Used for testing with finite streams.+drainInboxWithMetrics ::+  (IOE :> es, Tracing :> es) =>+  TVar ProcessorMetrics ->+  Handler es msg ->+  Inbox (Ingested es msg) ->+  Eff es ()+drainInboxWithMetrics metricsVar handler inbox = do+  haltRef <- liftIO $ newIORef Nothing+  go haltRef+  where+    go haltRef = do+      empty <- liftIO $ mailboxEmpty inbox+      unless empty $ do+        ingested <- liftIO $ receive inbox+        processOne metricsVar 1 haltRef handler ingested+        -- Check if halted+        halted <- liftIO $ readIORef haltRef+        case halted of+          Just reason -> throwIO $ ProcessorHalt reason+          Nothing -> go haltRef++-- | Process a single message with metrics tracking and tracing.+-- Thread-safe for concurrent execution.+processOne ::+  (IOE :> es, Tracing :> es) =>+  TVar ProcessorMetrics ->+  Int ->+  IORef (Maybe HaltReason) ->+  Handler es msg ->+  Ingested es msg ->+  Eff es ()+processOne metricsVar maxConc haltRef handler ingested = do+  -- Extract parent context from message headers for distributed tracing+  let parentCtx = ingested.envelope.traceContext >>= extractTraceContext++  withExtractedContext parentCtx $+    withSpan' processMessageSpanName consumerSpanArgs $ \traceSpan -> do+      -- Add messaging attributes+      let MessageId msgIdText = ingested.envelope.messageId+      addAttribute traceSpan attrMessagingSystem ("shibuya" :: Text)+      addAttribute traceSpan attrMessagingMessageId msgIdText++      -- Add partition if present+      case ingested.envelope.partition of+        Just p -> addAttribute traceSpan attrMessagingDestinationPartitionId p+        Nothing -> pure ()++      -- Increment in-flight and add inflight attributes+      now <- liftIO getCurrentTime+      currentInflight <- liftIO $ atomically $ do+        modifyTVar' metricsVar $ \m ->+          let current = case m.state of+                Processing info _ -> info.inFlight+                _ -> 0+           in m & #state .~ Processing (InFlightInfo (current + 1) maxConc) now+        m <- readTVar metricsVar+        pure $ case m.state of+          Processing info _ -> info.inFlight+          _ -> 1++      addAttribute traceSpan attrShibuyaInflightCount currentInflight+      addAttribute traceSpan attrShibuyaInflightMax maxConc++      -- Record handler start event+      addEvent traceSpan (mkEvent eventHandlerStarted [])++      -- Call handler and finalize+      result <-+        catchAny+          ( do+              decision <- handler ingested+              ingested.ack.finalize decision+              pure (Right decision)+          )+          ( \ex -> do+              recordException traceSpan ex+              addEvent traceSpan (mkEvent eventHandlerException [])+              pure $ Left $ HandlerException $ Text.pack $ show ex+          )++      -- Record completion event and set status+      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 ->+              setStatus traceSpan $ OTel.Error $ showDeadLetterReason 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++      -- Decrement in-flight and update stats+      now' <- liftIO getCurrentTime+      liftIO $ atomically $ modifyTVar' metricsVar $ decrementAndUpdate result now'++      -- Handle halt (set flag, don't throw - let stream drain)+      case result of+        Right (AckHalt reason) -> liftIO $ atomicWriteIORef haltRef (Just reason)+        _ -> pure ()+  where+    showAckDecision :: AckDecision -> Text+    showAckDecision AckOk = "ack_ok"+    showAckDecision (AckRetry _) = "ack_retry"+    showAckDecision (AckDeadLetter _) = "ack_dead_letter"+    showAckDecision (AckHalt _) = "ack_halt"++    showDeadLetterReason :: DeadLetterReason -> Text+    showDeadLetterReason (PoisonPill t) = "poison_pill: " <> t+    showDeadLetterReason (InvalidPayload t) = "invalid_payload: " <> t+    showDeadLetterReason MaxRetriesExceeded = "max_retries_exceeded"++    showHaltReason :: HaltReason -> Text+    showHaltReason (HaltOrderedStream t) = "halt_ordered_stream: " <> t+    showHaltReason (HaltFatal t) = "halt_fatal: " <> t++-- | Decrement in-flight count and update stats based on result.+decrementAndUpdate ::+  Either HandlerError AckDecision ->+  UTCTime ->+  ProcessorMetrics ->+  ProcessorMetrics+decrementAndUpdate result now m =+  let newState = case m.state of+        Processing info _ ->+          if info.inFlight <= 1+            then Idle+            else Processing (info {inFlight = info.inFlight - 1}) now+        other -> other+      newStats = case result of+        Right AckOk -> incProcessed m.stats+        Right (AckRetry _) -> incProcessed m.stats+        Right (AckDeadLetter _) -> incFailed m.stats+        Right (AckHalt _) -> m.stats -- Mark as failed with halt message+        Left _ -> incFailed m.stats+      finalState = case result of+        Right (AckHalt reason) -> Failed (haltReasonText reason) now+        Left err -> Failed (handlerErrorToText err) now+        _ -> newState+   in m {state = finalState, stats = newStats}+  where+    haltReasonText (HaltOrderedStream t) = t+    haltReasonText (HaltFatal t) = t
+ src/Shibuya/Stream.hs view
@@ -0,0 +1,42 @@+-- | Stream utilities for adapters.+-- Common stream transformations for use with adapters.+module Shibuya.Stream+  ( -- * Stream Sources+    pollingStream,++    -- * Transformations+    batchStream,+    filterStream,+  )+where++import Control.Concurrent (threadDelay)+import Effectful (Eff, IOE, liftIO, (:>))+import Streamly.Data.Fold qualified as Fold+import Streamly.Data.Stream (Stream)+import Streamly.Data.Stream qualified as Stream++-- | Create a polling stream from a poll action.+-- Polls at the given interval, yielding messages when available.+pollingStream ::+  (IOE :> es) =>+  -- | Poll interval in microseconds+  Int ->+  -- | Poll action+  Eff es (Maybe msg) ->+  Stream (Eff es) msg+pollingStream interval poll =+  Stream.catMaybes $+    Stream.repeatM $ do+      mmsg <- poll+      case mmsg of+        Nothing -> liftIO (threadDelay interval) >> pure Nothing+        Just msg -> pure (Just msg)++-- | Batch messages into chunks.+batchStream :: (Monad m) => Int -> Stream m msg -> Stream m [msg]+batchStream n = Stream.foldMany (Fold.take n Fold.toList)++-- | Filter messages.+filterStream :: (Monad m) => (msg -> Bool) -> Stream m msg -> Stream m msg+filterStream = Stream.filter
+ src/Shibuya/Telemetry.hs view
@@ -0,0 +1,42 @@+-- | OpenTelemetry tracing support for Shibuya.+--+-- This module re-exports all telemetry-related types and functions.+--+-- == Quick Start+--+-- 1. Enable tracing in your config:+--+-- @+-- config = defaultAppConfig+--   { tracing = defaultTracingConfig { enabled = True }+--   }+-- @+--+-- 2. Set environment variables for the OTLP exporter:+--+-- @+-- export OTEL_SERVICE_NAME="my-service"+-- export OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:4317"+-- @+--+-- 3. Spans are automatically created for message processing.+--    See "Shibuya.Telemetry.Effect" for custom instrumentation.+module Shibuya.Telemetry+  ( -- * Configuration+    module Shibuya.Telemetry.Config,++    -- * Effect+    module Shibuya.Telemetry.Effect,++    -- * Context Propagation+    module Shibuya.Telemetry.Propagation,++    -- * Semantic Conventions+    module Shibuya.Telemetry.Semantic,+  )+where++import Shibuya.Telemetry.Config+import Shibuya.Telemetry.Effect+import Shibuya.Telemetry.Propagation+import Shibuya.Telemetry.Semantic
+ src/Shibuya/Telemetry/Config.hs view
@@ -0,0 +1,33 @@+-- | Tracing configuration for Shibuya.+-- Most configuration is handled via OTEL_* environment variables+-- by the hs-opentelemetry SDK. This module provides the minimal+-- application-level config.+module Shibuya.Telemetry.Config+  ( TracingConfig (..),+    defaultTracingConfig,+  )+where++import Data.Text (Text)+import GHC.Generics (Generic)++-- | Tracing configuration.+-- The SDK reads most settings from OTEL_* environment variables.+data TracingConfig = TracingConfig+  { -- | Master switch for tracing. When False, runTracingNoop is used.+    enabled :: !Bool,+    -- | Service name (fallback if OTEL_SERVICE_NAME not set)+    serviceName :: !Text,+    -- | Service version for span attributes+    serviceVersion :: !(Maybe Text)+  }+  deriving stock (Eq, Show, Generic)++-- | Default tracing config with tracing disabled.+defaultTracingConfig :: TracingConfig+defaultTracingConfig =+  TracingConfig+    { enabled = False,+      serviceName = "shibuya",+      serviceVersion = Nothing+    }
+ src/Shibuya/Telemetry/Effect.hs view
@@ -0,0 +1,312 @@+{-# LANGUAGE TypeFamilies #-}++-- | Tracing effect for OpenTelemetry integration.+-- Provides a simple, low-overhead tracing API that wraps hs-opentelemetry.+module Shibuya.Telemetry.Effect+  ( -- * Effect+    Tracing,++    -- * Runners+    runTracing,+    runTracingNoop,++    -- * Span Operations+    withSpan,+    withSpan',++    -- * Span Modification+    addAttribute,+    addAttributes,+    addEvent,+    recordException,+    setStatus,++    -- * Context+    getTracer,+    isTracingEnabled,+    withExtractedContext,++    -- * Re-exports+    OTel.Span,+    OTel.SpanArguments (..),+    OTel.SpanKind (..),+    OTel.SpanStatus (..),+    OTel.NewEvent (..),+    OTel.Tracer,+    OTel.defaultSpanArguments,+    OTel.toAttribute,+  )+where++import Control.Exception (Exception, bracket_)+import Data.ByteString qualified as BS+import Data.HashMap.Strict (HashMap)+import Data.HashMap.Strict qualified as HashMap+import Data.Text (Text)+import Effectful (Dispatch (..), DispatchOf, Eff, Effect, IOE, liftIO, withEffToIO, (:>))+import Effectful.Dispatch.Static+  ( SideEffects (..),+    StaticRep,+    evalStaticRep,+    getStaticRep,+  )+import Effectful.Internal.Unlift (Limit (..), Persistence (..), UnliftStrategy (..))+import OpenTelemetry.Attributes (Attribute, ToAttribute)+import OpenTelemetry.Attributes qualified as Attr+import OpenTelemetry.Context qualified as Ctx+import OpenTelemetry.Context.ThreadLocal qualified as Ctx+import OpenTelemetry.Trace.Core qualified as OTel+import OpenTelemetry.Trace.Id qualified as OTel.Id+import OpenTelemetry.Trace.TraceState qualified as OTel.TraceState++--------------------------------------------------------------------------------+-- Effect Definition+--------------------------------------------------------------------------------++-- | Tracing effect for OpenTelemetry spans.+data Tracing :: Effect++type instance DispatchOf Tracing = 'Static 'WithSideEffects++-- | Internal representation carrying the Tracer and enabled flag.+data instance StaticRep Tracing = TracingRep+  { tracer :: !OTel.Tracer,+    tracingEnabled :: !Bool+  }++--------------------------------------------------------------------------------+-- Runners+--------------------------------------------------------------------------------++-- | Run with actual tracing enabled.+-- The Tracer should be obtained from a TracerProvider.+--+-- Example:+--+-- @+-- provider <- OTel.initializeTracerProvider+-- let tracer = OTel.makeTracer provider "shibuya" OTel.tracerOptions+-- runTracing tracer myAction+-- @+runTracing ::+  (IOE :> es) =>+  OTel.Tracer ->+  Eff (Tracing : es) a ->+  Eff es a+runTracing t = evalStaticRep (TracingRep t True)++-- | Run with tracing disabled (zero overhead).+-- All tracing operations become no-ops.+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+  where+    instrumentationLib =+      OTel.InstrumentationLibrary+        { OTel.libraryName = "shibuya-noop",+          OTel.libraryVersion = "",+          OTel.librarySchemaUrl = "",+          OTel.libraryAttributes = Attr.emptyAttributes+        }++--------------------------------------------------------------------------------+-- Span Operations+--------------------------------------------------------------------------------++-- | Create a span wrapping an action.+-- The span is automatically ended when the action completes.+--+-- Example:+--+-- @+-- withSpan "process.message" consumerSpanArgs $ do+--   -- your code here+-- @+withSpan ::+  (Tracing :> es, IOE :> es) =>+  Text ->+  OTel.SpanArguments ->+  Eff es a ->+  Eff es a+withSpan name args action = do+  TracingRep {tracer, tracingEnabled} <- getStaticRep+  if not tracingEnabled+    then action+    else withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->+      OTel.inSpan tracer name args (runInIO action)++-- | Create a span with access to the Span handle.+-- Use this when you need to add attributes or events to the span.+--+-- Example:+--+-- @+-- withSpan' "process.message" consumerSpanArgs $ \span -> do+--   addAttribute span "message.id" msgId+--   -- your code here+-- @+withSpan' ::+  (Tracing :> es, IOE :> es) =>+  Text ->+  OTel.SpanArguments ->+  (OTel.Span -> Eff es a) ->+  Eff es a+withSpan' name args f = do+  TracingRep {tracer, tracingEnabled} <- getStaticRep+  if not tracingEnabled+    then do+      -- Create a dropped span for the callback+      dummySpan <- liftIO mkDummySpan+      f dummySpan+    else withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->+      OTel.inSpan' tracer name args $ \traceSpan ->+        runInIO (f traceSpan)++--------------------------------------------------------------------------------+-- Span Modification+--------------------------------------------------------------------------------++-- | Add an attribute to a span.+addAttribute ::+  (Tracing :> es, IOE :> es, ToAttribute a) =>+  OTel.Span ->+  Text ->+  a ->+  Eff es ()+addAttribute traceSpan key val = do+  TracingRep {tracingEnabled} <- getStaticRep+  if not tracingEnabled+    then pure ()+    else liftIO $ OTel.addAttribute traceSpan key val++-- | Add multiple attributes to a span.+addAttributes ::+  (Tracing :> es, IOE :> es) =>+  OTel.Span ->+  HashMap Text Attribute ->+  Eff es ()+addAttributes traceSpan attrs = do+  TracingRep {tracingEnabled} <- getStaticRep+  if not tracingEnabled+    then pure ()+    else liftIO $ OTel.addAttributes traceSpan attrs++-- | Add an event to a span.+addEvent ::+  (Tracing :> es, IOE :> es) =>+  OTel.Span ->+  OTel.NewEvent ->+  Eff es ()+addEvent traceSpan event = do+  TracingRep {tracingEnabled} <- getStaticRep+  if not tracingEnabled+    then pure ()+    else liftIO $ OTel.addEvent traceSpan event++-- | Record an exception on a span.+-- This adds an "exception" event with stack trace information.+recordException ::+  (Tracing :> es, IOE :> es, Exception e) =>+  OTel.Span ->+  e ->+  Eff es ()+recordException traceSpan ex = do+  TracingRep {tracingEnabled} <- getStaticRep+  if not tracingEnabled+    then pure ()+    else liftIO $ OTel.recordException traceSpan HashMap.empty Nothing ex++-- | Set the status of a span.+setStatus ::+  (Tracing :> es, IOE :> es) =>+  OTel.Span ->+  OTel.SpanStatus ->+  Eff es ()+setStatus traceSpan status = do+  TracingRep {tracingEnabled} <- getStaticRep+  if not tracingEnabled+    then pure ()+    else liftIO $ OTel.setStatus traceSpan status++--------------------------------------------------------------------------------+-- Context Operations+--------------------------------------------------------------------------------++-- | Get the current tracer.+getTracer ::+  (Tracing :> es) =>+  Eff es OTel.Tracer+getTracer = do+  TracingRep {tracer} <- getStaticRep+  pure tracer++-- | Check if tracing is enabled.+isTracingEnabled ::+  (Tracing :> es) =>+  Eff es Bool+isTracingEnabled = do+  TracingRep {tracingEnabled} <- getStaticRep+  pure tracingEnabled++-- | Run an action with an extracted parent context.+-- Use this to link spans to parent traces from message headers.+--+-- Example:+--+-- @+-- let parentCtx = extractTraceContext messageHeaders+-- withExtractedContext parentCtx $ do+--   withSpan "child.span" consumerSpanArgs $ do+--     -- this span will be a child of parentCtx+-- @+withExtractedContext ::+  (Tracing :> es, IOE :> es) =>+  Maybe OTel.SpanContext ->+  Eff es a ->+  Eff es a+withExtractedContext Nothing action = action+withExtractedContext (Just parentCtx) action = do+  TracingRep {tracingEnabled} <- getStaticRep+  if not tracingEnabled+    then action+    else do+      -- Insert the parent span into thread-local context using bracket pattern+      let parentSpan = OTel.wrapSpanContext parentCtx+          newContext = Ctx.insertSpan parentSpan Ctx.empty+      withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->+        bracket_+          (Ctx.attachContext newContext)+          Ctx.detachContext+          (runInIO action)++--------------------------------------------------------------------------------+-- Internal Helpers+--------------------------------------------------------------------------------++-- | Create a dummy/dropped span for use when tracing is disabled.+-- All operations on this span are no-ops.+mkDummySpan :: IO OTel.Span+mkDummySpan = do+  -- Create a FrozenSpan with invalid (all-zeros) IDs+  -- This creates a "Dropped" span that no-ops all operations+  let dummyTraceId = case OTel.Id.bytesToTraceId (BS.replicate 16 0) of+        Right tid -> tid+        Left _ -> error "mkDummySpan: failed to create dummy TraceId"+      dummySpanId = case OTel.Id.bytesToSpanId (BS.replicate 8 0) of+        Right sid -> sid+        Left _ -> error "mkDummySpan: failed to create dummy SpanId"+      dummySpanContext =+        OTel.SpanContext+          { OTel.traceFlags = OTel.defaultTraceFlags,+            OTel.isRemote = False,+            OTel.traceId = dummyTraceId,+            OTel.spanId = dummySpanId,+            OTel.traceState = OTel.TraceState.empty+          }+  pure $ OTel.wrapSpanContext dummySpanContext
+ src/Shibuya/Telemetry/Propagation.hs view
@@ -0,0 +1,51 @@+-- | W3C Trace Context propagation for Shibuya.+-- Extracts and injects trace context from/to message headers.+module Shibuya.Telemetry.Propagation+  ( -- * Extraction+    extractTraceContext,++    -- * Injection+    injectTraceContext,++    -- * Re-export+    TraceHeaders,+  )+where++import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C+import OpenTelemetry.Trace.Core (Span, SpanContext)+import Shibuya.Core.Types (TraceHeaders)++-- | Extract SpanContext from W3C trace headers.+-- Returns Nothing if headers are missing or malformed.+--+-- Example:+--+-- @+-- let headers = [("traceparent", "00-abc123...-def456...-01")]+-- case extractTraceContext headers of+--   Nothing -> -- no valid trace context+--   Just ctx -> -- use ctx as parent+-- @+extractTraceContext :: TraceHeaders -> Maybe SpanContext+extractTraceContext headers =+  let traceparent = lookup "traceparent" headers+      tracestate = lookup "tracestate" headers+   in W3C.decodeSpanContext traceparent tracestate++-- | Inject current span's context into headers for propagation.+-- Use this when producing messages that should carry trace context.+--+-- Example:+--+-- @+-- headers <- injectTraceContext currentSpan+-- -- headers contains [("traceparent", "..."), ("tracestate", "...")]+-- @+injectTraceContext :: Span -> IO TraceHeaders+injectTraceContext otelSpan = do+  (traceparent, tracestate) <- W3C.encodeSpanContext otelSpan+  pure+    [ ("traceparent", traceparent),+      ("tracestate", tracestate)+    ]
+ src/Shibuya/Telemetry/Semantic.hs view
@@ -0,0 +1,162 @@+-- | Semantic conventions for Shibuya OpenTelemetry instrumentation.+-- Provides span names, attribute keys, and helpers following OTel conventions.+module Shibuya.Telemetry.Semantic+  ( -- * Span Names+    processMessageSpanName,+    ingestSpanName,++    -- * Attribute Keys (OTel Messaging Conventions)+    attrMessagingSystem,+    attrMessagingMessageId,+    attrMessagingDestinationName,+    attrMessagingDestinationPartitionId,++    -- * Attribute Keys (Shibuya-specific)+    attrShibuyaProcessorId,+    attrShibuyaInflightCount,+    attrShibuyaInflightMax,+    attrShibuyaAckDecision,++    -- * Event Names+    eventHandlerStarted,+    eventHandlerCompleted,+    eventHandlerException,+    eventAckDecision,++    -- * SpanArguments Helpers+    consumerSpanArgs,+    internalSpanArgs,++    -- * NewEvent Helper+    mkEvent,+  )+where++import Data.HashMap.Strict qualified as HashMap+import Data.Text (Text)+import OpenTelemetry.Attributes (Attribute)+import OpenTelemetry.Trace.Core+  ( NewEvent (..),+    SpanArguments (..),+    SpanKind (..),+  )++--------------------------------------------------------------------------------+-- Span Names+--------------------------------------------------------------------------------++-- | Span name for processing a single message.+processMessageSpanName :: Text+processMessageSpanName = "shibuya.process.message"++-- | Span name for the ingester lifecycle.+ingestSpanName :: Text+ingestSpanName = "shibuya.ingest"++--------------------------------------------------------------------------------+-- Attribute Keys (OTel Messaging Semantic Conventions)+-- See: https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/+--------------------------------------------------------------------------------++-- | The messaging system identifier.+attrMessagingSystem :: Text+attrMessagingSystem = "messaging.system"++-- | The message identifier.+attrMessagingMessageId :: Text+attrMessagingMessageId = "messaging.message.id"++-- | The destination (queue/topic) name.+attrMessagingDestinationName :: Text+attrMessagingDestinationName = "messaging.destination.name"++-- | The partition identifier.+attrMessagingDestinationPartitionId :: Text+attrMessagingDestinationPartitionId = "messaging.destination.partition.id"++--------------------------------------------------------------------------------+-- Attribute Keys (Shibuya-specific)+--------------------------------------------------------------------------------++-- | The processor identifier.+attrShibuyaProcessorId :: Text+attrShibuyaProcessorId = "shibuya.processor.id"++-- | Current number of in-flight messages.+attrShibuyaInflightCount :: Text+attrShibuyaInflightCount = "shibuya.inflight.count"++-- | Maximum concurrency limit.+attrShibuyaInflightMax :: Text+attrShibuyaInflightMax = "shibuya.inflight.max"++-- | The ack decision made by the handler.+attrShibuyaAckDecision :: Text+attrShibuyaAckDecision = "shibuya.ack.decision"++--------------------------------------------------------------------------------+-- Event Names+--------------------------------------------------------------------------------++-- | Event recorded when handler execution starts.+eventHandlerStarted :: Text+eventHandlerStarted = "handler.started"++-- | Event recorded when handler execution completes successfully.+eventHandlerCompleted :: Text+eventHandlerCompleted = "handler.completed"++-- | Event recorded when handler throws an exception.+eventHandlerException :: Text+eventHandlerException = "handler.exception"++-- | Event recorded when ack decision is made.+eventAckDecision :: Text+eventAckDecision = "ack.decision"++--------------------------------------------------------------------------------+-- SpanArguments Helpers+--------------------------------------------------------------------------------++-- | SpanArguments for message consumption (Consumer kind).+consumerSpanArgs :: SpanArguments+consumerSpanArgs =+  SpanArguments+    { kind = Consumer,+      attributes = HashMap.empty,+      links = [],+      startTime = Nothing+    }++-- | SpanArguments for internal operations (Internal kind).+internalSpanArgs :: SpanArguments+internalSpanArgs =+  SpanArguments+    { kind = Internal,+      attributes = HashMap.empty,+      links = [],+      startTime = Nothing+    }++--------------------------------------------------------------------------------+-- NewEvent Helper+--------------------------------------------------------------------------------++-- | Create a NewEvent with the given name and attributes.+-- Use 'toAttribute' to convert values to Attribute.+--+-- Example:+--+-- @+-- mkEvent "handler.completed"+--   [ ("duration_ms", toAttribute (42 :: Int))+--   , ("status", toAttribute ("ok" :: Text))+--   ]+-- @+mkEvent :: Text -> [(Text, Attribute)] -> NewEvent+mkEvent name attrs =+  NewEvent+    { newEventName = name,+      newEventAttributes = HashMap.fromList attrs,+      newEventTimestamp = Nothing+    }
+ test/Main.hs view
@@ -0,0 +1,22 @@+{-# LANGUAGE ImportQualifiedPost #-}++module Main (main) where++import Shibuya.Core.AckSpec qualified+import Shibuya.Core.TypesSpec qualified+import Shibuya.PolicySpec qualified+import Shibuya.Runner.SupervisedSpec qualified+import Shibuya.RunnerSpec qualified+import Shibuya.Telemetry.EffectSpec qualified+import Shibuya.Telemetry.PropagationSpec qualified+import Test.Hspec++main :: IO ()+main = hspec $ do+  describe "Shibuya.Core.Types" Shibuya.Core.TypesSpec.spec+  describe "Shibuya.Core.Ack" Shibuya.Core.AckSpec.spec+  describe "Shibuya.Policy" Shibuya.PolicySpec.spec+  describe "Shibuya.Runner" Shibuya.RunnerSpec.spec+  Shibuya.Runner.SupervisedSpec.spec+  Shibuya.Telemetry.EffectSpec.spec+  Shibuya.Telemetry.PropagationSpec.spec
+ test/Shibuya/Core/AckSpec.hs view
@@ -0,0 +1,90 @@+{-# LANGUAGE OverloadedStrings #-}++module Shibuya.Core.AckSpec (spec) where++import Data.Time (secondsToNominalDiffTime)+import Shibuya.Core.Ack+import Test.Hspec++spec :: Spec+spec = do+  describe "RetryDelay" $ do+    it "wraps NominalDiffTime" $ do+      let delay = RetryDelay (secondsToNominalDiffTime 30)+      delay.unRetryDelay `shouldBe` secondsToNominalDiffTime 30++    it "supports Eq" $ do+      let d1 = RetryDelay (secondsToNominalDiffTime 10)+          d2 = RetryDelay (secondsToNominalDiffTime 10)+          d3 = RetryDelay (secondsToNominalDiffTime 20)+      d1 `shouldBe` d2+      d1 `shouldNotBe` d3++  describe "DeadLetterReason" $ do+    it "distinguishes all constructors" $ do+      let r1 = PoisonPill "bad message"+          r2 = InvalidPayload "parse error"+          r3 = MaxRetriesExceeded+      r1 `shouldNotBe` r2+      r2 `shouldNotBe` r3+      r1 `shouldNotBe` r3++    it "PoisonPill carries message" $ do+      let r = PoisonPill "corrupt data"+      case r of+        PoisonPill msg -> msg `shouldBe` "corrupt data"+        _ -> expectationFailure "wrong constructor"++    it "InvalidPayload carries message" $ do+      let r = InvalidPayload "JSON decode failed"+      case r of+        InvalidPayload msg -> msg `shouldBe` "JSON decode failed"+        _ -> expectationFailure "wrong constructor"++  describe "HaltReason" $ do+    it "HaltOrderedStream carries message" $ do+      let r = HaltOrderedStream "ordering violation"+      case r of+        HaltOrderedStream msg -> msg `shouldBe` "ordering violation"+        _ -> expectationFailure "wrong constructor"++    it "HaltFatal carries message" $ do+      let r = HaltFatal "database connection lost"+      case r of+        HaltFatal msg -> msg `shouldBe` "database connection lost"+        _ -> expectationFailure "wrong constructor"++  describe "AckDecision" $ do+    it "distinguishes all constructors" $ do+      let d1 = AckOk+          d2 = AckRetry (RetryDelay 10)+          d3 = AckDeadLetter MaxRetriesExceeded+          d4 = AckHalt (HaltFatal "error")+      d1 `shouldNotBe` d2+      d2 `shouldNotBe` d3+      d3 `shouldNotBe` d4+      d1 `shouldNotBe` d4++    it "AckOk equals AckOk" $ do+      AckOk `shouldBe` AckOk++    it "AckRetry carries delay" $ do+      let delay = RetryDelay (secondsToNominalDiffTime 60)+          decision = AckRetry delay+      case decision of+        AckRetry d -> d `shouldBe` delay+        _ -> expectationFailure "wrong constructor"++    it "AckDeadLetter carries reason" $ do+      let reason = PoisonPill "unprocessable"+          decision = AckDeadLetter reason+      case decision of+        AckDeadLetter r -> r `shouldBe` reason+        _ -> expectationFailure "wrong constructor"++    it "AckHalt carries reason" $ do+      let reason = HaltOrderedStream "must stop"+          decision = AckHalt reason+      case decision of+        AckHalt r -> r `shouldBe` reason+        _ -> expectationFailure "wrong constructor"
+ test/Shibuya/Core/TypesSpec.hs view
@@ -0,0 +1,72 @@+{-# LANGUAGE OverloadedStrings #-}++module Shibuya.Core.TypesSpec (spec) where++import Data.Time (UTCTime (..), fromGregorian)+import Shibuya.Core.Types+import Test.Hspec++spec :: Spec+spec = do+  describe "MessageId" $ do+    it "supports Eq" $ do+      MessageId "abc" `shouldBe` MessageId "abc"+      MessageId "abc" `shouldNotBe` MessageId "xyz"++    it "supports Ord for Map keys" $ do+      MessageId "a" `compare` MessageId "b" `shouldBe` LT+      MessageId "b" `compare` MessageId "a" `shouldBe` GT+      MessageId "a" `compare` MessageId "a" `shouldBe` EQ++    it "supports IsString for OverloadedStrings" $ do+      let msgId :: MessageId = "test-message"+      msgId.unMessageId `shouldBe` "test-message"++  describe "Cursor" $ do+    it "CursorInt compares numerically" $ do+      CursorInt 1 `compare` CursorInt 2 `shouldBe` LT+      CursorInt 10 `compare` CursorInt 2 `shouldBe` GT++    it "CursorText compares lexicographically" $ do+      CursorText "a" `compare` CursorText "b" `shouldBe` LT+      CursorText "z" `compare` CursorText "a" `shouldBe` GT++    it "CursorInt comes before CursorText in Ord" $ do+      -- This is the derived Ord behavior+      CursorInt 999 `compare` CursorText "a" `shouldBe` LT++  describe "Envelope" $ do+    it "Functor: fmap id == id" $ do+      let env = testEnvelope (42 :: Int)+      fmap id env `shouldBe` env++    it "Functor: fmap (f . g) == fmap f . fmap g" $ do+      let env = testEnvelope (1 :: Int)+          f = (+ 1)+          g = (* 2)+      fmap (f . g) env `shouldBe` (fmap f . fmap g) env++    it "preserves all fields through fmap" $ do+      let env = testEnvelope ("hello" :: String)+          mapped = fmap length env+      mapped.messageId `shouldBe` env.messageId+      mapped.cursor `shouldBe` env.cursor+      mapped.partition `shouldBe` env.partition+      mapped.enqueuedAt `shouldBe` env.enqueuedAt+      mapped.traceContext `shouldBe` env.traceContext+      mapped.payload `shouldBe` 5++-- Test helper+testEnvelope :: msg -> Envelope msg+testEnvelope msg =+  Envelope+    { messageId = MessageId "test-id",+      cursor = Just (CursorInt 42),+      partition = Just "partition-0",+      enqueuedAt = Just testTime,+      traceContext = Nothing,+      payload = msg+    }++testTime :: UTCTime+testTime = UTCTime (fromGregorian 2024 1 1) 0
+ test/Shibuya/PolicySpec.hs view
@@ -0,0 +1,69 @@+{-# LANGUAGE OverloadedStrings #-}++module Shibuya.PolicySpec (spec) where++import Shibuya.Policy+import Test.Hspec+import Prelude hiding (Ordering)++spec :: Spec+spec = do+  describe "Ordering" $ do+    it "all constructors are distinguishable" $ do+      StrictInOrder `shouldNotBe` PartitionedInOrder+      PartitionedInOrder `shouldNotBe` Unordered+      StrictInOrder `shouldNotBe` Unordered++  describe "Concurrency" $ do+    it "all constructors are distinguishable" $ do+      Serial `shouldNotBe` Ahead 1+      Ahead 1 `shouldNotBe` Async 1+      Serial `shouldNotBe` Async 1++    it "Ahead preserves Int value" $ do+      let c = Ahead 42+      case c of+        Ahead n -> n `shouldBe` 42+        _ -> expectationFailure "wrong constructor"++    it "Async preserves Int value" $ do+      let c = Async 100+      case c of+        Async n -> n `shouldBe` 100+        _ -> expectationFailure "wrong constructor"++  describe "validatePolicy" $ do+    describe "StrictInOrder" $ do+      it "allows Serial" $ do+        validatePolicy StrictInOrder Serial `shouldBe` Right ()++      it "rejects Ahead" $ do+        validatePolicy StrictInOrder (Ahead 5) `shouldSatisfy` isLeft++      it "rejects Async" $ do+        validatePolicy StrictInOrder (Async 5) `shouldSatisfy` isLeft++    describe "PartitionedInOrder" $ do+      it "allows Serial" $ do+        validatePolicy PartitionedInOrder Serial `shouldBe` Right ()++      it "allows Ahead" $ do+        validatePolicy PartitionedInOrder (Ahead 10) `shouldBe` Right ()++      it "allows Async" $ do+        validatePolicy PartitionedInOrder (Async 10) `shouldBe` Right ()++    describe "Unordered" $ do+      it "allows Serial" $ do+        validatePolicy Unordered Serial `shouldBe` Right ()++      it "allows Ahead" $ do+        validatePolicy Unordered (Ahead 10) `shouldBe` Right ()++      it "allows Async" $ do+        validatePolicy Unordered (Async 10) `shouldBe` Right ()++-- Helper+isLeft :: Either a b -> Bool+isLeft (Left _) = True+isLeft (Right _) = False
+ test/Shibuya/Runner/SupervisedSpec.hs view
@@ -0,0 +1,905 @@+{-# LANGUAGE OverloadedStrings #-}++module Shibuya.Runner.SupervisedSpec (spec) where++import Control.Concurrent.NQE.Supervisor (Strategy (..))+import Control.Concurrent.STM (readTVarIO)+import Control.Monad (forM, replicateM)+import Data.IORef (IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef)+import Data.Text qualified as Text+import Data.Time (UTCTime (..), fromGregorian)+import Effectful (Eff, IOE, liftIO, runEff, (:>))+import Shibuya.Adapter (Adapter (..))+import Shibuya.App+  ( ShutdownConfig (..),+    SupervisionStrategy (..),+    mkProcessor,+    runApp,+    stopAppGracefully,+  )+import Shibuya.Core (ProcessorHalt (..))+import Shibuya.Core.Ack (AckDecision (..), HaltReason (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..))+import Shibuya.Handler (Handler)+import Shibuya.Policy (Concurrency (..))+import Shibuya.Runner.Master+  ( getAllMetrics,+    startMaster,+    stopMaster,+  )+import Shibuya.Runner.Metrics+  ( InFlightInfo (..),+    ProcessorId (..),+    ProcessorMetrics (..),+    ProcessorState (..),+    StreamStats (..),+  )+import Shibuya.Runner.Supervised+  ( SupervisedProcessor (..),+    getMetrics,+    isDone,+    runSupervised,+    runWithMetrics,+  )+import Shibuya.Telemetry.Effect (runTracingNoop)+import Streamly.Data.Stream qualified as Stream+import Test.Hspec+import UnliftIO (SomeException, try)+import UnliftIO.Concurrent (threadDelay)++spec :: Spec+spec = do+  describe "Shibuya.Runner.Master" $ do+    it "starts and stops cleanly" $ do+      result <- runEff $ runTracingNoop $ do+        master <- startMaster IgnoreAll+        stopMaster master+        pure ()+      result `shouldBe` ()++    it "returns empty metrics initially" $ do+      metrics <- runEff $ runTracingNoop $ do+        master <- startMaster IgnoreAll+        m <- getAllMetrics master+        stopMaster master+        pure m+      metrics `shouldSatisfy` null++  describe "Shibuya.Runner.Supervised" $ do+    describe "runWithMetrics" $ do+      it "processes messages and tracks metrics" $ do+        (finalMetrics, processedMsgs) <- runEff $ runTracingNoop $ do+          -- Track processed messages+          processedRef <- liftIO $ newIORef ([] :: [String])++          -- Create test messages+          messages <- createTestMessages 3++          -- Create adapter and handler+          let adapter = testAdapter messages+              handler = testHandler processedRef+              procId = ProcessorId "test-processor"++          -- Run with metrics+          sp <- runWithMetrics 10 procId adapter handler++          -- Get results+          metrics <- getMetrics sp+          msgs <- liftIO $ readIORef processedRef+          pure (metrics, msgs)++        -- Verify metrics+        finalMetrics.stats.received `shouldBe` 3+        finalMetrics.stats.processed `shouldBe` 3+        finalMetrics.stats.failed `shouldBe` 0+        finalMetrics.state `shouldBe` Idle++        -- Verify all messages were processed+        length processedMsgs `shouldBe` 3++      it "tracks failed messages" $ do+        finalMetrics <- runEff $ runTracingNoop $ do+          messages <- createTestMessages 3++          let adapter = testAdapter messages+              handler = failingHandler+              procId = ProcessorId "failing-processor"++          sp <- runWithMetrics 10 procId adapter handler+          getMetrics sp++        -- All should be failed (handler throws)+        finalMetrics.stats.failed `shouldBe` 3++      it "marks processor as done when stream exhausted" $ do+        done <- runEff $ runTracingNoop $ do+          messages <- createTestMessages 2++          let adapter = testAdapter messages+              handler = alwaysAckOk+              procId = ProcessorId "quick-processor"++          sp <- runWithMetrics 10 procId adapter handler+          isDone sp++        done `shouldBe` True++    describe "AckHalt behavior" $ do+      it "stops processing when handler returns AckHalt" $ do+        processedCount <- do+          processedRef <- newIORef (0 :: Int)++          let handler _ = do+                count <- liftIO $ readIORef processedRef+                liftIO $ modifyIORef' processedRef (+ 1)+                if count >= 2+                  then pure $ AckHalt (HaltFatal "stopping after 3")+                  else pure AckOk++          -- Try to run and catch ProcessorHalt+          result <- try $ runEff $ runTracingNoop $ do+            messages <- createTestMessages 10+            let adapter = testAdapter messages+                procId = ProcessorId "halting-processor"+            runWithMetrics 10 procId adapter handler++          -- Verify ProcessorHalt was thrown+          case result of+            Left (ProcessorHalt reason) ->+              reason `shouldBe` HaltFatal "stopping after 3"+            Right _ ->+              expectationFailure "Expected ProcessorHalt exception"++          readIORef processedRef++        -- Should have processed exactly 3 messages before halt+        processedCount `shouldBe` 3++      it "updates metrics to halted state on AckHalt" $ do+        finalMetrics <- runEff $ runTracingNoop $ do+          messages <- createTestMessages 5++          let handler _ = pure $ AckHalt (HaltFatal "immediate halt")+              adapter = testAdapter messages++          master <- startMaster IgnoreAll++          -- runSupervised catches ProcessorHalt, so we can check metrics+          sp <- runSupervised master 10 (ProcessorId "halt-metrics") Serial adapter handler++          -- Wait for processor to halt+          liftIO $ threadDelay 100000 -- 100ms+          m <- liftIO $ readTVarIO sp.metrics+          stopMaster master+          pure m++        case finalMetrics.state of+          Failed msg _ -> msg `shouldSatisfy` (Text.isInfixOf "immediate halt")+          other -> expectationFailure $ "Expected Failed state, got: " ++ show other++      it "halt in one supervised processor doesn't affect others" $ do+        (countA, countB) <- runEff $ runTracingNoop $ do+          countARef <- liftIO $ newIORef (0 :: Int)+          countBRef <- liftIO $ newIORef (0 :: Int)++          messagesA <- createTestMessages 10+          messagesB <- createTestMessages 10++          -- Handler A halts after 2 messages+          let handlerA _ = do+                count <- liftIO $ readIORef countARef+                liftIO $ modifyIORef' countARef (+ 1)+                if count >= 1+                  then pure $ AckHalt (HaltFatal "A stopping")+                  else pure AckOk++          -- Handler B processes all messages+          let handlerB _ = do+                liftIO $ modifyIORef' countBRef (+ 1)+                pure AckOk++          let adapterA = testAdapter messagesA+              adapterB = testAdapter messagesB++          master <- startMaster IgnoreAll++          _spA <- runSupervised master 10 (ProcessorId "A") Serial adapterA handlerA+          _spB <- runSupervised master 10 (ProcessorId "B") Serial adapterB handlerB++          -- Wait for both to complete+          liftIO $ threadDelay 500000 -- 500ms+          cA <- liftIO $ readIORef countARef+          cB <- liftIO $ readIORef countBRef++          stopMaster master+          pure (cA, cB)++        -- A should have stopped after 2 messages+        countA `shouldBe` 2+        -- B should have processed all 10+        countB `shouldBe` 10++    describe "Concurrency modes" $ do+      describe "Ahead mode" $ do+        it "processes all messages successfully" $ do+          countRef <- newIORef (0 :: Int)++          finalCount <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 5++            let handler _ = do+                  liftIO $ atomicModifyIORef' countRef (\n -> (n + 1, ()))+                  liftIO $ threadDelay 10000 -- 10ms+                  pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            _ <- runSupervised master 10 (ProcessorId "ahead") (Ahead 3) adapter handler++            -- Wait for completion+            liftIO $ threadDelay 300000 -- 300ms+            stopMaster master+            liftIO $ readIORef countRef++          finalCount `shouldBe` 5++        it "respects maxBuffer limit" $ do+          -- This tests that we don't start too many concurrent tasks+          maxInFlightRef <- newIORef (0 :: Int)+          currentInFlightRef <- newIORef (0 :: Int)++          _ <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 10++            let handler _ = do+                  -- Atomically increment current in-flight and get new value+                  cur <-+                    liftIO $+                      atomicModifyIORef' currentInFlightRef (\n -> let n' = n + 1 in (n', n'))+                  -- Update max+                  liftIO $ atomicModifyIORef' maxInFlightRef (\m -> (max m cur, ()))+                  -- Simulate some work+                  liftIO $ threadDelay 50000 -- 50ms+                  -- Decrement in-flight+                  liftIO $ atomicModifyIORef' currentInFlightRef (\c -> (c - 1, ()))+                  pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            -- Use Ahead with max 3 concurrent+            _sp <- runSupervised master 10 (ProcessorId "ahead-limit") (Ahead 3) adapter handler++            liftIO $ threadDelay 1000000 -- 1s to let it complete+            stopMaster master++          maxInFlight <- readIORef maxInFlightRef+          -- Max in-flight should not exceed buffer size + 1 (buffer + currently processing)+          maxInFlight `shouldSatisfy` (<= 4)++      describe "Async mode" $ do+        it "processes all messages concurrently" $ do+          countRef <- newIORef (0 :: Int)++          finalCount <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 5++            let handler _ = do+                  liftIO $ atomicModifyIORef' countRef (\n -> (n + 1, ()))+                  liftIO $ threadDelay 10000 -- 10ms+                  pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            _ <- runSupervised master 10 (ProcessorId "async") (Async 5) adapter handler++            -- Wait for completion+            liftIO $ threadDelay 300000 -- 300ms+            stopMaster master+            liftIO $ readIORef countRef++          finalCount `shouldBe` 5++        it "respects concurrency limit" $ do+          maxInFlightRef <- newIORef (0 :: Int)+          currentInFlightRef <- newIORef (0 :: Int)++          _ <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 10++            let handler _ = do+                  cur <-+                    liftIO $+                      atomicModifyIORef' currentInFlightRef (\n -> let n' = n + 1 in (n', n'))+                  liftIO $ atomicModifyIORef' maxInFlightRef (\m -> (max m cur, ()))+                  liftIO $ threadDelay 50000 -- 50ms+                  liftIO $ atomicModifyIORef' currentInFlightRef (\c -> (c - 1, ()))+                  pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            _sp <- runSupervised master 10 (ProcessorId "async-limit") (Async 3) adapter handler++            liftIO $ threadDelay 1000000 -- 1s+            stopMaster master++          maxInFlight <- readIORef maxInFlightRef+          maxInFlight `shouldSatisfy` (<= 4)++      describe "Halt with concurrency" $ do+        it "waits for in-flight to complete before halting" $ do+          processedRef <- newIORef ([] :: [Int])++          _ <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 10++            let handler ingested = do+                  let msgNum = case ingested.envelope.cursor of+                        Just (CursorInt n) -> n+                        _ -> 0+                  -- Simulate work+                  liftIO $ threadDelay 50000 -- 50ms+                  liftIO $ atomicModifyIORef' processedRef (\xs -> (msgNum : xs, ()))+                  -- Halt on message 3+                  if msgNum == 3+                    then pure $ AckHalt (HaltFatal "halt on 3")+                    else pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            _sp <- runSupervised master 10 (ProcessorId "halt-concurrent") (Async 3) adapter handler++            liftIO $ threadDelay 500000 -- 500ms+            stopMaster master++          processed <- readIORef processedRef+          -- With Async 3, messages 1, 2, 3 start together+          -- When 3 halts, 1 and 2 should complete+          length processed `shouldSatisfy` (>= 3)++        it "stops reading new messages after halt decision" $ do+          processedRef <- newIORef (0 :: Int)++          _ <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 20++            let handler _ = do+                  count <-+                    liftIO $+                      atomicModifyIORef' processedRef (\n -> let n' = n + 1 in (n', n'))+                  liftIO $ threadDelay 20000 -- 20ms+                  -- Halt after processing 5 messages+                  if count >= 5+                    then pure $ AckHalt (HaltFatal "halt at 5")+                    else pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            _sp <- runSupervised master 10 (ProcessorId "halt-stop-read") (Async 3) adapter handler++            liftIO $ threadDelay 500000 -- 500ms+            stopMaster master++          processed <- readIORef processedRef+          -- Should process around 5-8 messages (5 before halt + a few in-flight)+          -- but definitely not all 20+          processed `shouldSatisfy` (< 15)++      describe "Error handling" $ do+        it "handler exception doesn't affect other in-flight handlers" $ do+          resultsRef <- newIORef ([] :: [Either String Int])++          _ <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 5++            let handler ingested = do+                  let msgNum = case ingested.envelope.cursor of+                        Just (CursorInt n) -> n+                        _ -> 0+                  -- Message 2 throws an exception+                  if msgNum == 2+                    then do+                      liftIO $ atomicModifyIORef' resultsRef (\xs -> (Left "error on 2" : xs, ()))+                      error "Intentional failure on message 2"+                    else do+                      liftIO $ threadDelay 30000 -- 30ms+                      liftIO $ atomicModifyIORef' resultsRef (\xs -> (Right msgNum : xs, ()))+                      pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            _ <- runSupervised master 10 (ProcessorId "error-handling") (Async 3) adapter handler++            liftIO $ threadDelay 500000 -- 500ms+            stopMaster master++          results <- readIORef resultsRef+          -- Message 2 should have errored+          let errors = [e | Left e <- results]+          length errors `shouldBe` 1+          -- Other messages (1, 3, 4, 5) should have succeeded+          let successes = [n | Right n <- results]+          length successes `shouldSatisfy` (>= 3)++      describe "Metrics tracking" $ do+        it "tracks in-flight count during concurrent processing" $ do+          maxInFlightObserved <- newIORef (0 :: Int)++          _ <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 5++            let handler _ = do+                  liftIO $ threadDelay 100000 -- 100ms+                  pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            sp <- runSupervised master 10 (ProcessorId "metrics-test") (Async 3) adapter handler++            -- Check metrics while processing+            liftIO $ do+              threadDelay 50000 -- 50ms - should be in the middle of processing+              metrics <- readTVarIO sp.metrics+              case metrics.state of+                Processing info _ -> modifyIORef' maxInFlightObserved (max info.inFlight)+                _ -> pure ()++            liftIO $ threadDelay 600000 -- 600ms to complete+            stopMaster master++          maxObserved <- readIORef maxInFlightObserved+          -- Should have observed at least 2 concurrent handlers+          maxObserved `shouldSatisfy` (>= 2)++        it "reports correct maxConcurrency in metrics" $ do+          observedMaxConc <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 3++            let handler _ = do+                  liftIO $ threadDelay 50000 -- 50ms+                  pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            sp <- runSupervised master 10 (ProcessorId "max-conc") (Ahead 7) adapter handler++            -- Check metrics while processing+            result <- liftIO $ do+              threadDelay 25000 -- 25ms+              metrics <- readTVarIO sp.metrics+              case metrics.state of+                Processing info _ -> pure $ Just info.maxConcurrency+                _ -> pure Nothing++            liftIO $ threadDelay 300000+            stopMaster master+            pure result++          observedMaxConc `shouldBe` Just 7++      describe "Ahead mode ordering" $ do+        it "completes handlers and emits results in input order" $ do+          -- Track the order in which handlers START and COMPLETE+          startOrderRef <- newIORef ([] :: [Int])+          completeOrderRef <- newIORef ([] :: [Int])++          _ <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 5++            let handler ingested = do+                  let msgNum = case ingested.envelope.cursor of+                        Just (CursorInt n) -> n+                        _ -> 0+                  -- Record start+                  liftIO $ atomicModifyIORef' startOrderRef (\xs -> (msgNum : xs, ()))+                  -- Variable delay: message 1 is slowest, 5 is fastest+                  liftIO $ threadDelay ((6 - msgNum) * 20000)+                  -- Record completion+                  liftIO $ atomicModifyIORef' completeOrderRef (\xs -> (msgNum : xs, ()))+                  pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            _ <- runSupervised master 10 (ProcessorId "ahead-order") (Ahead 5) adapter handler++            liftIO $ threadDelay 500000 -- 500ms+            stopMaster master++          -- With Ahead mode, all 5 should complete+          completeOrder <- readIORef completeOrderRef+          length completeOrder `shouldBe` 5++    -- The key property: all messages were processed+    -- (Ordering of side effects may vary, but stream output is ordered)++    describe "Robustness" $ do+      describe "Adapter source exceptions" $ do+        it "handles adapter source throwing mid-stream" $ do+          processedRef <- newIORef (0 :: Int)++          _ <- runEff $ runTracingNoop $ do+            -- Create an adapter where source throws after 3 messages+            _goodMessages <- createTestMessages 3+            -- Use fromEffect to throw after yielding good messages+            let failingSource = Stream.unfoldrM step (0 :: Int)+                  where+                    step n+                      | n < 3 = do+                          msg <- liftIO $ createSingleMessage (n + 1)+                          pure $ Just (msg, n + 1)+                      | n == 3 = error "Network failure mid-stream"+                      | otherwise = pure Nothing++            let adapter =+                  Adapter+                    { adapterName = "test:failing-source",+                      source = failingSource,+                      shutdown = pure ()+                    }++            let handler _ = do+                  liftIO $ modifyIORef' processedRef (+ 1)+                  pure AckOk++            master <- startMaster IgnoreAll+            _ <- runSupervised master 10 (ProcessorId "failing-adapter") Serial adapter handler++            -- Wait for processing+            liftIO $ threadDelay 200000 -- 200ms+            stopMaster master++          -- Should have processed the 3 good messages before the error+          processed <- readIORef processedRef+          processed `shouldBe` 3++      describe "Rapid start/stop cycles" $ do+        it "handles rapid start/stop without resource leaks" $ do+          -- Run 50 rapid cycles+          results <- replicateM 50 $ runEff $ runTracingNoop $ do+            master <- startMaster IgnoreAll+            messages <- createTestMessages 5++            let adapter = testAdapter messages+                handler _ = pure AckOk++            sp <- runSupervised master 10 (ProcessorId "rapid") Serial adapter handler++            -- Minimal wait+            liftIO $ threadDelay 10000 -- 10ms+            stopMaster master++            -- Check it stopped cleanly+            isDone sp++          -- All should complete (no hangs, no crashes)+          length results `shouldBe` 50++        it "handles concurrent start/stop cycles" $ do+          -- Run 20 concurrent cycles+          countRef <- newIORef (0 :: Int)++          _ <- runEff $ runTracingNoop $ do+            master <- startMaster IgnoreAll++            -- Start 10 processors rapidly+            _sps <- forM [(1 :: Int) .. 10] $ \i -> do+              messages <- createTestMessages 3+              let adapter = testAdapter messages+                  handler _ = do+                    liftIO $ atomicModifyIORef' countRef (\n -> (n + 1, ()))+                    liftIO $ threadDelay 5000 -- 5ms+                    pure AckOk+              runSupervised master 10 (ProcessorId $ "concurrent-" <> Text.pack (show i)) Serial adapter handler++            -- Wait briefly then stop+            liftIO $ threadDelay 100000 -- 100ms+            stopMaster master++          -- Should have processed messages from multiple processors+          count <- readIORef countRef+          count `shouldSatisfy` (> 0)++      describe "KillAll supervision strategy" $ do+        it "stops all processors when adapter source fails with KillAll strategy" $ do+          countARef <- newIORef (0 :: Int)+          countBRef <- newIORef (0 :: Int)++          _ :: Either SomeException () <- try $ runEff $ runTracingNoop $ do+            messagesB <- createTestMessages 20++            -- Adapter A throws after 3 messages (ingester crash, not handler)+            let failingSourceA = Stream.unfoldrM step (0 :: Int)+                  where+                    step n+                      | n < 3 = do+                          msg <- liftIO $ createSingleMessage (n + 1)+                          pure $ Just (msg, n + 1)+                      | otherwise = error "Adapter A source failed!"++            let adapterA =+                  Adapter+                    { adapterName = "test:failing-A",+                      source = failingSourceA,+                      shutdown = pure ()+                    }++            -- Handler A counts messages+            let handlerA _ = do+                  liftIO $ modifyIORef' countARef (+ 1)+                  liftIO $ threadDelay 10000 -- 10ms+                  pure AckOk++            -- Handler B processes slowly+            let handlerB _ = do+                  liftIO $ modifyIORef' countBRef (+ 1)+                  liftIO $ threadDelay 30000 -- 30ms+                  pure AckOk++            let adapterB = testAdapter messagesB++            -- Use KillAll strategy+            master <- startMaster KillAll++            _spA <- runSupervised master 10 (ProcessorId "killall-A") Serial adapterA handlerA+            _spB <- runSupervised master 10 (ProcessorId "killall-B") Serial adapterB handlerB++            -- Wait for A to fail and trigger KillAll+            liftIO $ threadDelay 500000 -- 500ms+            stopMaster master++          -- A should have processed messages before adapter failed+          countA <- readIORef countARef+          countA `shouldSatisfy` (<= 3)++          -- B should have been killed (not processed all 20)+          countB <- readIORef countBRef+          countB `shouldSatisfy` (< 20)++      describe "Multiple concurrent halts" $ do+        it "handles multiple handlers returning AckHalt simultaneously" $ do+          haltCountRef <- newIORef (0 :: Int)+          processedRef <- newIORef (0 :: Int)++          _ <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 10++            let handler ingested = do+                  let msgNum = case ingested.envelope.cursor of+                        Just (CursorInt n) -> n+                        _ -> 0+                  liftIO $ atomicModifyIORef' processedRef (\n -> (n + 1, ()))+                  liftIO $ threadDelay 30000 -- 30ms+                  -- Messages 2, 3, 4 all return halt+                  if msgNum `elem` [2, 3, 4]+                    then do+                      liftIO $ atomicModifyIORef' haltCountRef (\n -> (n + 1, ()))+                      pure $ AckHalt (HaltFatal $ "halt from " <> Text.pack (show msgNum))+                    else pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            _ <- runSupervised master 10 (ProcessorId "multi-halt") (Async 5) adapter handler++            liftIO $ threadDelay 500000 -- 500ms+            stopMaster master++          -- Multiple halts should have been recorded+          haltCount <- readIORef haltCountRef+          haltCount `shouldSatisfy` (>= 1) -- At least one halt processed++          -- Should not have processed all messages+          processed <- readIORef processedRef+          processed `shouldSatisfy` (< 10)++      describe "Stability under load" $ do+        it "processes many messages without issues" $ do+          processedRef <- newIORef (0 :: Int)++          finalCount <- runEff $ runTracingNoop $ do+            -- 500 messages+            messages <- createTestMessages 500++            let handler _ = do+                  liftIO $ atomicModifyIORef' processedRef (\n -> (n + 1, ()))+                  pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            _sp <- runSupervised master 100 (ProcessorId "load-test") (Async 10) adapter handler++            -- Wait for completion+            liftIO $ threadDelay 2000000 -- 2s+            stopMaster master+            liftIO $ readIORef processedRef++          finalCount `shouldBe` 500++        it "handles mixed success and failure under load" $ do+          successRef <- newIORef (0 :: Int)+          failRef <- newIORef (0 :: Int)++          _ <- runEff $ runTracingNoop $ do+            messages <- createTestMessages 200++            let handler ingested = do+                  let msgNum = case ingested.envelope.cursor of+                        Just (CursorInt n) -> n+                        _ -> 0+                  -- Every 7th message fails+                  if msgNum `mod` 7 == 0+                    then do+                      liftIO $ atomicModifyIORef' failRef (\n -> (n + 1, ()))+                      error "Intentional failure"+                    else do+                      liftIO $ atomicModifyIORef' successRef (\n -> (n + 1, ()))+                      pure AckOk++            master <- startMaster IgnoreAll+            let adapter = testAdapter messages+            _ <- runSupervised master 50 (ProcessorId "mixed-load") (Async 5) adapter handler++            liftIO $ threadDelay 1000000 -- 1s+            stopMaster master++          success <- readIORef successRef+          failures <- readIORef failRef++          -- Should have processed many messages+          success `shouldSatisfy` (> 150)+          -- Should have recorded failures+          failures `shouldSatisfy` (> 20)++  describe "Graceful Shutdown" $ do+    describe "stopAppGracefully" $ do+      it "drains in-flight messages before stopping" $ do+        processedRef <- newIORef (0 :: Int)++        drained <- runEff $ runTracingNoop $ do+          messages <- createTestMessages 10++          let handler _ = do+                liftIO $ do+                  threadDelay 50000 -- 50ms per message+                  modifyIORef' processedRef (+ 1)+                pure AckOk++          let adapter = testAdapter messages+              processor = mkProcessor adapter handler++          result <- runApp IgnoreFailures 10 [(ProcessorId "drain-test", processor)]+          case result of+            Left _ -> pure False+            Right appHandle -> do+              -- Give some time for processing to start+              liftIO $ threadDelay 100000 -- 100ms++              -- Graceful shutdown with generous timeout+              let config = ShutdownConfig {drainTimeout = 5} -- 5 seconds+              stopAppGracefully config appHandle++        drained `shouldBe` True++        -- All messages should have been processed+        processed <- readIORef processedRef+        processed `shouldBe` 10++      it "respects timeout when processors are slow" $ do+        processedRef <- newIORef (0 :: Int)++        drained <- runEff $ runTracingNoop $ do+          messages <- createTestMessages 20++          let handler _ = do+                liftIO $ do+                  threadDelay 500000 -- 500ms per message - very slow+                  modifyIORef' processedRef (+ 1)+                pure AckOk++          let adapter = testAdapter messages+              processor = mkProcessor adapter handler++          result <- runApp IgnoreFailures 5 [(ProcessorId "timeout-test", processor)]+          case result of+            Left _ -> pure True -- Treat error as "drained" for simplicity+            Right appHandle -> do+              -- Start immediately+              liftIO $ threadDelay 100000 -- 100ms++              -- Very short timeout (0.3 seconds)+              let config = ShutdownConfig {drainTimeout = 0.3}+              stopAppGracefully config appHandle++        -- Should timeout (not all drained)+        drained `shouldBe` False++        -- Should have processed only some messages+        processed <- readIORef processedRef+        processed `shouldSatisfy` (< 20)++      it "returns True when all processors finish quickly" $ do+        drained <- runEff $ runTracingNoop $ do+          messages <- createTestMessages 5++          let handler _ = pure AckOk -- Instant processing+          let adapter = testAdapter messages+              processor = mkProcessor adapter handler++          result <- runApp IgnoreFailures 10 [(ProcessorId "quick-test", processor)]+          case result of+            Left _ -> pure False+            Right appHandle -> do+              -- Give time to complete+              liftIO $ threadDelay 200000 -- 200ms+              let config = ShutdownConfig {drainTimeout = 1}+              stopAppGracefully config appHandle++        drained `shouldBe` True++-- Test helpers++testTime :: UTCTime+testTime = UTCTime (fromGregorian 2026 1 1) 0++createTestMessages :: (IOE :> es) => Int -> Eff es [Ingested es String]+createTestMessages n = mapM createMessage [1 .. n]+  where+    createMessage i = do+      let msgId = MessageId $ "msg-" <> Text.pack (show i)+          env =+            Envelope+              { messageId = msgId,+                cursor = Just (CursorInt i),+                partition = Nothing,+                enqueuedAt = Just testTime,+                traceContext = Nothing,+                payload = "message-" <> show i+              }+          ackHandle = AckHandle $ \_ -> pure ()+      pure $+        Ingested+          { envelope = env,+            ack = ackHandle,+            lease = Nothing+          }++-- | Create a single message (for use in streaming contexts)+-- Polymorphic over effect stack for use with tracing.+createSingleMessage :: (IOE :> es) => Int -> IO (Ingested es String)+createSingleMessage i = do+  let msgId = MessageId $ "msg-" <> Text.pack (show i)+      env =+        Envelope+          { messageId = msgId,+            cursor = Just (CursorInt i),+            partition = Nothing,+            enqueuedAt = Just testTime,+            traceContext = Nothing,+            payload = "message-" <> show i+          }+      ackHandle = AckHandle $ \_ -> pure ()+  pure $+    Ingested+      { envelope = env,+        ack = ackHandle,+        lease = Nothing+      }++testAdapter :: [Ingested es String] -> Adapter es String+testAdapter messages =+  Adapter+    { adapterName = "test:supervised",+      source = Stream.fromList messages,+      shutdown = pure ()+    }++testHandler :: (IOE :> es) => IORef [String] -> Handler es String+testHandler ref ingested = do+  liftIO $ modifyIORef' ref (ingested.envelope.payload :)+  pure AckOk++failingHandler :: Handler es msg+failingHandler _ = error "Intentional failure"++alwaysAckOk :: Handler es msg+alwaysAckOk _ = pure AckOk
+ test/Shibuya/RunnerSpec.hs view
@@ -0,0 +1,258 @@+{-# LANGUAGE OverloadedStrings #-}++module Shibuya.RunnerSpec (spec) where++import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+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 (..), QueueProcessor (..), SupervisionStrategy (..), mkProcessor, runApp, waitApp)+import Shibuya.Core.Ack (AckDecision (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Error (PolicyError (..))+import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..))+import Shibuya.Handler (Handler)+import Shibuya.Policy (Concurrency (..), Ordering (..))+import Shibuya.Runner.Metrics (ProcessorId (..))+import Shibuya.Telemetry.Effect (runTracingNoop)+import Streamly.Data.Stream qualified as Stream+import Test.Hspec+import Prelude hiding (Ordering)++spec :: Spec+spec = do+  describe "runApp" $ do+    it "processes messages from mock adapter" $ do+      result <- runEff $ runTracingNoop $ do+        -- Track processed messages+        processedRef <- liftIO $ newIORef ([] :: [String])++        -- Create test messages+        messages <- createTestMessages 3++        -- Create adapter and handler+        let adapter = testAdapter messages+            handler = testHandler processedRef+            processor = mkProcessor adapter handler++        -- Run the app+        res <-+          runApp+            IgnoreFailures+            100+            [ (ProcessorId "test", processor)+            ]++        case res of+          Left err -> pure $ Left err+          Right appHandle -> do+            waitApp appHandle+            pure $ Right ()++      -- Verify result+      result `shouldBe` Right ()++    it "calls finalize for each message" $ do+      (decisions, result) <- runEff $ runTracingNoop $ do+        -- Track ack decisions+        tracking <- newTrackingAck++        -- Create test messages with tracking acks+        messages <- createTrackedMessages tracking 3++        -- Create adapter and handler+        let adapter = testAdapter messages+            handler = alwaysAckOk+            processor = mkProcessor adapter handler++        -- Run the app+        res <-+          runApp+            IgnoreFailures+            100+            [ (ProcessorId "test", processor)+            ]++        case res of+          Left err -> do+            decs <- liftIO $ readIORef tracking.trackedDecisions+            pure (decs, Left err)+          Right appHandle -> do+            waitApp appHandle+            decs <- liftIO $ readIORef tracking.trackedDecisions+            pure (decs, Right ())++      -- Verify all messages were acked+      result `shouldBe` Right ()+      length decisions `shouldBe` 3+      -- All should be AckOk (decisions are in reverse order)+      all ((== AckOk) . snd) decisions `shouldBe` True++    it "returns AppHandle for multiple processors" $ do+      result <- runEff $ runTracingNoop $ do+        messages1 <- createTestMessages 2+        messages2 <- createTestMessages 2++        let adapter1 = testAdapter messages1+            adapter2 = testAdapter messages2+            handler = alwaysAckOk+            proc1 = mkProcessor adapter1 handler+            proc2 = mkProcessor adapter2 handler++        res <-+          runApp+            IgnoreFailures+            100+            [ (ProcessorId "proc1", proc1),+              (ProcessorId "proc2", proc2)+            ]++        case res of+          Left err -> pure $ Left err+          Right appHandle -> do+            waitApp appHandle+            pure $ Right ()++      result `shouldBe` Right ()++  describe "Policy validation" $ do+    it "rejects StrictInOrder with Async" $ do+      result <- runEff $ runTracingNoop $ do+        messages <- createTestMessages 3+        let adapter = testAdapter messages+            handler = alwaysAckOk+            -- Invalid combination: StrictInOrder requires Serial+            processor = QueueProcessor adapter handler StrictInOrder (Async 5)++        runApp IgnoreFailures 100 [(ProcessorId "invalid", processor)]++      case result of+        Left (AppPolicyError (InvalidPolicyCombo _)) -> pure ()+        Left err -> expectationFailure $ "Expected AppPolicyError, got: " ++ show err+        Right _ -> expectationFailure "Expected policy validation to fail"++    it "rejects StrictInOrder with Ahead" $ do+      result <- runEff $ runTracingNoop $ do+        messages <- createTestMessages 3+        let adapter = testAdapter messages+            handler = alwaysAckOk+            processor = QueueProcessor adapter handler StrictInOrder (Ahead 5)++        runApp IgnoreFailures 100 [(ProcessorId "invalid", processor)]++      case result of+        Left (AppPolicyError (InvalidPolicyCombo _)) -> pure ()+        Left err -> expectationFailure $ "Expected AppPolicyError, got: " ++ show err+        Right _ -> expectationFailure "Expected policy validation to fail"++    it "accepts valid policy combinations" $ do+      result <- runEff $ runTracingNoop $ do+        messages <- createTestMessages 2+        let adapter = testAdapter messages+            handler = alwaysAckOk+            -- Valid combinations+            proc1 = QueueProcessor adapter handler Unordered (Async 3)+            proc2 = QueueProcessor adapter handler PartitionedInOrder (Ahead 3)++        res <-+          runApp+            IgnoreFailures+            100+            [ (ProcessorId "async", proc1),+              (ProcessorId "ahead", proc2)+            ]+        case res of+          Left err -> pure $ Left err+          Right appHandle -> do+            waitApp appHandle+            pure $ Right ()++      result `shouldBe` Right ()++  describe "mkProcessor" $ do+    it "creates processor with Unordered ordering" $ do+      messages <- runEff $ createTestMessages 1+      let adapter = testAdapter messages+          handler = alwaysAckOk+          QueueProcessor _ _ ordering _ = mkProcessor adapter handler+      ordering `shouldBe` Unordered++    it "creates processor with Serial concurrency" $ do+      messages <- runEff $ createTestMessages 1+      let adapter = testAdapter messages+          handler = alwaysAckOk+          QueueProcessor _ _ _ concurrency = mkProcessor adapter handler+      concurrency `shouldBe` Serial++-- Test helpers++testTime :: UTCTime+testTime = UTCTime (fromGregorian 2024 1 1) 0++-- | Create N test messages with simple string payloads+createTestMessages :: (IOE :> es) => Int -> Eff es [Ingested es String]+createTestMessages n = mapM createMessage [1 .. n]+  where+    createMessage i = do+      let msgId = MessageId $ "msg-" <> (if i < 10 then "0" else "") <> Text.pack (show i)+          env =+            Envelope+              { messageId = msgId,+                cursor = Just (CursorInt i),+                partition = Nothing,+                enqueuedAt = Just testTime,+                traceContext = Nothing,+                payload = "message-" <> show i+              }+          ackHandle = AckHandle $ \_ -> pure () -- No-op ack+      pure $+        Ingested+          { envelope = env,+            ack = ackHandle,+            lease = Nothing+          }++-- | Create N test messages with tracking acks+createTrackedMessages :: (IOE :> es) => TrackingAck -> Int -> Eff es [Ingested es String]+createTrackedMessages tracking n = mapM createMessage [1 .. n]+  where+    createMessage i = do+      let msgId = MessageId $ "msg-" <> (if i < 10 then "0" else "") <> Text.pack (show i)+          env =+            Envelope+              { messageId = msgId,+                cursor = Just (CursorInt i),+                partition = Nothing,+                enqueuedAt = Just testTime,+                traceContext = Nothing,+                payload = "message-" <> show i+              }+          ackHandle = trackingAckHandle tracking msgId+      pure $+        Ingested+          { envelope = env,+            ack = ackHandle,+            lease = Nothing+          }++-- | Create a test adapter from a list of messages+testAdapter :: [Ingested es String] -> Adapter es String+testAdapter messages =+  Adapter+    { adapterName = "test:mock",+      source = Stream.fromList messages,+      shutdown = pure ()+    }++-- | Handler that records processed messages+testHandler :: (IOE :> es) => IORef [String] -> Handler es String+testHandler ref ingested = do+  liftIO $ modifyIORef' ref (ingested.envelope.payload :)+  pure AckOk++-- | Handler that always returns AckOk+alwaysAckOk :: Handler es msg+alwaysAckOk _ = pure AckOk
+ test/Shibuya/Telemetry/EffectSpec.hs view
@@ -0,0 +1,108 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Tests for the Tracing effect.+module Shibuya.Telemetry.EffectSpec (spec) where++import Data.IORef (newIORef, readIORef, writeIORef)+import Data.Text (Text)+import Effectful (liftIO, runEff)+import OpenTelemetry.Trace.Core qualified as OTel+import Shibuya.Telemetry.Effect+  ( addAttribute,+    addEvent,+    getTracer,+    isTracingEnabled,+    runTracingNoop,+    setStatus,+    withSpan,+    withSpan',+  )+import Shibuya.Telemetry.Semantic (consumerSpanArgs, mkEvent)+import Test.Hspec++spec :: Spec+spec = describe "Shibuya.Telemetry.Effect" $ do+  describe "runTracingNoop" $ do+    it "executes actions and returns result" $ do+      result <- runEff $ runTracingNoop $ do+        pure (42 :: Int)+      result `shouldBe` 42++    it "executes nested withSpan without error" $ do+      result <- runEff $ runTracingNoop $ do+        withSpan "outer" consumerSpanArgs $ do+          withSpan "inner" consumerSpanArgs $ do+            pure ("success" :: String)+      result `shouldBe` "success"++    it "executes withSpan' and provides span handle" $ do+      result <- runEff $ runTracingNoop $ do+        withSpan' "test-span" consumerSpanArgs $ \traceSpan -> do+          -- These should be no-ops but not error+          addAttribute traceSpan "key" ("value" :: Text)+          addEvent traceSpan (mkEvent "test.event" [])+          setStatus traceSpan OTel.Ok+          pure (123 :: Int)+      result `shouldBe` 123++    it "reports tracing as disabled" $ do+      enabled <- runEff $ runTracingNoop $ isTracingEnabled+      enabled `shouldBe` False++    it "provides a tracer (noop)" $ do+      result <- runEff $ runTracingNoop $ do+        _tracer <- getTracer+        pure ("got tracer" :: String)+      result `shouldBe` "got tracer"++    it "handles exceptions in actions correctly" $ do+      ref <- newIORef (0 :: Int)+      result <- runEff $ runTracingNoop $ do+        withSpan "outer" consumerSpanArgs $ do+          liftIO $ writeIORef ref 1+          withSpan "inner" consumerSpanArgs $ do+            liftIO $ writeIORef ref 2+            pure ()+        liftIO $ readIORef ref+      result `shouldBe` 2++    it "works with effectful actions" $ do+      ref <- newIORef ([] :: [String])+      let modifyIORef' r f = do+            val <- readIORef r+            writeIORef r (f val)+      result <- runEff $ runTracingNoop $ do+        withSpan "span1" consumerSpanArgs $ do+          liftIO $ modifyIORef' ref ("span1" :)+        withSpan "span2" consumerSpanArgs $ do+          liftIO $ modifyIORef' ref ("span2" :)+        liftIO $ readIORef ref+      -- Order is reversed due to cons+      result `shouldBe` ["span2", "span1"]++  describe "Span operations (noop mode)" $ do+    it "addAttribute is a no-op" $ do+      result <- runEff $ runTracingNoop $ do+        withSpan' "test" consumerSpanArgs $ \traceSpan -> do+          addAttribute traceSpan "string" ("value" :: Text)+          addAttribute traceSpan "int" (42 :: Int)+          addAttribute traceSpan "bool" True+          pure ("done" :: String)+      result `shouldBe` ("done" :: String)++    it "addEvent is a no-op" $ do+      result <- runEff $ runTracingNoop $ do+        withSpan' "test" consumerSpanArgs $ \traceSpan -> do+          addEvent traceSpan (mkEvent "event1" [])+          addEvent traceSpan (mkEvent "event2" [("key", OTel.toAttribute ("val" :: Text))])+          pure ("done" :: String)+      result `shouldBe` ("done" :: String)++    it "setStatus is a no-op" $ do+      result <- runEff $ runTracingNoop $ do+        withSpan' "test" consumerSpanArgs $ \traceSpan -> do+          setStatus traceSpan OTel.Ok+          setStatus traceSpan (OTel.Error "test error")+          setStatus traceSpan OTel.Unset+          pure ("done" :: String)+      result `shouldBe` ("done" :: String)
+ test/Shibuya/Telemetry/PropagationSpec.hs view
@@ -0,0 +1,121 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Tests for W3C Trace Context propagation.+module Shibuya.Telemetry.PropagationSpec (spec) where++import Data.ByteArray.Encoding (Base (..))+import Data.ByteString (ByteString)+import OpenTelemetry.Trace.Core qualified as OTel+import OpenTelemetry.Trace.Id qualified as OTel.Id+import Shibuya.Telemetry.Propagation (extractTraceContext)+import Test.Hspec++spec :: Spec+spec = describe "Shibuya.Telemetry.Propagation" $ do+  describe "extractTraceContext" $ do+    it "extracts valid W3C traceparent header" $ do+      let headers :: [(ByteString, ByteString)]+          headers =+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")+            ]+      case extractTraceContext headers of+        Nothing -> expectationFailure "Expected to extract trace context"+        Just ctx -> do+          -- Verify trace ID+          OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "0af7651916cd43dd8448eb211c80319c"+          -- Verify span ID+          OTel.Id.spanIdBaseEncodedText Base16 (OTel.spanId ctx) `shouldBe` "b7ad6b7169203331"++    it "extracts traceparent with tracestate" $ do+      let headers :: [(ByteString, ByteString)]+          headers =+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"),+              ("tracestate", "congo=t61rcWkgMzE,rojo=00f067aa0ba902b7")+            ]+      case extractTraceContext headers of+        Nothing -> expectationFailure "Expected to extract trace context"+        Just ctx -> do+          OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "0af7651916cd43dd8448eb211c80319c"+          OTel.Id.spanIdBaseEncodedText Base16 (OTel.spanId ctx) `shouldBe` "b7ad6b7169203331"++    it "returns Nothing for missing traceparent" $ do+      let headers :: [(ByteString, ByteString)]+          headers = []+      extractTraceContext headers `shouldBe` Nothing++    it "returns Nothing for invalid traceparent format" $ do+      let headers :: [(ByteString, ByteString)]+          headers = [("traceparent", "invalid-format")]+      extractTraceContext headers `shouldBe` Nothing++    it "parses traceparent with non-standard version (permissive)" $ do+      -- Note: hs-opentelemetry propagator is permissive and accepts non-00 versions+      let headers :: [(ByteString, ByteString)]+          headers =+            [ ("traceparent", "ff-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")+            ]+      -- Library accepts this - validation is left to the application+      case extractTraceContext headers of+        Nothing -> expectationFailure "Library should parse non-standard version"+        Just _ctx -> pure ()++    it "parses traceparent with all-zero trace ID (permissive)" $ do+      -- Note: hs-opentelemetry propagator doesn't reject all-zero trace IDs+      let headers :: [(ByteString, ByteString)]+          headers =+            [ ("traceparent", "00-00000000000000000000000000000000-b7ad6b7169203331-01")+            ]+      -- Library accepts this - validation is left to the application+      case extractTraceContext headers of+        Nothing -> expectationFailure "Library should parse all-zero trace ID"+        Just ctx -> OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "00000000000000000000000000000000"++    it "parses traceparent with all-zero span ID (permissive)" $ do+      -- Note: hs-opentelemetry propagator doesn't reject all-zero span IDs+      let headers :: [(ByteString, ByteString)]+          headers =+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-0000000000000000-01")+            ]+      -- Library accepts this - validation is left to the application+      case extractTraceContext headers of+        Nothing -> expectationFailure "Library should parse all-zero span ID"+        Just ctx -> OTel.Id.spanIdBaseEncodedText Base16 (OTel.spanId ctx) `shouldBe` "0000000000000000"++    it "handles unsampled trace flag (00)" $ do+      let headers :: [(ByteString, ByteString)]+          headers =+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-00")+            ]+      case extractTraceContext headers of+        Nothing -> expectationFailure "Expected to extract trace context"+        Just _ctx -> pure () -- Just verify it parses+    it "handles sampled trace flag (01)" $ do+      let headers :: [(ByteString, ByteString)]+          headers =+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")+            ]+      case extractTraceContext headers of+        Nothing -> expectationFailure "Expected to extract trace context"+        Just _ctx -> pure () -- Just verify it parses+    it "ignores case-sensitivity of header names" $ do+      -- W3C spec says header names are case-insensitive+      let headers :: [(ByteString, ByteString)]+          headers =+            [ ("Traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")+            ]+      -- Note: This depends on how lookup handles case.+      -- The current implementation uses exact match, so this should return Nothing+      -- This is a known limitation - header normalization should happen at the adapter level+      extractTraceContext headers `shouldBe` Nothing++    it "extracts only traceparent when tracestate is malformed" $ do+      let headers :: [(ByteString, ByteString)]+          headers =+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"),+              ("tracestate", "malformed tracestate value!!!")+            ]+      -- Should still extract the traceparent even if tracestate is invalid+      case extractTraceContext headers of+        Nothing -> expectationFailure "Expected to extract trace context despite malformed tracestate"+        Just ctx -> do+          OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "0af7651916cd43dd8448eb211c80319c"