diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,24 @@ # Changelog +## 0.5.1.0 — 2026-08-15 + +### Other Changes + +* Requires `shibuya-core >=0.9 && <0.10`, which adds the + `ApplicationFailure DeadLetterCode Text` dead-letter reason. +* `toKirokuDeadLetterReason` now translates `ApplicationFailure` into Kiroku's + `DeadLetterOther`. The summary is Shibuya's canonical `code: detail` + rendering, and the structured `reason` JSONB is + `{"kind":"other","summary":…,"detail":{"code":…,"detail":…}}` so the + application-owned code is queryable as `reason->'detail'->>'code'` without + parsing the summary. +* The `PoisonPill`, `InvalidPayload`, and `MaxRetriesExceeded` translations are + unchanged, so existing `dead_letters` rows keep their encoding. No + `shibuya-kiroku-adapter` API changed. +* Build with `-Werror=incomplete-patterns`. This package translates between + dependency-owned sum types, where an upstream constructor addition otherwise + degrades a translation into a runtime failure behind a green `-Wall` build. + ## 0.5.0.2 — 2026-08-13 ### Other Changes 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.0.2 +version: 0.5.1.0 synopsis: Kiroku event store adapter for the Shibuya queue processing framework @@ -36,7 +36,7 @@ OverloadedLabels OverloadedStrings - ghc-options: -Wall + ghc-options: -Wall -Werror=incomplete-patterns library import: common @@ -51,7 +51,7 @@ , hs-opentelemetry-api ^>=1.0 , hs-opentelemetry-semantic-conventions ^>=1.40 , kiroku-store ^>=0.7 - , shibuya-core >=0.8 && <0.9 + , shibuya-core >=0.9 && <0.10 , stm >=2.5 && <2.6 , streamly-core >=0.3 && <0.4 , text >=2.0 && <2.2 @@ -81,7 +81,7 @@ , kiroku-store ^>=0.7 , kiroku-test-support , lens >=5.2 && <5.4 - , shibuya-core >=0.8 && <0.9 + , shibuya-core >=0.9 && <0.10 , shibuya-kiroku-adapter , stm >=2.5 && <2.6 , streamly-core >=0.3 && <0.4 diff --git a/src/Shibuya/Adapter/Kiroku/Convert.hs b/src/Shibuya/Adapter/Kiroku/Convert.hs --- a/src/Shibuya/Adapter/Kiroku/Convert.hs +++ b/src/Shibuya/Adapter/Kiroku/Convert.hs @@ -34,7 +34,7 @@ import Control.Concurrent.STM (atomically, tryPutTMVar) import Control.Monad (void) -import Data.Aeson (Value (..)) +import Data.Aeson (Value (..), object, (.=)) import Data.Aeson.Key qualified as Key import Data.Aeson.KeyMap qualified as KM import Data.HashMap.Strict (HashMap) @@ -152,12 +152,27 @@ AckDeadLetter reason -> DeadLetter (toKirokuDeadLetterReason attempt reason) AckHalt _ -> Continue --- | Translate a Shibuya 'Ack.DeadLetterReason' into a Kiroku 'DeadLetterReason'. +{- | 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'. diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -27,6 +27,7 @@ import Hasql.Session qualified as Session import Kiroku.Store import Kiroku.Store.SQL qualified as SQL +import Kiroku.Store.Subscription.Types qualified as KTypes import Kiroku.Store.Subscription.Worker (withFetchBatchHookForTest) import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres) import OpenTelemetry.Attributes (toAttribute) @@ -40,7 +41,12 @@ kirokuConsumerGroupProcessors, kirokuConsumerGroupProcessorsWith, ) -import Shibuya.Adapter.Kiroku.Convert (KirokuEnvelopeAttrs, kirokuEnvelopeAttrs, toEnvelope) +import Shibuya.Adapter.Kiroku.Convert ( + KirokuEnvelopeAttrs, + kirokuEnvelopeAttrs, + toEnvelope, + toKirokuDeadLetterReason, + ) import Shibuya.App ( ProcessorId (..), QueueProcessor (..), @@ -137,6 +143,35 @@ 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) @@ -982,8 +1017,16 @@ -- The group is presented as N processors, each pinned to the -- group-level PartitionedInOrder contract + per-member Serial. liftIO $ length processors `shouldBe` 4 - let policies = map (\(_, QueueProcessor _ _ ord conc) -> (ord, conc)) processors - liftIO $ policies `shouldBe` replicate 4 (PartitionedInOrder, Serial) + -- '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 $