shibuya-kiroku-adapter 0.5.1.2 → 0.5.1.3
raw patch · 6 files changed
+837/−52 lines, 6 filesdep +asyncdep +bytestringdep +vectordep ~effectfuldep ~effectful-coredep ~kiroku-test-supportnew-component:exe:lifecycle-livePVP ok
version bump matches the API change (PVP)
Dependencies added: async, bytestring, vector
Dependency ranges changed: effectful, effectful-core, kiroku-test-support, shibuya-core
API changes (from Hackage documentation)
Files
- CHANGELOG.md +30/−0
- app/LifecycleLive.hs +403/−0
- internal/Shibuya/Adapter/Kiroku/Internal.hs +58/−0
- shibuya-kiroku-adapter.cabal +69/−22
- src/Shibuya/Adapter/Kiroku.hs +20/−23
- test/Main.hs +257/−7
CHANGELOG.md view
@@ -1,5 +1,35 @@ # Changelog +## 0.5.1.3 — 2026-09-21++### Bug Fixes++* Consumer-group construction now uses a masked ownership ledger. Cancellation+ cannot strand a member between factory return and registration, every+ acquired member receives a shutdown attempt after a later failure, and a+ throwing shutdown no longer replaces the primary construction exception.+* The underlying ack-stream bridge now closes its own subscription-to-monitor+ ownership window and waits for both threads during idempotent shutdown.++### Other Changes++* Require `shibuya-core >=0.10 && <0.11` across the library, tests, lifecycle+ fixture, and benchmark for the coordinated lifecycle release candidate.+* Support `effectful-core` 2.6.1 and 2.7.1.1 or later, while excluding 2.7.0.0+ through 2.7.1.0 because upstream records a per-operation performance+ regression for dynamically dispatched effects.+* Real-store regressions cover duplicate acknowledgement, `AckHalt` replay,+ cancellation after reply but before checkpoint persistence, existing and+ missing checkpoint policies, source failure, retry, and leak-free shutdown.+* Keep the live lifecycle fixture enabled in the repository while making it an+ opt-in Cabal component for source-distribution consumers, so the published+ library does not require the unpublished `kiroku-test-support` package.+* The live restart fixture now streams identities externally, then emits and+ enforces a reconciled per-delivery ledger. This detects duplicate/loss+ compensation without retaining the identity set in the measured process heap.+* The handler-exception documentation now reflects Shibuya's supervised+ immediate-retry finalization behavior.+ ## 0.5.1.2 — 2026-09-18 ### Other Changes
+ app/LifecycleLive.hs view
@@ -0,0 +1,403 @@+-- | EP-45 live Kiroku performance, retained-memory, and restart fixture.+module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.STM (TVar, atomically, modifyTVar', newTVarIO, readTVarIO, writeTVar)+import Control.Monad (forever, unless, when)+import Data.Aeson (encode, object, withObject, (.:), (.=))+import Data.Aeson.Types (parseMaybe)+import Data.ByteString.Char8 qualified as BS+import Data.ByteString.Lazy qualified as LBS+import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)+import Data.Int (Int64)+import Data.Set (Set)+import Data.Set qualified as Set+import Data.Text qualified as Text+import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime)+import Data.Time.Format (defaultTimeLocale, formatTime)+import Data.Vector qualified as Vector+import Data.Word (Word64)+import Effectful (liftIO, runEff)+import EphemeralPg qualified as Pg+import GHC.Stats (GCDetails (..), RTSStats (..), getRTSStats, getRTSStatsEnabled)+import Kiroku.Store+import Kiroku.Test.Postgres (ephemeralConfig, migrateTestDatabase)+import Shibuya.Adapter.Kiroku (defaultKirokuAdapterConfig, kirokuAdapter)+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 (..))+import Shibuya.Telemetry.Effect (runTracingNoop)+import System.Environment (lookupEnv)+import System.Exit (exitFailure)+import System.IO (Handle, IOMode (..), SeekMode (AbsoluteSeek), hFlush, hPutStrLn, hSeek, hSetFileSize, withFile)+import System.Mem (performMajorGC)+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+ }++data Sample = Sample+ { 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)+ }++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 ()+ where+ validResourceChar 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+ }++envString :: String -> String -> IO String+envString key fallback = maybe fallback (\value -> value) <$> lookupEnv key++envInt :: String -> Int -> IO Int+envInt key fallback = maybe fallback (\value -> value) . (>>= readMaybe) <$> lookupEnv key++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++ 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"++ 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++ 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++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)++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"+ 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++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}++recordProduced :: DeliveryLedger -> Int -> IO ()+recordProduced ledger value = hPutStrLn ledger.producedHandle (show value)++recordProcessed :: DeliveryLedger -> Int -> IO ()+recordProcessed ledger value = hPutStrLn ledger.processedHandle (show value)++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++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))+ where+ parseIdentity 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)++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++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+ }++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)+ where+ isTargetCheckpoint (SubscriptionCheckpoint name member _ _) =+ 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)++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++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++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+ ]
+ internal/Shibuya/Adapter/Kiroku/Internal.hs view
@@ -0,0 +1,58 @@+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.+-}+acquireAllAndTransfer ::+ (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+ 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)++ cleanupAndRethrow primary owned = do+ mapM_ releaseIgnoringFailure owned+ Exception.throwIO primary++ releaseIgnoringFailure resource = do+ _ <- tryAny (release resource)+ pure ()++tryAny :: Eff es a -> Eff es (Either SomeException a)+tryAny = Exception.try
shibuya-kiroku-adapter.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: shibuya-kiroku-adapter-version: 0.5.1.2+version: 0.5.1.3 synopsis: Kiroku event store adapter for the Shibuya queue processing framework @@ -28,6 +28,11 @@ type: git location: https://github.com/shinzui/kiroku.git +flag lifecycle-live+ description: Build the repository-only lifecycle release fixture+ default: False+ manual: True+ common common default-language: GHC2024 default-extensions:@@ -38,6 +43,15 @@ ghc-options: -Wall -Werror=incomplete-patterns +library shibuya-kiroku-adapter-internal+ import: common+ visibility: private+ exposed-modules: Shibuya.Adapter.Kiroku.Internal+ hs-source-dirs: internal+ build-depends:+ , base >=4.18 && <5+ , effectful-core >=2.6.1 && <2.7 || >=2.7.1.1 && <2.8+ library import: common exposed-modules:@@ -47,11 +61,12 @@ build-depends: , aeson >=2.1 && <2.3 , base >=4.18 && <5- , effectful-core >=2.5 && <2.7+ , 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.8- , shibuya-core >=0.9 && <0.10+ , shibuya-core >=0.10 && <0.11+ , shibuya-kiroku-adapter-internal , stm >=2.5 && <2.6 , streamly-core >=0.3 && <0.4 , text >=2.0 && <2.2@@ -67,25 +82,57 @@ hs-source-dirs: test ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends:- , aeson >=2.1 && <2.3- , base >=4.18 && <5- , containers >=0.6 && <0.8+ , aeson >=2.1 && <2.3+ , base >=4.18 && <5+ , containers >=0.6 && <0.8 , directory- , effectful >=2.4 && <2.7- , ephemeral-pg >=0.3.1 && <0.4- , generic-lens >=2.2 && <2.4- , hasql >=1.10 && <1.11- , hasql-pool >=1.2 && <1.5- , hs-opentelemetry-api ^>=1.0- , hspec >=2.10 && <2.12+ , effectful >=2.6.1 && <2.8+ , effectful-core >=2.6.1 && <2.7 || >=2.7.1.1 && <2.8+ , ephemeral-pg >=0.3.1 && <0.4+ , generic-lens >=2.2 && <2.4+ , hasql >=1.10 && <1.11+ , hasql-pool >=1.2 && <1.5+ , hs-opentelemetry-api ^>=1.0+ , hspec >=2.10 && <2.12+ , kiroku-store ^>=0.8+ , kiroku-test-support ^>=0.1.0.0+ , lens >=5.2 && <5.4+ , shibuya-core >=0.10 && <0.11+ , shibuya-kiroku-adapter+ , shibuya-kiroku-adapter-internal+ , stm >=2.5 && <2.6+ , streamly-core >=0.3 && <0.4+ , text >=2.0 && <2.2+ , time >=1.12 && <1.15+ , unordered-containers >=0.2 && <0.3+ , uuid >=1.3 && <1.4++-- EP-45 live-store performance, restart, backlog, and retained-memory fixture.+-- The executable owns an ephemeral PostgreSQL instance and emits the common+-- adapter-soak CSV consumed by shibuya's release analyzer.+executable lifecycle-live+ import: common++ if !flag(lifecycle-live)+ buildable: False++ main-is: LifecycleLive.hs+ hs-source-dirs: app+ ghc-options: -threaded -rtsopts "-with-rtsopts=-N4 -T -A32m" -O2+ default-extensions: OverloadedRecordDot+ build-depends:+ , aeson >=2.1 && <2.3+ , async >=2.2 && <2.3+ , base >=4.18 && <5+ , bytestring >=0.11 && <0.13+ , 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.8- , kiroku-test-support- , lens >=5.2 && <5.4- , shibuya-core >=0.9 && <0.10+ , kiroku-test-support ^>=0.1.0.0+ , shibuya-core >=0.10 && <0.11 , shibuya-kiroku-adapter- , stm >=2.5 && <2.6- , streamly-core >=0.3 && <0.4- , text >=2.0 && <2.2- , time >=1.12 && <1.15- , unordered-containers >=0.2 && <0.3- , uuid >=1.3 && <1.4+ , stm >=2.5 && <2.6+ , text >=2.0 && <2.2+ , time >=1.12 && <1.15+ , vector >=0.13 && <0.14
src/Shibuya/Adapter/Kiroku.hs view
@@ -103,12 +103,12 @@ overflow policy, so a paused subscriber catches up from its checkpoint instead of being killed. -A Shibuya handler used with this adapter must not let synchronous exceptions-escape. Shibuya's supervised runner records handler exceptions without-finalizing the ack; with Kiroku's ack-coupled bridge, an unfinalized ack blocks-the Kiroku worker forever. Direct 'mkProcessor' users should wrap handlers in-'guardKirokuHandler' or handle exceptions themselves.-'kirokuConsumerGroupProcessors' applies 'guardKirokuHandler' automatically.+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. '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. -} module Shibuya.Adapter.Kiroku ( -- * Adapter@@ -139,7 +139,7 @@ import Data.Int (Int32) import Data.Text qualified as T import Effectful (Eff, IOE, liftIO, (:>))-import Effectful.Exception (catchSync, onException, throwIO)+import Effectful.Exception (catchSync, throwIO) import GHC.Generics (Generic) import Kiroku.Store.Connection (KirokuStore) import Kiroku.Store.Subscription.Stream (subscriptionAckStream)@@ -159,6 +159,7 @@ import Numeric.Natural (Natural) import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Kiroku.Convert (kirokuEnvelopeAttrs, toIngestedAck)+import Shibuya.Adapter.Kiroku.Internal (acquireAllAndTransfer) import Shibuya.App (ProcessorId (..), QueueProcessor (..)) import Shibuya.Core.Ack (AckDecision (..), RetryDelay (..)) import Shibuya.Core.Error (PolicyError (..))@@ -530,19 +531,15 @@ Left e -> pure (Left e) Right (ordering, conc) -> do let SubscriptionName name = subName- adapters <- createAdapters [] 0- let processors =- [ let pid = ProcessorId (name <> "-member-" <> T.pack (show m))- in (pid, QueueProcessor adapter (guardKirokuHandler handler) ordering conc)- | (m, adapter) <- zip [0 .. n - 1] adapters- ]- pure (Right processors)- where- createAdapters created m- | m >= n = pure (reverse created)- | otherwise = do- adapter <- mkMemberAdapter m `onException` shutdownCreated created- createAdapters (adapter : created) (m + 1)-- shutdownCreated =- mapM_ $ \Adapter{shutdown = shutdownAction} -> shutdownAction+ acquireAllAndTransfer+ 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 .. n - 1] adapters+ ]+ )
test/Main.hs view
@@ -1,8 +1,8 @@ module Main where -import Control.Concurrent (threadDelay)+import Control.Concurrent (forkIO, killThread, threadDelay) import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)-import Control.Concurrent.STM (atomically, newTVarIO, readTVar, registerDelay, writeTVar)+import Control.Concurrent.STM (atomically, newEmptyTMVarIO, newTVarIO, readTMVar, readTVar, registerDelay, writeTVar) import Control.Concurrent.STM qualified as STM import Control.Exception qualified as E import Control.Lens ((&), (.~), (^.))@@ -27,8 +27,9 @@ import Hasql.Session qualified as Session import Kiroku.Store import Kiroku.Store.SQL qualified as SQL+import Kiroku.Store.Subscription.Stream (AckItem (..)) import Kiroku.Store.Subscription.Types qualified as KTypes-import Kiroku.Store.Subscription.Worker (withFetchBatchHookForTest)+import Kiroku.Store.Subscription.Worker (withFetchBatchHookForTest, withSaveCheckpointHookForTest) import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres) import OpenTelemetry.Attributes (toAttribute) import Shibuya.Adapter (Adapter (..))@@ -45,8 +46,10 @@ KirokuEnvelopeAttrs, kirokuEnvelopeAttrs, toEnvelope,+ toIngestedAck, toKirokuDeadLetterReason, )+import Shibuya.Adapter.Kiroku.Internal (acquireAllAndTransfer) import Shibuya.App ( ProcessorId (..), QueueProcessor (..),@@ -56,12 +59,14 @@ runApp, stopApp, stopAppGracefully,+ waitApp, ) import Shibuya.App qualified as Shibuya-import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..))+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 (Message (..))+import Shibuya.Core.Ingested (Ingested (..), Message (..)) import Shibuya.Core.Metrics (ProcessorState (..)) import Shibuya.Core.Types (Attempt (..), Envelope (..)) import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))@@ -384,6 +389,83 @@ 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)@@ -654,6 +736,22 @@ 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@@ -690,6 +788,108 @@ 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 [])]@@ -918,7 +1118,7 @@ throwsSize 0 throwsSize (-1) - it "shuts down already-created member adapters after a later factory failure" $ \_store -> do+ 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@@ -928,7 +1128,11 @@ Adapter { adapterName = "stub-kiroku" , source = Stream.nil- , shutdown = liftIO $ modifyIORef' shutdowns (<> [m])+ , 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@@ -946,6 +1150,41 @@ 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@@ -1140,6 +1379,17 @@ 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 =