shibuya-kiroku-adapter 0.6.0.1 → 0.7.0.0

raw patch · 8 files changed

+2669/−2676 lines, 8 filesdep ~kiroku-storedep ~shibuya-corePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: kiroku-store, shibuya-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +24/−0
• app/LifecycleLive.hs +270/−270
• bench/WriteProbe.hs +155/−156
• internal/Shibuya/Adapter/Kiroku/Internal.hs +41/−41
• shibuya-kiroku-adapter.cabal +19/−19
• src/Shibuya/Adapter/Kiroku.hs +467/−485
• src/Shibuya/Adapter/Kiroku/Convert.hs +155/−161
• test/Main.hs +1538/−1544

@@ -1,5 +1,29 @@ # Changelog +## 0.7.0.0 — 2026-10-26++### Breaking Changes++- Require `kiroku-store ^>=0.12.0.0`. Re-exported `SubscriptionTarget` gains+  `CategoryNamespace Text`; update exhaustive matches.+- Require [shibuya-core](mori://shinzui/shibuya/packages/shibuya-core)+  `>=0.12 && <0.13` across the library, tests, lifecycle fixture, and benchmarks.+  Downstream direct `Ingested` record constructors must supply Shibuya's new+  `deliveryStatus` field.++### New Features++- Support namespace targets in single and grouped configurations through the+  underlying store worker. Namespace checkpoints may advance beyond the last+  matching delivery; each member scans bounded global windows. Apply migration+  0016 from kiroku-store-migrations 0.7.2.0 before starting namespace workers.++### Other Changes++- Initialize `deliveryStatus` to `pure DeliveryCurrent` for ack-coupled Kiroku+  deliveries. Adapter-specific function signatures and configuration fields are+  unchanged.+ ## 0.6.0.1 — 2026-10-11  ### Other Changes

@@ -24,15 +24,15 @@ import Kiroku.Store import Kiroku.Test.Postgres (ephemeralConfig, migrateTestDatabase) import Shibuya.Adapter.Kiroku (defaultKirokuAdapterConfig, kirokuAdapter)-import Shibuya.App (-    ProcessorId (..),+import Shibuya.App+  ( ProcessorId (..),     ShutdownConfig (drainTimeout, totalShutdownTimeout),     defaultAppConfig,     defaultShutdownConfig,     mkProcessor,     runApp,     stopAppGracefully,- )+  ) import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..)) import Shibuya.Core.Ingested (Message (..)) import Shibuya.Core.Types (Envelope (..))@@ -44,86 +44,86 @@ import Text.Read (readMaybe)  data Config = Config-    { durationSecs :: !Int-    , messagesPerSecond :: !Int-    , sampleIntervalSecs :: !Int-    , outputCsv :: !FilePath-    , outputLedger :: !FilePath-    , runId :: !String-    , restartAtSecs :: !Int-    , shutdownDrainSecs :: !Int-    , shutdownTotalSecs :: !Int-    }+  { durationSecs :: !Int,+    messagesPerSecond :: !Int,+    sampleIntervalSecs :: !Int,+    outputCsv :: !FilePath,+    outputLedger :: !FilePath,+    runId :: !String,+    restartAtSecs :: !Int,+    shutdownDrainSecs :: !Int,+    shutdownTotalSecs :: !Int+  }  data Sample = Sample-    { timestamp :: !UTCTime-    , elapsedSecs :: !Int-    , messagesProduced :: !Int-    , messagesProcessed :: !Int-    , messagesFailed :: !Int-    , queueDepth :: !Int64-    , retainedBytes :: !Word64-    , maxLiveBytes :: !Word64-    }+  { timestamp :: !UTCTime,+    elapsedSecs :: !Int,+    messagesProduced :: !Int,+    messagesProcessed :: !Int,+    messagesFailed :: !Int,+    queueDepth :: !Int64,+    retainedBytes :: !Word64,+    maxLiveBytes :: !Word64+  }  data DeliveryLedger = DeliveryLedger-    { producedHandle :: !Handle-    , processedHandle :: !Handle-    , producedPath :: !FilePath-    , processedPath :: !FilePath-    , malformedDeliveries :: !(IORef Int)-    }+  { producedHandle :: !Handle,+    processedHandle :: !Handle,+    producedPath :: !FilePath,+    processedPath :: !FilePath,+    malformedDeliveries :: !(IORef Int)+  }  main :: IO () main = do-    config <- loadConfig-    let suffix = Text.pack (filter validResourceChar config.runId)-        streamName = StreamName ("ep45-" <> suffix)-        subscriptionName = SubscriptionName ("ep45-" <> suffix)-    putStrLn "=== Shibuya Kiroku lifecycle live fixture ==="-    putStrLn $ "Duration: " <> show config.durationSecs <> " seconds"-    putStrLn $ "Target rate: " <> show config.messagesPerSecond <> " msg/s"-    putStrLn $ "Restart at: " <> show config.restartAtSecs <> " seconds"-    pgConfig <- ephemeralConfig-    result <- Pg.withCachedConfig pgConfig Pg.defaultCacheConfig $ \database -> do-        let connectionString = Pg.connectionString database-        migrateTestDatabase connectionString-        withStore (defaultConnectionSettings connectionString) $ \store ->-            runFixture config store streamName subscriptionName-    case result of-        Left err -> error $ "Failed to start ephemeral PostgreSQL: " <> show err-        Right () -> pure ()+  config <- loadConfig+  let suffix = Text.pack (filter validResourceChar config.runId)+      streamName = StreamName ("ep45-" <> suffix)+      subscriptionName = SubscriptionName ("ep45-" <> suffix)+  putStrLn "=== Shibuya Kiroku lifecycle live fixture ==="+  putStrLn $ "Duration: " <> show config.durationSecs <> " seconds"+  putStrLn $ "Target rate: " <> show config.messagesPerSecond <> " msg/s"+  putStrLn $ "Restart at: " <> show config.restartAtSecs <> " seconds"+  pgConfig <- ephemeralConfig+  result <- Pg.withCachedConfig pgConfig Pg.defaultCacheConfig $ \database -> do+    let connectionString = Pg.connectionString database+    migrateTestDatabase connectionString+    withStore (defaultConnectionSettings connectionString) $ \store ->+      runFixture config store streamName subscriptionName+  case result of+    Left err -> error $ "Failed to start ephemeral PostgreSQL: " <> show err+    Right () -> pure ()   where     validResourceChar char =-        ('a' <= char && char <= 'z')-            || ('A' <= char && char <= 'Z')-            || ('0' <= char && char <= '9')-            || char == '-'+      ('a' <= char && char <= 'z')+        || ('A' <= char && char <= 'Z')+        || ('0' <= char && char <= '9')+        || char == '-'  loadConfig :: IO Config loadConfig = do-    duration <- envInt "DURATION_SECS" 1800-    rate <- envInt "MESSAGES_PER_SECOND" 100-    interval <- envInt "SAMPLE_INTERVAL_SECS" 30-    output <- envString "OUTPUT_CSV" "kiroku-lifecycle.csv"-    ledger <- envString "OUTPUT_LEDGER" (output <> ".ledger.json")-    now <- getCurrentTime-    identifier <- envString "LIFECYCLE_RUN_ID" (formatTime defaultTimeLocale "%Y%m%d%H%M%S" now)-    restart <- envInt "RESTART_AT_SECS" (duration `div` 2)-    shutdownDrain <- envInt "SHUTDOWN_DRAIN_SECS" 30-    shutdownTotal <- envInt "SHUTDOWN_TOTAL_SECS" 60-    pure-        Config-            { durationSecs = duration-            , messagesPerSecond = rate-            , sampleIntervalSecs = interval-            , outputCsv = output-            , outputLedger = ledger-            , runId = identifier-            , restartAtSecs = max 1 (min (duration - 1) restart)-            , shutdownDrainSecs = shutdownDrain-            , shutdownTotalSecs = shutdownTotal-            }+  duration <- envInt "DURATION_SECS" 1800+  rate <- envInt "MESSAGES_PER_SECOND" 100+  interval <- envInt "SAMPLE_INTERVAL_SECS" 30+  output <- envString "OUTPUT_CSV" "kiroku-lifecycle.csv"+  ledger <- envString "OUTPUT_LEDGER" (output <> ".ledger.json")+  now <- getCurrentTime+  identifier <- envString "LIFECYCLE_RUN_ID" (formatTime defaultTimeLocale "%Y%m%d%H%M%S" now)+  restart <- envInt "RESTART_AT_SECS" (duration `div` 2)+  shutdownDrain <- envInt "SHUTDOWN_DRAIN_SECS" 30+  shutdownTotal <- envInt "SHUTDOWN_TOTAL_SECS" 60+  pure+    Config+      { durationSecs = duration,+        messagesPerSecond = rate,+        sampleIntervalSecs = interval,+        outputCsv = output,+        outputLedger = ledger,+        runId = identifier,+        restartAtSecs = max 1 (min (duration - 1) restart),+        shutdownDrainSecs = shutdownDrain,+        shutdownTotalSecs = shutdownTotal+      }  envString :: String -> String -> IO String envString key fallback = maybe fallback (\value -> value) <$> lookupEnv key@@ -133,114 +133,114 @@  runFixture :: Config -> KirokuStore -> StreamName -> SubscriptionName -> IO () runFixture config store streamName subscriptionName = do-    producedVar <- newTVarIO (0 :: Int)-    processedRef <- newIORef (0 :: Int)-    failedRef <- newIORef (0 :: Int)-    stopVar <- newTVarIO False-    startTime <- getCurrentTime+  producedVar <- newTVarIO (0 :: Int)+  processedRef <- newIORef (0 :: Int)+  failedRef <- newIORef (0 :: Int)+  stopVar <- newTVarIO False+  startTime <- getCurrentTime -    withDeliveryLedger config $ \ledger ->-        withFile config.outputCsv WriteMode $ \handle -> do-            hPutStrLn handle csvHeader-            producerThread <- Async.async $ runProducer config store streamName producedVar failedRef ledger stopVar-            samplerThread <- Async.async $ runSampler config store subscriptionName startTime producedVar processedRef failedRef handle+  withDeliveryLedger config $ \ledger ->+    withFile config.outputCsv WriteMode $ \handle -> do+      hPutStrLn handle csvHeader+      producerThread <- Async.async $ runProducer config store streamName producedVar failedRef ledger stopVar+      samplerThread <- Async.async $ runSampler config store subscriptionName startTime producedVar processedRef failedRef handle -            runConsumerSegment config store subscriptionName processedRef failedRef ledger $ do-                threadDelay (config.restartAtSecs * 1_000_000)-                putStrLn "Graceful midpoint stop"+      runConsumerSegment config store subscriptionName processedRef failedRef ledger $ do+        threadDelay (config.restartAtSecs * 1_000_000)+        putStrLn "Graceful midpoint stop" -            putStrLn "Restarting with the same durable subscription"-            runConsumerSegment config store subscriptionName processedRef failedRef ledger $ do-                threadDelay ((config.durationSecs - config.restartAtSecs) * 1_000_000)-                atomically $ writeTVar stopVar True-                Async.wait producerThread-                waitForDrain producedVar processedRef 60-                waitForCheckpointDrain store subscriptionName 30+      putStrLn "Restarting with the same durable subscription"+      runConsumerSegment config store subscriptionName processedRef failedRef ledger $ do+        threadDelay ((config.durationSecs - config.restartAtSecs) * 1_000_000)+        atomically $ writeTVar stopVar True+        Async.wait producerThread+        waitForDrain producedVar processedRef 60+        waitForCheckpointDrain store subscriptionName 30 -            Async.cancel samplerThread-            finalSample <- sampleMetrics store subscriptionName startTime producedVar processedRef failedRef-            hPutStrLn handle $ sampleToCsv finalSample-            hFlush handle+      Async.cancel samplerThread+      finalSample <- sampleMetrics store subscriptionName startTime producedVar processedRef failedRef+      hPutStrLn handle $ sampleToCsv finalSample+      hFlush handle -            ledgerPassed <- writeDeliveryLedger config ledger+      ledgerPassed <- writeDeliveryLedger config ledger -            let passed =-                    finalSample.messagesFailed == 0-                        && finalSample.messagesProcessed == finalSample.messagesProduced-                        && finalSample.queueDepth == 0-                        && ledgerPassed-            putStrLn $ "Produced: " <> show finalSample.messagesProduced-            putStrLn $ "Processed: " <> show finalSample.messagesProcessed-            putStrLn $ "Durable backlog: " <> show finalSample.queueDepth-            unless passed exitFailure+      let passed =+            finalSample.messagesFailed == 0+              && finalSample.messagesProcessed == finalSample.messagesProduced+              && finalSample.queueDepth == 0+              && ledgerPassed+      putStrLn $ "Produced: " <> show finalSample.messagesProduced+      putStrLn $ "Processed: " <> show finalSample.messagesProcessed+      putStrLn $ "Durable backlog: " <> show finalSample.queueDepth+      unless passed exitFailure  runProducer :: Config -> KirokuStore -> StreamName -> TVar Int -> IORef Int -> DeliveryLedger -> TVar Bool -> IO () runProducer config store streamName producedVar failedRef ledger stopVar = loop (0 :: Int)   where     delayMicros = 1_000_000 `div` max 1 config.messagesPerSecond     loop index = do-        shouldStop <- readTVarIO stopVar-        unless shouldStop $ do-            let event =-                    EventData-                        { eventId = Nothing-                        , eventType = EventType "Ep45Lifecycle"-                        , payload = object ["sequence" .= index]-                        , metadata = Nothing-                        , causationId = Nothing-                        , correlationId = Nothing-                        }-            result <- runStoreIO store $ appendToStream streamName AnyVersion [event]-            case result of-                Left _ -> atomicModifyIORef' failedRef $ \count -> (count + 1, ())-                Right _ -> do-                    atomically $ modifyTVar' producedVar (+ 1)-                    recordProduced ledger index-            when (delayMicros > 0) $ threadDelay delayMicros-            loop (index + 1)+      shouldStop <- readTVarIO stopVar+      unless shouldStop $ do+        let event =+              EventData+                { eventId = Nothing,+                  eventType = EventType "Ep45Lifecycle",+                  payload = object ["sequence" .= index],+                  metadata = Nothing,+                  causationId = Nothing,+                  correlationId = Nothing+                }+        result <- runStoreIO store $ appendToStream streamName AnyVersion [event]+        case result of+          Left _ -> atomicModifyIORef' failedRef $ \count -> (count + 1, ())+          Right _ -> do+            atomically $ modifyTVar' producedVar (+ 1)+            recordProduced ledger index+        when (delayMicros > 0) $ threadDelay delayMicros+        loop (index + 1)  runConsumerSegment :: Config -> KirokuStore -> SubscriptionName -> IORef Int -> IORef Int -> DeliveryLedger -> IO () -> IO () runConsumerSegment config store subscriptionName processedRef failedRef ledger action =-    runEff $ runTracingNoop $ do-        adapter <- kirokuAdapter store (defaultKirokuAdapterConfig subscriptionName AllStreams)-        result <- runApp defaultAppConfig [(ProcessorId "kiroku-ep45", mkProcessor adapter handler)]-        case result of-            Left err -> liftIO $ error $ "runApp failed: " <> show err-            Right appHandle -> do-                liftIO action-                let shutdownConfig =-                        defaultShutdownConfig-                            { drainTimeout = fromIntegral config.shutdownDrainSecs-                            , totalShutdownTimeout = fromIntegral config.shutdownTotalSecs-                            }-                drained <- stopAppGracefully shutdownConfig appHandle-                unless drained $ liftIO $ error "Shibuya application required forced shutdown"+  runEff $ runTracingNoop $ do+    adapter <- kirokuAdapter store (defaultKirokuAdapterConfig subscriptionName AllStreams)+    result <- runApp defaultAppConfig [(ProcessorId "kiroku-ep45", mkProcessor adapter handler)]+    case result of+      Left err -> liftIO $ error $ "runApp failed: " <> show err+      Right appHandle -> do+        liftIO action+        let shutdownConfig =+              defaultShutdownConfig+                { drainTimeout = fromIntegral config.shutdownDrainSecs,+                  totalShutdownTimeout = fromIntegral config.shutdownTotalSecs+                }+        drained <- stopAppGracefully shutdownConfig appHandle+        unless drained $ liftIO $ error "Shibuya application required forced shutdown"   where     handler message = do-        let Message{envelope = Envelope{payload = recorded}} = message-            sequenceNumber = parseMaybe (withObject "EP-45 event" (.: "sequence")) recorded.payload-        case sequenceNumber of-            Nothing -> do-                liftIO $ do-                    atomicModifyIORef' failedRef $ \count -> (count + 1, ())-                    atomicModifyIORef' ledger.malformedDeliveries $ \count -> (count + 1, ())-                pure $ AckDeadLetter (InvalidPayload "EP-45 ledger sequence missing")-            Just value -> do-                liftIO $ do-                    atomicModifyIORef' processedRef $ \count -> (count + 1, ())-                    recordProcessed ledger value-                pure AckOk+      let Message {envelope = Envelope {payload = recorded}} = message+          sequenceNumber = parseMaybe (withObject "EP-45 event" (.: "sequence")) recorded.payload+      case sequenceNumber of+        Nothing -> do+          liftIO $ do+            atomicModifyIORef' failedRef $ \count -> (count + 1, ())+            atomicModifyIORef' ledger.malformedDeliveries $ \count -> (count + 1, ())+          pure $ AckDeadLetter (InvalidPayload "EP-45 ledger sequence missing")+        Just value -> do+          liftIO $ do+            atomicModifyIORef' processedRef $ \count -> (count + 1, ())+            recordProcessed ledger value+          pure AckOk  withDeliveryLedger :: Config -> (DeliveryLedger -> IO a) -> IO a withDeliveryLedger config action =-    let producedPath = config.outputLedger <> ".produced.ids"-        processedPath = config.outputLedger <> ".processed.ids"-     in withFile producedPath ReadWriteMode $ \producedHandle ->-            withFile processedPath ReadWriteMode $ \processedHandle -> do-                hSetFileSize producedHandle 0-                hSetFileSize processedHandle 0-                malformedDeliveries <- newIORef 0-                action DeliveryLedger{producedHandle, processedHandle, producedPath, processedPath, malformedDeliveries}+  let producedPath = config.outputLedger <> ".produced.ids"+      processedPath = config.outputLedger <> ".processed.ids"+   in withFile producedPath ReadWriteMode $ \producedHandle ->+        withFile processedPath ReadWriteMode $ \processedHandle -> do+          hSetFileSize producedHandle 0+          hSetFileSize processedHandle 0+          malformedDeliveries <- newIORef 0+          action DeliveryLedger {producedHandle, processedHandle, producedPath, processedPath, malformedDeliveries}  recordProduced :: DeliveryLedger -> Int -> IO () recordProduced ledger value = hPutStrLn ledger.producedHandle (show value)@@ -250,154 +250,154 @@  writeDeliveryLedger :: Config -> DeliveryLedger -> IO Bool writeDeliveryLedger config ledger = do-    producedValues <- readIdentityHandle ledger.producedPath ledger.producedHandle-    processedValues <- readIdentityHandle ledger.processedPath ledger.processedHandle-    malformed <- readIORef ledger.malformedDeliveries-    let produced = Set.fromList producedValues-        processed = Set.fromList processedValues-        duplicates = duplicateValues processedValues-    let missing = produced `Set.difference` processed-        unexpected = processed `Set.difference` produced-        passed = Set.null missing && Set.null unexpected && Set.null duplicates && malformed == 0-        artifact =-            object-                [ "schemaVersion" .= (1 :: Int)-                , "adapter" .= ("kiroku" :: String)-                , "runId" .= config.runId-                , "status" .= if passed then ("pass" :: String) else "fail"-                , "producedIds" .= Set.toAscList produced-                , "processedIds" .= Set.toAscList processed-                , "duplicateIds" .= Set.toAscList duplicates-                , "missingIds" .= Set.toAscList missing-                , "unexpectedIds" .= Set.toAscList unexpected-                , "malformedDeliveries" .= malformed-                ]-    LBS.writeFile config.outputLedger (encode artifact)-    putStrLn $ "Delivery ledger: " <> config.outputLedger <> " (" <> if passed then "pass)" else "fail)"-    pure passed+  producedValues <- readIdentityHandle ledger.producedPath ledger.producedHandle+  processedValues <- readIdentityHandle ledger.processedPath ledger.processedHandle+  malformed <- readIORef ledger.malformedDeliveries+  let produced = Set.fromList producedValues+      processed = Set.fromList processedValues+      duplicates = duplicateValues processedValues+  let missing = produced `Set.difference` processed+      unexpected = processed `Set.difference` produced+      passed = Set.null missing && Set.null unexpected && Set.null duplicates && malformed == 0+      artifact =+        object+          [ "schemaVersion" .= (1 :: Int),+            "adapter" .= ("kiroku" :: String),+            "runId" .= config.runId,+            "status" .= if passed then ("pass" :: String) else "fail",+            "producedIds" .= Set.toAscList produced,+            "processedIds" .= Set.toAscList processed,+            "duplicateIds" .= Set.toAscList duplicates,+            "missingIds" .= Set.toAscList missing,+            "unexpectedIds" .= Set.toAscList unexpected,+            "malformedDeliveries" .= malformed+          ]+  LBS.writeFile config.outputLedger (encode artifact)+  putStrLn $ "Delivery ledger: " <> config.outputLedger <> " (" <> if passed then "pass)" else "fail)"+  pure passed  readIdentityHandle :: FilePath -> Handle -> IO [Int] readIdentityHandle path handle = do-    hFlush handle-    hSeek handle AbsoluteSeek 0-    contents <- BS.hGetContents handle-    traverse parseIdentity (filter (not . BS.null) (BS.lines contents))+  hFlush handle+  hSeek handle AbsoluteSeek 0+  contents <- BS.hGetContents handle+  traverse parseIdentity (filter (not . BS.null) (BS.lines contents))   where     parseIdentity raw =-        maybe (ioError $ userError $ "Invalid delivery identity in " <> path) pure (readMaybe $ BS.unpack raw)+      maybe (ioError $ userError $ "Invalid delivery identity in " <> path) pure (readMaybe $ BS.unpack raw)  duplicateValues :: [Int] -> Set Int duplicateValues = snd . foldl' step (Set.empty, Set.empty)   where     step (seen, duplicates) value-        | Set.member value seen = (seen, Set.insert value duplicates)-        | otherwise = (Set.insert value seen, duplicates)+      | Set.member value seen = (seen, Set.insert value duplicates)+      | otherwise = (Set.insert value seen, duplicates)  runSampler :: Config -> KirokuStore -> SubscriptionName -> UTCTime -> TVar Int -> IORef Int -> IORef Int -> Handle -> IO () runSampler config store subscriptionName startTime producedVar processedRef failedRef handle = forever $ do-    threadDelay (config.sampleIntervalSecs * 1_000_000)-    sample <- sampleMetrics store subscriptionName startTime producedVar processedRef failedRef-    hPutStrLn handle $ sampleToCsv sample-    hFlush handle-    putStrLn $-        "["-            <> show sample.elapsedSecs-            <> "s] produced="-            <> show sample.messagesProduced-            <> " processed="-            <> show sample.messagesProcessed-            <> " backlog="-            <> show sample.queueDepth-            <> " retained="-            <> show sample.retainedBytes+  threadDelay (config.sampleIntervalSecs * 1_000_000)+  sample <- sampleMetrics store subscriptionName startTime producedVar processedRef failedRef+  hPutStrLn handle $ sampleToCsv sample+  hFlush handle+  putStrLn $+    "["+      <> show sample.elapsedSecs+      <> "s] produced="+      <> show sample.messagesProduced+      <> " processed="+      <> show sample.messagesProcessed+      <> " backlog="+      <> show sample.queueDepth+      <> " retained="+      <> show sample.retainedBytes  sampleMetrics :: KirokuStore -> SubscriptionName -> UTCTime -> TVar Int -> IORef Int -> IORef Int -> IO Sample sampleMetrics store subscriptionName startTime producedVar processedRef failedRef = do-    now <- getCurrentTime-    produced <- readTVarIO producedVar-    processed <- readIORef processedRef-    failed <- readIORef failedRef-    backlog <- checkpointBacklog store subscriptionName-    (retained, highWater) <- getMemoryBytes-    pure-        Sample-            { timestamp = now-            , elapsedSecs = round $ diffUTCTime now startTime-            , messagesProduced = produced-            , messagesProcessed = processed-            , messagesFailed = failed-            , queueDepth = backlog-            , retainedBytes = retained-            , maxLiveBytes = highWater-            }+  now <- getCurrentTime+  produced <- readTVarIO producedVar+  processed <- readIORef processedRef+  failed <- readIORef failedRef+  backlog <- checkpointBacklog store subscriptionName+  (retained, highWater) <- getMemoryBytes+  pure+    Sample+      { timestamp = now,+        elapsedSecs = round $ diffUTCTime now startTime,+        messagesProduced = produced,+        messagesProcessed = processed,+        messagesFailed = failed,+        queueDepth = backlog,+        retainedBytes = retained,+        maxLiveBytes = highWater+      }  checkpointBacklog :: KirokuStore -> SubscriptionName -> IO Int64 checkpointBacklog store subscriptionName = do-    result <- runStoreIO store subscriptionCheckpointInventory-    inventory <- case result of-        Left err -> error $ "Checkpoint inventory failed: " <> show err-        Right value -> pure value-    let GlobalPosition storePosition = inventory.storePosition-        checkpointPosition =-            case Vector.find isTargetCheckpoint inventory.checkpoints of-                Nothing -> 0-                Just checkpoint ->-                    let GlobalPosition position = checkpoint.checkpointPosition-                     in position-    pure $ max 0 (storePosition - checkpointPosition)+  result <- runStoreIO store subscriptionCheckpointInventory+  inventory <- case result of+    Left err -> error $ "Checkpoint inventory failed: " <> show err+    Right value -> pure value+  let GlobalPosition storePosition = inventory.storePosition+      checkpointPosition =+        case Vector.find isTargetCheckpoint inventory.checkpoints of+          Nothing -> 0+          Just checkpoint ->+            let GlobalPosition position = checkpoint.checkpointPosition+             in position+  pure $ max 0 (storePosition - checkpointPosition)   where     isTargetCheckpoint (SubscriptionCheckpoint name member _ _) =-        name == subscriptionName && member == 0+      name == subscriptionName && member == 0  getMemoryBytes :: IO (Word64, Word64) getMemoryBytes = do-    enabled <- getRTSStatsEnabled-    if enabled-        then do-            performMajorGC-            stats <- getRTSStats-            pure (gcdetails_live_bytes stats.gc, max_live_bytes stats)-        else pure (0, 0)+  enabled <- getRTSStatsEnabled+  if enabled+    then do+      performMajorGC+      stats <- getRTSStats+      pure (gcdetails_live_bytes stats.gc, max_live_bytes stats)+    else pure (0, 0)  waitForDrain :: TVar Int -> IORef Int -> Int -> IO () waitForDrain producedVar processedRef timeoutSecs = loop (timeoutSecs * 10)   where     loop remaining = do-        produced <- readTVarIO producedVar-        processed <- readIORef processedRef-        if processed >= produced-            then pure ()-            else-                if remaining > 0-                    then threadDelay 100_000 >> loop (remaining - 1)-                    else error $ "Timed out draining produced events: produced=" <> show produced <> " processed=" <> show processed+      produced <- readTVarIO producedVar+      processed <- readIORef processedRef+      if processed >= produced+        then pure ()+        else+          if remaining > 0+            then threadDelay 100_000 >> loop (remaining - 1)+            else error $ "Timed out draining produced events: produced=" <> show produced <> " processed=" <> show processed  waitForCheckpointDrain :: KirokuStore -> SubscriptionName -> Int -> IO () waitForCheckpointDrain store subscriptionName timeoutSecs = loop (timeoutSecs * 10)   where     loop remaining = do-        backlog <- checkpointBacklog store subscriptionName-        if backlog <= 0-            then pure ()-            else-                if remaining > 0-                    then threadDelay 100_000 >> loop (remaining - 1)-                    else error $ "Timed out draining Kiroku checkpoint backlog: backlog=" <> show backlog+      backlog <- checkpointBacklog store subscriptionName+      if backlog <= 0+        then pure ()+        else+          if remaining > 0+            then threadDelay 100_000 >> loop (remaining - 1)+            else error $ "Timed out draining Kiroku checkpoint backlog: backlog=" <> show backlog  csvHeader :: String csvHeader = "timestamp,elapsed_secs,produced,processed,failed,queue_depth,retained_bytes,max_live_bytes"  sampleToCsv :: Sample -> String sampleToCsv sample =-    Text.unpack $-        Text.intercalate-            ","-            [ Text.pack $ formatTime defaultTimeLocale "%Y-%m-%d %H:%M:%S" sample.timestamp-            , Text.pack $ show sample.elapsedSecs-            , Text.pack $ show sample.messagesProduced-            , Text.pack $ show sample.messagesProcessed-            , Text.pack $ show sample.messagesFailed-            , Text.pack $ show sample.queueDepth-            , Text.pack $ show sample.retainedBytes-            , Text.pack $ show sample.maxLiveBytes-            ]+  Text.unpack $+    Text.intercalate+      ","+      [ Text.pack $ formatTime defaultTimeLocale "%Y-%m-%d %H:%M:%S" sample.timestamp,+        Text.pack $ show sample.elapsedSecs,+        Text.pack $ show sample.messagesProduced,+        Text.pack $ show sample.messagesProcessed,+        Text.pack $ show sample.messagesFailed,+        Text.pack $ show sample.queueDepth,+        Text.pack $ show sample.retainedBytes,+        Text.pack $ show sample.maxLiveBytes+      ]

