diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/app/LifecycleLive.hs b/app/LifecycleLive.hs new file mode 100644 --- /dev/null +++ b/app/LifecycleLive.hs @@ -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 + ] diff --git a/internal/Shibuya/Adapter/Kiroku/Internal.hs b/internal/Shibuya/Adapter/Kiroku/Internal.hs new file mode 100644 --- /dev/null +++ b/internal/Shibuya/Adapter/Kiroku/Internal.hs @@ -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 diff --git a/shibuya-kiroku-adapter.cabal b/shibuya-kiroku-adapter.cabal --- a/shibuya-kiroku-adapter.cabal +++ b/shibuya-kiroku-adapter.cabal @@ -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 diff --git a/src/Shibuya/Adapter/Kiroku.hs b/src/Shibuya/Adapter/Kiroku.hs --- a/src/Shibuya/Adapter/Kiroku.hs +++ b/src/Shibuya/Adapter/Kiroku.hs @@ -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 + ] + ) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 =