diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,22 @@ # Changelog -## Unreleased +## 0.5.0.0 — 2026-08-11 + +### Breaking Changes + +* `KirokuAdapterConfig` and `KirokuConsumerGroupConfig` gain + `missingCheckpointPolicy`. Their smart constructors default it to + `FromBeginning`; exhaustive record literals must choose a policy or use the + smart constructors. +* Requires `kiroku-store ^>=0.5`. + +### New Features + +* The adapter re-exports `MissingCheckpointPolicy` and forwards it to every + underlying Kiroku worker. `FromCurrentHead` supports future-only Shibuya + processors without historical side effects, while `FailIfMissing` preserves + mandatory checkpoint provisioning. A consumer-group policy is applied + independently to every member key; existing rows always win. ## 0.4.0.1 — 2026-08-09 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.4.0.1 +version: 0.5.0.0 synopsis: Kiroku event store adapter for the Shibuya queue processing framework @@ -50,7 +50,7 @@ , effectful-core >=2.5 && <2.7 , hs-opentelemetry-api ^>=1.0 , hs-opentelemetry-semantic-conventions ^>=1.40 - , kiroku-store ^>=0.4 + , kiroku-store ^>=0.5 , shibuya-core >=0.8 && <0.9 , stm >=2.5 && <2.6 , streamly-core >=0.3 && <0.4 @@ -78,7 +78,7 @@ , hasql-pool >=1.2 && <1.5 , hs-opentelemetry-api ^>=1.0 , hspec >=2.10 && <2.12 - , kiroku-store ^>=0.4 + , kiroku-store ^>=0.5 , kiroku-test-support , lens >=5.2 && <5.4 , shibuya-core >=0.8 && <0.9 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 @@ -132,6 +132,7 @@ SubscriptionTarget (..), ConsumerGroup (..), EventTypeFilter (..), + MissingCheckpointPolicy (..), ) where import Control.Exception (SomeException) @@ -146,6 +147,7 @@ ConsumerGroup (..), EventTypeFilter (..), InvalidConsumerGroup (..), + MissingCheckpointPolicy (..), SubscriptionConfig, SubscriptionName (..), SubscriptionResult (..), @@ -204,6 +206,13 @@ the underlying 'Kiroku.Store.Subscription.subscribe' call, which throws 'Kiroku.Store.Subscription.Types.InvalidConsumerGroup' on violation. -} + , 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 @@ -230,7 +239,8 @@ {- | A 'KirokuAdapterConfig' with sensible defaults: @batchSize = 100@, @bufferSize = 256@, @queueCapacity = 16@, @consumerGroup = 'Nothing'@ -(ordinary single-consumer subscription), @eventTypeFilter = 'AllEventTypes'@ +(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. @@ -255,6 +265,7 @@ , bufferSize = 256 , queueCapacity = 16 , consumerGroup = Nothing + , missingCheckpointPolicy = FromBeginning , eventTypeFilter = AllEventTypes , selector = Nothing } @@ -303,7 +314,7 @@ KirokuStore -> KirokuAdapterConfig -> Eff es (Adapter es RecordedEvent) -kirokuAdapter store KirokuAdapterConfig{subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, consumerGroup = cg, eventTypeFilter = etf, selector = sel} = do +kirokuAdapter store KirokuAdapterConfig{subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, 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 @@ -314,6 +325,7 @@ { Sub.batchSize = bs , Sub.queueCapacity = qCap , Sub.consumerGroup = cg + , Sub.missingCheckpointPolicy = checkpointPolicy , Sub.eventTypeFilter = etf , Sub.selector = sel } @@ -380,6 +392,11 @@ -} , 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 @@ -404,7 +421,8 @@ {- | A 'KirokuConsumerGroupConfig' with sensible defaults: @memberConcurrency = 'Serial'@ (the only legal per-member concurrency), @batchSize = 100@, -@bufferSize = 256@, @queueCapacity = 16@, @eventTypeFilter = 'AllEventTypes'@ +@bufferSize = 256@, @queueCapacity = 16@, @missingCheckpointPolicy = +'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@ (deliver every type), @selector = 'Nothing'@ (no extra predicate filtering). Supply the subscription name, target, and group size. -} @@ -419,6 +437,7 @@ , bufferSize = 256 , queueCapacity = 16 , memberConcurrency = Serial + , missingCheckpointPolicy = FromBeginning , eventTypeFilter = AllEventTypes , selector = Nothing } @@ -466,7 +485,7 @@ 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, eventTypeFilter = etf, selector = sel} handler = +kirokuConsumerGroupProcessors store cfg@KirokuConsumerGroupConfig{subscriptionName = subName, subscriptionTarget = subTarget, groupSize = n, batchSize = bs, bufferSize = buf, queueCapacity = qCap, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} handler = kirokuConsumerGroupProcessorsWith mkMemberAdapter cfg handler where mkMemberAdapter m = @@ -479,6 +498,7 @@ , bufferSize = buf , queueCapacity = qCap , consumerGroup = Just (ConsumerGroup{member = m, size = n}) + , missingCheckpointPolicy = checkpointPolicy , eventTypeFilter = etf , selector = sel } diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -304,6 +304,51 @@ 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 "delivers live events through Shibuya pipeline" $ \store -> do ref <- newIORef ([] :: [RecordedEvent]) countVar <- newTVarIO (0 :: Int) @@ -1037,6 +1082,21 @@ 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 makeEvent :: Text -> Value -> EventData makeEvent typ p =