@@ -1,10 +1,9 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedRecordDot #-} -{- | MP-12's durable append/subscription comparison probe. Build this identical-workload against the original control and candidate; only the legacy group-constructor needs a compatibility branch. No production instrumentation.--}+-- | MP-12's durable append/subscription comparison probe. Build this identical+-- workload against the original control and candidate; only the legacy group+-- constructor needs a compatibility branch. No production instrumentation. module Main (main) where  import Contravariant.Extras (contrazip2)@@ -43,69 +42,69 @@ import Text.Read (readMaybe)  data Workload = Workload-    { seconds :: !Int-    , mode :: !String-    , width :: !Int-    , appendBatch :: !Int-    , checkpointBatch :: !Int32-    , fresh :: !Bool-    , offered :: !Int-    }+  { seconds :: !Int,+    mode :: !String,+    width :: !Int,+    appendBatch :: !Int,+    checkpointBatch :: !Int32,+    fresh :: !Bool,+    offered :: !Int+  }  main :: IO () main = do-    args <- getArgs-    workload <- case args of-        [duration, mode, width, batch, checkpoint, fresh, offered] ->-            Workload <$> number duration <*> pure mode <*> number width <*> number batch <*> number checkpoint <*> number fresh <*> number offered-        _ -> fail "usage: write-probe SECONDS MODE WIDTH APPEND_BATCH CHECKPOINT_BATCH FRESH OFFERED_CALLS_PER_SEC (MODE=none|all|category|group|adapter)"-    unless (workload.seconds > 0 && workload.width > 0 && workload.appendBatch > 0 && workload.checkpointBatch > 0 && workload.offered >= 0 && workload.mode `elem` ["none", "all", "category", "group", "adapter"]) (fail "invalid workload")-    original <- ephemeralConfig-    let overridden = ["fsync", "full_page_writes", "synchronous_commit", "shared_buffers", "wal_level"]-        durable = original{Pg.postgresSettings = filter (\(key, _) -> key `notElem` overridden) original.postgresSettings ++ [("fsync", "on"), ("synchronous_commit", "on"), ("full_page_writes", "on"), ("shared_buffers", "128MB"), ("wal_level", "replica")]}-    result <- Pg.withCachedConfig durable Pg.defaultCacheConfig $ \database -> do-        migrateTestDatabase (Pg.connectionString database)-        live <- newIORef (0 :: Int)-        failures <- newIORef (0 :: Int)-        batches <- newIORef (0 :: Int)-        let observe = \case-                KirokuEventSubscriptionCaughtUp{} -> atomicModifyIORef' live (\n -> (n + 1, ()))-                KirokuEventSubscriptionDbError{} -> atomicModifyIORef' failures (\n -> (n + 1, ()))-                KirokuEventSubscriptionDelivered{} -> atomicModifyIORef' batches (\n -> (n + 1, ()))-                _ -> pure ()-            settings = (defaultConnectionSettings (Pg.connectionString database)){poolSize = 10, eventHandler = Just observe}-        withStore settings $ \store -> do-            let event = EventData Nothing (EventType "Probe") (object ["body" .= Text.replicate 512 "x"]) Nothing Nothing Nothing-            -- Identical existing-stream fixtures and explicit future-only live entry.-            forM_ [0 .. 3] $ \writer -> forM_ [0 .. workload.width - 1] $ \slot ->-                void $ append store [(stream writer slot 0 False, AnyVersion, [event])]-            delivered <- newIORef (0 :: Int)-            let handler _ = atomicModifyIORef' delivered (\n -> (n + 1, ())) >> pure Continue-                config m =-                    (defaultSubscriptionConfig (SubscriptionName "probe") (workloadTarget workload) handler)-                        { missingCheckpointPolicy = FromCurrentHead-                        , batchSize = either (error . show) Prelude.id (mkBatchSize workload.checkpointBatch)-                        , consumerGroup = if workload.mode == "group" then Just (membership m) else Nothing-                        }-                members = if workload.mode == "group" then [0, 1, 2, 3] else [0]-                native = bracket (mapM (subscribe store . config) members) (mapM_ cancel) $ \_ -> measure workload store event delivered live failures batches (length members)-            case workload.mode of-                "none" -> measure workload store event delivered live failures batches 0-                "adapter" -> runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            (defaultKirokuAdapterConfig (SubscriptionName "probe") AllStreams)-                                { Adapter.missingCheckpointPolicy = FromCurrentHead-                                , Adapter.batchSize = either (error . show) Prelude.id (mkBatchSize workload.checkpointBatch)-                                }-                    app <- runApp defaultAppConfig [(ProcessorId "probe", mkProcessor adapter (\_ -> liftIO (atomicModifyIORef' delivered (\n -> (n + 1, ()))) >> pure AckOk))]-                    case app of-                        Left err -> liftIO $ fail (show err)-                        Right handle -> do-                            liftIO $ measure workload store event delivered live failures batches 1-                            stopApp handle-                _ -> native-    either (fail . show) pure result+  args <- getArgs+  workload <- case args of+    [duration, mode, width, batch, checkpoint, fresh, offered] ->+      Workload <$> number duration <*> pure mode <*> number width <*> number batch <*> number checkpoint <*> number fresh <*> number offered+    _ -> fail "usage: write-probe SECONDS MODE WIDTH APPEND_BATCH CHECKPOINT_BATCH FRESH OFFERED_CALLS_PER_SEC (MODE=none|all|category|group|adapter)"+  unless (workload.seconds > 0 && workload.width > 0 && workload.appendBatch > 0 && workload.checkpointBatch > 0 && workload.offered >= 0 && workload.mode `elem` ["none", "all", "category", "group", "adapter"]) (fail "invalid workload")+  original <- ephemeralConfig+  let overridden = ["fsync", "full_page_writes", "synchronous_commit", "shared_buffers", "wal_level"]+      durable = original {Pg.postgresSettings = filter (\(key, _) -> key `notElem` overridden) original.postgresSettings ++ [("fsync", "on"), ("synchronous_commit", "on"), ("full_page_writes", "on"), ("shared_buffers", "128MB"), ("wal_level", "replica")]}+  result <- Pg.withCachedConfig durable Pg.defaultCacheConfig $ \database -> do+    migrateTestDatabase (Pg.connectionString database)+    live <- newIORef (0 :: Int)+    failures <- newIORef (0 :: Int)+    batches <- newIORef (0 :: Int)+    let observe = \case+          KirokuEventSubscriptionCaughtUp {} -> atomicModifyIORef' live (\n -> (n + 1, ()))+          KirokuEventSubscriptionDbError {} -> atomicModifyIORef' failures (\n -> (n + 1, ()))+          KirokuEventSubscriptionDelivered {} -> atomicModifyIORef' batches (\n -> (n + 1, ()))+          _ -> pure ()+        settings = (defaultConnectionSettings (Pg.connectionString database)) {poolSize = 10, eventHandler = Just observe}+    withStore settings $ \store -> do+      let event = EventData Nothing (EventType "Probe") (object ["body" .= Text.replicate 512 "x"]) Nothing Nothing Nothing+      -- Identical existing-stream fixtures and explicit future-only live entry.+      forM_ [0 .. 3] $ \writer -> forM_ [0 .. workload.width - 1] $ \slot ->+        void $ append store [(stream writer slot 0 False, AnyVersion, [event])]+      delivered <- newIORef (0 :: Int)+      let handler _ = atomicModifyIORef' delivered (\n -> (n + 1, ())) >> pure Continue+          config m =+            (defaultSubscriptionConfig (SubscriptionName "probe") (workloadTarget workload) handler)+              { missingCheckpointPolicy = FromCurrentHead,+                batchSize = either (error . show) Prelude.id (mkBatchSize workload.checkpointBatch),+                consumerGroup = if workload.mode == "group" then Just (membership m) else Nothing+              }+          members = if workload.mode == "group" then [0, 1, 2, 3] else [0]+          native = bracket (mapM (subscribe store . config) members) (mapM_ cancel) $ \_ -> measure workload store event delivered live failures batches (length members)+      case workload.mode of+        "none" -> measure workload store event delivered live failures batches 0+        "adapter" -> runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              (defaultKirokuAdapterConfig (SubscriptionName "probe") AllStreams)+                { Adapter.missingCheckpointPolicy = FromCurrentHead,+                  Adapter.batchSize = either (error . show) Prelude.id (mkBatchSize workload.checkpointBatch)+                }+          app <- runApp defaultAppConfig [(ProcessorId "probe", mkProcessor adapter (\_ -> liftIO (atomicModifyIORef' delivered (\n -> (n + 1, ()))) >> pure AckOk))]+          case app of+            Left err -> liftIO $ fail (show err)+            Right handle -> do+              liftIO $ measure workload store event delivered live failures batches 1+              stopApp handle+        _ -> native+  either (fail . show) pure result   where     number text = maybe (fail ("invalid argument " <> text)) pure (readMaybe text) @@ -127,98 +126,98 @@  measure :: Workload -> KirokuStore -> EventData -> IORef Int -> IORef Int -> IORef Int -> IORef Int -> Int -> IO () measure workload store event delivered live failures batches members = do-    when (members > 0) $ await "live entry" (fmap (>= members) (readIORef live))-    -- Warm up the same append shape for two seconds, then fully drain before-    -- measuring. The output contains only the declared measurement interval.-    void $ writers 2-    when (members > 0) $ await "warmup checkpoint drain" durable-    flushStats store-    writeIORef delivered 0-    writeIORef batches 0-    performMajorGC-    stats0 <- getRTSStats-    wal0 <- scalar store "SELECT pg_current_wal_insert_lsn()::text"-    tables0 <- tableStats store-    start <- getMonotonicTimeNSec-    samples <- writers workload.seconds-    finish <- getMonotonicTimeNSec-    let calls = sum [length latencies | (latencies, _) <- samples]-        events = calls * workload.width * workload.appendBatch-        elapsed = secondsBetween start finish-        sorted = sort (concat [latencies | (latencies, _) <- samples])-    when (members > 0) $ do-        await "delivery drain" (fmap (== events) (readIORef delivered))-        await "durable progress drain" durable-    performMajorGC-    stats1 <- getRTSStats-    errors <- readIORef failures-    unless (errors == 0) (fail "database errors during workload")-    flushStats store-    wal1 <- scalar store "SELECT pg_current_wal_insert_lsn()::text"-    walBytes <- scalarInt store ("SELECT pg_wal_lsn_diff('" <> wal1 <> "'::pg_lsn, '" <> wal0 <> "'::pg_lsn)::bigint")-    tables1 <- tableStats store-    count <- readIORef delivered-    batchCount <- readIORef batches-    server <- scalar store "SELECT version()"-    durability <- scalar store "SELECT current_setting('fsync') || ',' || current_setting('synchronous_commit') || ',' || current_setting('full_page_writes')"-    let (saves0, hot0) = tables0-        (saves1, hot1) = tables1-    Bytes.putStrLn $-        encode $-            object-                [ "workload" .= object ["mode" .= workload.mode, "width" .= workload.width, "append_batch" .= workload.appendBatch, "checkpoint_batch" .= workload.checkpointBatch, "fresh" .= workload.fresh, "offered" .= workload.offered, "seconds" .= workload.seconds]-                , "server" .= server-                , "durability" .= durability-                , "calls" .= calls-                , "events" .= events-                , "elapsed" .= elapsed-                , "events_per_second" .= (fromIntegral events / elapsed :: Double)-                , "append_p50_ms" .= percentile sorted 0.50-                , "append_p95_ms" .= percentile sorted 0.95-                , "append_p99_ms" .= percentile sorted 0.99-                , "delivered" .= count-                , "delivery_batches" .= batchCount-                , "durable_drained" .= True-                , "wal_bytes" .= walBytes-                , "checkpoint_updates" .= (saves1 - saves0)-                , "checkpoint_hot_updates" .= (hot1 - hot0)-                , "allocated_bytes" .= (stats1.allocated_bytes - stats0.allocated_bytes)-                , "gc_cpu_ns" .= (stats1.gc_cpu_ns - stats0.gc_cpu_ns)-                , "gc_elapsed_ns" .= (stats1.gc_elapsed_ns - stats0.gc_elapsed_ns)-                , "max_live_bytes" .= stats1.max_live_bytes-                ]+  when (members > 0) $ await "live entry" (fmap (>= members) (readIORef live))+  -- Warm up the same append shape for two seconds, then fully drain before+  -- measuring. The output contains only the declared measurement interval.+  void $ writers 2+  when (members > 0) $ await "warmup checkpoint drain" durable+  flushStats store+  writeIORef delivered 0+  writeIORef batches 0+  performMajorGC+  stats0 <- getRTSStats+  wal0 <- scalar store "SELECT pg_current_wal_insert_lsn()::text"+  tables0 <- tableStats store+  start <- getMonotonicTimeNSec+  samples <- writers workload.seconds+  finish <- getMonotonicTimeNSec+  let calls = sum [length latencies | (latencies, _) <- samples]+      events = calls * workload.width * workload.appendBatch+      elapsed = secondsBetween start finish+      sorted = sort (concat [latencies | (latencies, _) <- samples])+  when (members > 0) $ do+    await "delivery drain" (fmap (== events) (readIORef delivered))+    await "durable progress drain" durable+  performMajorGC+  stats1 <- getRTSStats+  errors <- readIORef failures+  unless (errors == 0) (fail "database errors during workload")+  flushStats store+  wal1 <- scalar store "SELECT pg_current_wal_insert_lsn()::text"+  walBytes <- scalarInt store ("SELECT pg_wal_lsn_diff('" <> wal1 <> "'::pg_lsn, '" <> wal0 <> "'::pg_lsn)::bigint")+  tables1 <- tableStats store+  count <- readIORef delivered+  batchCount <- readIORef batches+  server <- scalar store "SELECT version()"+  durability <- scalar store "SELECT current_setting('fsync') || ',' || current_setting('synchronous_commit') || ',' || current_setting('full_page_writes')"+  let (saves0, hot0) = tables0+      (saves1, hot1) = tables1+  Bytes.putStrLn $+    encode $+      object+        [ "workload" .= object ["mode" .= workload.mode, "width" .= workload.width, "append_batch" .= workload.appendBatch, "checkpoint_batch" .= workload.checkpointBatch, "fresh" .= workload.fresh, "offered" .= workload.offered, "seconds" .= workload.seconds],+          "server" .= server,+          "durability" .= durability,+          "calls" .= calls,+          "events" .= events,+          "elapsed" .= elapsed,+          "events_per_second" .= (fromIntegral events / elapsed :: Double),+          "append_p50_ms" .= percentile sorted 0.50,+          "append_p95_ms" .= percentile sorted 0.95,+          "append_p99_ms" .= percentile sorted 0.99,+          "delivered" .= count,+          "delivery_batches" .= batchCount,+          "durable_drained" .= True,+          "wal_bytes" .= walBytes,+          "checkpoint_updates" .= (saves1 - saves0),+          "checkpoint_hot_updates" .= (hot1 - hot0),+          "allocated_bytes" .= (stats1.allocated_bytes - stats0.allocated_bytes),+          "gc_cpu_ns" .= (stats1.gc_cpu_ns - stats0.gc_cpu_ns),+          "gc_elapsed_ns" .= (stats1.gc_elapsed_ns - stats0.gc_elapsed_ns),+          "max_live_bytes" .= stats1.max_live_bytes+        ]   where     durable = do-        Right inventory <- runStoreIO store subscriptionCheckpointInventory-        -- Group members have sparse partitions, so their own final event may-        -- precede the global head; verify against the last matching fetch below.-        if workload.mode /= "group"-            then pure (all (\row -> row.checkpointPosition >= inventory.storePosition) (Vector.toList inventory.checkpoints))-            else-                and-                    <$> mapM-                        ( \row -> do-                            n <- runSession store $ Session.statement (let { GlobalPosition p = row.checkpointPosition } in p, row.consumerGroupMember) groupRemaining-                            pure (n == 0)-                        )-                        (Vector.toList inventory.checkpoints)+      Right inventory <- runStoreIO store subscriptionCheckpointInventory+      -- Group members have sparse partitions, so their own final event may+      -- precede the global head; verify against the last matching fetch below.+      if workload.mode /= "group"+        then pure (all (\row -> row.checkpointPosition >= inventory.storePosition) (Vector.toList inventory.checkpoints))+        else+          and+            <$> mapM+              ( \row -> do+                  n <- runSession store $ Session.statement (let { GlobalPosition p = row.checkpointPosition } in p, row.consumerGroupMember) groupRemaining+                  pure (n == 0)+              )+              (Vector.toList inventory.checkpoints)     writers duration = do-        begin <- getMonotonicTimeNSec-        let end = begin + fromIntegral duration * 1_000_000_000-        Async.mapConcurrently (writer begin end) [0 .. 3]+      begin <- getMonotonicTimeNSec+      let end = begin + fromIntegral duration * 1_000_000_000+      Async.mapConcurrently (writer begin end) [0 .. 3]     writer begin end writerId = loop 0 []       where         spacing = if workload.offered == 0 then 0 else 4_000_000_000 `div` fromIntegral workload.offered         loop i collected = do-            now <- getMonotonicTimeNSec-            let scheduled = if spacing == 0 then now else begin + fromIntegral i * spacing-            if now >= end || scheduled >= end-                then pure (collected, ())-                else do-                    when (scheduled > now) (threadDelay (fromIntegral ((scheduled - now) `div` 1000)))-                    void $ append store [(stream writerId slot (i + fromIntegral begin) workload.fresh, AnyVersion, replicate workload.appendBatch event) | slot <- [0 .. workload.width - 1]]-                    completed <- getMonotonicTimeNSec-                    loop (i + 1) (fromIntegral (completed - scheduled) / 1_000_000 : collected)+          now <- getMonotonicTimeNSec+          let scheduled = if spacing == 0 then now else begin + fromIntegral i * spacing+          if now >= end || scheduled >= end+            then pure (collected, ())+            else do+              when (scheduled > now) (threadDelay (fromIntegral ((scheduled - now) `div` 1000)))+              void $ append store [(stream writerId slot (i + fromIntegral begin) workload.fresh, AnyVersion, replicate workload.appendBatch event) | slot <- [0 .. workload.width - 1]]+              completed <- getMonotonicTimeNSec+              loop (i + 1) (fromIntegral (completed - scheduled) / 1_000_000 : collected)  secondsBetween :: Word64 -> Word64 -> Double secondsBetween start finish = fromIntegral (finish - start) / 1_000_000_000@@ -246,10 +245,10 @@  groupRemaining :: Statement (Int64, Int32) Int64 groupRemaining =-    preparable-        "SELECT count(*) FROM kiroku.stream_events WHERE stream_id = 0 AND category = 'probe' AND stream_version > $1 AND (((hashtextextended(original_stream_id::text, 0) % 4) + 4) % 4) = $2"-        (contrazip2 (E.param (E.nonNullable E.int8)) (E.param (E.nonNullable E.int4)))-        (D.singleRow (D.column (D.nonNullable D.int8)))+  preparable+    "SELECT count(*) FROM kiroku.stream_events WHERE stream_id = 0 AND category = 'probe' AND stream_version > $1 AND (((hashtextextended(original_stream_id::text, 0) % 4) + 4) % 4) = $2"+    (contrazip2 (E.param (E.nonNullable E.int8)) (E.param (E.nonNullable E.int4)))+    (D.singleRow (D.column (D.nonNullable D.int8)))  -- Flush backend-local table statistics outside the timed interval. Ten -- concurrent sessions occupy the ten pool slots, including idle writer and@@ -257,6 +256,6 @@ -- into the measured HOT-update deltas. flushStats :: KirokuStore -> IO () flushStats store =-    Async.replicateConcurrently_ 10 $-        runSession store $-            Session.script "SELECT pg_stat_force_next_flush(); SELECT pg_sleep(0.2)"+  Async.replicateConcurrently_ 10 $+    runSession store $+      Session.script "SELECT pg_stat_force_next_flush(); SELECT pg_sleep(0.2)"

@@ -1,58 +1,58 @@-module Shibuya.Adapter.Kiroku.Internal (-    acquireAllAndTransfer,-) where+module Shibuya.Adapter.Kiroku.Internal+  ( acquireAllAndTransfer,+  )+where  import Control.Exception (SomeException) import Data.Int (Int32) import Effectful (Eff, IOE, (:>)) import Effectful.Exception qualified as Exception -{- | Acquire a fixed set of resources and atomically transfer their ownership.--The acquisition itself is restored to the caller's masking state because it may-perform interruptible startup work. As soon as an acquisition returns, the new-resource is added to the masked ownership ledger before the next interruptible-operation can run. If acquisition or transfer fails, every owned resource is-given one release attempt in reverse acquisition order and the original-exception is rethrown even when a release also fails.--The post-acquire hook runs while the new resource is already in the ledger. It-exists so tests can stop at the exact cancellation boundary without adding a-hook to the public adapter API; production callers pass a no-op.--}+-- | Acquire a fixed set of resources and atomically transfer their ownership.+--+-- The acquisition itself is restored to the caller's masking state because it may+-- perform interruptible startup work. As soon as an acquisition returns, the new+-- resource is added to the masked ownership ledger before the next interruptible+-- operation can run. If acquisition or transfer fails, every owned resource is+-- given one release attempt in reverse acquisition order and the original+-- exception is rethrown even when a release also fails.+--+-- The post-acquire hook runs while the new resource is already in the ledger. It+-- exists so tests can stop at the exact cancellation boundary without adding a+-- hook to the public adapter API; production callers pass a no-op. acquireAllAndTransfer ::-    (IOE :> es) =>-    Int32 ->-    (Int32 -> Eff es resource) ->-    (resource -> Eff es ()) ->-    (Int32 -> resource -> Eff es ()) ->-    ([resource] -> Eff es result) ->-    Eff es result+  (IOE :> es) =>+  Int32 ->+  (Int32 -> Eff es resource) ->+  (resource -> Eff es ()) ->+  (Int32 -> resource -> Eff es ()) ->+  ([resource] -> Eff es result) ->+  Eff es result acquireAllAndTransfer count acquire release afterAcquire transfer =-    Exception.mask $ \restore -> go restore [] 0+  Exception.mask $ \restore -> go restore [] 0   where     go restore owned member-        | member >= count = do-            outcome <- tryAny (transfer (reverse owned))-            either (`cleanupAndRethrow` owned) pure outcome-        | otherwise = do-            acquisition <- tryAny (restore (acquire member))-            case acquisition of-                Left primary -> cleanupAndRethrow primary owned-                Right resource -> do-                    let owned' = resource : owned-                    handoff <- tryAny (afterAcquire member resource)-                    case handoff of-                        Left primary -> cleanupAndRethrow primary owned'-                        Right () -> go restore owned' (member + 1)+      | member >= count = do+          outcome <- tryAny (transfer (reverse owned))+          either (`cleanupAndRethrow` owned) pure outcome+      | otherwise = do+          acquisition <- tryAny (restore (acquire member))+          case acquisition of+            Left primary -> cleanupAndRethrow primary owned+            Right resource -> do+              let owned' = resource : owned+              handoff <- tryAny (afterAcquire member resource)+              case handoff of+                Left primary -> cleanupAndRethrow primary owned'+                Right () -> go restore owned' (member + 1)      cleanupAndRethrow primary owned = do-        mapM_ releaseIgnoringFailure owned-        Exception.throwIO primary+      mapM_ releaseIgnoringFailure owned+      Exception.throwIO primary      releaseIgnoringFailure resource = do-        _ <- tryAny (release resource)-        pure ()+      _ <- tryAny (release resource)+      pure ()  tryAny :: Eff es a -> Eff es (Either SomeException a) tryAny = Exception.try

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            shibuya-kiroku-adapter-version:         0.6.0.1+version:         0.7.0.0 synopsis:   Kiroku event store adapter for the Shibuya queue processing framework @@ -65,8 +65,8 @@     , effectful-core                         >=2.6.1     && <2.7  || >=2.7.1.1 && <2.8     , hs-opentelemetry-api                   ^>=1.0     , hs-opentelemetry-semantic-conventions  ^>=1.40-    , kiroku-store                           ^>=0.11.0.0-    , shibuya-core                           >=0.10      && <0.11+    , kiroku-store                           ^>=0.12.0.0+    , shibuya-core                           >=0.12      && <0.13     , shibuya-kiroku-adapter-internal     , stm                                    >=2.5       && <2.6     , streamly-core                          >=0.3       && <0.4@@ -96,10 +96,10 @@     , hasql-pool                       >=1.2       && <1.5     , hs-opentelemetry-api             ^>=1.0     , hspec                            >=2.10      && <2.12-    , kiroku-store                     ^>=0.11.0.0+    , kiroku-store                     ^>=0.12.0.0     , kiroku-test-support              ^>=0.1.0.0     , lens                             >=5.2       && <5.4-    , shibuya-core                     >=0.10      && <0.11+    , shibuya-core                     >=0.12      && <0.13     , shibuya-kiroku-adapter     , shibuya-kiroku-adapter-internal     , stm                              >=2.5       && <2.6@@ -130,9 +130,9 @@     , containers              >=0.6       && <0.8     , effectful-core          >=2.6.1     && <2.7  || >=2.7.1.1 && <2.8     , ephemeral-pg            >=0.3.1     && <0.4-    , kiroku-store            ^>=0.11.0.0+    , kiroku-store            ^>=0.12.0.0     , kiroku-test-support     ^>=0.1.0.0-    , shibuya-core            >=0.10      && <0.11+    , shibuya-core            >=0.12      && <0.13     , shibuya-kiroku-adapter     , stm                     >=2.5       && <2.6     , text                    >=2.0       && <2.2@@ -148,18 +148,18 @@   default-extensions: OverloadedRecordDot   ghc-options:        -threaded -rtsopts "-with-rtsopts=-N4 -T -A32m" -O2   build-depends:-    , aeson                   >=2.1   && <2.3-    , async                   >=2.2   && <2.3-    , base                    >=4.18  && <5-    , bytestring              >=0.11  && <0.13+    , aeson                   >=2.1       && <2.3+    , async                   >=2.2       && <2.3+    , base                    >=4.18      && <5+    , bytestring              >=0.11      && <0.13     , contravariant-extras    >=0.3-    , effectful-core          >=2.6.1 && <2.7  || >=2.7.1.1 && <2.8-    , ephemeral-pg            >=0.3.1 && <0.4-    , hasql                   >=1.10  && <1.11-    , hasql-pool              >=1.2   && <1.5-    , kiroku-store+    , effectful-core          >=2.6.1     && <2.7  || >=2.7.1.1 && <2.8+    , ephemeral-pg            >=0.3.1     && <0.4+    , hasql                   >=1.10      && <1.11+    , hasql-pool              >=1.2       && <1.5+    , kiroku-store            ^>=0.12.0.0     , kiroku-test-support-    , shibuya-core            >=0.10  && <0.11+    , shibuya-core            >=0.12      && <0.13     , shibuya-kiroku-adapter-    , text                    >=2.0   && <2.2-    , vector                  >=0.13  && <0.14+    , text                    >=2.0       && <2.2+    , vector                  >=0.13      && <0.14

@@ -1,125 +1,124 @@-{- | Kiroku event store adapter for the Shibuya queue processing framework.--This adapter wraps Kiroku's push-based subscriptions into Shibuya's-pull-based 'Adapter' interface. Events are bridged through a bounded-'TBQueue' (via @kiroku-store@'s 'subscriptionStream') and lifted into the-effectful stack with @Stream.morphInner@.--== Example--@-import Effectful (runEff)-import Kiroku.Store (withStore, defaultConnectionSettings)-import Shibuya.Adapter.Kiroku-import Shibuya.App-import Shibuya.Core.Ack (AckDecision (..))-import Shibuya.Telemetry.Effect (runTracingNoop)--main :: IO ()-main = withStore settings $ \\store ->-    runEff $ runTracingNoop $ do-        let cfg = defaultKirokuAdapterConfig (SubscriptionName \"my-projection\") AllStreams-        adapter <- kirokuAdapter store cfg--        let handler ingested = do-                -- process ingested.envelope.payload :: RecordedEvent-                pure AckOk--        Right appHandle <- runApp defaultAppConfig-            [(ProcessorId \"my-projection\", kirokuProcessor adapter handler)]--        waitApp appHandle-@--== Consumer-Group Example (size 4)--A consumer group splits one logical subscription across @N@ members. Each-originating stream is deterministically assigned to exactly one member (by a-hash computed in PostgreSQL), so same-stream events stay ordered while distinct-streams are processed in parallel. To run a whole group in one process, use-'kirokuConsumerGroupProcessors': one call yields @N@ named processors, each a-member adapter pinned to the group-level @('PartitionedInOrder', 'Serial')@-policy — no manual @[0..N-1]@ wiring.--@-main :: IO ()-main = withStore settings $ \\store ->-    runEff $ runTracingNoop $ do-        groupSize <- either (fail . show) pure (mkConsumerGroupSize 4)-        let cfg = defaultConsumerGroupConfig-                (SubscriptionName \"orders-projection\")-                (Category (CategoryName \"orders\"))-                groupSize--        Right processors <- kirokuConsumerGroupProcessors store cfg handler-        Right appHandle <- runApp defaultAppConfig processors-        waitApp appHandle-  where-    handler ingested = do-        -- process ingested.envelope.payload :: RecordedEvent-        pure AckOk-@--To run members across separate processes instead, give each process one-'kirokuAdapter' with its own 'member' index and the same 'subscriptionName'.-Kiroku's per-member checkpoint (keyed by @(subscriptionName, member)@) lets each-process resume from its own position after a restart. Exactly one live process-must own each member index at a time.--== Ack Semantics--The adapter bridges through @kiroku-store@'s __ack-coupled__ stream-('subscriptionAckStream'): for each event the Kiroku worker blocks until the-Shibuya handler's 'AckDecision' is finalized, then acts on it. The handler's-decision therefore drives Kiroku checkpointing per event:--* 'AckOk' — the worker checkpoints past the event (the normal case).-* 'AckRetry' @delay@ — the worker redelivers the /same/ event after @delay@,-  bounded by the subscription's retry policy-  ('Kiroku.Store.Subscription.Types.RetryPolicy', default five attempts); on-  exhaustion the event is dead-lettered with-  'Kiroku.Store.Subscription.Types.DeadLetterMaxAttempts'.-* 'AckDeadLetter' @reason@ — the worker records the event in-  @kiroku.dead_letters@ (with the reason translated to a Kiroku-native-  'Kiroku.Store.Subscription.Types.DeadLetterReason') and atomically advances-  the checkpoint past it.-* 'AckHalt' — cancels the underlying Kiroku subscription (no checkpoint advance,-  so the halting event replays on restart).--The envelope's @attempt@ reports the zero-based redelivery count, so a handler-can observe how many times Kiroku has redelivered an event.--== Backpressure and Handler Exceptions--Because delivery is ack-coupled, the Kiroku subscription worker blocks on each-event until the Shibuya handler finalizes its decision, providing natural-backpressure. The @bufferSize@ field is only the capacity of the bridge queue-between the worker and the Shibuya stream consumer; for this adapter the-effective depth is at most one event because the worker waits for an ack before-delivering the next event, and @bufferSize@ must be at least 1.--The @queueCapacity@ field is the publisher-side burst knob: it is the number of-publisher batches, up to 1000 events each, that can be buffered before Kiroku-pauses this subscriber. The adapter uses Kiroku's lossless @PauseAndResume@-overflow policy, so a paused subscriber catches up from its checkpoint instead-of being killed.--Shibuya's supervised runner converts a synchronous handler exception to an-immediate 'AckRetry' and finalizes it, so the ack-coupled Kiroku worker cannot be-left blocked by an abandoned reply. 'kirokuProcessor' applies the one-second-paced guard for a single processor. 'guardKirokuHandlerWith' remains useful when-the application wants a different exception disposition, and-'kirokuConsumerGroupProcessors' applies the adapter's default guard-automatically. Asynchronous cancellation is never converted into an ack.--Raw consumers of @adapter.source@ must finalize every item. Leaving one pending-blocks delivery and checkpoint advancement. Opt in with a positive-@handlerStallWarnAfter@ to receive advisory store handler-stall events; warnings-never finalize, retry or checkpoint the item. @retryPolicy@ controls total-deliveries independently of the delay chosen by 'AckRetry'.--}-module Shibuya.Adapter.Kiroku (-    -- * Adapter+-- | Kiroku event store adapter for the Shibuya queue processing framework.+--+-- This adapter wraps Kiroku's push-based subscriptions into Shibuya's+-- pull-based 'Adapter' interface. Events are bridged through a bounded+-- 'TBQueue' (via @kiroku-store@'s 'subscriptionStream') and lifted into the+-- effectful stack with @Stream.morphInner@.+--+-- == Example+--+-- @+-- import Effectful (runEff)+-- import Kiroku.Store (withStore, defaultConnectionSettings)+-- import Shibuya.Adapter.Kiroku+-- import Shibuya.App+-- import Shibuya.Core.Ack (AckDecision (..))+-- import Shibuya.Telemetry.Effect (runTracingNoop)+--+-- main :: IO ()+-- main = withStore settings $ \\store ->+--     runEff $ runTracingNoop $ do+--         let cfg = defaultKirokuAdapterConfig (SubscriptionName \"my-projection\") AllStreams+--         adapter <- kirokuAdapter store cfg+--+--         let handler ingested = do+--                 -- process ingested.envelope.payload :: RecordedEvent+--                 pure AckOk+--+--         Right appHandle <- runApp defaultAppConfig+--             [(ProcessorId \"my-projection\", kirokuProcessor adapter handler)]+--+--         waitApp appHandle+-- @+--+-- == Consumer-Group Example (size 4)+--+-- A consumer group splits one logical subscription across @N@ members. Each+-- originating stream is deterministically assigned to exactly one member (by a+-- hash computed in PostgreSQL), so same-stream events stay ordered while distinct+-- streams are processed in parallel. To run a whole group in one process, use+-- 'kirokuConsumerGroupProcessors': one call yields @N@ named processors, each a+-- member adapter pinned to the group-level @('PartitionedInOrder', 'Serial')@+-- policy — no manual @[0..N-1]@ wiring.+--+-- @+-- main :: IO ()+-- main = withStore settings $ \\store ->+--     runEff $ runTracingNoop $ do+--         groupSize <- either (fail . show) pure (mkConsumerGroupSize 4)+--         let cfg = defaultConsumerGroupConfig+--                 (SubscriptionName \"orders-projection\")+--                 (Category (CategoryName \"orders\"))+--                 groupSize+--+--         Right processors <- kirokuConsumerGroupProcessors store cfg handler+--         Right appHandle <- runApp defaultAppConfig processors+--         waitApp appHandle+--   where+--     handler ingested = do+--         -- process ingested.envelope.payload :: RecordedEvent+--         pure AckOk+-- @+--+-- To run members across separate processes instead, give each process one+-- 'kirokuAdapter' with its own 'member' index and the same 'subscriptionName'.+-- Kiroku's per-member checkpoint (keyed by @(subscriptionName, member)@) lets each+-- process resume from its own position after a restart. Exactly one live process+-- must own each member index at a time.+--+-- == Ack Semantics+--+-- The adapter bridges through @kiroku-store@'s __ack-coupled__ stream+-- ('subscriptionAckStream'): for each event the Kiroku worker blocks until the+-- Shibuya handler's 'AckDecision' is finalized, then acts on it. The handler's+-- decision therefore drives Kiroku checkpointing per event:+--+-- * 'AckOk' — the worker checkpoints past the event (the normal case).+-- * 'AckRetry' @delay@ — the worker redelivers the /same/ event after @delay@,+--   bounded by the subscription's retry policy+--   ('Kiroku.Store.Subscription.Types.RetryPolicy', default five attempts); on+--   exhaustion the event is dead-lettered with+--   'Kiroku.Store.Subscription.Types.DeadLetterMaxAttempts'.+-- * 'AckDeadLetter' @reason@ — the worker records the event in+--   @kiroku.dead_letters@ (with the reason translated to a Kiroku-native+--   'Kiroku.Store.Subscription.Types.DeadLetterReason') and atomically advances+--   the checkpoint past it.+-- * 'AckHalt' — cancels the underlying Kiroku subscription (no checkpoint advance,+--   so the halting event replays on restart).+--+-- The envelope's @attempt@ reports the zero-based redelivery count, so a handler+-- can observe how many times Kiroku has redelivered an event.+--+-- == Backpressure and Handler Exceptions+--+-- Because delivery is ack-coupled, the Kiroku subscription worker blocks on each+-- event until the Shibuya handler finalizes its decision, providing natural+-- backpressure. The @bufferSize@ field is only the capacity of the bridge queue+-- between the worker and the Shibuya stream consumer; for this adapter the+-- effective depth is at most one event because the worker waits for an ack before+-- delivering the next event, and @bufferSize@ must be at least 1.+--+-- The @queueCapacity@ field is the publisher-side burst knob: it is the number of+-- publisher batches, up to 1000 events each, that can be buffered before Kiroku+-- pauses this subscriber. The adapter uses Kiroku's lossless @PauseAndResume@+-- overflow policy, so a paused subscriber catches up from its checkpoint instead+-- of being killed.+--+-- Shibuya's supervised runner converts a synchronous handler exception to an+-- immediate 'AckRetry' and finalizes it, so the ack-coupled Kiroku worker cannot be+-- left blocked by an abandoned reply. 'kirokuProcessor' applies the one-second+-- paced guard for a single processor. 'guardKirokuHandlerWith' remains useful when+-- the application wants a different exception disposition, and+-- 'kirokuConsumerGroupProcessors' applies the adapter's default guard+-- automatically. Asynchronous cancellation is never converted into an ack.+--+-- Raw consumers of @adapter.source@ must finalize every item. Leaving one pending+-- blocks delivery and checkpoint advancement. Opt in with a positive+-- @handlerStallWarnAfter@ to receive advisory store handler-stall events; warnings+-- never finalize, retry or checkpoint the item. @retryPolicy@ controls total+-- deliveries independently of the delay chosen by 'AckRetry'.+module Shibuya.Adapter.Kiroku+  ( -- * Adapter     kirokuAdapter,     kirokuProcessor,     guardKirokuHandlerWith,@@ -148,7 +147,8 @@     size,     EventTypeFilter (..),     MissingCheckpointPolicy (..),-) where+  )+where  import Control.Exception (SomeException) import Data.Int (Int32)@@ -159,8 +159,8 @@ import GHC.Generics (Generic) import Kiroku.Store.Connection (KirokuStore) import Kiroku.Store.Subscription.Stream (StreamBufferSize, defaultStreamBufferSize, subscriptionAckStream)-import Kiroku.Store.Subscription.Types (-    ConsumerGroup,+import Kiroku.Store.Subscription.Types+  ( ConsumerGroup,     ConsumerGroupSize,     EventTypeFilter (..),     MissingCheckpointPolicy (..),@@ -174,7 +174,7 @@     mkConsumerGroup,     mkConsumerGroupSize,     size,- )+  ) import Kiroku.Store.Subscription.Types qualified as Sub import Kiroku.Store.Types (RecordedEvent) import Numeric.Natural (Natural)@@ -188,395 +188,377 @@ import Shibuya.Policy (Concurrency (..), OrderingPolicy (..), validatePolicy) import Streamly.Data.Stream qualified as Stream -{- | Configuration for creating a Kiroku adapter.--@subscriptionName@ must be unique across all active subscriptions — it-identifies the checkpoint row in the @subscriptions@ table.--@bufferSize@ is the bridge queue capacity and must be at least 1. With this-ack-coupled adapter the effective depth is at most one event, because the-worker blocks until the handler's decision is finalized.--@queueCapacity@ is the publisher-side burst capacity in batches. When a burst-exceeds it, the adapter relies on Kiroku's default @PauseAndResume@ policy:-Kiroku pauses the subscriber and later resumes losslessly from its checkpoint.--}+-- | Configuration for creating a Kiroku adapter.+--+-- @subscriptionName@ must be unique across all active subscriptions — it+-- identifies the checkpoint row in the @subscriptions@ table.+--+-- @bufferSize@ is the bridge queue capacity and must be at least 1. With this+-- ack-coupled adapter the effective depth is at most one event, because the+-- worker blocks until the handler's decision is finalized.+--+-- @queueCapacity@ is the publisher-side burst capacity in batches. When a burst+-- exceeds it, the adapter relies on Kiroku's default @PauseAndResume@ policy:+-- Kiroku pauses the subscriber and later resumes losslessly from its checkpoint. data KirokuAdapterConfig = KirokuAdapterConfig-    { subscriptionName :: !SubscriptionName-    -- ^ Unique subscription identifier (checkpoint key)-    , subscriptionTarget :: !SubscriptionTarget-    -- ^ 'AllStreams' or @'Category' categoryName@-    , batchSize :: !Sub.BatchSize-    -- ^ Events per database fetch during catch-up-    , bufferSize :: !StreamBufferSize-    -- ^ Bridge 'TBQueue' capacity; must be at least 1.-    , retryPolicy :: !Sub.RetryPolicy-    -- ^ Total delivery attempts, default five; each AckRetry chooses its delay.-    , handlerStallWarnAfter :: !(Maybe NominalDiffTime)-    -- ^ Optional positive advisory warning interval, disabled by default.-    , queueCapacity :: !Natural-    {- ^ Publisher-side capacity in batches, where each batch contains up to-    Kiroku's publisher batch size (currently 1000 events). When this fills,-    Kiroku pauses and later resumes the subscriber losslessly.-    -}-    , consumerGroup :: !(Maybe ConsumerGroup)-    {- ^ Optional consumer-group membership for this adapter instance.-    'Nothing' (the default) = ordinary single-consumer subscription.-    @'Just' cg@ (built with 'mkConsumerGroup') = this adapter is-    member @m@ of a group of size @n@, receiving only the events whose-    originating stream hashes to slot @m@ (in global-position order). To run a-    full size-@n@ group, create @n@ adapters with the same 'subscriptionName'-    and distinct 'member' indices, each backed by its own Shibuya processor.--    'mkConsumerGroupSize' and 'mkConsumerGroup' validate this invariant before-    an adapter can be configured.-    -}-    , missingCheckpointPolicy :: !MissingCheckpointPolicy-    {- ^ What the underlying Kiroku worker does when this adapter's exact-    @(subscriptionName, consumer-group member)@ checkpoint row is absent.-    'FromBeginning' is the compatibility default; use 'FromCurrentHead' for a-    future-only processor or 'FailIfMissing' when prior provisioning is-    mandatory. Existing checkpoints always win.-    -}-    , eventTypeFilter :: !EventTypeFilter-    {- ^ Which event types this adapter delivers. Pass 'AllEventTypes' (deliver-    everything) or @'OnlyEventTypes' s@ to receive only events whose type is in-    @s@. Forwarded into the underlying subscription; filtering is worker-side-    (before the ack-coupled bridge), so a filtered-out event never reaches the-    Shibuya handler, is never retried or dead-lettered, and the checkpoint still-    advances past it. 'Shibuya.Adapter.Kiroku.Convert' and the 'AckHandle' are-    unaffected.-    -}-    , selector :: !(Maybe (RecordedEvent -> Bool))-    {- ^ Optional opaque per-event predicate, the escape hatch for filtering this-    adapter's stream on a property 'eventTypeFilter' cannot express (e.g.-    payload, metadata, or correlation\/causation ids). Default 'Nothing' (no-    extra filtering). Forwarded into the underlying subscription and composed-    with 'eventTypeFilter' as a logical AND: an event reaches the Shibuya handler-    only when it passes both. Like 'eventTypeFilter' it is applied worker-side-    before the ack-coupled bridge, so a rejected event is never retried or-    dead-lettered and the checkpoint still advances past it. See-    'Kiroku.Store.Subscription.Types.selector' for when to prefer it over the-    introspectable 'eventTypeFilter'.-    -}-    }-    deriving stock (Generic)--{- | A 'KirokuAdapterConfig' with sensible defaults: @batchSize = Sub.defaultBatchSize@,-@bufferSize = defaultStreamBufferSize@, @queueCapacity = 16@, @consumerGroup = 'Nothing'@-(ordinary single-consumer subscription), @missingCheckpointPolicy =-'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@-(deliver every type), and @selector = 'Nothing'@ (no extra predicate-filtering). Supply the subscription name and target; override individual fields-with record-update syntax.--Prefer this over a full record literal so that any field added to-'KirokuAdapterConfig' later is inherited at its default automatically:+  { -- | Unique subscription identifier (checkpoint key)+    subscriptionName :: !SubscriptionName,+    -- | 'AllStreams', exact @'Category' categoryName@, or literal+    -- @'CategoryNamespace' "pm"@ (category @pm@ and its @pm:*@ descendants).+    -- Groups use the same source-stream partitioning for every target.+    -- Namespace checkpoints can advance beyond the last delivered match;+    -- each member scans bounded global windows and periodically checks for+    -- missed notifications. Requires migration 0016 before starting workers.+    subscriptionTarget :: !SubscriptionTarget,+    -- | Events per database fetch during catch-up+    batchSize :: !Sub.BatchSize,+    -- | Bridge 'TBQueue' capacity; must be at least 1.+    bufferSize :: !StreamBufferSize,+    -- | Total delivery attempts, default five; each AckRetry chooses its delay.+    retryPolicy :: !Sub.RetryPolicy,+    -- | Optional positive advisory warning interval, disabled by default.+    handlerStallWarnAfter :: !(Maybe NominalDiffTime),+    -- | Publisher-side capacity in batches, where each batch contains up to+    --     Kiroku's publisher batch size (currently 1000 events). When this fills,+    --     Kiroku pauses and later resumes the subscriber losslessly.+    queueCapacity :: !Natural,+    -- | Optional consumer-group membership for this adapter instance.+    --     'Nothing' (the default) = ordinary single-consumer subscription.+    --     @'Just' cg@ (built with 'mkConsumerGroup') = this adapter is+    --     member @m@ of a group of size @n@, receiving only the events whose+    --     originating stream hashes to slot @m@ (in global-position order). To run a+    --     full size-@n@ group, create @n@ adapters with the same 'subscriptionName'+    --     and distinct 'member' indices, each backed by its own Shibuya processor.+    --+    --     'mkConsumerGroupSize' and 'mkConsumerGroup' validate this invariant before+    --     an adapter can be configured.+    consumerGroup :: !(Maybe ConsumerGroup),+    -- | What the underlying Kiroku worker does when this adapter's exact+    --     @(subscriptionName, consumer-group member)@ checkpoint row is absent.+    --     'FromBeginning' is the compatibility default; use 'FromCurrentHead' for a+    --     future-only processor or 'FailIfMissing' when prior provisioning is+    --     mandatory. Existing checkpoints always win.+    missingCheckpointPolicy :: !MissingCheckpointPolicy,+    -- | Which event types this adapter delivers. Pass 'AllEventTypes' (deliver+    --     everything) or @'OnlyEventTypes' s@ to receive only events whose type is in+    --     @s@. Forwarded into the underlying subscription; filtering is worker-side+    --     (before the ack-coupled bridge), so a filtered-out event never reaches the+    --     Shibuya handler, is never retried or dead-lettered, and the checkpoint still+    --     advances past it. 'Shibuya.Adapter.Kiroku.Convert' and the 'AckHandle' are+    --     unaffected.+    eventTypeFilter :: !EventTypeFilter,+    -- | Optional opaque per-event predicate, the escape hatch for filtering this+    --     adapter's stream on a property 'eventTypeFilter' cannot express (e.g.+    --     payload, metadata, or correlation\/causation ids). Default 'Nothing' (no+    --     extra filtering). Forwarded into the underlying subscription and composed+    --     with 'eventTypeFilter' as a logical AND: an event reaches the Shibuya handler+    --     only when it passes both. Like 'eventTypeFilter' it is applied worker-side+    --     before the ack-coupled bridge, so a rejected event is never retried or+    --     dead-lettered and the checkpoint still advances past it. See+    --     'Kiroku.Store.Subscription.Types.selector' for when to prefer it over the+    --     introspectable 'eventTypeFilter'.+    selector :: !(Maybe (RecordedEvent -> Bool))+  }+  deriving stock (Generic) -@-let cfg =-        (defaultKirokuAdapterConfig "my-projection" 'AllStreams')-            { eventTypeFilter = 'OnlyEventTypes' (Set.fromList [EventType "OrderPlaced"]) }-adapter <- kirokuAdapter store cfg-@--}+-- | A 'KirokuAdapterConfig' with sensible defaults: @batchSize = Sub.defaultBatchSize@,+-- @bufferSize = defaultStreamBufferSize@, @queueCapacity = 16@, @consumerGroup = 'Nothing'@+-- (ordinary single-consumer subscription), @missingCheckpointPolicy =+-- 'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@+-- (deliver every type), and @selector = 'Nothing'@ (no extra predicate+-- filtering). Supply the subscription name and target; override individual fields+-- with record-update syntax.+--+-- Prefer this over a full record literal so that any field added to+-- 'KirokuAdapterConfig' later is inherited at its default automatically:+--+-- @+-- let cfg =+--         (defaultKirokuAdapterConfig "my-projection" 'AllStreams')+--             { eventTypeFilter = 'OnlyEventTypes' (Set.fromList [EventType "OrderPlaced"]) }+-- adapter <- kirokuAdapter store cfg+-- @ defaultKirokuAdapterConfig ::-    SubscriptionName -> SubscriptionTarget -> KirokuAdapterConfig+  SubscriptionName -> SubscriptionTarget -> KirokuAdapterConfig defaultKirokuAdapterConfig name target =-    KirokuAdapterConfig-        { subscriptionName = name-        , subscriptionTarget = target-        , batchSize = Sub.defaultBatchSize-        , bufferSize = defaultStreamBufferSize-        , retryPolicy = Sub.defaultRetryPolicy-        , handlerStallWarnAfter = Nothing-        , queueCapacity = 16-        , consumerGroup = Nothing-        , missingCheckpointPolicy = FromBeginning-        , eventTypeFilter = AllEventTypes-        , selector = Nothing-        }--{- | Convert any synchronous exception thrown by a Shibuya handler into an-'AckDecision'.+  KirokuAdapterConfig+    { subscriptionName = name,+      subscriptionTarget = target,+      batchSize = Sub.defaultBatchSize,+      bufferSize = defaultStreamBufferSize,+      retryPolicy = Sub.defaultRetryPolicy,+      handlerStallWarnAfter = Nothing,+      queueCapacity = 16,+      consumerGroup = Nothing,+      missingCheckpointPolicy = FromBeginning,+      eventTypeFilter = AllEventTypes,+      selector = Nothing+    } -This ensures Shibuya still finalizes the ack. Asynchronous exceptions such as-thread cancellation are not caught.--}+-- | Convert any synchronous exception thrown by a Shibuya handler into an+-- 'AckDecision'.+--+-- This ensures Shibuya still finalizes the ack. Asynchronous exceptions such as+-- thread cancellation are not caught. guardKirokuHandlerWith ::-    (SomeException -> AckDecision) ->-    Handler es msg ->-    Handler es msg+  (SomeException -> AckDecision) ->+  Handler es msg ->+  Handler es msg guardKirokuHandlerWith handleException h ingested =-    h ingested `catchSync` (pure . handleException)--{- | Recommended handler guard for this adapter.+  h ingested `catchSync` (pure . handleException) -A synchronous exception becomes @'AckRetry' ('RetryDelay' 1)@. Kiroku then-redelivers after one second and eventually dead-letters persistent failures-according to the subscription retry policy.--}+-- | Recommended handler guard for this adapter.+--+-- A synchronous exception becomes @'AckRetry' ('RetryDelay' 1)@. Kiroku then+-- redelivers after one second and eventually dead-letters persistent failures+-- according to the subscription retry policy. guardKirokuHandler :: Handler es msg -> Handler es msg guardKirokuHandler = guardKirokuHandlerWith (const (AckRetry (RetryDelay 1))) -{- | Recommended single-processor constructor. Like mkProcessor, defaults to-unordered policy and serial concurrency, with the paced exception guard.--}+-- | Recommended single-processor constructor. Like mkProcessor, defaults to+-- unordered policy and serial concurrency, with the paced exception guard. kirokuProcessor :: Adapter es RecordedEvent -> Handler es RecordedEvent -> QueueProcessor es kirokuProcessor adapter handler = mkProcessor adapter (guardKirokuHandler handler) -{- | Create a Shibuya 'Adapter' backed by a Kiroku subscription.--The adapter:--1. Calls 'subscriptionStream' to start a Kiroku subscription with a-   ack-coupled bounded bridge.-2. Lifts the @Stream IO RecordedEvent@ to @Stream (Eff es)@ via-   @Stream.morphInner liftIO@.-3. Wraps each 'RecordedEvent' into an 'Ingested' value with an-   'Envelope' (mapping event ID → message ID, global position → cursor)-   and an 'AckHandle' whose finalized decision drives Kiroku checkpointing,-   retries, dead-lettering, or halt.--The returned adapter's @shutdown@ action cancels the underlying-subscription and wakes any blocked stream reader. If the subscription worker-dies with an exception, @source@ terminates with that exception.--}+-- | Create a Shibuya 'Adapter' backed by a Kiroku subscription.+--+-- The adapter:+--+-- 1. Calls 'subscriptionStream' to start a Kiroku subscription with a+--    ack-coupled bounded bridge.+-- 2. Lifts the @Stream IO RecordedEvent@ to @Stream (Eff es)@ via+--    @Stream.morphInner liftIO@.+-- 3. Wraps each 'RecordedEvent' into an 'Ingested' value with an+--    'Envelope' (mapping event ID → message ID, global position → cursor)+--    and an 'AckHandle' whose finalized decision drives Kiroku checkpointing,+--    retries, dead-lettering, or halt.+--+-- The returned adapter's @shutdown@ action cancels the underlying+-- subscription and wakes any blocked stream reader. If the subscription worker+-- dies with an exception, @source@ terminates with that exception. kirokuAdapter ::-    (IOE :> es) =>-    KirokuStore ->-    KirokuAdapterConfig ->-    Eff es (Adapter es RecordedEvent)-kirokuAdapter store KirokuAdapterConfig{subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, consumerGroup = cg, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} = do-    -- Build from 'defaultSubscriptionConfig' and override only the non-default-    -- fields. Using the smart constructor (rather than a full record literal)-    -- means any future field added to 'SubscriptionConfigM' is inherited at its-    -- default automatically — e.g. EP-2's 'consumerGroupGuard', left 'False' here.-    let subConfig :: SubscriptionConfig-        subConfig =-            (defaultSubscriptionConfig subName subTarget (\_ -> pure Continue))-                { Sub.batchSize = bs-                , Sub.retryPolicy = attempts-                , Sub.handlerStallWarnAfter = stallWarn-                , Sub.queueCapacity = qCap-                , Sub.consumerGroup = cg-                , Sub.missingCheckpointPolicy = checkpointPolicy-                , Sub.eventTypeFilter = etf-                , Sub.selector = sel-                }--    (ioStream, cancelAction) <- liftIO $ subscriptionAckStream store subConfig buf--    -- The subscription name and consumer-group member are known only here (not on-    -- the RecordedEvent), so thread them into the conversion as OTel attributes-    -- that ride onto Shibuya's per-message span (EP-5 M2).-    let SubscriptionName subNameText = subName-        -- Precompute the constant kiroku.* attributes once per adapter (not per-        -- event): kirokuEnvelopeAttrs builds the base map, and the per-event-        -- conversion only inserts the event type and global position.-        envAttrs =-            kirokuEnvelopeAttrs-                subNameText-                (fmap (fromIntegral . member) cg)-        ingestedStream = fmap (toIngestedAck envAttrs cancelAction) (Stream.morphInner liftIO ioStream)+  (IOE :> es) =>+  KirokuStore ->+  KirokuAdapterConfig ->+  Eff es (Adapter es RecordedEvent)+kirokuAdapter store KirokuAdapterConfig {subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, consumerGroup = cg, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} = do+  -- Build from 'defaultSubscriptionConfig' and override only the non-default+  -- fields. Using the smart constructor (rather than a full record literal)+  -- means any future field added to 'SubscriptionConfigM' is inherited at its+  -- default automatically — e.g. EP-2's 'consumerGroupGuard', left 'False' here.+  let subConfig :: SubscriptionConfig+      subConfig =+        (defaultSubscriptionConfig subName subTarget (\_ -> pure Continue))+          { Sub.batchSize = bs,+            Sub.retryPolicy = attempts,+            Sub.handlerStallWarnAfter = stallWarn,+            Sub.queueCapacity = qCap,+            Sub.consumerGroup = cg,+            Sub.missingCheckpointPolicy = checkpointPolicy,+            Sub.eventTypeFilter = etf,+            Sub.selector = sel+          } -    pure-        Adapter-            { adapterName = "kiroku"-            , source = ingestedStream-            , shutdown = liftIO cancelAction-            }+  (ioStream, cancelAction) <- liftIO $ subscriptionAckStream store subConfig buf -{- | Configuration for a whole kiroku consumer group presented as a single-Shibuya partitioned-ordering unit.+  -- The subscription name and consumer-group member are known only here (not on+  -- the RecordedEvent), so thread them into the conversion as OTel attributes+  -- that ride onto Shibuya's per-message span (EP-5 M2).+  let SubscriptionName subNameText = subName+      -- Precompute the constant kiroku.* attributes once per adapter (not per+      -- event): kirokuEnvelopeAttrs builds the base map, and the per-event+      -- conversion only inserts the event type and global position.+      envAttrs =+        kirokuEnvelopeAttrs+          subNameText+          (fmap (fromIntegral . member) cg)+      ingestedStream = fmap (toIngestedAck envAttrs cancelAction) (Stream.morphInner liftIO ioStream) -Unlike 'KirokuAdapterConfig' (which describes one member), this describes the-__entire__ group: 'groupSize' members of one subscription, each receiving the-streams whose originating-stream hash maps to its slot. Hand to-'kirokuConsumerGroupProcessors' to obtain @groupSize@ ready-to-run Shibuya-processors with no manual @[0..N-1]@ wiring.+  pure+    Adapter+      { adapterName = "kiroku",+        source = ingestedStream,+        shutdown = liftIO cancelAction+      } -@memberConcurrency@ is the per-member concurrency. Because kiroku delivers each-member a single strictly global-position-ordered stream, only 'Serial' honestly-preserves per-stream ordering; any 'Ahead'/'Async' is rejected by-'consumerGroupPolicy' before any subscription opens. The /group/ as a whole is-'PartitionedInOrder' (ordered within each member's partition, parallel across-members).--}+-- | Configuration for a whole kiroku consumer group presented as a single+-- Shibuya partitioned-ordering unit.+--+-- Unlike 'KirokuAdapterConfig' (which describes one member), this describes the+-- __entire__ group: 'groupSize' members of one subscription, each receiving the+-- streams whose originating-stream hash maps to its slot. Hand to+-- 'kirokuConsumerGroupProcessors' to obtain @groupSize@ ready-to-run Shibuya+-- processors with no manual @[0..N-1]@ wiring.+--+-- @memberConcurrency@ is the per-member concurrency. Because kiroku delivers each+-- member a single strictly global-position-ordered stream, only 'Serial' honestly+-- preserves per-stream ordering; any 'Ahead'/'Async' is rejected by+-- 'consumerGroupPolicy' before any subscription opens. The /group/ as a whole is+-- 'PartitionedInOrder' (ordered within each member's partition, parallel across+-- members). data KirokuConsumerGroupConfig = KirokuConsumerGroupConfig-    { subscriptionName :: !SubscriptionName-    {- ^ Shared subscription identifier; each member checkpoints under-    @(subscriptionName, member)@.-    -}-    , subscriptionTarget :: !SubscriptionTarget-    {- ^ 'AllStreams' or @'Category' categoryName@ — the same source for every-    member; kiroku partitions it across members in SQL.-    -}-    , groupSize :: !ConsumerGroupSize-    {- ^ @N@ members; must be @>= 1@ (enforced by the underlying-    'mkConsumerGroupSize' at construction).-    -}-    , batchSize :: !Sub.BatchSize-    -- ^ Events per database fetch during catch-up (per member).-    , bufferSize :: !StreamBufferSize-    -- ^ Per-member bridge 'TBQueue' capacity; must be at least 1.-    , retryPolicy :: !Sub.RetryPolicy-    -- ^ Total delivery attempts, default five; each AckRetry chooses its delay.-    , handlerStallWarnAfter :: !(Maybe NominalDiffTime)-    -- ^ Optional positive advisory warning interval, disabled by default.-    , queueCapacity :: !Natural-    {- ^ Per-member publisher-side capacity in batches. When this fills, Kiroku-    pauses and later resumes the member losslessly.-    -}-    , memberConcurrency :: !Concurrency-    -- ^ Per-member concurrency; must be 'Serial' (validated).-    , missingCheckpointPolicy :: !MissingCheckpointPolicy-    {- ^ Missing-checkpoint policy applied independently to every member key.-    Existing member rows always win; a new 'FromCurrentHead' group seeds every-    member at the head each member observes during startup.-    -}-    , eventTypeFilter :: !EventTypeFilter-    {- ^ Event-type filter applied to /every/ member (the same filter on each).-    'AllEventTypes' delivers everything; @'OnlyEventTypes' s@ delivers only the-    named types. Forwarded into each per-member 'KirokuAdapterConfig', so a-    filtered partitioned group behaves like a filtered single subscription:-    filtering is worker-side and per member, the checkpoint still advances past-    filtered events, and the partition's completeness is preserved over the-    delivered types.-    -}-    , selector :: !(Maybe (RecordedEvent -> Bool))-    {- ^ Optional opaque per-event predicate applied to /every/ member (the same-    predicate on each), the escape hatch for filtering a property-    'eventTypeFilter' cannot express. Default 'Nothing'. Forwarded into each-    per-member 'KirokuAdapterConfig' and composed with 'eventTypeFilter' as a-    logical AND, so a selector-filtered partitioned group behaves like a-    selector-filtered single subscription (worker-side, per member, checkpoint-    still advances past rejected events). See-    'Kiroku.Store.Subscription.Types.selector'.-    -}-    }-    deriving stock (Generic)+  { -- | Shared subscription identifier; each member checkpoints under+    --     @(subscriptionName, member)@.+    subscriptionName :: !SubscriptionName,+    -- | 'AllStreams' or @'Category' categoryName@ — the same source for every+    --     member; kiroku partitions it across members in SQL.+    subscriptionTarget :: !SubscriptionTarget,+    -- | @N@ members; must be @>= 1@ (enforced by the underlying+    --     'mkConsumerGroupSize' at construction).+    groupSize :: !ConsumerGroupSize,+    -- | Events per database fetch during catch-up (per member).+    batchSize :: !Sub.BatchSize,+    -- | Per-member bridge 'TBQueue' capacity; must be at least 1.+    bufferSize :: !StreamBufferSize,+    -- | Total delivery attempts, default five; each AckRetry chooses its delay.+    retryPolicy :: !Sub.RetryPolicy,+    -- | Optional positive advisory warning interval, disabled by default.+    handlerStallWarnAfter :: !(Maybe NominalDiffTime),+    -- | Per-member publisher-side capacity in batches. When this fills, Kiroku+    --     pauses and later resumes the member losslessly.+    queueCapacity :: !Natural,+    -- | Per-member concurrency; must be 'Serial' (validated).+    memberConcurrency :: !Concurrency,+    -- | Missing-checkpoint policy applied independently to every member key.+    --     Existing member rows always win; a new 'FromCurrentHead' group seeds every+    --     member at the head each member observes during startup.+    missingCheckpointPolicy :: !MissingCheckpointPolicy,+    -- | Event-type filter applied to /every/ member (the same filter on each).+    --     'AllEventTypes' delivers everything; @'OnlyEventTypes' s@ delivers only the+    --     named types. Forwarded into each per-member 'KirokuAdapterConfig', so a+    --     filtered partitioned group behaves like a filtered single subscription:+    --     filtering is worker-side and per member, the checkpoint still advances past+    --     filtered events, and the partition's completeness is preserved over the+    --     delivered types.+    eventTypeFilter :: !EventTypeFilter,+    -- | Optional opaque per-event predicate applied to /every/ member (the same+    --     predicate on each), the escape hatch for filtering a property+    --     'eventTypeFilter' cannot express. Default 'Nothing'. Forwarded into each+    --     per-member 'KirokuAdapterConfig' and composed with 'eventTypeFilter' as a+    --     logical AND, so a selector-filtered partitioned group behaves like a+    --     selector-filtered single subscription (worker-side, per member, checkpoint+    --     still advances past rejected events). See+    --     'Kiroku.Store.Subscription.Types.selector'.+    selector :: !(Maybe (RecordedEvent -> Bool))+  }+  deriving stock (Generic) -{- | A 'KirokuConsumerGroupConfig' with sensible defaults: @memberConcurrency =-'Serial'@ (the only legal per-member concurrency), @batchSize = Sub.defaultBatchSize@,-@bufferSize = defaultStreamBufferSize@, @queueCapacity = 16@, @missingCheckpointPolicy =-'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@-(deliver every type), @selector = 'Nothing'@ (no extra predicate filtering).-Supply the subscription name, target, and group size.--}+-- | A 'KirokuConsumerGroupConfig' with sensible defaults: @memberConcurrency =+-- 'Serial'@ (the only legal per-member concurrency), @batchSize = Sub.defaultBatchSize@,+-- @bufferSize = defaultStreamBufferSize@, @queueCapacity = 16@, @missingCheckpointPolicy =+-- 'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@+-- (deliver every type), @selector = 'Nothing'@ (no extra predicate filtering).+-- Supply the subscription name, target, and group size. defaultConsumerGroupConfig ::-    SubscriptionName -> SubscriptionTarget -> ConsumerGroupSize -> KirokuConsumerGroupConfig+  SubscriptionName -> SubscriptionTarget -> ConsumerGroupSize -> KirokuConsumerGroupConfig defaultConsumerGroupConfig name target n =-    KirokuConsumerGroupConfig-        { subscriptionName = name-        , subscriptionTarget = target-        , groupSize = n-        , batchSize = Sub.defaultBatchSize-        , bufferSize = defaultStreamBufferSize-        , retryPolicy = Sub.defaultRetryPolicy-        , handlerStallWarnAfter = Nothing-        , queueCapacity = 16-        , memberConcurrency = Serial-        , missingCheckpointPolicy = FromBeginning-        , eventTypeFilter = AllEventTypes-        , selector = Nothing-        }--{- | Map a requested per-member concurrency onto the group's validated Shibuya-@('OrderingPolicy', 'Concurrency')@.+  KirokuConsumerGroupConfig+    { subscriptionName = name,+      subscriptionTarget = target,+      groupSize = n,+      batchSize = Sub.defaultBatchSize,+      bufferSize = defaultStreamBufferSize,+      retryPolicy = Sub.defaultRetryPolicy,+      handlerStallWarnAfter = Nothing,+      queueCapacity = 16,+      memberConcurrency = Serial,+      missingCheckpointPolicy = FromBeginning,+      eventTypeFilter = AllEventTypes,+      selector = Nothing+    } -The group's ordering contract is always 'PartitionedInOrder'; a member's own-ordered stream must be processed serially. This reuses Shibuya's own-'validatePolicy' rule (@'StrictInOrder' => 'Serial'@) so the adapter never-invents its own legality check: 'Ahead'/'Async' yield-@'Left' ('InvalidPolicyCombo' ...)@ and 'Serial' yields-@'Right' ('PartitionedInOrder', 'Serial')@. The returned-@('PartitionedInOrder', 'Serial')@ also passes 'validatePolicy', so 'runApp'-will not reject it later.--}+-- | Map a requested per-member concurrency onto the group's validated Shibuya+-- @('OrderingPolicy', 'Concurrency')@.+--+-- The group's ordering contract is always 'PartitionedInOrder'; a member's own+-- ordered stream must be processed serially. This reuses Shibuya's own+-- 'validatePolicy' rule (@'StrictInOrder' => 'Serial'@) so the adapter never+-- invents its own legality check: 'Ahead'/'Async' yield+-- @'Left' ('InvalidPolicyCombo' ...)@ and 'Serial' yields+-- @'Right' ('PartitionedInOrder', 'Serial')@. The returned+-- @('PartitionedInOrder', 'Serial')@ also passes 'validatePolicy', so 'runApp'+-- will not reject it later. consumerGroupPolicy :: Concurrency -> Either PolicyError (OrderingPolicy, Concurrency) consumerGroupPolicy conc = do-    -- A member delivers one strictly-ordered stream; only Serial is honest.-    validatePolicy StrictInOrder conc -- rejects Ahead/Async with PolicyError-    pure (PartitionedInOrder, conc)--{- | Present a whole kiroku consumer group as a single 'PartitionedInOrder' unit:-one call yields @groupSize@ named 'QueueProcessor's, each backed by its own-member adapter and each pinned to @('PartitionedInOrder', 'Serial')@.-The supplied handler is wrapped in 'guardKirokuHandler' automatically so a-synchronous handler exception finalizes a retry disposition instead of-abandoning Kiroku's ack reply.--This replaces the manual @mapM mkMemberAdapter [0 .. N-1]@ boilerplate. The-member→policy mapping is validated once up front via 'consumerGroupPolicy'; if-the caller requests a 'memberConcurrency' kiroku cannot honor per member-('Ahead'/'Async'), the result is @'Left' ('InvalidPolicyCombo' ...)@ and __no-kiroku subscription is opened__.+  -- A member delivers one strictly-ordered stream; only Serial is honest.+  validatePolicy StrictInOrder conc -- rejects Ahead/Async with PolicyError+  pure (PartitionedInOrder, conc) -Each processor's 'ProcessorId' is-@\"\<subscriptionName\>-member-\<m\>\"@ so member identity is readable off the id-and two members never collide. The group size is validated by 'mkConsumerGroupSize' at construction.--}+-- | Present a whole kiroku consumer group as a single 'PartitionedInOrder' unit:+-- one call yields @groupSize@ named 'QueueProcessor's, each backed by its own+-- member adapter and each pinned to @('PartitionedInOrder', 'Serial')@.+-- The supplied handler is wrapped in 'guardKirokuHandler' automatically so a+-- synchronous handler exception finalizes a retry disposition instead of+-- abandoning Kiroku's ack reply.+--+-- This replaces the manual @mapM mkMemberAdapter [0 .. N-1]@ boilerplate. The+-- member→policy mapping is validated once up front via 'consumerGroupPolicy'; if+-- the caller requests a 'memberConcurrency' kiroku cannot honor per member+-- ('Ahead'/'Async'), the result is @'Left' ('InvalidPolicyCombo' ...)@ and __no+-- kiroku subscription is opened__.+--+-- Each processor's 'ProcessorId' is+-- @\"\<subscriptionName\>-member-\<m\>\"@ so member identity is readable off the id+-- and two members never collide. The group size is validated by 'mkConsumerGroupSize' at construction. kirokuConsumerGroupProcessors ::-    (IOE :> es) =>-    KirokuStore ->-    KirokuConsumerGroupConfig ->-    Handler es RecordedEvent ->-    Eff es (Either PolicyError [(ProcessorId, QueueProcessor es)])-kirokuConsumerGroupProcessors store cfg@KirokuConsumerGroupConfig{subscriptionName = subName, subscriptionTarget = subTarget, groupSize = n, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} handler =-    kirokuConsumerGroupProcessorsWith mkMemberAdapter cfg handler+  (IOE :> es) =>+  KirokuStore ->+  KirokuConsumerGroupConfig ->+  Handler es RecordedEvent ->+  Eff es (Either PolicyError [(ProcessorId, QueueProcessor es)])+kirokuConsumerGroupProcessors store cfg@KirokuConsumerGroupConfig {subscriptionName = subName, subscriptionTarget = subTarget, groupSize = n, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} handler =+  kirokuConsumerGroupProcessorsWith mkMemberAdapter cfg handler   where     mkMemberAdapter m =-        kirokuAdapter-            store-            KirokuAdapterConfig-                { subscriptionName = subName-                , subscriptionTarget = subTarget-                , batchSize = bs-                , bufferSize = buf-                , retryPolicy = attempts-                , handlerStallWarnAfter = stallWarn-                , queueCapacity = qCap-                , consumerGroup = Just (either (error . show) Prelude.id (mkConsumerGroup m n))-                , missingCheckpointPolicy = checkpointPolicy-                , eventTypeFilter = etf-                , selector = sel-                }--{- | Factory-parameterized core for 'kirokuConsumerGroupProcessors'.+      kirokuAdapter+        store+        KirokuAdapterConfig+          { subscriptionName = subName,+            subscriptionTarget = subTarget,+            batchSize = bs,+            bufferSize = buf,+            retryPolicy = attempts,+            handlerStallWarnAfter = stallWarn,+            queueCapacity = qCap,+            consumerGroup = Just (either (error . show) Prelude.id (mkConsumerGroup m n)),+            missingCheckpointPolicy = checkpointPolicy,+            eventTypeFilter = etf,+            selector = sel+          } -This is exported primarily for tests and advanced integration points. It-validates the group configuration, creates each member adapter with the-provided factory, and shuts down already-created adapters if a later factory-call throws before ownership can be returned to the caller.--}+-- | Factory-parameterized core for 'kirokuConsumerGroupProcessors'.+--+-- This is exported primarily for tests and advanced integration points. It+-- validates the group configuration, creates each member adapter with the+-- provided factory, and shuts down already-created adapters if a later factory+-- call throws before ownership can be returned to the caller. kirokuConsumerGroupProcessorsWith ::-    (IOE :> es) =>-    (Int32 -> Eff es (Adapter es RecordedEvent)) ->-    KirokuConsumerGroupConfig ->-    Handler es RecordedEvent ->-    Eff es (Either PolicyError [(ProcessorId, QueueProcessor es)])+  (IOE :> es) =>+  (Int32 -> Eff es (Adapter es RecordedEvent)) ->+  KirokuConsumerGroupConfig ->+  Handler es RecordedEvent ->+  Eff es (Either PolicyError [(ProcessorId, QueueProcessor es)]) kirokuConsumerGroupProcessorsWith-    mkMemberAdapter-    KirokuConsumerGroupConfig-        { subscriptionName = subName-        , groupSize = n-        , memberConcurrency = mc-        }-    handler =-        case consumerGroupPolicy mc of-            Left e -> pure (Left e)-            Right (ordering, conc) -> do-                let SubscriptionName name = subName-                acquireAllAndTransfer-                    (consumerGroupSizeValue n)-                    mkMemberAdapter-                    (\Adapter{shutdown = shutdownAction} -> shutdownAction)-                    (\_ _ -> pure ())-                    ( \adapters ->-                        pure . Right $-                            [ let pid = ProcessorId (name <> "-member-" <> T.pack (show m))-                               in (pid, QueueProcessor adapter (guardKirokuHandler handler) ordering conc)-                            | (m, adapter) <- zip [0 .. consumerGroupSizeValue n - 1] adapters-                            ]-                    )+  mkMemberAdapter+  KirokuConsumerGroupConfig+    { subscriptionName = subName,+      groupSize = n,+      memberConcurrency = mc+    }+  handler =+    case consumerGroupPolicy mc of+      Left e -> pure (Left e)+      Right (ordering, conc) -> do+        let SubscriptionName name = subName+        acquireAllAndTransfer+          (consumerGroupSizeValue n)+          mkMemberAdapter+          (\Adapter {shutdown = shutdownAction} -> shutdownAction)+          (\_ _ -> pure ())+          ( \adapters ->+              pure . Right $+                [ let pid = ProcessorId (name <> "-member-" <> T.pack (show m))+                   in (pid, QueueProcessor adapter (guardKirokuHandler handler) ordering conc)+                | (m, adapter) <- zip [0 .. consumerGroupSizeValue n - 1] adapters+                ]+          )

@@ -1,25 +1,24 @@-{- | Conversion from Kiroku's 'RecordedEvent' to Shibuya's 'Ingested' and-'Envelope' types.--== Envelope Mapping--@-RecordedEvent field   →  Envelope field-─────────────────────────────────────────-eventId (UUID)        →  messageId (Text)-globalPosition        →  cursor (CursorInt)-createdAt             →  enqueuedAt-metadata.traceparent  →  traceContext-(the event itself)    →  payload-(none)                →  partition = Nothing-@--The adapter preserves W3C trace-context metadata when @metadata@ is a JSON-object containing a string @traceparent@ key. A string @tracestate@ key is-included when present.--}-module Shibuya.Adapter.Kiroku.Convert (-    -- * Conversion+-- | Conversion from Kiroku's 'RecordedEvent' to Shibuya's 'Ingested' and+-- 'Envelope' types.+--+-- == Envelope Mapping+--+-- @+-- RecordedEvent field   →  Envelope field+-- ─────────────────────────────────────────+-- eventId (UUID)        →  messageId (Text)+-- globalPosition        →  cursor (CursorInt)+-- createdAt             →  enqueuedAt+-- metadata.traceparent  →  traceContext+-- (the event itself)    →  payload+-- (none)                →  partition = Nothing+-- @+--+-- The adapter preserves W3C trace-context metadata when @metadata@ is a JSON+-- object containing a string @traceparent@ key. A string @tracestate@ key is+-- included when present.+module Shibuya.Adapter.Kiroku.Convert+  ( -- * Conversion     toIngestedAck,     toEnvelope, @@ -30,7 +29,8 @@     -- * Ack-decision translation     toKirokuResult,     toKirokuDeadLetterReason,-) where+  )+where  import Control.Concurrent.STM (atomically, tryPutTMVar) import Control.Monad (void)@@ -46,168 +46,162 @@ import Data.UUID qualified as UUID import Effectful (IOE, liftIO, (:>)) import Kiroku.Store.Subscription.Stream (AckItem (..))-import Kiroku.Store.Subscription.Types (-    DeadLetterReason (..),+import Kiroku.Store.Subscription.Types+  ( DeadLetterReason (..),     RetryDelay (..),     SubscriptionResult (..),- )-import Kiroku.Store.Types (-    EventId (..),+  )+import Kiroku.Store.Types+  ( EventId (..),     EventType (..),     GlobalPosition (..),     RecordedEvent (..),- )+  ) import OpenTelemetry.Attributes (Attribute, AttributeKey (..), toAttribute) import OpenTelemetry.SemanticConventions qualified as Sem import Shibuya.Core.Ack (AckDecision (..)) import Shibuya.Core.Ack qualified as Ack import Shibuya.Core.AckHandle (AckHandle (..))-import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..)) import Shibuya.Core.Types (Attempt (..), Cursor (..), Envelope (..), MessageId (..), TraceHeaders) -{- | The kiroku identity stamped onto each Shibuya 'Envelope' as OpenTelemetry-attributes, so the kiroku subscription is visible on Shibuya's per-message span.--Construct it once per adapter with 'kirokuEnvelopeAttrs'. It holds the-/constant/ per-subscription attributes — the subscription name and, for a-grouped subscription, the consumer-group member — already built into an-attribute map, so the per-event conversion path only inserts this event's type-and global position rather than rebuilding the whole map for every event. The-subscription name and member are not carried on a 'RecordedEvent' (they are-known only at the adapter-config level), so the adapter threads them in through-this value; the event type and global position come from the 'RecordedEvent'-itself.--}+-- | The kiroku identity stamped onto each Shibuya 'Envelope' as OpenTelemetry+-- attributes, so the kiroku subscription is visible on Shibuya's per-message span.+--+-- Construct it once per adapter with 'kirokuEnvelopeAttrs'. It holds the+-- /constant/ per-subscription attributes — the subscription name and, for a+-- grouped subscription, the consumer-group member — already built into an+-- attribute map, so the per-event conversion path only inserts this event's type+-- and global position rather than rebuilding the whole map for every event. The+-- subscription name and member are not carried on a 'RecordedEvent' (they are+-- known only at the adapter-config level), so the adapter threads them in through+-- this value; the event type and global position come from the 'RecordedEvent'+-- itself. newtype KirokuEnvelopeAttrs = KirokuEnvelopeAttrs-    { baseAttributes :: HashMap Text Attribute-    -- ^ The constant per-subscription attributes, precomputed once.-    }+  { -- | The constant per-subscription attributes, precomputed once.+    baseAttributes :: HashMap Text Attribute+  } -{- | Build a 'KirokuEnvelopeAttrs' from a subscription name and an optional-consumer-group member index (@'Nothing'@ for a non-grouped subscription),-precomputing the constant @kiroku.*@ attribute map once. The member attribute is-included only when a member index is given.--}+-- | Build a 'KirokuEnvelopeAttrs' from a subscription name and an optional+-- consumer-group member index (@'Nothing'@ for a non-grouped subscription),+-- precomputing the constant @kiroku.*@ attribute map once. The member attribute is+-- included only when a member index is given. kirokuEnvelopeAttrs :: Text -> Maybe Int -> KirokuEnvelopeAttrs kirokuEnvelopeAttrs subscriptionName member =-    KirokuEnvelopeAttrs $-        HashMap.fromList $-            (attrKirokuSubscriptionName, toAttribute subscriptionName)-                : (attrMessagingSystem, toAttribute ("kiroku" :: Text))-                : (attrMessagingDestinationName, toAttribute subscriptionName)-                : maybe-                    []-                    (\m -> [(attrKirokuConsumerGroupMember, toAttribute (fromIntegral m :: Int64))])-                    member--{- | Wrap an ack-coupled 'AckItem' (from @kiroku-store@'s 'subscriptionAckStream')-into an 'Ingested' value suitable for Shibuya handlers.--The underlying Kiroku worker is blocked waiting for this item's reply, so the-'AckHandle.finalize' translates the Shibuya 'AckDecision' into a Kiroku-'SubscriptionResult' and writes it back — driving the worker's checkpointing:--* 'AckOk' — reply 'Continue'; the worker checkpoints past the event.-* 'AckRetry' @delay@ — reply 'Retry'; the worker redelivers the same event after-  @delay@, bounded by the subscription's retry policy, then dead-letters it.-* 'AckDeadLetter' @reason@ — reply 'DeadLetter'; the worker records the event in-  @kiroku.dead_letters@ and advances past it.-* 'AckHalt' — cancels the underlying Kiroku subscription (no checkpoint advance,-  so the halting event replays on restart), preserving the prior adapter-  behavior; the blocked worker is interrupted by the cancellation.--'finalize' is idempotent: the reply is written with 'tryPutTMVar' so a second-call is a no-op.+  KirokuEnvelopeAttrs $+    HashMap.fromList $+      (attrKirokuSubscriptionName, toAttribute subscriptionName)+        : (attrMessagingSystem, toAttribute ("kiroku" :: Text))+        : (attrMessagingDestinationName, toAttribute subscriptionName)+        : maybe+          []+          (\m -> [(attrKirokuConsumerGroupMember, toAttribute (fromIntegral m :: Int64))])+          member -The envelope's @attempt@ is set from the item's redelivery counter so a Shibuya-handler can observe how many times Kiroku has redelivered the event.--}+-- | Wrap an ack-coupled 'AckItem' (from @kiroku-store@'s 'subscriptionAckStream')+-- into an 'Ingested' value suitable for Shibuya handlers.+--+-- The underlying Kiroku worker is blocked waiting for this item's reply, so the+-- 'AckHandle.finalize' translates the Shibuya 'AckDecision' into a Kiroku+-- 'SubscriptionResult' and writes it back — driving the worker's checkpointing:+--+-- * 'AckOk' — reply 'Continue'; the worker checkpoints past the event.+-- * 'AckRetry' @delay@ — reply 'Retry'; the worker redelivers the same event after+--   @delay@, bounded by the subscription's retry policy, then dead-letters it.+-- * 'AckDeadLetter' @reason@ — reply 'DeadLetter'; the worker records the event in+--   @kiroku.dead_letters@ and advances past it.+-- * 'AckHalt' — cancels the underlying Kiroku subscription (no checkpoint advance,+--   so the halting event replays on restart), preserving the prior adapter+--   behavior; the blocked worker is interrupted by the cancellation.+--+-- 'finalize' is idempotent: the reply is written with 'tryPutTMVar' so a second+-- call is a no-op.+--+-- The envelope's @attempt@ is set from the item's redelivery counter so a Shibuya+-- handler can observe how many times Kiroku has redelivered the event. toIngestedAck :: (IOE :> es) => KirokuEnvelopeAttrs -> IO () -> AckItem -> Ingested es RecordedEvent toIngestedAck attrs cancelAction (AckItem event attempt reply) =-    Ingested-        { envelope = (toEnvelope attrs event){attempt = Just (Attempt attempt)}-        , ack =-            AckHandle-                { finalize = \case-                    AckHalt _ -> liftIO cancelAction-                    decision ->-                        liftIO $-                            atomically $-                                void $-                                    tryPutTMVar reply (toKirokuResult attempt decision)-                }-        , lease = Nothing-        }+  Ingested+    { envelope = (toEnvelope attrs event) {attempt = Just (Attempt attempt)},+      ack =+        AckHandle+          { finalize = \case+              AckHalt _ -> liftIO cancelAction+              decision ->+                liftIO $+                  atomically $+                    void $+                      tryPutTMVar reply (toKirokuResult attempt decision)+          },+      lease = Nothing,+      deliveryStatus = pure DeliveryCurrent+    } -{- | Translate a non-halt Shibuya 'AckDecision' into a Kiroku-'SubscriptionResult'. 'AckHalt' is handled separately (it cancels the-subscription) and maps to 'Continue' here only defensively. The 'Word' is the-event's current redelivery attempt, used to annotate a 'MaxRetriesExceeded'-reason.--}+-- | Translate a non-halt Shibuya 'AckDecision' into a Kiroku+-- 'SubscriptionResult'. 'AckHalt' is handled separately (it cancels the+-- subscription) and maps to 'Continue' here only defensively. The 'Word' is the+-- event's current redelivery attempt, used to annotate a 'MaxRetriesExceeded'+-- reason. toKirokuResult :: Word -> AckDecision -> SubscriptionResult toKirokuResult attempt = \case-    AckOk -> Continue-    AckRetry (Ack.RetryDelay d) -> Retry (RetryDelay d)-    AckDeadLetter reason -> DeadLetter (toKirokuDeadLetterReason attempt reason)-    AckHalt _ -> Continue--{- | Translate a Shibuya 'Ack.DeadLetterReason' into a Kiroku 'DeadLetterReason'.--The three framework-owned reasons keep their dedicated Kiroku constructors, so-their @reason_summary@ and @reason@ JSONB encodings are unchanged.+  AckOk -> Continue+  AckRetry (Ack.RetryDelay d) -> Retry (RetryDelay d)+  AckDeadLetter reason -> DeadLetter (toKirokuDeadLetterReason attempt reason)+  AckHalt _ -> Continue -'Ack.ApplicationFailure' (added in @shibuya-core@ 0.9) has no Kiroku-counterpart — Kiroku deliberately knows nothing about application-owned-dead-letter codes — so it maps to 'DeadLetterOther'. The summary is Shibuya's-canonical @code: detail@ rendering, and the structured JSON keeps the code in-its own field so operators can query @reason->'detail'->>'code'@ without-parsing the summary text.--}+-- | Translate a Shibuya 'Ack.DeadLetterReason' into a Kiroku 'DeadLetterReason'.+--+-- The three framework-owned reasons keep their dedicated Kiroku constructors, so+-- their @reason_summary@ and @reason@ JSONB encodings are unchanged.+--+-- 'Ack.ApplicationFailure' (added in @shibuya-core@ 0.9) has no Kiroku+-- counterpart — Kiroku deliberately knows nothing about application-owned+-- dead-letter codes — so it maps to 'DeadLetterOther'. The summary is Shibuya's+-- canonical @code: detail@ rendering, and the structured JSON keeps the code in+-- its own field so operators can query @reason->'detail'->>'code'@ without+-- parsing the summary text. toKirokuDeadLetterReason :: Word -> Ack.DeadLetterReason -> DeadLetterReason toKirokuDeadLetterReason attempt = \case-    Ack.PoisonPill detail -> DeadLetterPoison detail-    Ack.InvalidPayload detail -> DeadLetterInvalid detail-    Ack.MaxRetriesExceeded -> DeadLetterMaxAttempts (fromIntegral attempt)-    reason@(Ack.ApplicationFailure code detail) ->-        DeadLetterOther-            (Ack.renderDeadLetterReason reason)-            (object ["code" .= Ack.deadLetterCodeText code, "detail" .= detail])--{- | Convert a 'RecordedEvent' to a Shibuya 'Envelope'.+  Ack.PoisonPill detail -> DeadLetterPoison detail+  Ack.InvalidPayload detail -> DeadLetterInvalid detail+  Ack.MaxRetriesExceeded -> DeadLetterMaxAttempts (fromIntegral attempt)+  reason@(Ack.ApplicationFailure code detail) ->+    DeadLetterOther+      (Ack.renderDeadLetterReason reason)+      (object ["code" .= Ack.deadLetterCodeText code, "detail" .= detail]) -The event's UUID is formatted as text for the 'MessageId', and the-global position is used as an integer 'Cursor' for ordering. The 'Envelope's-@attributes@ are populated with the kiroku identity (subscription name,-consumer-group member, event type, global position) from 'KirokuEnvelopeAttrs'-and the event, so Shibuya's per-message span carries the kiroku context.--}+-- | Convert a 'RecordedEvent' to a Shibuya 'Envelope'.+--+-- The event's UUID is formatted as text for the 'MessageId', and the+-- global position is used as an integer 'Cursor' for ordering. The 'Envelope's+-- @attributes@ are populated with the kiroku identity (subscription name,+-- consumer-group member, event type, global position) from 'KirokuEnvelopeAttrs'+-- and the event, so Shibuya's per-message span carries the kiroku context. toEnvelope :: KirokuEnvelopeAttrs -> RecordedEvent -> Envelope RecordedEvent toEnvelope attrs event =-    let RecordedEvent{eventId = EventId uuid, eventType = EventType etype, globalPosition = GlobalPosition pos, createdAt = ts, metadata = meta} = event-     in Envelope-            { messageId = MessageId (T.pack (UUID.toString uuid))-            , cursor = Just (CursorInt (fromIntegral pos))-            , partition = Nothing-            , enqueuedAt = Just ts-            , traceContext = metadataTraceContext meta-            , headers = Nothing-            , attempt = Nothing-            , attributes = eventAttributes attrs etype pos-            , payload = event-            }+  let RecordedEvent {eventId = EventId uuid, eventType = EventType etype, globalPosition = GlobalPosition pos, createdAt = ts, metadata = meta} = event+   in Envelope+        { messageId = MessageId (T.pack (UUID.toString uuid)),+          cursor = Just (CursorInt (fromIntegral pos)),+          partition = Nothing,+          enqueuedAt = Just ts,+          traceContext = metadataTraceContext meta,+          headers = Nothing,+          attempt = Nothing,+          attributes = eventAttributes attrs etype pos,+          payload = event+        } -{- | The per-event @kiroku.*@ attribute map: the precomputed constant attributes-('baseAttributes') plus this event's type and global position. Only the two-per-event keys are inserted, so the constant attributes are not rebuilt for-every event. Keys mirror the native-span attribute keys in-@Kiroku.Otel.Subscription@ so a trace reads consistently across the kiroku and-Shibuya sides.--}+-- | The per-event @kiroku.*@ attribute map: the precomputed constant attributes+-- ('baseAttributes') plus this event's type and global position. Only the two+-- per-event keys are inserted, so the constant attributes are not rebuilt for+-- every event. Keys mirror the native-span attribute keys in+-- @Kiroku.Otel.Subscription@ so a trace reads consistently across the kiroku and+-- Shibuya sides. eventAttributes :: KirokuEnvelopeAttrs -> Text -> Int64 -> HashMap Text Attribute eventAttributes attrs etype pos =-    HashMap.insert attrKirokuEventType (toAttribute etype) $-        HashMap.insert attrKirokuEventGlobalPosition (toAttribute pos) (baseAttributes attrs)+  HashMap.insert attrKirokuEventType (toAttribute etype) $+    HashMap.insert attrKirokuEventGlobalPosition (toAttribute pos) (baseAttributes attrs)  attrMessagingSystem :: Text attrMessagingSystem = unkey Sem.messaging_system@@ -229,11 +223,11 @@  metadataTraceContext :: Maybe Value -> Maybe TraceHeaders metadataTraceContext (Just (Object metadata)) = do-    String traceparent <- KM.lookup (Key.fromString "traceparent") metadata-    let traceparentHeader = ("traceparent", TE.encodeUtf8 traceparent)-        traceHeaders =-            case KM.lookup (Key.fromString "tracestate") metadata of-                Just (String tracestate) -> [traceparentHeader, ("tracestate", TE.encodeUtf8 tracestate)]-                _ -> [traceparentHeader]-    pure traceHeaders+  String traceparent <- KM.lookup (Key.fromString "traceparent") metadata+  let traceparentHeader = ("traceparent", TE.encodeUtf8 traceparent)+      traceHeaders =+        case KM.lookup (Key.fromString "tracestate") metadata of+          Just (String tracestate) -> [traceparentHeader, ("tracestate", TE.encodeUtf8 tracestate)]+          _ -> [traceparentHeader]+  pure traceHeaders metadataTraceContext _ = Nothing

@@ -37,1547 +37,1541 @@ import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres) import OpenTelemetry.Attributes (toAttribute) import Shibuya.Adapter (Adapter (..))-import Shibuya.Adapter.Kiroku (-    consumerGroupPolicy,-    defaultConsumerGroupConfig,-    defaultKirokuAdapterConfig,-    guardKirokuHandlerWith,-    kirokuAdapter,-    kirokuConsumerGroupProcessors,-    kirokuConsumerGroupProcessorsWith,-    kirokuProcessor,- )-import Shibuya.Adapter.Kiroku.Convert (-    KirokuEnvelopeAttrs,-    kirokuEnvelopeAttrs,-    toEnvelope,-    toIngestedAck,-    toKirokuDeadLetterReason,- )-import Shibuya.Adapter.Kiroku.Internal (acquireAllAndTransfer)-import Shibuya.App (-    ProcessorId (..),-    QueueProcessor (..),-    defaultAppConfig,-    getAppMetrics,-    mkProcessor,-    runApp,-    stopApp,-    stopAppGracefully,-    waitApp,- )-import Shibuya.App qualified as Shibuya-import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..))-import Shibuya.Core.Ack qualified as Ack-import Shibuya.Core.AckHandle (AckHandle (..))-import Shibuya.Core.Error (PolicyError (..))-import Shibuya.Core.Ingested (Ingested (..), Message (..))-import Shibuya.Core.Metrics (ProcessorState (..))-import Shibuya.Core.Types (Attempt (..), Envelope (..))-import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))-import Shibuya.Telemetry.Effect (runTracingNoop)-import Streamly.Data.Fold qualified as Fold-import Streamly.Data.Stream qualified as Stream-import System.Timeout (timeout)-import Test.Hspec--main :: IO ()-main = withSharedMigratedPostgres $ hspec $ do-    describe "toEnvelope" $ do-        it "copies W3C trace metadata into Shibuya trace headers" $ do-            let traceparent :: Text-                traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"-                tracestate :: Text-                tracestate = "rojo=00f067aa0ba902b7"-                Envelope{traceContext} =-                    toEnvelope-                        sampleEnvelopeAttrs-                        ( makeRecordedEvent-                            ( Just $-                                Aeson.object-                                    [ "traceparent" Aeson..= traceparent-                                    , "tracestate" Aeson..= tracestate-                                    , "other" Aeson..= ("preserved" :: Text)-                                    ]-                            )-                        )--            traceContext-                `shouldBe` Just-                    [ ("traceparent", "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")-                    , ("tracestate", "rojo=00f067aa0ba902b7")-                    ]--        it "omits trace headers when traceparent is absent or not a string" $ do-            let Envelope{traceContext = missingTraceparent} =-                    toEnvelope sampleEnvelopeAttrs (makeRecordedEvent (Just (Aeson.object ["tracestate" Aeson..= ("state" :: Text)])))-                Envelope{traceContext = nonStringTraceparent} =-                    toEnvelope sampleEnvelopeAttrs (makeRecordedEvent (Just (Aeson.object ["traceparent" Aeson..= Aeson.Number 1])))--            missingTraceparent `shouldBe` Nothing-            nonStringTraceparent `shouldBe` Nothing--        it "does not surface broker headers (headers is Nothing), even with trace metadata" $ do-            let Envelope{headers = noMeta} =-                    toEnvelope sampleEnvelopeAttrs (makeRecordedEvent Nothing)-                Envelope{headers = withTraceMeta} =-                    toEnvelope-                        sampleEnvelopeAttrs-                        (makeRecordedEvent (Just (Aeson.object ["traceparent" Aeson..= ("00-abc-def-01" :: Text)])))--            noMeta `shouldBe` Nothing-            withTraceMeta `shouldBe` Nothing--        it "stamps kiroku identity attributes for a non-grouped subscription (EP-5 M2)" $ do-            let attrs = kirokuEnvelopeAttrs "orders-proj" Nothing-                Envelope{attributes} = toEnvelope attrs (makeRecordedEvent Nothing)-            -- makeRecordedEvent: eventType "TraceEvent", globalPosition 1.-            HashMap.lookup "kiroku.subscription.name" attributes-                `shouldBe` Just (toAttribute ("orders-proj" :: Text))-            HashMap.lookup "messaging.system" attributes-                `shouldBe` Just (toAttribute ("kiroku" :: Text))-            HashMap.lookup "messaging.destination.name" attributes-                `shouldBe` Just (toAttribute ("orders-proj" :: Text))-            HashMap.lookup "kiroku.event.type" attributes-                `shouldBe` Just (toAttribute ("TraceEvent" :: Text))-            HashMap.lookup "kiroku.event.global_position" attributes-                `shouldBe` Just (toAttribute (1 :: Int64))-            -- No member key for a non-grouped subscription.-            HashMap.lookup "kiroku.consumer_group.member" attributes `shouldBe` Nothing--        it "stamps the consumer-group member attribute for a grouped subscription (EP-5 M2)" $ do-            let attrs = kirokuEnvelopeAttrs "orders-proj" (Just 2)-                Envelope{attributes} = toEnvelope attrs (makeRecordedEvent Nothing)-            HashMap.lookup "kiroku.subscription.name" attributes-                `shouldBe` Just (toAttribute ("orders-proj" :: Text))-            HashMap.lookup "kiroku.consumer_group.member" attributes-                `shouldBe` Just (toAttribute (2 :: Int64))--    describe "toKirokuDeadLetterReason" $ do-        it "keeps the framework reasons on their dedicated Kiroku constructors" $ do-            toKirokuDeadLetterReason 3 (PoisonPill "bad event")-                `shouldBe` KTypes.DeadLetterPoison "bad event"-            toKirokuDeadLetterReason 3 (InvalidPayload "not json")-                `shouldBe` KTypes.DeadLetterInvalid "not json"-            toKirokuDeadLetterReason 3 MaxRetriesExceeded-                `shouldBe` KTypes.DeadLetterMaxAttempts 3--        it "maps an application failure to DeadLetterOther with a queryable code (shibuya-core 0.9)" $ do-            code <--                either (E.throwIO . userError . T.unpack) pure $-                    Ack.mkDeadLetterCode "billing.card_declined"-            let reason =-                    toKirokuDeadLetterReason 1 (Ack.ApplicationFailure code "card declined by issuer")--            KTypes.deadLetterSummary reason-                `shouldBe` "billing.card_declined: card declined by issuer"-            KTypes.deadLetterReasonJson reason-                `shouldBe` Aeson.object-                    [ "kind" Aeson..= ("other" :: Text)-                    , "summary" Aeson..= ("billing.card_declined: card declined by issuer" :: Text)-                    , "detail"-                        Aeson..= Aeson.object-                            [ "code" Aeson..= ("billing.card_declined" :: Text)-                            , "detail" Aeson..= ("card declined by issuer" :: Text)-                            ]-                    ]--    describe "consumer group policy" $ do-        it "accepts Serial member concurrency as (PartitionedInOrder, Serial)" $-            consumerGroupPolicy Serial `shouldBe` Right (PartitionedInOrder, Serial)--        it "rejects Ahead member concurrency with a PolicyError" $-            consumerGroupPolicy (Ahead 4)-                `shouldBe` Left (InvalidPolicyCombo "StrictInOrder requires Serial concurrency")--        it "rejects Async member concurrency with a PolicyError" $-            consumerGroupPolicy (Async 4)-                `shouldBe` Left (InvalidPolicyCombo "StrictInOrder requires Serial concurrency")--    acknowledgementSpec-    around withTestStore $ do-        describe "kirokuAdapter" $ do-            it "delivers only matching event types when an eventTypeFilter is set (EP-43)" $ \store -> do-                -- A, B, A, B, A on one stream (global positions 1..5). The adapter-                -- is filtered to type A; the Shibuya handler must see only the As.-                let events =-                        [ makeEvent "A" (Aeson.object [])-                        , makeEvent "B" (Aeson.object [])-                        , makeEvent "A" (Aeson.object [])-                        , makeEvent "B" (Aeson.object [])-                        , makeEvent "A" (Aeson.object [])-                        ]-                Right _ <- runStoreIO store $ appendToStream (StreamName "etf-adapter-1") NoStream events-                ref <- newIORef ([] :: [RecordedEvent])-                countVar <- newTVarIO (0 :: Int)-                runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "etf-adapter") AllStreams-                                & #eventTypeFilter .~ OnlyEventTypes (Set.fromList [EventType "A"])-                    let handler ingested = do-                            liftIO $ do-                                modifyIORef' ref (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar countVar-                                    writeTVar countVar (c + 1)-                            pure AckOk-                    res <- runApp defaultAppConfig [(ProcessorId "etf", mkProcessor adapter handler)]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ waitForCount countVar 3 10_000_000-                            stopApp appHandle--                collected <- readIORef ref-                -- Only the three A events reached the handler; no Bs.-                map (^. #eventType) collected `shouldBe` replicate 3 (EventType "A")-                map globalPos (reverse collected) `shouldBe` [1, 3, 5]--            it "delivers only matching types across a filtered consumer group (EP-43)" $ \store -> do-                -- 20 streams in category "etfg", each [A, B]; A at odd global-                -- positions 1,3,..,39. A size-4 group filtered to type A must-                -- deliver exactly the 20 As, disjoint and complete.-                let streams = [0 .. 19 :: Int]-                mapM_-                    ( \i -> do-                        let sn = StreamName ("etfg-" <> T.pack (show i))-                        Right _ <--                            runStoreIO store $-                                appendToStream sn NoStream [makeEvent "A" (Aeson.object []), makeEvent "B" (Aeson.object [])]-                        pure ()-                    )-                    streams-                ref <- newIORef ([] :: [RecordedEvent])-                countVar <- newTVarIO (0 :: Int)-                runEff $ runTracingNoop $ do-                    let cfg =-                            defaultConsumerGroupConfig (SubscriptionName "etfg") (Category (CategoryName "etfg")) (validGroupSize 4)-                                & #eventTypeFilter .~ OnlyEventTypes (Set.fromList [EventType "A"])-                        handler ingested = do-                            liftIO $ do-                                modifyIORef' ref (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar countVar-                                    writeTVar countVar (c + 1)-                            pure AckOk-                    res <- kirokuConsumerGroupProcessors store cfg handler-                    case res of-                        Left err -> liftIO $ expectationFailure ("kirokuConsumerGroupProcessors failed: " <> show err)-                        Right processors -> do-                            appRes <- runApp defaultAppConfig processors-                            case appRes of-                                Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                                Right appHandle -> do-                                    liftIO $ waitForCount countVar 20 15_000_000-                                    stopApp appHandle--                collected <- readIORef ref-                -- Every delivered event is an A (filter honored per member)...-                map (^. #eventType) collected `shouldBe` replicate 20 (EventType "A")-                -- ...and the union over members is the complete, disjoint set of A-                -- positions [1,3,..,39] — no A dropped, none delivered twice.-                sort (map globalPos collected) `shouldBe` [1, 3 .. 39]--            it "delivers only events matching an opaque selector (EP-43 follow-up)" $ \store -> do-                -- All one type "A" (so eventTypeFilter cannot distinguish them);-                -- only the payload tag {keep} differs. keep, skip, keep, skip, keep-                -- at positions 1..5. The selector admits only keep=True.-                let keepObj = Aeson.object [("keep", Aeson.Bool True)]-                    skipObj = Aeson.object [("keep", Aeson.Bool False)]-                    events =-                        [ makeEvent "A" keepObj-                        , makeEvent "A" skipObj-                        , makeEvent "A" keepObj-                        , makeEvent "A" skipObj-                        , makeEvent "A" keepObj-                        ]-                Right _ <- runStoreIO store $ appendToStream (StreamName "sel-adapter-1") NoStream events-                ref <- newIORef ([] :: [RecordedEvent])-                countVar <- newTVarIO (0 :: Int)-                runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "sel-adapter") AllStreams-                                & #selector .~ Just (\e -> (e ^. #payload) == keepObj)-                    let handler ingested = do-                            liftIO $ do-                                modifyIORef' ref (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar countVar-                                    writeTVar countVar (c + 1)-                            pure AckOk-                    res <- runApp defaultAppConfig [(ProcessorId "sel", mkProcessor adapter handler)]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ waitForCount countVar 3 10_000_000-                            stopApp appHandle--                collected <- readIORef ref-                -- Only the three keep events reached the handler (positions 1,3,5);-                -- the two skip events were filtered worker-side and never delivered.-                map globalPos (reverse collected) `shouldBe` [1, 3, 5]--            it "delivers catch-up events through Shibuya pipeline" $ \store -> do-                let events = map (\i -> makeEvent ("CU" <> T.pack (show i)) (Aeson.object [])) [1 .. 10 :: Int]-                Right _ <- runStoreIO store $ appendToStream (StreamName "shibuya-catchup-1") NoStream events-                threadDelay 200_000--                ref <- newIORef ([] :: [RecordedEvent])-                countVar <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "shibuya-catchup-test") AllStreams--                    let handler ingested = do-                            liftIO $ do-                                modifyIORef' ref (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar countVar-                                    writeTVar countVar (c + 1)-                            pure AckOk--                    res <- runApp defaultAppConfig [(ProcessorId "catchup", mkProcessor adapter handler)]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ waitForCount countVar 10 10_000_000-                            stopApp appHandle--                collected <- readIORef ref-                length collected `shouldBe` 10--            it "supports a future-only FromCurrentHead adapter" $ \store -> do-                Right _ <--                    runStoreIO store $-                        appendToStream-                            (StreamName "shibuya-future-only-events")-                            NoStream-                            (map (\i -> makeEvent ("Historical" <> T.pack (show i)) (Aeson.object [])) [1 .. 4 :: Int])-                ref <- newIORef ([] :: [RecordedEvent])-                countVar <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "shibuya-future-only") AllStreams-                                & #missingCheckpointPolicy .~ FromCurrentHead-                    liftIO $-                        waitForCheckpointPosition-                            store-                            (SubscriptionCheckpointKey (SubscriptionName "shibuya-future-only") 0)-                            (GlobalPosition 4)-                    liftIO $ do-                        Right _ <--                            runStoreIO store $-                                appendToStream-                                    (StreamName "shibuya-future-only-events")-                                    StreamExists-                                    [makeEvent "Future" (Aeson.object [])]-                        pure ()-                    let handler ingested = do-                            liftIO $ do-                                modifyIORef' ref (envelopePayload ingested :)-                                atomically $ do-                                    count <- readTVar countVar-                                    writeTVar countVar (count + 1)-                            pure AckOk-                    result <- runApp defaultAppConfig [(ProcessorId "future-only", mkProcessor adapter handler)]-                    case result of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ waitForCount countVar 1 10_000_000-                            stopApp appHandle--                collected <- reverse <$> readIORef ref-                map globalPos collected `shouldBe` [5]--            it "preserves an existing checkpoint when restarted with FromCurrentHead" $ \store -> do-                let subscriptionName = SubscriptionName "shibuya-existing-checkpoint"-                    checkpointKey = SubscriptionCheckpointKey subscriptionName 0-                    config policy =-                        defaultKirokuAdapterConfig subscriptionName AllStreams-                            & #missingCheckpointPolicy .~ policy-                Right _ <- runStoreIO store $ appendToStream (StreamName "shibuya-existing-checkpoint-events") NoStream [makeEvent "First" (Aeson.object [])]-                firstSeen <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    adapter <- kirokuAdapter store (config FromBeginning)-                    result <--                        runApp-                            defaultAppConfig-                            [-                                ( ProcessorId "existing-checkpoint-first"-                                , mkProcessor adapter $ \_ -> do-                                    liftIO $ atomically $ writeTVar firstSeen 1-                                    pure AckOk-                                )-                            ]-                    case result of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ waitForCount firstSeen 1 10_000_000-                            liftIO $ waitForCheckpointPosition store checkpointKey (GlobalPosition 1)-                            stopApp appHandle--                Right _ <--                    runStoreIO store $-                        appendToStream-                            (StreamName "shibuya-existing-checkpoint-events")-                            StreamExists-                            [makeEvent "Second" (Aeson.object [])]-                replayed <- newIORef ([] :: [Int64])-                within "existing checkpoint restart" $-                    runEff $-                        runTracingNoop $ do-                            adapter <- kirokuAdapter store (config FromCurrentHead)-                            result <--                                runApp-                                    defaultAppConfig-                                    [-                                        ( ProcessorId "existing-checkpoint-restart"-                                        , mkProcessor adapter $ \ingested -> do-                                            liftIO $ modifyIORef' replayed (globalPos (envelopePayload ingested) :)-                                            pure (AckHalt (HaltFatal "existing checkpoint observed"))-                                        )-                                    ]-                            case result of-                                Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                                Right appHandle -> waitApp appHandle--                readIORef replayed `shouldReturn` [2]--            it "surfaces FailIfMissing without leaving a subscription registered" $ \store -> do-                let subscriptionName = SubscriptionName "shibuya-required-checkpoint"-                    checkpointKey = SubscriptionCheckpointKey subscriptionName 0-                    config =-                        defaultKirokuAdapterConfig subscriptionName AllStreams-                            & #missingCheckpointPolicy .~ FailIfMissing-                outcome <--                    E.try $-                        within "FailIfMissing adapter refusal" $-                            runEff $-                                runTracingNoop $ do-                                    adapter <- kirokuAdapter store config-                                    let Adapter{source = sourceStream} = adapter-                                    Stream.fold Fold.drain sourceStream-                case outcome of-                    Left exception-                        | Just (SubscriptionCheckpointMissing actual) <- E.fromException (exception :: E.SomeException) ->-                            actual `shouldBe` checkpointKey-                    Left exception -> expectationFailure ("expected SubscriptionCheckpointMissing, got: " <> show exception)-                    Right () -> expectationFailure "expected a missing-checkpoint refusal"-                Map.null <$> subscriptionStates store `shouldReturn` True--            it "delivers live events through Shibuya pipeline" $ \store -> do-                ref <- newIORef ([] :: [RecordedEvent])-                countVar <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "shibuya-live-test") AllStreams--                    let handler ingested = do-                            liftIO $ do-                                modifyIORef' ref (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar countVar-                                    writeTVar countVar (c + 1)-                            pure AckOk--                    res <- runApp defaultAppConfig [(ProcessorId "live", mkProcessor adapter handler)]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ threadDelay 200_000-                            liftIO $ do-                                let events = map (\i -> makeEvent ("Live" <> T.pack (show i)) (Aeson.object [])) [1 .. 5 :: Int]-                                Right _ <- runStoreIO store $ appendToStream (StreamName "shibuya-live-1") NoStream events-                                pure ()-                            liftIO $ waitForCount countVar 5 10_000_000-                            stopApp appHandle--                collected <- readIORef ref-                length collected `shouldBe` 5--            it "survives an append burst while the handler is paused" $ \store -> do-                firstSeen <- newEmptyMVar-                release <- newEmptyMVar-                countVar <- newTVarIO (0 :: Int)-                seenRef <- newIORef ([] :: [RecordedEvent])--                runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "burst-pause-resume") AllStreams-                                & #queueCapacity .~ 1--                    let handler ingested = do-                            n <- liftIO $ do-                                modifyIORef' seenRef (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar countVar-                                    let c' = c + 1-                                    writeTVar countVar c'-                                    pure c'-                            if n == 1-                                then liftIO $ putMVar firstSeen () >> takeMVar release-                                else pure ()-                            pure AckOk--                    res <- runApp defaultAppConfig [(ProcessorId "burst", mkProcessor adapter handler)]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ do-                                let appendOne i =-                                        runStoreIO store $-                                            appendToStream-                                                (StreamName ("burst-" <> T.pack (show (i :: Int))))-                                                NoStream-                                                [makeEvent ("Burst" <> T.pack (show i)) (Aeson.object [])]-                                Right _ <- appendOne 1-                                takeMVar firstSeen-                                mapM_ (\i -> appendOne i >>= (`shouldSatisfy` either (const False) (const True))) [2 .. 40 :: Int]-                                putMVar release ()-                                waitForCount countVar 40 15_000_000-                            stopApp appHandle--                seen <- readIORef seenRef-                sort (map globalPos seen) `shouldBe` [1 .. 40]--            it "ends adapter.source cleanly after shutdown" $ \store -> do-                within "adapter source clean shutdown" $-                    runEff $-                        runTracingNoop $ do-                            adapter <- kirokuAdapter store $ defaultKirokuAdapterConfig (SubscriptionName "source-clean-end") AllStreams-                            let Adapter{source = sourceStream, shutdown = shutdownAction} = adapter-                            shutdownAction-                            Stream.fold Fold.drain sourceStream--            it "rethrows worker crashes through adapter.source" $ \store -> do-                Right _ <- runStoreIO store $ appendToStream (StreamName "source-crash-1") NoStream [makeEvent "SourceCrash" (Aeson.object [])]-                threadDelay 200_000-                let injectBoom _ _ = E.throwIO (userError "adapter source worker boom")-                result <--                    withFetchBatchHookForTest injectBoom $-                        E.try $-                            within "adapter source worker crash" $-                                runEff $-                                    runTracingNoop $ do-                                        adapter <- kirokuAdapter store $ defaultKirokuAdapterConfig (SubscriptionName "source-crash") AllStreams-                                        let Adapter{source = sourceStream} = adapter-                                        Stream.fold Fold.drain sourceStream-                case result of-                    Left (e :: E.SomeException) -> show e `shouldContain` "adapter source worker boom"-                    Right _ -> expectationFailure "expected adapter.source to rethrow the worker exception"--            it "runs multiple category subscriptions concurrently" $ \store -> do-                Right _ <- runStoreIO store $ appendToStream (StreamName "orders-s1") NoStream [makeEvent "OrderCreated" (Aeson.object [])]-                Right _ <- runStoreIO store $ appendToStream (StreamName "orders-s2") NoStream [makeEvent "OrderShipped" (Aeson.object [])]-                Right _ <- runStoreIO store $ appendToStream (StreamName "payments-s1") NoStream [makeEvent "PaymentReceived" (Aeson.object [])]-                Right _ <- runStoreIO store $ appendToStream (StreamName "inventory-s1") NoStream [makeEvent "StockUpdated" (Aeson.object [])]-                Right _ <- runStoreIO store $ appendToStream (StreamName "inventory-s2") NoStream [makeEvent "StockDepleted" (Aeson.object [])]-                threadDelay 200_000--                ordersRef <- newIORef ([] :: [RecordedEvent])-                paymentsRef <- newIORef ([] :: [RecordedEvent])-                inventoryRef <- newIORef ([] :: [RecordedEvent])-                ordersCount <- newTVarIO (0 :: Int)-                paymentsCount <- newTVarIO (0 :: Int)-                inventoryCount <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    let mkCatAdapter nm cat =-                            kirokuAdapter store $-                                defaultKirokuAdapterConfig (SubscriptionName nm) (Category (CategoryName cat))-                    let mkHandler ref' cVar ingested = do-                            liftIO $ do-                                modifyIORef' ref' (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar cVar-                                    writeTVar cVar (c + 1)-                            pure AckOk--                    ordersAdapter <- mkCatAdapter "orders-proj" "orders"-                    paymentsAdapter <- mkCatAdapter "payments-proj" "payments"-                    inventoryAdapter <- mkCatAdapter "inventory-proj" "inventory"--                    res <--                        runApp-                            defaultAppConfig-                            [ (ProcessorId "orders", mkProcessor ordersAdapter (mkHandler ordersRef ordersCount))-                            , (ProcessorId "payments", mkProcessor paymentsAdapter (mkHandler paymentsRef paymentsCount))-                            , (ProcessorId "inventory", mkProcessor inventoryAdapter (mkHandler inventoryRef inventoryCount))-                            ]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ do-                                waitForCount ordersCount 2 10_000_000-                                waitForCount paymentsCount 1 10_000_000-                                waitForCount inventoryCount 2 10_000_000-                            stopApp appHandle--                ordersCollected <- readIORef ordersRef-                paymentsCollected <- readIORef paymentsRef-                inventoryCollected <- readIORef inventoryRef-                length ordersCollected `shouldBe` 2-                length paymentsCollected `shouldBe` 1-                length inventoryCollected `shouldBe` 2--            it "isolates a failing subscription from healthy ones" $ \store -> do-                Right _ <- runStoreIO store $ appendToStream (StreamName "good-s1") NoStream (map (\i -> makeEvent ("Good" <> T.pack (show i)) (Aeson.object [])) [1 .. 3 :: Int])-                Right _ <- runStoreIO store $ appendToStream (StreamName "bad-s1") NoStream [makeEvent "Bad1" (Aeson.object []), makeEvent "Bad2" (Aeson.object [])]-                Right _ <- runStoreIO store $ appendToStream (StreamName "other-s1") NoStream [makeEvent "Other1" (Aeson.object []), makeEvent "Other2" (Aeson.object [])]-                threadDelay 200_000--                goodRef <- newIORef ([] :: [RecordedEvent])-                otherRef <- newIORef ([] :: [RecordedEvent])-                goodCount <- newTVarIO (0 :: Int)-                otherCount <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    goodAdapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "good-proj") (Category (CategoryName "good"))-                    badAdapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "bad-proj") (Category (CategoryName "bad"))-                    otherAdapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "other-proj") (Category (CategoryName "other"))--                    let goodHandler ingested = do-                            liftIO $ do-                                modifyIORef' goodRef (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar goodCount-                                    writeTVar goodCount (c + 1)-                            pure AckOk--                    let badHandler _ingested = do-                            liftIO $ error "handler crash!"--                    let otherHandler ingested = do-                            liftIO $ do-                                modifyIORef' otherRef (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar otherCount-                                    writeTVar otherCount (c + 1)-                            pure AckOk--                    res <--                        runApp-                            defaultAppConfig-                            [ (ProcessorId "good", mkProcessor goodAdapter goodHandler)-                            , (ProcessorId "bad", mkProcessor badAdapter badHandler)-                            , (ProcessorId "other", mkProcessor otherAdapter otherHandler)-                            ]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ do-                                waitForCount goodCount 3 10_000_000-                                waitForCount otherCount 2 10_000_000-                                threadDelay 500_000-                            metrics <- getAppMetrics appHandle-                            case Map.lookup (ProcessorId "bad") metrics of-                                Just m -> case m ^. #state of-                                    Failed _ _ -> pure ()-                                    other -> liftIO $ expectationFailure ("Expected Failed state, got: " <> show other)-                                Nothing -> liftIO $ expectationFailure "No metrics for bad processor"-                            stopApp appHandle--                goodCollected <- readIORef goodRef-                otherCollected <- readIORef otherRef-                length goodCollected `shouldBe` 3-                length otherCollected `shouldBe` 2--            it "shuts down all subscriptions coordinately" $ \store -> do-                Right _ <- runStoreIO store $ appendToStream (StreamName "shut-a") NoStream [makeEvent "A" (Aeson.object [])]-                Right _ <- runStoreIO store $ appendToStream (StreamName "shut-b") NoStream [makeEvent "B" (Aeson.object [])]-                threadDelay 200_000--                countA <- newTVarIO (0 :: Int)-                countB <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    adapterA <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "shut-a-proj") (Category (CategoryName "shut"))-                    adapterB <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "shut-b-proj") (Category (CategoryName "shut"))--                    let handlerA _ingested = do-                            liftIO $ atomically $ do-                                c <- readTVar countA-                                writeTVar countA (c + 1)-                            pure AckOk--                    let handlerB _ingested = do-                            liftIO $ atomically $ do-                                c <- readTVar countB-                                writeTVar countB (c + 1)-                            pure AckOk--                    res <--                        runApp-                            defaultAppConfig-                            [ (ProcessorId "a", mkProcessor adapterA handlerA)-                            , (ProcessorId "b", mkProcessor adapterB handlerB)-                            ]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ do-                                waitForCount countA 2 10_000_000-                                waitForCount countB 2 10_000_000-                            drained <- stopAppGracefully Shibuya.defaultShutdownConfig appHandle-                            liftIO $ drained `shouldBe` True--        describe "ack dispositions" $ do-            it "keeps the first Kiroku reply when an ack is finalized twice" $ \_store -> do-                reply <- newEmptyTMVarIO-                cancellations <- newIORef (0 :: Int)-                let Ingested{ack = AckHandle{finalize}} =-                        toIngestedAck-                            sampleEnvelopeAttrs-                            (modifyIORef' cancellations (+ 1))-                            (AckItem (makeRecordedEvent Nothing) 0 reply)--                runEff $ do-                    finalize AckOk-                    finalize (AckDeadLetter (PoisonPill "late duplicate"))--                atomically (readTMVar reply) `shouldReturn` Continue-                readIORef cancellations `shouldReturn` 0--            it "AckRetry redelivers the same event, then AckOk advances (EP-40 M3)" $ \store -> do-                Right _ <- runStoreIO store $ appendToStream (StreamName "ackretry-1") NoStream [makeEvent "R1" (Aeson.object [])]-                threadDelay 200_000--                deliveries <- newIORef (0 :: Int)-                countVar <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "ackretry-proj") AllStreams--                    let handler _ingested = do-                            n <- liftIO $ do-                                modifyIORef' deliveries (+ 1)-                                atomically $ do-                                    c <- readTVar countVar-                                    writeTVar countVar (c + 1)-                                    pure (c + 1)-                            -- Retry the first delivery, accept the second.-                            pure (if n == 1 then AckRetry (Ack.RetryDelay 0) else AckOk)--                    res <- runApp defaultAppConfig [(ProcessorId "ackretry", mkProcessor adapter handler)]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ waitForCount countVar 2 10_000_000-                            stopApp appHandle--                -- The one event was delivered twice (initial + one retry).-                n <- readIORef deliveries-                n `shouldBe` 2-                -- It was not dead-lettered (the retry succeeded).-                dls <- readDeadLetters store "ackretry-proj"-                length dls `shouldBe` 0--            it "AckHalt leaves the checkpoint behind the event and restart replays it" $ \store -> do-                let subscriptionName = SubscriptionName "ackhalt-replay-proj"-                    checkpointKey = SubscriptionCheckpointKey subscriptionName 0-                Right _ <- runStoreIO store $ appendToStream (StreamName "ackhalt-replay-1") NoStream [makeEvent "H1" (Aeson.object [])]-                threadDelay 200_000--                deliveries <- newIORef ([] :: [Int64])-                let runHaltingProcessor label =-                        within label $-                            runEff $-                                runTracingNoop $ do-                                    adapter <--                                        kirokuAdapter store $-                                            defaultKirokuAdapterConfig subscriptionName AllStreams-                                    result <--                                        runApp-                                            defaultAppConfig-                                            [-                                                ( ProcessorId "ackhalt-replay"-                                                , mkProcessor adapter $ \ingested -> do-                                                    liftIO $ modifyIORef' deliveries (globalPos (envelopePayload ingested) :)-                                                    pure (AckHalt (HaltFatal "intentional halt"))-                                                )-                                            ]-                                    case result of-                                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                                        Right appHandle -> waitApp appHandle--                runHaltingProcessor "first AckHalt"-                readCheckpointPosition store checkpointKey `shouldReturn` Just (GlobalPosition 0)--                runHaltingProcessor "AckHalt restart"-                reverse <$> readIORef deliveries `shouldReturn` [1, 1]-                readCheckpointPosition store checkpointKey `shouldReturn` Just (GlobalPosition 0)--            it "shutdown after an ack but before checkpoint persistence replays without leaking" $ \store -> do-                let subscriptionName = SubscriptionName "checkpoint-cancel-replay-proj"-                    checkpointKey = SubscriptionCheckpointKey subscriptionName 0-                Right _ <- runStoreIO store $ appendToStream (StreamName "checkpoint-cancel-replay-1") NoStream [makeEvent "C1" (Aeson.object [])]-                threadDelay 200_000--                saveStarted <- newEmptyMVar-                holdSave <- newEmptyMVar-                firstDelivery <- newTVarIO (0 :: Int)-                let pauseSave config _-                        | KTypes.name config == subscriptionName = putMVar saveStarted () >> takeMVar holdSave-                        | otherwise = pure ()--                withSaveCheckpointHookForTest pauseSave $-                    within "shutdown during checkpoint persistence" $-                        runEff $-                            runTracingNoop $ do-                                adapter <--                                    kirokuAdapter store $-                                        defaultKirokuAdapterConfig subscriptionName AllStreams-                                result <--                                    runApp-                                        defaultAppConfig-                                        [-                                            ( ProcessorId "checkpoint-cancel"-                                            , mkProcessor adapter $ \_ -> do-                                                liftIO $ atomically $ do-                                                    count <- readTVar firstDelivery-                                                    writeTVar firstDelivery (count + 1)-                                                pure AckOk-                                            )-                                        ]-                                case result of-                                    Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                                    Right appHandle -> do-                                        liftIO $ waitForCount firstDelivery 1 10_000_000-                                        liftIO $ takeMVar saveStarted-                                        stopApp appHandle--                readCheckpointPosition store checkpointKey `shouldReturn` Just (GlobalPosition 0)-                Map.null <$> subscriptionStates store `shouldReturn` True--                replayed <- newIORef ([] :: [Int64])-                within "restart after interrupted checkpoint" $-                    runEff $-                        runTracingNoop $ do-                            adapter <--                                kirokuAdapter store $-                                    defaultKirokuAdapterConfig subscriptionName AllStreams-                            result <--                                runApp-                                    defaultAppConfig-                                    [-                                        ( ProcessorId "checkpoint-replay"-                                        , mkProcessor adapter $ \ingested -> do-                                            liftIO $ modifyIORef' replayed (globalPos (envelopePayload ingested) :)-                                            pure (AckHalt (HaltFatal "replay observed"))-                                        )-                                    ]-                            case result of-                                Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                                Right appHandle -> waitApp appHandle--                readIORef replayed `shouldReturn` [1]-                readCheckpointPosition store checkpointKey `shouldReturn` Just (GlobalPosition 0)-                Map.null <$> subscriptionStates store `shouldReturn` True--            it "AckDeadLetter records the event and the next event continues (EP-40 M3)" $ \store -> do-                Right _ <- runStoreIO store $ appendToStream (StreamName "ackdl-1") NoStream [makeEvent "D1" (Aeson.object [])]-                Right _ <- runStoreIO store $ appendToStream (StreamName "ackdl-2") NoStream [makeEvent "D2" (Aeson.object [])]-                threadDelay 200_000--                okPayloads <- newIORef ([] :: [RecordedEvent])-                countVar <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "ackdl-proj") AllStreams--                    let handler ingested = do-                            n <- liftIO $ atomically $ do-                                c <- readTVar countVar-                                writeTVar countVar (c + 1)-                                pure (c + 1)-                            -- Dead-letter the first event; accept the second.-                            if n == 1-                                then pure (AckDeadLetter (PoisonPill "poison"))-                                else do-                                    liftIO $ modifyIORef' okPayloads (envelopePayload ingested :)-                                    pure AckOk--                    res <- runApp defaultAppConfig [(ProcessorId "ackdl", mkProcessor adapter handler)]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ waitForCount countVar 2 10_000_000-                            stopApp appHandle--                -- The first event (global position 1) is recorded as dead-letter.-                dls <- readDeadLetters store "ackdl-proj"-                map SQL.deadLetterGlobalPosition dls `shouldBe` [1]-                -- The second event was delivered and accepted.-                oks <- readIORef okPayloads-                map globalPos oks `shouldBe` [2]--            it "guardKirokuHandlerWith retries a transient handler exception and preserves attempts" $ \store -> do-                Right _ <- runStoreIO store $ appendToStream (StreamName "guardretry-1") NoStream [makeEvent "GR1" (Aeson.object [])]-                threadDelay 200_000--                deliveries <- newIORef (0 :: Int)-                attempts <- newIORef ([] :: [Maybe Attempt])-                countVar <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "guardretry-proj") AllStreams--                    let throwingOnce ingested = do-                            let Message{envelope = Envelope{attempt = deliveryAttempt}} = ingested-                            n <- liftIO $ do-                                modifyIORef' attempts (deliveryAttempt :)-                                modifyIORef' deliveries (+ 1)-                                atomically $ do-                                    c <- readTVar countVar-                                    let c' = c + 1-                                    writeTVar countVar c'-                                    pure c'-                            if n == 1-                                then liftIO $ E.throwIO (userError "transient projection failure")-                                else pure AckOk-                        handler = guardKirokuHandlerWith (const (AckRetry (Ack.RetryDelay 0))) throwingOnce--                    res <- runApp defaultAppConfig [(ProcessorId "guardretry", mkProcessor adapter handler)]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ waitForCount countVar 2 10_000_000-                            stopApp appHandle--                n <- readIORef deliveries-                n `shouldBe` 2-                reverse <$> readIORef attempts `shouldReturn` [Just (Attempt 0), Just (Attempt 1)]-                dls <- readDeadLetters store "guardretry-proj"-                length dls `shouldBe` 0--            it "guardKirokuHandlerWith dead-letters a persistent handler exception and continues" $ \store -> do-                Right _ <- runStoreIO store $ appendToStream (StreamName "guarddl-1") NoStream [makeEvent "GDL1" (Aeson.object [])]-                Right _ <- runStoreIO store $ appendToStream (StreamName "guarddl-2") NoStream [makeEvent "GDL2" (Aeson.object [])]-                threadDelay 200_000--                nextSeen <- newTVarIO (0 :: Int)--                runEff $ runTracingNoop $ do-                    adapter <--                        kirokuAdapter store $-                            defaultKirokuAdapterConfig (SubscriptionName "guarddl-proj") AllStreams--                    let alwaysThrowFirst ingested =-                            if globalPos (envelopePayload ingested) == 1-                                then liftIO $ E.throwIO (userError "persistent projection failure")-                                else do-                                    liftIO $ atomically $ do-                                        c <- readTVar nextSeen-                                        writeTVar nextSeen (c + 1)-                                    pure AckOk-                        handler = guardKirokuHandlerWith (const (AckRetry (Ack.RetryDelay 0))) alwaysThrowFirst--                    res <- runApp defaultAppConfig [(ProcessorId "guarddl", mkProcessor adapter handler)]-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ waitForCount nextSeen 1 15_000_000-                            stopApp appHandle--                dls <- readDeadLetters store "guarddl-proj"-                map SQL.deadLetterGlobalPosition dls `shouldBe` [1]--        describe "consumer groups" $ do-            it "four-member group delivers a disjoint, complete partition of the stream" $ \store -> do-                -- 20 streams × 2 events = 40 events, global positions 1..40, category "cg".-                let streams = ["cg-" <> T.pack (show i) | i <- [1 .. 20 :: Int]]-                mapM_-                    ( \sn -> do-                        let evs = [makeEvent ("EV" <> T.pack (show k)) (Aeson.object []) | k <- [1 .. 2 :: Int]]-                        Right _ <- runStoreIO store $ appendToStream (StreamName sn) NoStream evs-                        pure ()-                    )-                    streams-                threadDelay 200_000--                -- One IORef + one TVar counter per member.-                refs <- mapM (const (newIORef ([] :: [RecordedEvent]))) [0 .. 3 :: Int]-                cvars <- mapM (const (newTVarIO (0 :: Int))) [0 .. 3 :: Int]--                runEff $ runTracingNoop $ do-                    -- Build one adapter per member index; each carries the same-                    -- subscriptionName and a distinct member of a size-4 group.-                    adapters <--                        mapM-                            ( \m ->-                                kirokuAdapter store $-                                    defaultKirokuAdapterConfig (SubscriptionName "cg-shibuya-group") (Category (CategoryName "cg"))-                                        & #consumerGroup .~ Just (validConsumerGroup m 4)-                            )-                            [0, 1, 2, 3]--                    let mkHandler ref' cvar ingested = do-                            liftIO $ do-                                modifyIORef' ref' (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar cvar-                                    writeTVar cvar (c + 1)-                            pure AckOk--                        processors =-                            [ ( ProcessorId ("cg-member-" <> T.pack (show m))-                              , mkProcessor (adapters !! m) (mkHandler (refs !! m) (cvars !! m))-                              )-                            | m <- [0 .. 3 :: Int]-                            ]--                    res <- runApp defaultAppConfig processors-                    case res of-                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                        Right appHandle -> do-                            liftIO $ waitForTotal cvars 40 15_000_000-                            stopApp appHandle--                collected <- mapM readIORef refs-                -- Disjoint + complete: the union of every member's delivered global-                -- positions is exactly [1..40] — no event dropped, none delivered twice.-                let allPositions = sort (concatMap (map globalPos) collected)-                allPositions `shouldBe` [1 .. 40]-                -- No member starved: 20 streams over 4 members gives each ≥ 1 stream.-                mapM_ (\c -> length c `shouldSatisfy` (>= 1)) collected--        describe "consumer group policy" $ do-            it "rejects an invalid member concurrency before opening any subscription" $ \store -> do-                -- A request the helper cannot honor per member returns Left without-                -- starting any worker. The (unused) store proves no subscription opens.-                result <--                    runEff $-                        runTracingNoop $-                            kirokuConsumerGroupProcessors-                                store-                                ( defaultConsumerGroupConfig (SubscriptionName "cgp-reject") (Category (CategoryName "cgp-reject")) (validGroupSize 4)-                                    & #memberConcurrency .~ Async 4-                                )-                                ( \ingested -> do-                                    let _ = envelopePayload ingested-                                    pure AckOk-                                )-                case result of-                    Left err -> err `shouldBe` InvalidPolicyCombo "StrictInOrder requires Serial concurrency"-                    Right _ -> expectationFailure "expected Left PolicyError for Async member concurrency"--                withResult <--                    runEff $-                        runTracingNoop $-                            kirokuConsumerGroupProcessorsWith-                                ( \_ -> do-                                    liftIO $ expectationFailure "factory should not be called for an invalid policy"-                                    pure stubAdapter-                                )-                                ( defaultConsumerGroupConfig (SubscriptionName "cgp-reject-with") (Category (CategoryName "cgp-reject-with")) (validGroupSize 4)-                                    & #memberConcurrency .~ Async 4-                                )-                                ( \ingested -> do-                                    let _ = envelopePayload ingested-                                    pure AckOk-                                )-                case withResult of-                    Left err -> err `shouldBe` InvalidPolicyCombo "StrictInOrder requires Serial concurrency"-                    Right _ -> expectationFailure "expected Left PolicyError for Async member concurrency"--            it "rejects non-positive group sizes at construction" $ \_store -> do-                mkConsumerGroupSize 0 `shouldBe` Left (InvalidConsumerGroup 0 0)-                mkConsumerGroupSize (-1) `shouldBe` Left (InvalidConsumerGroup 0 (-1))--            it "shuts down every created member and preserves the factory failure when cleanup throws" $ \_store -> do-                shutdowns <- newIORef ([] :: [Int32])-                let sentinel = userError "member 2 failed"-                    factory m-                        | m == 2 = liftIO $ E.throwIO sentinel-                        | otherwise =-                            pure-                                Adapter-                                    { adapterName = "stub-kiroku"-                                    , source = Stream.nil-                                    , shutdown = liftIO $ do-                                        modifyIORef' shutdowns (<> [m])-                                        if m == 1-                                            then E.throwIO (userError "member 1 cleanup failed")-                                            else pure ()-                                    }-                    handler ingested = do-                        let _ = envelopePayload ingested-                        pure AckOk-                result <--                    E.try $-                        runEff $-                            runTracingNoop $-                                kirokuConsumerGroupProcessorsWith-                                    factory-                                    (defaultConsumerGroupConfig (SubscriptionName "cgp-partial-cleanup") AllStreams (validGroupSize 3))-                                    handler-                case result of-                    Left (e :: E.SomeException) -> show e `shouldContain` "member 2 failed"-                    Right _ -> expectationFailure "expected the member factory failure to rethrow"-                readIORef shutdowns `shouldReturn` [1, 0]--            it "owns a returned member before cancellation can interrupt construction" $ \_store -> do-                acquired <- newEmptyMVar-                holdTransfer <- newEmptyMVar-                finished <- newEmptyMVar-                shutdowns <- newIORef ([] :: [Int32])-                maskingStates <- newIORef ([] :: [E.MaskingState])-                worker <--                    forkIO $ do-                        outcome <--                            E.try $-                                runEff $-                                    acquireAllAndTransfer-                                        2-                                        pure-                                        (\member -> liftIO $ modifyIORef' shutdowns (<> [member]))-                                        ( \member _ -> liftIO $ do-                                            state <- E.getMaskingState-                                            modifyIORef' maskingStates (<> [state])-                                            if member == 0-                                                then putMVar acquired () >> takeMVar holdTransfer-                                                else pure ()-                                        )-                                        pure-                        putMVar finished (outcome :: Either E.SomeException [Int32])--                takeMVar acquired-                killThread worker-                outcome <- takeMVar finished-                case outcome of-                    Left exception ->-                        E.fromException exception `shouldBe` Just E.ThreadKilled-                    Right _ -> expectationFailure "expected cancellation during ownership transfer"-                readIORef maskingStates `shouldReturn` [E.MaskedInterruptible]-                readIORef shutdowns `shouldReturn` [0]--            it "wraps consumer-group handlers so exceptions finalize a retry decision" $ \store -> do-                Right _ <- runStoreIO store $ appendToStream (StreamName "cgp-guard-1") NoStream [makeEvent "CGGuard" (Aeson.object [])]-                threadDelay 200_000--                countVar <- newTVarIO (0 :: Int)-                runEff $-                    runTracingNoop $ do-                        result <--                            kirokuConsumerGroupProcessors-                                store-                                (defaultConsumerGroupConfig (SubscriptionName "cgp-guard") AllStreams (validGroupSize 1))-                                ( \_ -> do-                                    n <- liftIO $ atomically $ do-                                        c <- readTVar countVar-                                        let c' = c + 1-                                        writeTVar countVar c'-                                        pure c'-                                    if n == 1-                                        then liftIO $ E.throwIO (userError "guard me")-                                        else pure AckOk-                                )-                        case result of-                            Left err -> liftIO $ expectationFailure ("kirokuConsumerGroupProcessors failed: " <> show err)-                            Right processors -> do-                                appRes <- runApp defaultAppConfig processors-                                case appRes of-                                    Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                                    Right appHandle -> do-                                        liftIO $ waitForCount countVar 2 15_000_000-                                        stopApp appHandle--            it "one call yields N PartitionedInOrder processors; members partition the stream disjointly" $ \store -> do-                -- 20 streams × 2 events = 40 events, global positions 1..40, category "cgp".-                let streams = ["cgp-" <> T.pack (show i) | i <- [1 .. 20 :: Int]]-                mapM_-                    ( \sn -> do-                        let evs = [makeEvent ("EV" <> T.pack (show k)) (Aeson.object []) | k <- [1 .. 2 :: Int]]-                        Right _ <- runStoreIO store $ appendToStream (StreamName sn) NoStream evs-                        pure ()-                    )-                    streams-                threadDelay 200_000--                -- A single shared handler records every delivery (consumer-group-                -- members all run the same handler). One counter for the whole group.-                ref <- newIORef ([] :: [RecordedEvent])-                countVar <- newTVarIO (0 :: Int)--                let cfg =-                        defaultConsumerGroupConfig-                            (SubscriptionName "cgp-shibuya-group")-                            (Category (CategoryName "cgp"))-                            (validGroupSize 4)--                runEff $ runTracingNoop $ do-                    let handler ingested = do-                            liftIO $ do-                                modifyIORef' ref (envelopePayload ingested :)-                                atomically $ do-                                    c <- readTVar countVar-                                    writeTVar countVar (c + 1)-                            pure AckOk--                    -- One call replaces the manual `mapM mkMemberAdapter [0..3]` wiring.-                    res <- kirokuConsumerGroupProcessors store cfg handler-                    case res of-                        Left err -> liftIO $ expectationFailure ("kirokuConsumerGroupProcessors failed: " <> show err)-                        Right processors -> do-                            -- The group is presented as N processors, each pinned to the-                            -- group-level PartitionedInOrder contract + per-member Serial.-                            liftIO $ length processors `shouldBe` 4-                            -- 'Nothing' marks a batching processor, which the-                            -- consumer group must never produce.-                            let policies =-                                    map-                                        ( \(_, p) -> case p of-                                            QueueProcessor{ordering, concurrency} -> Just (ordering, concurrency)-                                            BatchingProcessor{} -> Nothing-                                        )-                                        processors-                            liftIO $ policies `shouldBe` replicate 4 (Just (PartitionedInOrder, Serial))-                            -- The member index is readable off the ProcessorId.-                            let pids = map (\(ProcessorId p, _) -> p) processors-                            liftIO $-                                pids-                                    `shouldBe` ["cgp-shibuya-group-member-" <> T.pack (show m) | m <- [0 .. 3 :: Int]]--                            appRes <- runApp defaultAppConfig processors-                            case appRes of-                                Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)-                                Right appHandle -> do-                                    liftIO $ waitForCount countVar 40 15_000_000-                                    stopApp appHandle--                collected <- readIORef ref-                -- Disjoint + complete through the helper: the union of every member's-                -- delivered global positions is exactly [1..40] — no event dropped, none-                -- delivered twice. No duplicates ⇒ the member partitions are disjoint;-                -- ==[1..40] ⇒ the union is the complete source. This is the same-                -- disjoint-complete property MasterPlan 4 used, now asserted through the-                -- single-call helper and the Shibuya pipeline.-                sort (map globalPos collected) `shouldBe` [1 .. 40]-                -- Every one of the 20 originating streams was covered.-                length (nub (map origStreamId collected)) `shouldBe` 20---- Helpers--envelopePayload :: Message es RecordedEvent -> RecordedEvent-envelopePayload ing = let Message{envelope = env} = ing in env ^. #payload--stubAdapter :: Adapter es RecordedEvent-stubAdapter =-    Adapter-        { adapterName = "stub-kiroku"-        , source = Stream.nil-        , shutdown = pure ()-        }---- | Read the dead letters recorded for a non-group subscription (member 0).-readDeadLetters :: KirokuStore -> Text -> IO [SQL.DeadLetterRecord]-readDeadLetters store subName = readMemberDeadLetters store subName 0--readMemberDeadLetters :: KirokuStore -> Text -> Int32 -> IO [SQL.DeadLetterRecord]-readMemberDeadLetters store subName memberIndex = do-    result <- Pool.use (store ^. #pool) (Session.statement (subName, memberIndex) SQL.readDeadLettersStmt)-    case result of-        Left err -> error ("readDeadLetters failed: " <> show err)-        Right v -> pure (toList v)---- | The raw global position of a recorded event (for disjoint/complete checks).-globalPos :: RecordedEvent -> Int64-globalPos e = case e ^. #globalPosition of GlobalPosition p -> p---- | The originating stream's surrogate id (for stream-coverage checks).-origStreamId :: RecordedEvent -> Int64-origStreamId e = case e ^. #originalStreamId of StreamId p -> p---- | Wait until the sum of all TVar counts reaches the target or the timeout fires.-waitForTotal :: [STM.TVar Int] -> Int -> Int -> IO ()-waitForTotal vars target timeoutMicros = do-    timeoutVar <- registerDelay timeoutMicros-    result <--        atomically $-            ( do-                total <- sum <$> mapM readTVar vars-                STM.check (total >= target)-                pure True-            )-                `STM.orElse` ( do-                                t <- readTVar timeoutVar-                                STM.check t-                                pure False-                             )-    unless result $ do-        actual <- atomically $ sum <$> mapM readTVar vars-        expectationFailure ("Timed out waiting for total " <> show target <> ", got " <> show actual)--waitForCount :: STM.TVar Int -> Int -> Int -> IO ()-waitForCount countVar target timeoutMicros = do-    timeoutVar <- registerDelay timeoutMicros-    result <--        atomically $-            ( do-                c <- readTVar countVar-                STM.check (c >= target)-                pure True-            )-                `STM.orElse` ( do-                                t <- readTVar timeoutVar-                                STM.check t-                                pure False-                             )-    unless result $ do-        actual <- atomically $ readTVar countVar-        expectationFailure ("Timed out waiting for count " <> show target <> ", got " <> show actual)--within :: String -> IO a -> IO a-within label action = do-    result <- timeout 10_000_000 action-    case result of-        Nothing -> fail ("Timed out waiting for " <> label)-        Just a -> pure a--waitForCheckpointPosition :: KirokuStore -> SubscriptionCheckpointKey -> GlobalPosition -> IO ()-waitForCheckpointPosition store expectedKey expectedPosition =-    within "subscription checkpoint initialization" loop-  where-    loop = do-        Right (SubscriptionCheckpointInventory _ checkpoints) <--            runStoreIO store subscriptionCheckpointInventory-        let positions =-                [ (SubscriptionCheckpointKey name member, position)-                | SubscriptionCheckpoint name member position _ <- toList checkpoints-                ]-        if (expectedKey, expectedPosition) `elem` positions-            then pure ()-            else threadDelay 20_000 >> loop--readCheckpointPosition :: KirokuStore -> SubscriptionCheckpointKey -> IO (Maybe GlobalPosition)-readCheckpointPosition store expectedKey = do-    Right (SubscriptionCheckpointInventory _ checkpoints) <--        runStoreIO store subscriptionCheckpointInventory-    pure $ case [ position-                | SubscriptionCheckpoint name member position _ <- toList checkpoints-                , SubscriptionCheckpointKey name member == expectedKey-                ] of-        position : _ -> Just position-        [] -> Nothing--makeEvent :: Text -> Value -> EventData-makeEvent typ p =-    EventData-        { eventId = Nothing-        , eventType = EventType typ-        , payload = p-        , metadata = Nothing-        , causationId = Nothing-        , correlationId = Nothing-        }--{- | A throwaway attribute source for the trace-header tests, whose assertions-do not depend on the kiroku identity attributes.--}-sampleEnvelopeAttrs :: KirokuEnvelopeAttrs-sampleEnvelopeAttrs = kirokuEnvelopeAttrs "test-sub" Nothing--makeRecordedEvent :: Maybe Value -> RecordedEvent-makeRecordedEvent meta =-    RecordedEvent-        { eventId = EventId UUID.nil-        , eventType = EventType "TraceEvent"-        , streamVersion = StreamVersion 1-        , globalPosition = GlobalPosition 1-        , originalStreamId = StreamId 1-        , originalVersion = StreamVersion 1-        , payload = Aeson.object []-        , metadata = meta-        , causationId = Nothing-        , correlationId = Nothing-        , createdAt = UTCTime (fromGregorian 2026 5 16) 0-        }--withTestStore :: (KirokuStore -> IO ()) -> IO ()-withTestStore action =-    withMigratedTestDatabase $ \connStr ->-        withStore (defaultConnectionSettings connStr) action--validGroupSize :: Int32 -> ConsumerGroupSize-validGroupSize = either (error . show) Prelude.id . mkConsumerGroupSize--validConsumerGroup :: Int32 -> Int32 -> ConsumerGroup-validConsumerGroup m n = either (error . show) Prelude.id (mkConsumerGroup m (validGroupSize n))--acknowledgementSpec :: Spec-acknowledgementSpec = do-    describe "acknowledgement liveness" $ do-        forM_ [False, True] $ \guarded ->-            it (if guarded then "guarded single processor retries with pacing and continues" else "standard Shibuya handler exceptions are finalized") $-                withTestStore $ \store -> do-                    Right _ <--                        runStoreIO store $-                            appendToStream-                                (StreamName "ack-liveness")-                                NoStream-                                [makeEvent "First" (Aeson.object []), makeEvent "Second" (Aeson.object [])]-                    attempts <- newIORef ([] :: [(Int64, Maybe Attempt, Word64)])-                    count <- newTVarIO (0 :: Int)-                    within "runner finalization" $ runEff $ runTracingNoop $ do-                        adapter <- kirokuAdapter store (defaultKirokuAdapterConfig (SubscriptionName "ack-liveness") AllStreams)-                        let handler ingested = do-                                now <- liftIO getMonotonicTimeNSec-                                let pos = globalPos (envelopePayload ingested)-                                    Message{envelope = Envelope{attempt = attempt'}} = ingested-                                n <- liftIO $ do-                                    modifyIORef' attempts ((pos, attempt', now) :)-                                    atomically $ do-                                        old <- readTVar count-                                        writeTVar count (old + 1)-                                        pure old-                                if n == 0 then liftIO (E.throwIO (userError "one-shot handler failure")) else pure AckOk-                            processor = if guarded then kirokuProcessor adapter handler else mkProcessor adapter handler-                        app <- runApp defaultAppConfig [(ProcessorId "ack-liveness", processor)] >>= either (liftIO . fail . show) pure-                        liftIO $ waitForCount count 3 5_000_000-                        liftIO $ waitForCheckpointPosition store (SubscriptionCheckpointKey (SubscriptionName "ack-liveness") 0) (GlobalPosition 2)-                        stopApp app-                    recorded <- reverse <$> readIORef attempts-                    map (\(pos, attempt', _) -> (pos, attempt')) recorded-                        `shouldBe` [(1, Just (Attempt 0)), (1, Just (Attempt 1)), (2, Just (Attempt 0))]-                    case recorded of-                        (_, _, firstAt) : (_, _, retriedAt) : _ ->-                            if guarded-                                then retriedAt - firstAt `shouldSatisfy` (>= 900_000_000)-                                else retriedAt - firstAt `shouldSatisfy` (< 900_000_000)-                        _ -> expectationFailure "expected three finalized deliveries"--        forM_ [Nothing, Just 0.05] $ \warnAfter ->-            it ("keeps a raw unfinalized item pending and forwards warning policy " <> show warnAfter) $ do-                warnings <- newTVarIO (0 :: Int)-                warning <- STM.newEmptyTMVarIO-                let observe event = case event of-                        KirokuEventSubscriptionHandlerStalled _ pos _ elapsed _ -> atomically $ do-                            STM.modifyTVar' warnings (+ 1)-                            _ <- STM.tryPutTMVar warning (pos, elapsed)-                            pure ()-                        _ -> pure ()-                withMigratedTestDatabase $ \connStr ->-                    withStore (defaultConnectionSettings connStr & #eventHandler .~ Just observe) $ \store -> do-                        Right _ <--                            runStoreIO store $-                                appendToStream-                                    (StreamName "ack-raw")-                                    NoStream-                                    [makeEvent "First" (Aeson.object []), makeEvent "Second" (Aeson.object [])]-                        within "raw acknowledgement ownership" $ runEff $ runTracingNoop $ Timeout.runTimeout $ do-                            adapter <--                                kirokuAdapter-                                    store-                                    ( (defaultKirokuAdapterConfig (SubscriptionName "ack-raw") AllStreams)-                                        & #handlerStallWarnAfter .~ warnAfter-                                    )-                            let Adapter{source = sourceStream, shutdown = shutdownAction} = adapter-                            first <- Stream.uncons sourceStream-                            case first of-                                Nothing -> liftIO (expectationFailure "expected first item")-                                Just (item, rest) -> do-                                    -- The raw consumer intentionally leaves item unfinalized.-                                    secondRead <- Timeout.timeout 150_000 (Stream.uncons rest)-                                    liftIO (isNothing secondRead `shouldBe` True)-                                    liftIO $ readCheckpointPosition store (SubscriptionCheckpointKey (SubscriptionName "ack-raw") 0) `shouldReturn` Just (GlobalPosition 0)-                                    case warnAfter of-                                        Nothing -> liftIO $ STM.readTVarIO warnings `shouldReturn` 0-                                        Just threshold -> liftIO $ do-                                            (pos, elapsed) <- within "raw ack warning" (atomically (STM.readTMVar warning))-                                            pos `shouldBe` GlobalPosition 1-                                            elapsed `shouldSatisfy` (>= threshold)-                                    -- Finalization remains explicitly owned by this consumer.-                                    let Ingested{ack = AckHandle{finalize = finalizeFirst}} = item-                                    finalizeFirst AckOk-                                    (second, _) <- Stream.uncons rest >>= maybe (liftIO (fail "expected second item")) pure-                                    let Ingested{envelope = Envelope{payload = secondEvent}} = second-                                    liftIO $ globalPos secondEvent `shouldBe` 2-                                    let Ingested{ack = AckHandle{finalize = finalizeSecond}} = second-                                    finalizeSecond AckOk-                                    liftIO $ waitForCheckpointPosition store (SubscriptionCheckpointKey (SubscriptionName "ack-raw") 0) (GlobalPosition 2)-                            shutdownAction-                        stoppedCount <- STM.readTVarIO warnings-                        threadDelay 100_000-                        STM.readTVarIO warnings `shouldReturn` stoppedCount--    describe "retry policy" $ do-        forM_ [Nothing, Just 2] $ \attemptLimit ->-            it ("honors single-adapter total delivery limit " <> show attemptLimit) $ withTestStore $ \store -> do-                Right _ <- runStoreIO store $ appendToStream (StreamName "ack-policy") NoStream [makeEvent "Retry" (Aeson.object [])]-                count <- newTVarIO (0 :: Int)-                runEff $ runTracingNoop $ do-                    let config =-                            (defaultKirokuAdapterConfig (SubscriptionName "ack-policy") AllStreams)-                                & #retryPolicy .~ maybe defaultRetryPolicy RetryPolicy attemptLimit-                    adapter <- kirokuAdapter store config-                    let handler _ = liftIO (atomically (STM.modifyTVar' count (+ 1))) >> pure (AckRetry (Ack.RetryDelay 0))-                    app <- runApp defaultAppConfig [(ProcessorId "ack-policy", kirokuProcessor adapter handler)] >>= either (liftIO . fail . show) pure-                    liftIO $ waitForCheckpointPosition store (SubscriptionCheckpointKey (SubscriptionName "ack-policy") 0) (GlobalPosition 1)-                    stopApp app-                STM.readTVarIO count `shouldReturn` maybe 5 Prelude.id attemptLimit-                letters <- readDeadLetters store "ack-policy"-                map SQL.deadLetterAttemptCount letters `shouldBe` [fromIntegral (maybe 5 Prelude.id attemptLimit)]--        it "forwards two attempts and stall policy to both consumer-group members" $ do-            warned <- newTVarIO Set.empty-            release <- STM.newEmptyTMVarIO-            let observe event = case event of-                    KirokuEventSubscriptionHandlerStalled _ _ _ _ (GroupMember member' 2) -> atomically (STM.modifyTVar' warned (Set.insert member'))-                    _ -> pure ()-            withMigratedTestDatabase $ \connStr ->-                withStore (defaultConnectionSettings connStr & #eventHandler .~ Just observe) $ \store -> do-                    -- Enough distinct streams to exercise both PostgreSQL hash slots;-                    -- require both warnings before release so an empty slot fails.-                    forM_ [1 .. 20 :: Int] $ \i -> do-                        Right _ <- runStoreIO store $ appendToStream (StreamName ("ack-group-" <> T.pack (show i))) NoStream [makeEvent "Retry" (Aeson.object [])]-                        pure ()-                    seen <- newIORef Map.empty-                    runEff $ runTracingNoop $ do-                        let config =-                                (defaultConsumerGroupConfig (SubscriptionName "ack-policy-group") AllStreams (validGroupSize 2))-                                    & #retryPolicy .~ RetryPolicy 2-                                    & #handlerStallWarnAfter .~ Just 0.05-                            handler ingested = do-                                liftIO $ atomically (STM.readTMVar release)-                                liftIO $ atomicModifyIORef' seen (\old -> (Map.insertWith (+) (globalPos (envelopePayload ingested)) (1 :: Int) old, ()))-                                pure (AckRetry (Ack.RetryDelay 0))-                        processors <- kirokuConsumerGroupProcessors store config handler >>= either (liftIO . fail . show) pure-                        app <- runApp defaultAppConfig processors >>= either (liftIO . fail . show) pure-                        liftIO $ within "both group members stalled" $ atomically $ readTVar warned >>= STM.check . (== Set.fromList [0, 1])-                        liftIO $ atomically (STM.putTMVar release ())-                        liftIO $ within "all group dead letters" $ do-                            let loop = do-                                    letters <- concat <$> mapM (readMemberDeadLetters store "ack-policy-group") [0, 1]-                                    if length letters == 20 then pure () else threadDelay 10_000 >> loop-                            loop-                        stopApp app-                    Map.elems <$> readIORef seen `shouldReturn` replicate 20 2-                    letters <- concat <$> mapM (readMemberDeadLetters store "ack-policy-group") [0, 1]-                    sort (map SQL.deadLetterGlobalPosition letters) `shouldBe` [1 .. 20]-                    map SQL.deadLetterAttemptCount letters `shouldBe` replicate 20 2+import Shibuya.Adapter.Kiroku+  ( consumerGroupPolicy,+    defaultConsumerGroupConfig,+    defaultKirokuAdapterConfig,+    guardKirokuHandlerWith,+    kirokuAdapter,+    kirokuConsumerGroupProcessors,+    kirokuConsumerGroupProcessorsWith,+    kirokuProcessor,+  )+import Shibuya.Adapter.Kiroku.Convert+  ( KirokuEnvelopeAttrs,+    kirokuEnvelopeAttrs,+    toEnvelope,+    toIngestedAck,+    toKirokuDeadLetterReason,+  )+import Shibuya.Adapter.Kiroku.Internal (acquireAllAndTransfer)+import Shibuya.App+  ( ProcessorId (..),+    QueueProcessor (..),+    defaultAppConfig,+    getAppMetrics,+    mkProcessor,+    runApp,+    stopApp,+    stopAppGracefully,+    waitApp,+  )+import Shibuya.App qualified as Shibuya+import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..))+import Shibuya.Core.Ack qualified as Ack+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Error (PolicyError (..))+import Shibuya.Core.Ingested (Ingested (..), Message (..))+import Shibuya.Core.Metrics (ProcessorState (..))+import Shibuya.Core.Types (Attempt (..), Envelope (..))+import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))+import Shibuya.Telemetry.Effect (runTracingNoop)+import Streamly.Data.Fold qualified as Fold+import Streamly.Data.Stream qualified as Stream+import System.Timeout (timeout)+import Test.Hspec++main :: IO ()+main = withSharedMigratedPostgres $ hspec $ do+  describe "toEnvelope" $ do+    it "copies W3C trace metadata into Shibuya trace headers" $ do+      let traceparent :: Text+          traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"+          tracestate :: Text+          tracestate = "rojo=00f067aa0ba902b7"+          Envelope {traceContext} =+            toEnvelope+              sampleEnvelopeAttrs+              ( makeRecordedEvent+                  ( Just $+                      Aeson.object+                        [ "traceparent" Aeson..= traceparent,+                          "tracestate" Aeson..= tracestate,+                          "other" Aeson..= ("preserved" :: Text)+                        ]+                  )+              )++      traceContext+        `shouldBe` Just+          [ ("traceparent", "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"),+            ("tracestate", "rojo=00f067aa0ba902b7")+          ]++    it "omits trace headers when traceparent is absent or not a string" $ do+      let Envelope {traceContext = missingTraceparent} =+            toEnvelope sampleEnvelopeAttrs (makeRecordedEvent (Just (Aeson.object ["tracestate" Aeson..= ("state" :: Text)])))+          Envelope {traceContext = nonStringTraceparent} =+            toEnvelope sampleEnvelopeAttrs (makeRecordedEvent (Just (Aeson.object ["traceparent" Aeson..= Aeson.Number 1])))++      missingTraceparent `shouldBe` Nothing+      nonStringTraceparent `shouldBe` Nothing++    it "does not surface broker headers (headers is Nothing), even with trace metadata" $ do+      let Envelope {headers = noMeta} =+            toEnvelope sampleEnvelopeAttrs (makeRecordedEvent Nothing)+          Envelope {headers = withTraceMeta} =+            toEnvelope+              sampleEnvelopeAttrs+              (makeRecordedEvent (Just (Aeson.object ["traceparent" Aeson..= ("00-abc-def-01" :: Text)])))++      noMeta `shouldBe` Nothing+      withTraceMeta `shouldBe` Nothing++    it "stamps kiroku identity attributes for a non-grouped subscription (EP-5 M2)" $ do+      let attrs = kirokuEnvelopeAttrs "orders-proj" Nothing+          Envelope {attributes} = toEnvelope attrs (makeRecordedEvent Nothing)+      -- makeRecordedEvent: eventType "TraceEvent", globalPosition 1.+      HashMap.lookup "kiroku.subscription.name" attributes+        `shouldBe` Just (toAttribute ("orders-proj" :: Text))+      HashMap.lookup "messaging.system" attributes+        `shouldBe` Just (toAttribute ("kiroku" :: Text))+      HashMap.lookup "messaging.destination.name" attributes+        `shouldBe` Just (toAttribute ("orders-proj" :: Text))+      HashMap.lookup "kiroku.event.type" attributes+        `shouldBe` Just (toAttribute ("TraceEvent" :: Text))+      HashMap.lookup "kiroku.event.global_position" attributes+        `shouldBe` Just (toAttribute (1 :: Int64))+      -- No member key for a non-grouped subscription.+      HashMap.lookup "kiroku.consumer_group.member" attributes `shouldBe` Nothing++    it "stamps the consumer-group member attribute for a grouped subscription (EP-5 M2)" $ do+      let attrs = kirokuEnvelopeAttrs "orders-proj" (Just 2)+          Envelope {attributes} = toEnvelope attrs (makeRecordedEvent Nothing)+      HashMap.lookup "kiroku.subscription.name" attributes+        `shouldBe` Just (toAttribute ("orders-proj" :: Text))+      HashMap.lookup "kiroku.consumer_group.member" attributes+        `shouldBe` Just (toAttribute (2 :: Int64))++  describe "toKirokuDeadLetterReason" $ do+    it "keeps the framework reasons on their dedicated Kiroku constructors" $ do+      toKirokuDeadLetterReason 3 (PoisonPill "bad event")+        `shouldBe` KTypes.DeadLetterPoison "bad event"+      toKirokuDeadLetterReason 3 (InvalidPayload "not json")+        `shouldBe` KTypes.DeadLetterInvalid "not json"+      toKirokuDeadLetterReason 3 MaxRetriesExceeded+        `shouldBe` KTypes.DeadLetterMaxAttempts 3++    it "maps an application failure to DeadLetterOther with a queryable code (shibuya-core 0.9)" $ do+      code <-+        either (E.throwIO . userError . T.unpack) pure $+          Ack.mkDeadLetterCode "billing.card_declined"+      let reason =+            toKirokuDeadLetterReason 1 (Ack.ApplicationFailure code "card declined by issuer")++      KTypes.deadLetterSummary reason+        `shouldBe` "billing.card_declined: card declined by issuer"+      KTypes.deadLetterReasonJson reason+        `shouldBe` Aeson.object+          [ "kind" Aeson..= ("other" :: Text),+            "summary" Aeson..= ("billing.card_declined: card declined by issuer" :: Text),+            "detail"+              Aeson..= Aeson.object+                [ "code" Aeson..= ("billing.card_declined" :: Text),+                  "detail" Aeson..= ("card declined by issuer" :: Text)+                ]+          ]++  describe "consumer group policy" $ do+    it "accepts Serial member concurrency as (PartitionedInOrder, Serial)" $+      consumerGroupPolicy Serial `shouldBe` Right (PartitionedInOrder, Serial)++    it "rejects Ahead member concurrency with a PolicyError" $+      consumerGroupPolicy (Ahead 4)+        `shouldBe` Left (InvalidPolicyCombo "StrictInOrder requires Serial concurrency")++    it "rejects Async member concurrency with a PolicyError" $+      consumerGroupPolicy (Async 4)+        `shouldBe` Left (InvalidPolicyCombo "StrictInOrder requires Serial concurrency")++  acknowledgementSpec+  around withTestStore $ do+    describe "kirokuAdapter" $ do+      it "delivers only matching event types when an eventTypeFilter is set (EP-43)" $ \store -> do+        -- A, B, A, B, A on one stream (global positions 1..5). The adapter+        -- is filtered to type A; the Shibuya handler must see only the As.+        let events =+              [ makeEvent "A" (Aeson.object []),+                makeEvent "B" (Aeson.object []),+                makeEvent "A" (Aeson.object []),+                makeEvent "B" (Aeson.object []),+                makeEvent "A" (Aeson.object [])+              ]+        Right _ <- runStoreIO store $ appendToStream (StreamName "etf-adapter-1") NoStream events+        ref <- newIORef ([] :: [RecordedEvent])+        countVar <- newTVarIO (0 :: Int)+        runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "etf-adapter") AllStreams+                & #eventTypeFilter .~ OnlyEventTypes (Set.fromList [EventType "A"])+          let handler ingested = do+                liftIO $ do+                  modifyIORef' ref (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar countVar+                    writeTVar countVar (c + 1)+                pure AckOk+          res <- runApp defaultAppConfig [(ProcessorId "etf", mkProcessor adapter handler)]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ waitForCount countVar 3 10_000_000+              stopApp appHandle++        collected <- readIORef ref+        -- Only the three A events reached the handler; no Bs.+        map (^. #eventType) collected `shouldBe` replicate 3 (EventType "A")+        map globalPos (reverse collected) `shouldBe` [1, 3, 5]++      it "delivers only matching types across a filtered consumer group (EP-43)" $ \store -> do+        -- 20 streams in category "etfg", each [A, B]; A at odd global+        -- positions 1,3,..,39. A size-4 group filtered to type A must+        -- deliver exactly the 20 As, disjoint and complete.+        let streams = [0 .. 19 :: Int]+        mapM_+          ( \i -> do+              let sn = StreamName ("etfg-" <> T.pack (show i))+              Right _ <-+                runStoreIO store $+                  appendToStream sn NoStream [makeEvent "A" (Aeson.object []), makeEvent "B" (Aeson.object [])]+              pure ()+          )+          streams+        ref <- newIORef ([] :: [RecordedEvent])+        countVar <- newTVarIO (0 :: Int)+        runEff $ runTracingNoop $ do+          let cfg =+                defaultConsumerGroupConfig (SubscriptionName "etfg") (Category (CategoryName "etfg")) (validGroupSize 4)+                  & #eventTypeFilter .~ OnlyEventTypes (Set.fromList [EventType "A"])+              handler ingested = do+                liftIO $ do+                  modifyIORef' ref (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar countVar+                    writeTVar countVar (c + 1)+                pure AckOk+          res <- kirokuConsumerGroupProcessors store cfg handler+          case res of+            Left err -> liftIO $ expectationFailure ("kirokuConsumerGroupProcessors failed: " <> show err)+            Right processors -> do+              appRes <- runApp defaultAppConfig processors+              case appRes of+                Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                Right appHandle -> do+                  liftIO $ waitForCount countVar 20 15_000_000+                  stopApp appHandle++        collected <- readIORef ref+        -- Every delivered event is an A (filter honored per member)...+        map (^. #eventType) collected `shouldBe` replicate 20 (EventType "A")+        -- ...and the union over members is the complete, disjoint set of A+        -- positions [1,3,..,39] — no A dropped, none delivered twice.+        sort (map globalPos collected) `shouldBe` [1, 3 .. 39]++      it "delivers only events matching an opaque selector (EP-43 follow-up)" $ \store -> do+        -- All one type "A" (so eventTypeFilter cannot distinguish them);+        -- only the payload tag {keep} differs. keep, skip, keep, skip, keep+        -- at positions 1..5. The selector admits only keep=True.+        let keepObj = Aeson.object [("keep", Aeson.Bool True)]+            skipObj = Aeson.object [("keep", Aeson.Bool False)]+            events =+              [ makeEvent "A" keepObj,+                makeEvent "A" skipObj,+                makeEvent "A" keepObj,+                makeEvent "A" skipObj,+                makeEvent "A" keepObj+              ]+        Right _ <- runStoreIO store $ appendToStream (StreamName "sel-adapter-1") NoStream events+        ref <- newIORef ([] :: [RecordedEvent])+        countVar <- newTVarIO (0 :: Int)+        runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "sel-adapter") AllStreams+                & #selector .~ Just (\e -> (e ^. #payload) == keepObj)+          let handler ingested = do+                liftIO $ do+                  modifyIORef' ref (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar countVar+                    writeTVar countVar (c + 1)+                pure AckOk+          res <- runApp defaultAppConfig [(ProcessorId "sel", mkProcessor adapter handler)]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ waitForCount countVar 3 10_000_000+              stopApp appHandle++        collected <- readIORef ref+        -- Only the three keep events reached the handler (positions 1,3,5);+        -- the two skip events were filtered worker-side and never delivered.+        map globalPos (reverse collected) `shouldBe` [1, 3, 5]++      it "delivers catch-up events through Shibuya pipeline" $ \store -> do+        let events = map (\i -> makeEvent ("CU" <> T.pack (show i)) (Aeson.object [])) [1 .. 10 :: Int]+        Right _ <- runStoreIO store $ appendToStream (StreamName "shibuya-catchup-1") NoStream events+        threadDelay 200_000++        ref <- newIORef ([] :: [RecordedEvent])+        countVar <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "shibuya-catchup-test") AllStreams++          let handler ingested = do+                liftIO $ do+                  modifyIORef' ref (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar countVar+                    writeTVar countVar (c + 1)+                pure AckOk++          res <- runApp defaultAppConfig [(ProcessorId "catchup", mkProcessor adapter handler)]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ waitForCount countVar 10 10_000_000+              stopApp appHandle++        collected <- readIORef ref+        length collected `shouldBe` 10++      it "supports a future-only FromCurrentHead adapter" $ \store -> do+        Right _ <-+          runStoreIO store $+            appendToStream+              (StreamName "shibuya-future-only-events")+              NoStream+              (map (\i -> makeEvent ("Historical" <> T.pack (show i)) (Aeson.object [])) [1 .. 4 :: Int])+        ref <- newIORef ([] :: [RecordedEvent])+        countVar <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "shibuya-future-only") AllStreams+                & #missingCheckpointPolicy .~ FromCurrentHead+          liftIO $+            waitForCheckpointPosition+              store+              (SubscriptionCheckpointKey (SubscriptionName "shibuya-future-only") 0)+              (GlobalPosition 4)+          liftIO $ do+            Right _ <-+              runStoreIO store $+                appendToStream+                  (StreamName "shibuya-future-only-events")+                  StreamExists+                  [makeEvent "Future" (Aeson.object [])]+            pure ()+          let handler ingested = do+                liftIO $ do+                  modifyIORef' ref (envelopePayload ingested :)+                  atomically $ do+                    count <- readTVar countVar+                    writeTVar countVar (count + 1)+                pure AckOk+          result <- runApp defaultAppConfig [(ProcessorId "future-only", mkProcessor adapter handler)]+          case result of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ waitForCount countVar 1 10_000_000+              stopApp appHandle++        collected <- reverse <$> readIORef ref+        map globalPos collected `shouldBe` [5]++      it "preserves an existing checkpoint when restarted with FromCurrentHead" $ \store -> do+        let subscriptionName = SubscriptionName "shibuya-existing-checkpoint"+            checkpointKey = SubscriptionCheckpointKey subscriptionName 0+            config policy =+              defaultKirokuAdapterConfig subscriptionName AllStreams+                & #missingCheckpointPolicy .~ policy+        Right _ <- runStoreIO store $ appendToStream (StreamName "shibuya-existing-checkpoint-events") NoStream [makeEvent "First" (Aeson.object [])]+        firstSeen <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          adapter <- kirokuAdapter store (config FromBeginning)+          result <-+            runApp+              defaultAppConfig+              [ ( ProcessorId "existing-checkpoint-first",+                  mkProcessor adapter $ \_ -> do+                    liftIO $ atomically $ writeTVar firstSeen 1+                    pure AckOk+                )+              ]+          case result of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ waitForCount firstSeen 1 10_000_000+              liftIO $ waitForCheckpointPosition store checkpointKey (GlobalPosition 1)+              stopApp appHandle++        Right _ <-+          runStoreIO store $+            appendToStream+              (StreamName "shibuya-existing-checkpoint-events")+              StreamExists+              [makeEvent "Second" (Aeson.object [])]+        replayed <- newIORef ([] :: [Int64])+        within "existing checkpoint restart" $+          runEff $+            runTracingNoop $ do+              adapter <- kirokuAdapter store (config FromCurrentHead)+              result <-+                runApp+                  defaultAppConfig+                  [ ( ProcessorId "existing-checkpoint-restart",+                      mkProcessor adapter $ \ingested -> do+                        liftIO $ modifyIORef' replayed (globalPos (envelopePayload ingested) :)+                        pure (AckHalt (HaltFatal "existing checkpoint observed"))+                    )+                  ]+              case result of+                Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                Right appHandle -> waitApp appHandle++        readIORef replayed `shouldReturn` [2]++      it "surfaces FailIfMissing without leaving a subscription registered" $ \store -> do+        let subscriptionName = SubscriptionName "shibuya-required-checkpoint"+            checkpointKey = SubscriptionCheckpointKey subscriptionName 0+            config =+              defaultKirokuAdapterConfig subscriptionName AllStreams+                & #missingCheckpointPolicy .~ FailIfMissing+        outcome <-+          E.try $+            within "FailIfMissing adapter refusal" $+              runEff $+                runTracingNoop $ do+                  adapter <- kirokuAdapter store config+                  let Adapter {source = sourceStream} = adapter+                  Stream.fold Fold.drain sourceStream+        case outcome of+          Left exception+            | Just (SubscriptionCheckpointMissing actual) <- E.fromException (exception :: E.SomeException) ->+                actual `shouldBe` checkpointKey+          Left exception -> expectationFailure ("expected SubscriptionCheckpointMissing, got: " <> show exception)+          Right () -> expectationFailure "expected a missing-checkpoint refusal"+        Map.null <$> subscriptionStates store `shouldReturn` True++      it "delivers live events through Shibuya pipeline" $ \store -> do+        ref <- newIORef ([] :: [RecordedEvent])+        countVar <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "shibuya-live-test") AllStreams++          let handler ingested = do+                liftIO $ do+                  modifyIORef' ref (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar countVar+                    writeTVar countVar (c + 1)+                pure AckOk++          res <- runApp defaultAppConfig [(ProcessorId "live", mkProcessor adapter handler)]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ threadDelay 200_000+              liftIO $ do+                let events = map (\i -> makeEvent ("Live" <> T.pack (show i)) (Aeson.object [])) [1 .. 5 :: Int]+                Right _ <- runStoreIO store $ appendToStream (StreamName "shibuya-live-1") NoStream events+                pure ()+              liftIO $ waitForCount countVar 5 10_000_000+              stopApp appHandle++        collected <- readIORef ref+        length collected `shouldBe` 5++      it "survives an append burst while the handler is paused" $ \store -> do+        firstSeen <- newEmptyMVar+        release <- newEmptyMVar+        countVar <- newTVarIO (0 :: Int)+        seenRef <- newIORef ([] :: [RecordedEvent])++        runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "burst-pause-resume") AllStreams+                & #queueCapacity .~ 1++          let handler ingested = do+                n <- liftIO $ do+                  modifyIORef' seenRef (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar countVar+                    let c' = c + 1+                    writeTVar countVar c'+                    pure c'+                if n == 1+                  then liftIO $ putMVar firstSeen () >> takeMVar release+                  else pure ()+                pure AckOk++          res <- runApp defaultAppConfig [(ProcessorId "burst", mkProcessor adapter handler)]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ do+                let appendOne i =+                      runStoreIO store $+                        appendToStream+                          (StreamName ("burst-" <> T.pack (show (i :: Int))))+                          NoStream+                          [makeEvent ("Burst" <> T.pack (show i)) (Aeson.object [])]+                Right _ <- appendOne 1+                takeMVar firstSeen+                mapM_ (\i -> appendOne i >>= (`shouldSatisfy` either (const False) (const True))) [2 .. 40 :: Int]+                putMVar release ()+                waitForCount countVar 40 15_000_000+              stopApp appHandle++        seen <- readIORef seenRef+        sort (map globalPos seen) `shouldBe` [1 .. 40]++      it "ends adapter.source cleanly after shutdown" $ \store -> do+        within "adapter source clean shutdown" $+          runEff $+            runTracingNoop $ do+              adapter <- kirokuAdapter store $ defaultKirokuAdapterConfig (SubscriptionName "source-clean-end") AllStreams+              let Adapter {source = sourceStream, shutdown = shutdownAction} = adapter+              shutdownAction+              Stream.fold Fold.drain sourceStream++      it "rethrows worker crashes through adapter.source" $ \store -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "source-crash-1") NoStream [makeEvent "SourceCrash" (Aeson.object [])]+        threadDelay 200_000+        let injectBoom _ _ = E.throwIO (userError "adapter source worker boom")+        result <-+          withFetchBatchHookForTest injectBoom $+            E.try $+              within "adapter source worker crash" $+                runEff $+                  runTracingNoop $ do+                    adapter <- kirokuAdapter store $ defaultKirokuAdapterConfig (SubscriptionName "source-crash") AllStreams+                    let Adapter {source = sourceStream} = adapter+                    Stream.fold Fold.drain sourceStream+        case result of+          Left (e :: E.SomeException) -> show e `shouldContain` "adapter source worker boom"+          Right _ -> expectationFailure "expected adapter.source to rethrow the worker exception"++      it "runs multiple category subscriptions concurrently" $ \store -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "orders-s1") NoStream [makeEvent "OrderCreated" (Aeson.object [])]+        Right _ <- runStoreIO store $ appendToStream (StreamName "orders-s2") NoStream [makeEvent "OrderShipped" (Aeson.object [])]+        Right _ <- runStoreIO store $ appendToStream (StreamName "payments-s1") NoStream [makeEvent "PaymentReceived" (Aeson.object [])]+        Right _ <- runStoreIO store $ appendToStream (StreamName "inventory-s1") NoStream [makeEvent "StockUpdated" (Aeson.object [])]+        Right _ <- runStoreIO store $ appendToStream (StreamName "inventory-s2") NoStream [makeEvent "StockDepleted" (Aeson.object [])]+        threadDelay 200_000++        ordersRef <- newIORef ([] :: [RecordedEvent])+        paymentsRef <- newIORef ([] :: [RecordedEvent])+        inventoryRef <- newIORef ([] :: [RecordedEvent])+        ordersCount <- newTVarIO (0 :: Int)+        paymentsCount <- newTVarIO (0 :: Int)+        inventoryCount <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          let mkCatAdapter nm cat =+                kirokuAdapter store $+                  defaultKirokuAdapterConfig (SubscriptionName nm) (Category (CategoryName cat))+          let mkHandler ref' cVar ingested = do+                liftIO $ do+                  modifyIORef' ref' (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar cVar+                    writeTVar cVar (c + 1)+                pure AckOk++          ordersAdapter <- mkCatAdapter "orders-proj" "orders"+          paymentsAdapter <- mkCatAdapter "payments-proj" "payments"+          inventoryAdapter <- mkCatAdapter "inventory-proj" "inventory"++          res <-+            runApp+              defaultAppConfig+              [ (ProcessorId "orders", mkProcessor ordersAdapter (mkHandler ordersRef ordersCount)),+                (ProcessorId "payments", mkProcessor paymentsAdapter (mkHandler paymentsRef paymentsCount)),+                (ProcessorId "inventory", mkProcessor inventoryAdapter (mkHandler inventoryRef inventoryCount))+              ]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ do+                waitForCount ordersCount 2 10_000_000+                waitForCount paymentsCount 1 10_000_000+                waitForCount inventoryCount 2 10_000_000+              stopApp appHandle++        ordersCollected <- readIORef ordersRef+        paymentsCollected <- readIORef paymentsRef+        inventoryCollected <- readIORef inventoryRef+        length ordersCollected `shouldBe` 2+        length paymentsCollected `shouldBe` 1+        length inventoryCollected `shouldBe` 2++      it "isolates a failing subscription from healthy ones" $ \store -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "good-s1") NoStream (map (\i -> makeEvent ("Good" <> T.pack (show i)) (Aeson.object [])) [1 .. 3 :: Int])+        Right _ <- runStoreIO store $ appendToStream (StreamName "bad-s1") NoStream [makeEvent "Bad1" (Aeson.object []), makeEvent "Bad2" (Aeson.object [])]+        Right _ <- runStoreIO store $ appendToStream (StreamName "other-s1") NoStream [makeEvent "Other1" (Aeson.object []), makeEvent "Other2" (Aeson.object [])]+        threadDelay 200_000++        goodRef <- newIORef ([] :: [RecordedEvent])+        otherRef <- newIORef ([] :: [RecordedEvent])+        goodCount <- newTVarIO (0 :: Int)+        otherCount <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          goodAdapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "good-proj") (Category (CategoryName "good"))+          badAdapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "bad-proj") (Category (CategoryName "bad"))+          otherAdapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "other-proj") (Category (CategoryName "other"))++          let goodHandler ingested = do+                liftIO $ do+                  modifyIORef' goodRef (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar goodCount+                    writeTVar goodCount (c + 1)+                pure AckOk++          let badHandler _ingested = do+                liftIO $ error "handler crash!"++          let otherHandler ingested = do+                liftIO $ do+                  modifyIORef' otherRef (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar otherCount+                    writeTVar otherCount (c + 1)+                pure AckOk++          res <-+            runApp+              defaultAppConfig+              [ (ProcessorId "good", mkProcessor goodAdapter goodHandler),+                (ProcessorId "bad", mkProcessor badAdapter badHandler),+                (ProcessorId "other", mkProcessor otherAdapter otherHandler)+              ]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ do+                waitForCount goodCount 3 10_000_000+                waitForCount otherCount 2 10_000_000+                threadDelay 500_000+              metrics <- getAppMetrics appHandle+              case Map.lookup (ProcessorId "bad") metrics of+                Just m -> case m ^. #state of+                  Failed _ _ -> pure ()+                  other -> liftIO $ expectationFailure ("Expected Failed state, got: " <> show other)+                Nothing -> liftIO $ expectationFailure "No metrics for bad processor"+              stopApp appHandle++        goodCollected <- readIORef goodRef+        otherCollected <- readIORef otherRef+        length goodCollected `shouldBe` 3+        length otherCollected `shouldBe` 2++      it "shuts down all subscriptions coordinately" $ \store -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "shut-a") NoStream [makeEvent "A" (Aeson.object [])]+        Right _ <- runStoreIO store $ appendToStream (StreamName "shut-b") NoStream [makeEvent "B" (Aeson.object [])]+        threadDelay 200_000++        countA <- newTVarIO (0 :: Int)+        countB <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          adapterA <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "shut-a-proj") (Category (CategoryName "shut"))+          adapterB <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "shut-b-proj") (Category (CategoryName "shut"))++          let handlerA _ingested = do+                liftIO $ atomically $ do+                  c <- readTVar countA+                  writeTVar countA (c + 1)+                pure AckOk++          let handlerB _ingested = do+                liftIO $ atomically $ do+                  c <- readTVar countB+                  writeTVar countB (c + 1)+                pure AckOk++          res <-+            runApp+              defaultAppConfig+              [ (ProcessorId "a", mkProcessor adapterA handlerA),+                (ProcessorId "b", mkProcessor adapterB handlerB)+              ]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ do+                waitForCount countA 2 10_000_000+                waitForCount countB 2 10_000_000+              drained <- stopAppGracefully Shibuya.defaultShutdownConfig appHandle+              liftIO $ drained `shouldBe` True++    describe "ack dispositions" $ do+      it "keeps the first Kiroku reply when an ack is finalized twice" $ \_store -> do+        reply <- newEmptyTMVarIO+        cancellations <- newIORef (0 :: Int)+        let Ingested {ack = AckHandle {finalize}} =+              toIngestedAck+                sampleEnvelopeAttrs+                (modifyIORef' cancellations (+ 1))+                (AckItem (makeRecordedEvent Nothing) 0 reply)++        runEff $ do+          finalize AckOk+          finalize (AckDeadLetter (PoisonPill "late duplicate"))++        atomically (readTMVar reply) `shouldReturn` Continue+        readIORef cancellations `shouldReturn` 0++      it "AckRetry redelivers the same event, then AckOk advances (EP-40 M3)" $ \store -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "ackretry-1") NoStream [makeEvent "R1" (Aeson.object [])]+        threadDelay 200_000++        deliveries <- newIORef (0 :: Int)+        countVar <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "ackretry-proj") AllStreams++          let handler _ingested = do+                n <- liftIO $ do+                  modifyIORef' deliveries (+ 1)+                  atomically $ do+                    c <- readTVar countVar+                    writeTVar countVar (c + 1)+                    pure (c + 1)+                -- Retry the first delivery, accept the second.+                pure (if n == 1 then AckRetry (Ack.RetryDelay 0) else AckOk)++          res <- runApp defaultAppConfig [(ProcessorId "ackretry", mkProcessor adapter handler)]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ waitForCount countVar 2 10_000_000+              stopApp appHandle++        -- The one event was delivered twice (initial + one retry).+        n <- readIORef deliveries+        n `shouldBe` 2+        -- It was not dead-lettered (the retry succeeded).+        dls <- readDeadLetters store "ackretry-proj"+        length dls `shouldBe` 0++      it "AckHalt leaves the checkpoint behind the event and restart replays it" $ \store -> do+        let subscriptionName = SubscriptionName "ackhalt-replay-proj"+            checkpointKey = SubscriptionCheckpointKey subscriptionName 0+        Right _ <- runStoreIO store $ appendToStream (StreamName "ackhalt-replay-1") NoStream [makeEvent "H1" (Aeson.object [])]+        threadDelay 200_000++        deliveries <- newIORef ([] :: [Int64])+        let runHaltingProcessor label =+              within label $+                runEff $+                  runTracingNoop $ do+                    adapter <-+                      kirokuAdapter store $+                        defaultKirokuAdapterConfig subscriptionName AllStreams+                    result <-+                      runApp+                        defaultAppConfig+                        [ ( ProcessorId "ackhalt-replay",+                            mkProcessor adapter $ \ingested -> do+                              liftIO $ modifyIORef' deliveries (globalPos (envelopePayload ingested) :)+                              pure (AckHalt (HaltFatal "intentional halt"))+                          )+                        ]+                    case result of+                      Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                      Right appHandle -> waitApp appHandle++        runHaltingProcessor "first AckHalt"+        readCheckpointPosition store checkpointKey `shouldReturn` Just (GlobalPosition 0)++        runHaltingProcessor "AckHalt restart"+        reverse <$> readIORef deliveries `shouldReturn` [1, 1]+        readCheckpointPosition store checkpointKey `shouldReturn` Just (GlobalPosition 0)++      it "shutdown after an ack but before checkpoint persistence replays without leaking" $ \store -> do+        let subscriptionName = SubscriptionName "checkpoint-cancel-replay-proj"+            checkpointKey = SubscriptionCheckpointKey subscriptionName 0+        Right _ <- runStoreIO store $ appendToStream (StreamName "checkpoint-cancel-replay-1") NoStream [makeEvent "C1" (Aeson.object [])]+        threadDelay 200_000++        saveStarted <- newEmptyMVar+        holdSave <- newEmptyMVar+        firstDelivery <- newTVarIO (0 :: Int)+        let pauseSave config _+              | KTypes.name config == subscriptionName = putMVar saveStarted () >> takeMVar holdSave+              | otherwise = pure ()++        withSaveCheckpointHookForTest pauseSave $+          within "shutdown during checkpoint persistence" $+            runEff $+              runTracingNoop $ do+                adapter <-+                  kirokuAdapter store $+                    defaultKirokuAdapterConfig subscriptionName AllStreams+                result <-+                  runApp+                    defaultAppConfig+                    [ ( ProcessorId "checkpoint-cancel",+                        mkProcessor adapter $ \_ -> do+                          liftIO $ atomically $ do+                            count <- readTVar firstDelivery+                            writeTVar firstDelivery (count + 1)+                          pure AckOk+                      )+                    ]+                case result of+                  Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                  Right appHandle -> do+                    liftIO $ waitForCount firstDelivery 1 10_000_000+                    liftIO $ takeMVar saveStarted+                    stopApp appHandle++        readCheckpointPosition store checkpointKey `shouldReturn` Just (GlobalPosition 0)+        Map.null <$> subscriptionStates store `shouldReturn` True++        replayed <- newIORef ([] :: [Int64])+        within "restart after interrupted checkpoint" $+          runEff $+            runTracingNoop $ do+              adapter <-+                kirokuAdapter store $+                  defaultKirokuAdapterConfig subscriptionName AllStreams+              result <-+                runApp+                  defaultAppConfig+                  [ ( ProcessorId "checkpoint-replay",+                      mkProcessor adapter $ \ingested -> do+                        liftIO $ modifyIORef' replayed (globalPos (envelopePayload ingested) :)+                        pure (AckHalt (HaltFatal "replay observed"))+                    )+                  ]+              case result of+                Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                Right appHandle -> waitApp appHandle++        readIORef replayed `shouldReturn` [1]+        readCheckpointPosition store checkpointKey `shouldReturn` Just (GlobalPosition 0)+        Map.null <$> subscriptionStates store `shouldReturn` True++      it "AckDeadLetter records the event and the next event continues (EP-40 M3)" $ \store -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "ackdl-1") NoStream [makeEvent "D1" (Aeson.object [])]+        Right _ <- runStoreIO store $ appendToStream (StreamName "ackdl-2") NoStream [makeEvent "D2" (Aeson.object [])]+        threadDelay 200_000++        okPayloads <- newIORef ([] :: [RecordedEvent])+        countVar <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "ackdl-proj") AllStreams++          let handler ingested = do+                n <- liftIO $ atomically $ do+                  c <- readTVar countVar+                  writeTVar countVar (c + 1)+                  pure (c + 1)+                -- Dead-letter the first event; accept the second.+                if n == 1+                  then pure (AckDeadLetter (PoisonPill "poison"))+                  else do+                    liftIO $ modifyIORef' okPayloads (envelopePayload ingested :)+                    pure AckOk++          res <- runApp defaultAppConfig [(ProcessorId "ackdl", mkProcessor adapter handler)]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ waitForCount countVar 2 10_000_000+              stopApp appHandle++        -- The first event (global position 1) is recorded as dead-letter.+        dls <- readDeadLetters store "ackdl-proj"+        map SQL.deadLetterGlobalPosition dls `shouldBe` [1]+        -- The second event was delivered and accepted.+        oks <- readIORef okPayloads+        map globalPos oks `shouldBe` [2]++      it "guardKirokuHandlerWith retries a transient handler exception and preserves attempts" $ \store -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "guardretry-1") NoStream [makeEvent "GR1" (Aeson.object [])]+        threadDelay 200_000++        deliveries <- newIORef (0 :: Int)+        attempts <- newIORef ([] :: [Maybe Attempt])+        countVar <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "guardretry-proj") AllStreams++          let throwingOnce ingested = do+                let Message {envelope = Envelope {attempt = deliveryAttempt}} = ingested+                n <- liftIO $ do+                  modifyIORef' attempts (deliveryAttempt :)+                  modifyIORef' deliveries (+ 1)+                  atomically $ do+                    c <- readTVar countVar+                    let c' = c + 1+                    writeTVar countVar c'+                    pure c'+                if n == 1+                  then liftIO $ E.throwIO (userError "transient projection failure")+                  else pure AckOk+              handler = guardKirokuHandlerWith (const (AckRetry (Ack.RetryDelay 0))) throwingOnce++          res <- runApp defaultAppConfig [(ProcessorId "guardretry", mkProcessor adapter handler)]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ waitForCount countVar 2 10_000_000+              stopApp appHandle++        n <- readIORef deliveries+        n `shouldBe` 2+        reverse <$> readIORef attempts `shouldReturn` [Just (Attempt 0), Just (Attempt 1)]+        dls <- readDeadLetters store "guardretry-proj"+        length dls `shouldBe` 0++      it "guardKirokuHandlerWith dead-letters a persistent handler exception and continues" $ \store -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "guarddl-1") NoStream [makeEvent "GDL1" (Aeson.object [])]+        Right _ <- runStoreIO store $ appendToStream (StreamName "guarddl-2") NoStream [makeEvent "GDL2" (Aeson.object [])]+        threadDelay 200_000++        nextSeen <- newTVarIO (0 :: Int)++        runEff $ runTracingNoop $ do+          adapter <-+            kirokuAdapter store $+              defaultKirokuAdapterConfig (SubscriptionName "guarddl-proj") AllStreams++          let alwaysThrowFirst ingested =+                if globalPos (envelopePayload ingested) == 1+                  then liftIO $ E.throwIO (userError "persistent projection failure")+                  else do+                    liftIO $ atomically $ do+                      c <- readTVar nextSeen+                      writeTVar nextSeen (c + 1)+                    pure AckOk+              handler = guardKirokuHandlerWith (const (AckRetry (Ack.RetryDelay 0))) alwaysThrowFirst++          res <- runApp defaultAppConfig [(ProcessorId "guarddl", mkProcessor adapter handler)]+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ waitForCount nextSeen 1 15_000_000+              stopApp appHandle++        dls <- readDeadLetters store "guarddl-proj"+        map SQL.deadLetterGlobalPosition dls `shouldBe` [1]++    describe "consumer groups" $ do+      it "four-member group delivers a disjoint, complete partition of the stream" $ \store -> do+        -- 20 streams × 2 events = 40 events, global positions 1..40, category "cg".+        let streams = ["cg-" <> T.pack (show i) | i <- [1 .. 20 :: Int]]+        mapM_+          ( \sn -> do+              let evs = [makeEvent ("EV" <> T.pack (show k)) (Aeson.object []) | k <- [1 .. 2 :: Int]]+              Right _ <- runStoreIO store $ appendToStream (StreamName sn) NoStream evs+              pure ()+          )+          streams+        threadDelay 200_000++        -- One IORef + one TVar counter per member.+        refs <- mapM (const (newIORef ([] :: [RecordedEvent]))) [0 .. 3 :: Int]+        cvars <- mapM (const (newTVarIO (0 :: Int))) [0 .. 3 :: Int]++        runEff $ runTracingNoop $ do+          -- Build one adapter per member index; each carries the same+          -- subscriptionName and a distinct member of a size-4 group.+          adapters <-+            mapM+              ( \m ->+                  kirokuAdapter store $+                    defaultKirokuAdapterConfig (SubscriptionName "cg-shibuya-group") (Category (CategoryName "cg"))+                      & #consumerGroup .~ Just (validConsumerGroup m 4)+              )+              [0, 1, 2, 3]++          let mkHandler ref' cvar ingested = do+                liftIO $ do+                  modifyIORef' ref' (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar cvar+                    writeTVar cvar (c + 1)+                pure AckOk++              processors =+                [ ( ProcessorId ("cg-member-" <> T.pack (show m)),+                    mkProcessor (adapters !! m) (mkHandler (refs !! m) (cvars !! m))+                  )+                | m <- [0 .. 3 :: Int]+                ]++          res <- runApp defaultAppConfig processors+          case res of+            Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+            Right appHandle -> do+              liftIO $ waitForTotal cvars 40 15_000_000+              stopApp appHandle++        collected <- mapM readIORef refs+        -- Disjoint + complete: the union of every member's delivered global+        -- positions is exactly [1..40] — no event dropped, none delivered twice.+        let allPositions = sort (concatMap (map globalPos) collected)+        allPositions `shouldBe` [1 .. 40]+        -- No member starved: 20 streams over 4 members gives each ≥ 1 stream.+        mapM_ (\c -> length c `shouldSatisfy` (>= 1)) collected++    describe "consumer group policy" $ do+      it "rejects an invalid member concurrency before opening any subscription" $ \store -> do+        -- A request the helper cannot honor per member returns Left without+        -- starting any worker. The (unused) store proves no subscription opens.+        result <-+          runEff $+            runTracingNoop $+              kirokuConsumerGroupProcessors+                store+                ( defaultConsumerGroupConfig (SubscriptionName "cgp-reject") (Category (CategoryName "cgp-reject")) (validGroupSize 4)+                    & #memberConcurrency .~ Async 4+                )+                ( \ingested -> do+                    let _ = envelopePayload ingested+                    pure AckOk+                )+        case result of+          Left err -> err `shouldBe` InvalidPolicyCombo "StrictInOrder requires Serial concurrency"+          Right _ -> expectationFailure "expected Left PolicyError for Async member concurrency"++        withResult <-+          runEff $+            runTracingNoop $+              kirokuConsumerGroupProcessorsWith+                ( \_ -> do+                    liftIO $ expectationFailure "factory should not be called for an invalid policy"+                    pure stubAdapter+                )+                ( defaultConsumerGroupConfig (SubscriptionName "cgp-reject-with") (Category (CategoryName "cgp-reject-with")) (validGroupSize 4)+                    & #memberConcurrency .~ Async 4+                )+                ( \ingested -> do+                    let _ = envelopePayload ingested+                    pure AckOk+                )+        case withResult of+          Left err -> err `shouldBe` InvalidPolicyCombo "StrictInOrder requires Serial concurrency"+          Right _ -> expectationFailure "expected Left PolicyError for Async member concurrency"++      it "rejects non-positive group sizes at construction" $ \_store -> do+        mkConsumerGroupSize 0 `shouldBe` Left (InvalidConsumerGroup 0 0)+        mkConsumerGroupSize (-1) `shouldBe` Left (InvalidConsumerGroup 0 (-1))++      it "shuts down every created member and preserves the factory failure when cleanup throws" $ \_store -> do+        shutdowns <- newIORef ([] :: [Int32])+        let sentinel = userError "member 2 failed"+            factory m+              | m == 2 = liftIO $ E.throwIO sentinel+              | otherwise =+                  pure+                    Adapter+                      { adapterName = "stub-kiroku",+                        source = Stream.nil,+                        shutdown = liftIO $ do+                          modifyIORef' shutdowns (<> [m])+                          if m == 1+                            then E.throwIO (userError "member 1 cleanup failed")+                            else pure ()+                      }+            handler ingested = do+              let _ = envelopePayload ingested+              pure AckOk+        result <-+          E.try $+            runEff $+              runTracingNoop $+                kirokuConsumerGroupProcessorsWith+                  factory+                  (defaultConsumerGroupConfig (SubscriptionName "cgp-partial-cleanup") AllStreams (validGroupSize 3))+                  handler+        case result of+          Left (e :: E.SomeException) -> show e `shouldContain` "member 2 failed"+          Right _ -> expectationFailure "expected the member factory failure to rethrow"+        readIORef shutdowns `shouldReturn` [1, 0]++      it "owns a returned member before cancellation can interrupt construction" $ \_store -> do+        acquired <- newEmptyMVar+        holdTransfer <- newEmptyMVar+        finished <- newEmptyMVar+        shutdowns <- newIORef ([] :: [Int32])+        maskingStates <- newIORef ([] :: [E.MaskingState])+        worker <-+          forkIO $ do+            outcome <-+              E.try $+                runEff $+                  acquireAllAndTransfer+                    2+                    pure+                    (\member -> liftIO $ modifyIORef' shutdowns (<> [member]))+                    ( \member _ -> liftIO $ do+                        state <- E.getMaskingState+                        modifyIORef' maskingStates (<> [state])+                        if member == 0+                          then putMVar acquired () >> takeMVar holdTransfer+                          else pure ()+                    )+                    pure+            putMVar finished (outcome :: Either E.SomeException [Int32])++        takeMVar acquired+        killThread worker+        outcome <- takeMVar finished+        case outcome of+          Left exception ->+            E.fromException exception `shouldBe` Just E.ThreadKilled+          Right _ -> expectationFailure "expected cancellation during ownership transfer"+        readIORef maskingStates `shouldReturn` [E.MaskedInterruptible]+        readIORef shutdowns `shouldReturn` [0]++      it "wraps consumer-group handlers so exceptions finalize a retry decision" $ \store -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "cgp-guard-1") NoStream [makeEvent "CGGuard" (Aeson.object [])]+        threadDelay 200_000++        countVar <- newTVarIO (0 :: Int)+        runEff $+          runTracingNoop $ do+            result <-+              kirokuConsumerGroupProcessors+                store+                (defaultConsumerGroupConfig (SubscriptionName "cgp-guard") AllStreams (validGroupSize 1))+                ( \_ -> do+                    n <- liftIO $ atomically $ do+                      c <- readTVar countVar+                      let c' = c + 1+                      writeTVar countVar c'+                      pure c'+                    if n == 1+                      then liftIO $ E.throwIO (userError "guard me")+                      else pure AckOk+                )+            case result of+              Left err -> liftIO $ expectationFailure ("kirokuConsumerGroupProcessors failed: " <> show err)+              Right processors -> do+                appRes <- runApp defaultAppConfig processors+                case appRes of+                  Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                  Right appHandle -> do+                    liftIO $ waitForCount countVar 2 15_000_000+                    stopApp appHandle++      it "one call yields N PartitionedInOrder processors; members partition the stream disjointly" $ \store -> do+        -- 20 streams × 2 events = 40 events, global positions 1..40, category "cgp".+        let streams = ["cgp-" <> T.pack (show i) | i <- [1 .. 20 :: Int]]+        mapM_+          ( \sn -> do+              let evs = [makeEvent ("EV" <> T.pack (show k)) (Aeson.object []) | k <- [1 .. 2 :: Int]]+              Right _ <- runStoreIO store $ appendToStream (StreamName sn) NoStream evs+              pure ()+          )+          streams+        threadDelay 200_000++        -- A single shared handler records every delivery (consumer-group+        -- members all run the same handler). One counter for the whole group.+        ref <- newIORef ([] :: [RecordedEvent])+        countVar <- newTVarIO (0 :: Int)++        let cfg =+              defaultConsumerGroupConfig+                (SubscriptionName "cgp-shibuya-group")+                (Category (CategoryName "cgp"))+                (validGroupSize 4)++        runEff $ runTracingNoop $ do+          let handler ingested = do+                liftIO $ do+                  modifyIORef' ref (envelopePayload ingested :)+                  atomically $ do+                    c <- readTVar countVar+                    writeTVar countVar (c + 1)+                pure AckOk++          -- One call replaces the manual `mapM mkMemberAdapter [0..3]` wiring.+          res <- kirokuConsumerGroupProcessors store cfg handler+          case res of+            Left err -> liftIO $ expectationFailure ("kirokuConsumerGroupProcessors failed: " <> show err)+            Right processors -> do+              -- The group is presented as N processors, each pinned to the+              -- group-level PartitionedInOrder contract + per-member Serial.+              liftIO $ length processors `shouldBe` 4+              -- 'Nothing' marks a batching processor, which the+              -- consumer group must never produce.+              let policies =+                    map+                      ( \(_, p) -> case p of+                          QueueProcessor {ordering, concurrency} -> Just (ordering, concurrency)+                          BatchingProcessor {} -> Nothing+                      )+                      processors+              liftIO $ policies `shouldBe` replicate 4 (Just (PartitionedInOrder, Serial))+              -- The member index is readable off the ProcessorId.+              let pids = map (\(ProcessorId p, _) -> p) processors+              liftIO $+                pids+                  `shouldBe` ["cgp-shibuya-group-member-" <> T.pack (show m) | m <- [0 .. 3 :: Int]]++              appRes <- runApp defaultAppConfig processors+              case appRes of+                Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                Right appHandle -> do+                  liftIO $ waitForCount countVar 40 15_000_000+                  stopApp appHandle++        collected <- readIORef ref+        -- Disjoint + complete through the helper: the union of every member's+        -- delivered global positions is exactly [1..40] — no event dropped, none+        -- delivered twice. No duplicates ⇒ the member partitions are disjoint;+        -- ==[1..40] ⇒ the union is the complete source. This is the same+        -- disjoint-complete property MasterPlan 4 used, now asserted through the+        -- single-call helper and the Shibuya pipeline.+        sort (map globalPos collected) `shouldBe` [1 .. 40]+        -- Every one of the 20 originating streams was covered.+        length (nub (map origStreamId collected)) `shouldBe` 20++-- Helpers++envelopePayload :: Message es RecordedEvent -> RecordedEvent+envelopePayload ing = let Message {envelope = env} = ing in env ^. #payload++stubAdapter :: Adapter es RecordedEvent+stubAdapter =+  Adapter+    { adapterName = "stub-kiroku",+      source = Stream.nil,+      shutdown = pure ()+    }++-- | Read the dead letters recorded for a non-group subscription (member 0).+readDeadLetters :: KirokuStore -> Text -> IO [SQL.DeadLetterRecord]+readDeadLetters store subName = readMemberDeadLetters store subName 0++readMemberDeadLetters :: KirokuStore -> Text -> Int32 -> IO [SQL.DeadLetterRecord]+readMemberDeadLetters store subName memberIndex = do+  result <- Pool.use (store ^. #pool) (Session.statement (subName, memberIndex) SQL.readDeadLettersStmt)+  case result of+    Left err -> error ("readDeadLetters failed: " <> show err)+    Right v -> pure (toList v)++-- | The raw global position of a recorded event (for disjoint/complete checks).+globalPos :: RecordedEvent -> Int64+globalPos e = case e ^. #globalPosition of GlobalPosition p -> p++-- | The originating stream's surrogate id (for stream-coverage checks).+origStreamId :: RecordedEvent -> Int64+origStreamId e = case e ^. #originalStreamId of StreamId p -> p++-- | Wait until the sum of all TVar counts reaches the target or the timeout fires.+waitForTotal :: [STM.TVar Int] -> Int -> Int -> IO ()+waitForTotal vars target timeoutMicros = do+  timeoutVar <- registerDelay timeoutMicros+  result <-+    atomically $+      ( do+          total <- sum <$> mapM readTVar vars+          STM.check (total >= target)+          pure True+      )+        `STM.orElse` ( do+                         t <- readTVar timeoutVar+                         STM.check t+                         pure False+                     )+  unless result $ do+    actual <- atomically $ sum <$> mapM readTVar vars+    expectationFailure ("Timed out waiting for total " <> show target <> ", got " <> show actual)++waitForCount :: STM.TVar Int -> Int -> Int -> IO ()+waitForCount countVar target timeoutMicros = do+  timeoutVar <- registerDelay timeoutMicros+  result <-+    atomically $+      ( do+          c <- readTVar countVar+          STM.check (c >= target)+          pure True+      )+        `STM.orElse` ( do+                         t <- readTVar timeoutVar+                         STM.check t+                         pure False+                     )+  unless result $ do+    actual <- atomically $ readTVar countVar+    expectationFailure ("Timed out waiting for count " <> show target <> ", got " <> show actual)++within :: String -> IO a -> IO a+within label action = do+  result <- timeout 10_000_000 action+  case result of+    Nothing -> fail ("Timed out waiting for " <> label)+    Just a -> pure a++waitForCheckpointPosition :: KirokuStore -> SubscriptionCheckpointKey -> GlobalPosition -> IO ()+waitForCheckpointPosition store expectedKey expectedPosition =+  within "subscription checkpoint initialization" loop+  where+    loop = do+      Right (SubscriptionCheckpointInventory _ checkpoints) <-+        runStoreIO store subscriptionCheckpointInventory+      let positions =+            [ (SubscriptionCheckpointKey name member, position)+            | SubscriptionCheckpoint name member position _ <- toList checkpoints+            ]+      if (expectedKey, expectedPosition) `elem` positions+        then pure ()+        else threadDelay 20_000 >> loop++readCheckpointPosition :: KirokuStore -> SubscriptionCheckpointKey -> IO (Maybe GlobalPosition)+readCheckpointPosition store expectedKey = do+  Right (SubscriptionCheckpointInventory _ checkpoints) <-+    runStoreIO store subscriptionCheckpointInventory+  pure $ case [ position+              | SubscriptionCheckpoint name member position _ <- toList checkpoints,+                SubscriptionCheckpointKey name member == expectedKey+              ] of+    position : _ -> Just position+    [] -> Nothing++makeEvent :: Text -> Value -> EventData+makeEvent typ p =+  EventData+    { eventId = Nothing,+      eventType = EventType typ,+      payload = p,+      metadata = Nothing,+      causationId = Nothing,+      correlationId = Nothing+    }++-- | A throwaway attribute source for the trace-header tests, whose assertions+-- do not depend on the kiroku identity attributes.+sampleEnvelopeAttrs :: KirokuEnvelopeAttrs+sampleEnvelopeAttrs = kirokuEnvelopeAttrs "test-sub" Nothing++makeRecordedEvent :: Maybe Value -> RecordedEvent+makeRecordedEvent meta =+  RecordedEvent+    { eventId = EventId UUID.nil,+      eventType = EventType "TraceEvent",+      streamVersion = StreamVersion 1,+      globalPosition = GlobalPosition 1,+      originalStreamId = StreamId 1,+      originalVersion = StreamVersion 1,+      payload = Aeson.object [],+      metadata = meta,+      causationId = Nothing,+      correlationId = Nothing,+      createdAt = UTCTime (fromGregorian 2026 5 16) 0+    }++withTestStore :: (KirokuStore -> IO ()) -> IO ()+withTestStore action =+  withMigratedTestDatabase $ \connStr ->+    withStore (defaultConnectionSettings connStr) action++validGroupSize :: Int32 -> ConsumerGroupSize+validGroupSize = either (error . show) Prelude.id . mkConsumerGroupSize++validConsumerGroup :: Int32 -> Int32 -> ConsumerGroup+validConsumerGroup m n = either (error . show) Prelude.id (mkConsumerGroup m (validGroupSize n))++acknowledgementSpec :: Spec+acknowledgementSpec = do+  describe "acknowledgement liveness" $ do+    forM_ [False, True] $ \guarded ->+      it (if guarded then "guarded single processor retries with pacing and continues" else "standard Shibuya handler exceptions are finalized") $+        withTestStore $ \store -> do+          Right _ <-+            runStoreIO store $+              appendToStream+                (StreamName "ack-liveness")+                NoStream+                [makeEvent "First" (Aeson.object []), makeEvent "Second" (Aeson.object [])]+          attempts <- newIORef ([] :: [(Int64, Maybe Attempt, Word64)])+          count <- newTVarIO (0 :: Int)+          within "runner finalization" $ runEff $ runTracingNoop $ do+            adapter <- kirokuAdapter store (defaultKirokuAdapterConfig (SubscriptionName "ack-liveness") AllStreams)+            let handler ingested = do+                  now <- liftIO getMonotonicTimeNSec+                  let pos = globalPos (envelopePayload ingested)+                      Message {envelope = Envelope {attempt = attempt'}} = ingested+                  n <- liftIO $ do+                    modifyIORef' attempts ((pos, attempt', now) :)+                    atomically $ do+                      old <- readTVar count+                      writeTVar count (old + 1)+                      pure old+                  if n == 0 then liftIO (E.throwIO (userError "one-shot handler failure")) else pure AckOk+                processor = if guarded then kirokuProcessor adapter handler else mkProcessor adapter handler+            app <- runApp defaultAppConfig [(ProcessorId "ack-liveness", processor)] >>= either (liftIO . fail . show) pure+            liftIO $ waitForCount count 3 5_000_000+            liftIO $ waitForCheckpointPosition store (SubscriptionCheckpointKey (SubscriptionName "ack-liveness") 0) (GlobalPosition 2)+            stopApp app+          recorded <- reverse <$> readIORef attempts+          map (\(pos, attempt', _) -> (pos, attempt')) recorded+            `shouldBe` [(1, Just (Attempt 0)), (1, Just (Attempt 1)), (2, Just (Attempt 0))]+          case recorded of+            (_, _, firstAt) : (_, _, retriedAt) : _ ->+              if guarded+                then retriedAt - firstAt `shouldSatisfy` (>= 900_000_000)+                else retriedAt - firstAt `shouldSatisfy` (< 900_000_000)+            _ -> expectationFailure "expected three finalized deliveries"++    forM_ [Nothing, Just 0.05] $ \warnAfter ->+      it ("keeps a raw unfinalized item pending and forwards warning policy " <> show warnAfter) $ do+        warnings <- newTVarIO (0 :: Int)+        warning <- STM.newEmptyTMVarIO+        let observe event = case event of+              KirokuEventSubscriptionHandlerStalled _ pos _ elapsed _ -> atomically $ do+                STM.modifyTVar' warnings (+ 1)+                _ <- STM.tryPutTMVar warning (pos, elapsed)+                pure ()+              _ -> pure ()+        withMigratedTestDatabase $ \connStr ->+          withStore (defaultConnectionSettings connStr & #eventHandler .~ Just observe) $ \store -> do+            Right _ <-+              runStoreIO store $+                appendToStream+                  (StreamName "ack-raw")+                  NoStream+                  [makeEvent "First" (Aeson.object []), makeEvent "Second" (Aeson.object [])]+            within "raw acknowledgement ownership" $ runEff $ runTracingNoop $ Timeout.runTimeout $ do+              adapter <-+                kirokuAdapter+                  store+                  ( (defaultKirokuAdapterConfig (SubscriptionName "ack-raw") AllStreams)+                      & #handlerStallWarnAfter .~ warnAfter+                  )+              let Adapter {source = sourceStream, shutdown = shutdownAction} = adapter+              first <- Stream.uncons sourceStream+              case first of+                Nothing -> liftIO (expectationFailure "expected first item")+                Just (item, rest) -> do+                  -- The raw consumer intentionally leaves item unfinalized.+                  secondRead <- Timeout.timeout 150_000 (Stream.uncons rest)+                  liftIO (isNothing secondRead `shouldBe` True)+                  liftIO $ readCheckpointPosition store (SubscriptionCheckpointKey (SubscriptionName "ack-raw") 0) `shouldReturn` Just (GlobalPosition 0)+                  case warnAfter of+                    Nothing -> liftIO $ STM.readTVarIO warnings `shouldReturn` 0+                    Just threshold -> liftIO $ do+                      (pos, elapsed) <- within "raw ack warning" (atomically (STM.readTMVar warning))+                      pos `shouldBe` GlobalPosition 1+                      elapsed `shouldSatisfy` (>= threshold)+                  -- Finalization remains explicitly owned by this consumer.+                  let Ingested {ack = AckHandle {finalize = finalizeFirst}} = item+                  finalizeFirst AckOk+                  (second, _) <- Stream.uncons rest >>= maybe (liftIO (fail "expected second item")) pure+                  let Ingested {envelope = Envelope {payload = secondEvent}} = second+                  liftIO $ globalPos secondEvent `shouldBe` 2+                  let Ingested {ack = AckHandle {finalize = finalizeSecond}} = second+                  finalizeSecond AckOk+                  liftIO $ waitForCheckpointPosition store (SubscriptionCheckpointKey (SubscriptionName "ack-raw") 0) (GlobalPosition 2)+              shutdownAction+            stoppedCount <- STM.readTVarIO warnings+            threadDelay 100_000+            STM.readTVarIO warnings `shouldReturn` stoppedCount++  describe "retry policy" $ do+    forM_ [Nothing, Just 2] $ \attemptLimit ->+      it ("honors single-adapter total delivery limit " <> show attemptLimit) $ withTestStore $ \store -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "ack-policy") NoStream [makeEvent "Retry" (Aeson.object [])]+        count <- newTVarIO (0 :: Int)+        runEff $ runTracingNoop $ do+          let config =+                (defaultKirokuAdapterConfig (SubscriptionName "ack-policy") AllStreams)+                  & #retryPolicy .~ maybe defaultRetryPolicy RetryPolicy attemptLimit+          adapter <- kirokuAdapter store config+          let handler _ = liftIO (atomically (STM.modifyTVar' count (+ 1))) >> pure (AckRetry (Ack.RetryDelay 0))+          app <- runApp defaultAppConfig [(ProcessorId "ack-policy", kirokuProcessor adapter handler)] >>= either (liftIO . fail . show) pure+          liftIO $ waitForCheckpointPosition store (SubscriptionCheckpointKey (SubscriptionName "ack-policy") 0) (GlobalPosition 1)+          stopApp app+        STM.readTVarIO count `shouldReturn` maybe 5 Prelude.id attemptLimit+        letters <- readDeadLetters store "ack-policy"+        map SQL.deadLetterAttemptCount letters `shouldBe` [fromIntegral (maybe 5 Prelude.id attemptLimit)]++    it "forwards two attempts and stall policy to both consumer-group members" $ do+      warned <- newTVarIO Set.empty+      release <- STM.newEmptyTMVarIO+      let observe event = case event of+            KirokuEventSubscriptionHandlerStalled _ _ _ _ (GroupMember member' 2) -> atomically (STM.modifyTVar' warned (Set.insert member'))+            _ -> pure ()+      withMigratedTestDatabase $ \connStr ->+        withStore (defaultConnectionSettings connStr & #eventHandler .~ Just observe) $ \store -> do+          -- Enough distinct streams to exercise both PostgreSQL hash slots;+          -- require both warnings before release so an empty slot fails.+          forM_ [1 .. 20 :: Int] $ \i -> do+            Right _ <- runStoreIO store $ appendToStream (StreamName ("ack-group-" <> T.pack (show i))) NoStream [makeEvent "Retry" (Aeson.object [])]+            pure ()+          seen <- newIORef Map.empty+          runEff $ runTracingNoop $ do+            let config =+                  (defaultConsumerGroupConfig (SubscriptionName "ack-policy-group") AllStreams (validGroupSize 2))+                    & #retryPolicy .~ RetryPolicy 2+                    & #handlerStallWarnAfter .~ Just 0.05+                handler ingested = do+                  liftIO $ atomically (STM.readTMVar release)+                  liftIO $ atomicModifyIORef' seen (\old -> (Map.insertWith (+) (globalPos (envelopePayload ingested)) (1 :: Int) old, ()))+                  pure (AckRetry (Ack.RetryDelay 0))+            processors <- kirokuConsumerGroupProcessors store config handler >>= either (liftIO . fail . show) pure+            app <- runApp defaultAppConfig processors >>= either (liftIO . fail . show) pure+            liftIO $ within "both group members stalled" $ atomically $ readTVar warned >>= STM.check . (== Set.fromList [0, 1])+            liftIO $ atomically (STM.putTMVar release ())+            liftIO $ within "all group dead letters" $ do+              let loop = do+                    letters <- concat <$> mapM (readMemberDeadLetters store "ack-policy-group") [0, 1]+                    if length letters == 20 then pure () else threadDelay 10_000 >> loop+              loop+            stopApp app+          Map.elems <$> readIORef seen `shouldReturn` replicate 20 2+          letters <- concat <$> mapM (readMemberDeadLetters store "ack-policy-group") [0, 1]+          sort (map SQL.deadLetterGlobalPosition letters) `shouldBe` [1 .. 20]+          map SQL.deadLetterAttemptCount letters `shouldBe` replicate 20 2