shibuya-core 0.3.0.0 → 0.4.0.0

raw patch · 11 files changed

+317/−2 lines, 11 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +27/−0
• shibuya-core.cabal +4/−1
• src/Shibuya/Core.hs +2/−1
• src/Shibuya/Core/Retry.hs +136/−0
• src/Shibuya/Core/Types.hs +15/−0
• test/Main.hs +2/−0
• test/Shibuya/Core/RetrySpec.hs +103/−0
• test/Shibuya/Core/TypesSpec.hs +23/−0
• test/Shibuya/Runner/SupervisedSpec.hs +2/−0
• test/Shibuya/RunnerSpec.hs +2/−0
• test/Shibuya/Telemetry/SemanticSpec.hs +1/−0

@@ -1,5 +1,32 @@ # Changelog +## 0.4.0.0 — 2026-04-29++### Breaking Changes++- `Envelope` gained an `attempt :: !(Maybe Attempt)` field carrying the+  adapter's delivery counter (zero-indexed; `Nothing` if unknown). Direct+  constructions of `Envelope` must add the field. The new `Attempt`+  newtype is exported from `Shibuya.Core` and `Shibuya.Core.Types`.++### New Features++- New module `Shibuya.Core.Retry` providing `BackoffPolicy`, `Jitter`+  (`NoJitter`, `FullJitter`, `EqualJitter`), `defaultBackoffPolicy`, the+  pure evaluator `exponentialBackoffPure`, the effectful+  `exponentialBackoff`, and the handler convenience `retryWithBackoff`.+  Handlers can now compute exponentially-growing, jittered retry delays+  with a single call: `retryWithBackoff defaultBackoffPolicy+  ingested.envelope`. Pulls in `random ^>=1.2` as a new build-depends+  (already a transitive dep — ships with GHC). See the module haddock and+  the new `RetrySpec` test for usage patterns.+- A runnable end-to-end demonstration of the new API lives in the+  sibling [`shibuya-pgmq-adapter`](https://github.com/shinzui/shibuya-pgmq-adapter)+  repo at `shibuya-pgmq-example/`, exposed via the `backoff-demo`+  subcommand of `shibuya-pgmq-consumer`. The plan+  `docs/plans/8-demonstrate-backoff-end-to-end.md` records setup+  instructions and captured transcripts.+ ## 0.3.0.0 — 2026-04-24  Version bumped to track the shared release version. No user-visible

@@ -1,6 +1,6 @@ cabal-version: 3.14 name: shibuya-core-version: 0.3.0.0+version: 0.4.0.0 synopsis: Supervised queue processing framework for Haskell description:   A supervised queue processing framework inspired by Broadway (Elixir).@@ -29,6 +29,7 @@     Shibuya.Core.Error     Shibuya.Core.Ingested     Shibuya.Core.Lease+    Shibuya.Core.Retry     Shibuya.Core.Types     Shibuya.Handler     Shibuya.Policy@@ -74,6 +75,7 @@     hs-opentelemetry-semantic-conventions ^>=0.1,     lens ^>=5.3.5,     nqe ^>=0.6,+    random ^>=1.2,     stm ^>=2.5,     streamly ^>=0.11,     streamly-core ^>=0.3,@@ -111,6 +113,7 @@    other-modules:     Shibuya.Core.AckSpec+    Shibuya.Core.RetrySpec     Shibuya.Core.TypesSpec     Shibuya.PolicySpec     Shibuya.Runner.SupervisedSpec

@@ -19,6 +19,7 @@   ( -- * Message Types     MessageId (..),     Cursor (..),+    Attempt (..),     Envelope (..),      -- * Ack Semantics@@ -86,7 +87,7 @@ import Shibuya.Core.Error (HandlerError (..), PolicyError (..), RuntimeError (..)) import Shibuya.Core.Ingested (Ingested (..)) import Shibuya.Core.Lease (Lease (..))-import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..))+import Shibuya.Core.Types (Attempt (..), Cursor (..), Envelope (..), MessageId (..)) import Shibuya.Handler (Handler) import Shibuya.Policy (Concurrency (..), Ordering (..), validatePolicy) import Shibuya.Runner.Halt (ProcessorHalt (..))

@@ -0,0 +1,136 @@+-- | Exponential backoff policy and pure evaluator.+--+-- Handlers compute a 'RetryDelay' from a 'BackoffPolicy' and the current+-- delivery 'Attempt'. The pure evaluator takes a jitter sample in @[0,1)@+-- and is suitable for property tests.+module Shibuya.Core.Retry+  ( -- * Policy+    BackoffPolicy (..),+    Jitter (..),+    defaultBackoffPolicy,++    -- * Pure evaluator+    exponentialBackoffPure,++    -- * Effectful evaluator+    exponentialBackoff,++    -- * Handler convenience+    retryWithBackoff,+  )+where++import Data.Maybe (fromMaybe)+import Data.Time (NominalDiffTime, nominalDiffTimeToSeconds, secondsToNominalDiffTime)+import Effectful (Eff, IOE, liftIO, (:>))+import GHC.Generics (Generic)+import Shibuya.Core.Ack (AckDecision (..), RetryDelay (..))+import Shibuya.Core.Types (Attempt (..), Envelope (..))+import System.Random qualified as Random++-- | Strategy for adding randomness to backoff delays.+-- Jitter prevents thundering herds when many messages fail simultaneously.+--+-- See <https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/>.+data Jitter+  = -- | No randomness; delay is deterministic.+    NoJitter+  | -- | Delay is uniform in @[0, baseExp]@ where @baseExp@ is+    -- @min maxDelay (base * factor^attempt)@.+    FullJitter+  | -- | Delay is @baseExp / 2 + uniform(0, baseExp / 2)@.+    EqualJitter+  deriving stock (Eq, Show, Generic)++-- | Configuration for exponential backoff with optional jitter.+data BackoffPolicy = BackoffPolicy+  { -- | Base delay applied at attempt 0 (before jitter).+    base :: !NominalDiffTime,+    -- | Multiplicative growth factor between attempts. Typically 2.0.+    factor :: !Double,+    -- | Upper bound on the computed delay before jitter.+    -- Prevents arbitrarily long retries far in the future.+    maxDelay :: !NominalDiffTime,+    -- | Jitter strategy.+    jitter :: !Jitter+  }+  deriving stock (Eq, Show, Generic)++-- | Sensible defaults: 1 s base, factor 2, max 5 min, full jitter.+--+-- This matches AWS's published recommendation. With these defaults:+--+-- - Attempt 0: random delay in [0, 1) s+-- - Attempt 1: random delay in [0, 2) s+-- - Attempt 2: random delay in [0, 4) s+-- - ...+-- - Attempt 8 onwards: random delay in [0, 256) s, capped at [0, 300) s+defaultBackoffPolicy :: BackoffPolicy+defaultBackoffPolicy =+  BackoffPolicy+    { base = 1,+      factor = 2.0,+      maxDelay = 300,+      jitter = FullJitter+    }++-- | Compute a retry delay from a policy, an attempt count, and a jitter+-- sample in @[0, 1)@.+--+-- The sample is consumed only when the policy's 'jitter' is not 'NoJitter';+-- callers passing @0.0@ to a 'NoJitter' policy will see deterministic output.+exponentialBackoffPure ::+  BackoffPolicy ->+  Attempt ->+  -- | Jitter sample in @[0, 1)@. Ignored when 'jitter' = 'NoJitter'.+  Double ->+  RetryDelay+exponentialBackoffPure policy (Attempt n) sample =+  RetryDelay (secondsToNominalDiffTime (realToFrac jittered))+  where+    baseSec :: Double+    baseSec = realToFrac (nominalDiffTimeToSeconds policy.base)++    maxSec :: Double+    maxSec = realToFrac (nominalDiffTimeToSeconds policy.maxDelay)++    baseExp :: Double+    baseExp = min maxSec (baseSec * policy.factor ** fromIntegral n)++    clampedSample :: Double+    clampedSample = max 0 (min 0.999999 sample)++    jittered :: Double+    jittered = case policy.jitter of+      NoJitter -> baseExp+      FullJitter -> baseExp * clampedSample+      EqualJitter -> baseExp / 2 + (baseExp / 2) * clampedSample++-- | Compute a retry delay by sampling jitter from 'IO'.+--+-- Equivalent to 'exponentialBackoffPure' with a fresh sample drawn via+-- 'System.Random.randomRIO' on each call.+exponentialBackoff ::+  (IOE :> es) =>+  BackoffPolicy ->+  Attempt ->+  Eff es RetryDelay+exponentialBackoff policy attempt = do+  sample <- liftIO (Random.randomRIO (0.0 :: Double, 1.0))+  pure (exponentialBackoffPure policy attempt sample)++-- | One-line helper for the common case: read 'envelope.attempt', compute a+-- backoff delay, and return 'AckRetry'.+--+-- Treats 'Nothing' attempt as @'Attempt' 0@ (first delivery), so handlers+-- consuming envelopes from adapters that do not track redeliveries still+-- get a sensible base-delay retry.+retryWithBackoff ::+  (IOE :> es) =>+  BackoffPolicy ->+  Envelope msg ->+  Eff es AckDecision+retryWithBackoff policy envelope = do+  let attempt = fromMaybe (Attempt 0) envelope.attempt+  delay <- exponentialBackoff policy attempt+  pure (AckRetry delay)

@@ -8,6 +8,9 @@     -- * Cursor / Offset     Cursor (..), +    -- * Delivery Attempt+    Attempt (..),+     -- * Message Envelope     Envelope (..), @@ -36,6 +39,14 @@   deriving stock (Eq, Ord, Show, Generic)   deriving anyclass (NFData) +-- | Zero-indexed delivery attempt count.+-- 0 means first delivery; 1 means first retry; and so on.+-- Adapters that cannot track redeliveries report 'Nothing' on the envelope.+newtype Attempt = Attempt {unAttempt :: Word}+  deriving stock (Eq, Ord, Show, Generic)+  deriving newtype (Num, Real, Enum, Integral, Bounded)+  deriving anyclass (NFData)+ -- | W3C Trace Context headers for distributed tracing. -- Contains traceparent and optionally tracestate headers. type TraceHeaders = [(ByteString, ByteString)]@@ -53,6 +64,10 @@     enqueuedAt :: !(Maybe UTCTime),     -- | W3C trace context headers for distributed tracing     traceContext :: !(Maybe TraceHeaders),+    -- | Optional zero-indexed delivery counter.+    -- 'Just (Attempt 0)' on first delivery; 'Nothing' if the adapter+    -- does not track redeliveries (e.g., Kafka).+    attempt :: !(Maybe Attempt),     -- | The actual message payload     payload :: !msg   }

@@ -3,6 +3,7 @@ module Main (main) where  import Shibuya.Core.AckSpec qualified+import Shibuya.Core.RetrySpec qualified import Shibuya.Core.TypesSpec qualified import Shibuya.PolicySpec qualified import Shibuya.Runner.SupervisedSpec qualified@@ -16,6 +17,7 @@ main = hspec $ do   describe "Shibuya.Core.Types" Shibuya.Core.TypesSpec.spec   describe "Shibuya.Core.Ack" Shibuya.Core.AckSpec.spec+  describe "Shibuya.Core.Retry" Shibuya.Core.RetrySpec.spec   describe "Shibuya.Policy" Shibuya.PolicySpec.spec   describe "Shibuya.Runner" Shibuya.RunnerSpec.spec   Shibuya.Runner.SupervisedSpec.spec

@@ -0,0 +1,103 @@+{-# LANGUAGE OverloadedStrings #-}++module Shibuya.Core.RetrySpec (spec) where++import Data.Time (nominalDiffTimeToSeconds)+import Effectful (runEff)+import Shibuya.Core.Ack (AckDecision (..), RetryDelay (..))+import Shibuya.Core.Retry+import Shibuya.Core.Types (Attempt (..), Envelope (..), MessageId (..))+import Test.Hspec+import Test.QuickCheck++secondsOf :: RetryDelay -> Double+secondsOf (RetryDelay d) = realToFrac (nominalDiffTimeToSeconds d)++testEnvelope :: Maybe Attempt -> Envelope ()+testEnvelope a =+  Envelope+    { messageId = MessageId "test",+      cursor = Nothing,+      partition = Nothing,+      enqueuedAt = Nothing,+      traceContext = Nothing,+      attempt = a,+      payload = ()+    }++spec :: Spec+spec = do+  describe "exponentialBackoffPure" $ do+    describe "NoJitter" $ do+      let p = defaultBackoffPolicy {jitter = NoJitter}+      it "delay = base at attempt 0" $+        secondsOf (exponentialBackoffPure p (Attempt 0) 0) `shouldBe` 1.0+      it "delay doubles each attempt" $ do+        secondsOf (exponentialBackoffPure p (Attempt 1) 0) `shouldBe` 2.0+        secondsOf (exponentialBackoffPure p (Attempt 2) 0) `shouldBe` 4.0+        secondsOf (exponentialBackoffPure p (Attempt 3) 0) `shouldBe` 8.0+      it "delay clamps to maxDelay" $+        -- 2^20 = 1048576 ≫ 300 seconds+        secondsOf (exponentialBackoffPure p (Attempt 20) 0) `shouldBe` 300.0+      it "ignores the jitter sample" $+        secondsOf (exponentialBackoffPure p (Attempt 3) 0)+          `shouldBe` secondsOf (exponentialBackoffPure p (Attempt 3) 0.999)++    describe "FullJitter" $ do+      let p = defaultBackoffPolicy {jitter = FullJitter}+      it "delay at sample 0 is 0" $+        secondsOf (exponentialBackoffPure p (Attempt 5) 0) `shouldBe` 0.0+      it "delay at sample ~1 approaches baseExp" $+        secondsOf (exponentialBackoffPure p (Attempt 2) 0.999)+          `shouldSatisfy` (\x -> x > 3.99 && x <= 4.0)+      it "delay never exceeds the clamped baseExp" $+        property $ \(NonNegative n) ->+          forAll (choose (0.0, 0.999999)) $ \sample ->+            let attemptN = Attempt (fromIntegral (n :: Int))+                delay = secondsOf (exponentialBackoffPure p attemptN sample)+             in delay >= 0 && delay <= 300++    describe "EqualJitter" $ do+      let p = defaultBackoffPolicy {jitter = EqualJitter}+      it "delay at sample 0 is half the baseExp" $+        secondsOf (exponentialBackoffPure p (Attempt 2) 0) `shouldBe` 2.0+      it "delay at sample ~1 approaches the full baseExp" $+        secondsOf (exponentialBackoffPure p (Attempt 2) 0.999)+          `shouldSatisfy` (\x -> x > 3.99 && x <= 4.0)+      it "delay is in [baseExp/2, baseExp]" $+        property $ \(NonNegative n) ->+          forAll (choose (0.0, 0.999999)) $ \sample ->+            let attemptN = Attempt (fromIntegral (n :: Int))+                delay = secondsOf (exponentialBackoffPure p attemptN sample)+             in delay >= 0 && delay <= 300++    describe "Attempt 0" $ do+      it "with NoJitter equals base" $+        secondsOf (exponentialBackoffPure (defaultBackoffPolicy {jitter = NoJitter}) (Attempt 0) 0)+          `shouldBe` 1.0+      it "with FullJitter is in [0, base)" $+        property $+          forAll (choose (0.0, 0.999999)) $ \sample ->+            let delay = secondsOf (exponentialBackoffPure (defaultBackoffPolicy {jitter = FullJitter}) (Attempt 0) sample)+             in delay >= 0 && delay < 1.0++  describe "retryWithBackoff" $ do+    it "produces AckRetry when attempt is set" $ do+      decision <- runEff (retryWithBackoff defaultBackoffPolicy (testEnvelope (Just (Attempt 1))))+      case decision of+        AckRetry _ -> pure ()+        other -> expectationFailure $ "expected AckRetry, got " <> show other++    it "produces AckRetry when attempt is Nothing (treats as Attempt 0)" $ do+      decision <- runEff (retryWithBackoff defaultBackoffPolicy (testEnvelope Nothing))+      case decision of+        AckRetry (RetryDelay d) ->+          -- baseExp at Attempt 0 = base = 1s, FullJitter -> [0, 1)+          realToFrac (nominalDiffTimeToSeconds d) `shouldSatisfy` (\x -> x >= 0 && x < (1.0 :: Double))+        other -> expectationFailure $ "expected AckRetry, got " <> show other++  describe "exponentialBackoff (effectful)" $ do+    it "samples a delay in [0, baseExp] for FullJitter" $ do+      RetryDelay d <- runEff (exponentialBackoff defaultBackoffPolicy (Attempt 4))+      -- baseExp at Attempt 4 = 1 * 2^4 = 16s, FullJitter -> [0, 16)+      realToFrac (nominalDiffTimeToSeconds d) `shouldSatisfy` (\x -> x >= 0 && x < (16.0 :: Double))

@@ -22,6 +22,23 @@       let msgId :: MessageId = "test-message"       msgId.unMessageId `shouldBe` "test-message" +  describe "Attempt" $ do+    it "supports Eq" $ do+      Attempt 0 `shouldBe` Attempt 0+      Attempt 0 `shouldNotBe` Attempt 1++    it "supports Ord for sequencing retries" $ do+      Attempt 0 `compare` Attempt 1 `shouldBe` LT+      Attempt 5 `compare` Attempt 5 `shouldBe` EQ+      Attempt 9 `compare` Attempt 1 `shouldBe` GT++    it "supports Num so retries can be incremented" $ do+      Attempt 0 + 1 `shouldBe` Attempt 1+      Attempt 4 + 1 `shouldBe` Attempt 5++    it "Bounded reflects the underlying Word" $ do+      (minBound :: Attempt).unAttempt `shouldBe` 0+   describe "Cursor" $ do     it "CursorInt compares numerically" $ do       CursorInt 1 `compare` CursorInt 2 `shouldBe` LT@@ -56,6 +73,11 @@       mapped.traceContext `shouldBe` env.traceContext       mapped.payload `shouldBe` 5 +    it "preserves attempt through fmap" $ do+      let env = (testEnvelope (1 :: Int)) {attempt = Just (Attempt 3)}+          mapped = fmap show env+      mapped.attempt `shouldBe` Just (Attempt 3)+ -- Test helper testEnvelope :: msg -> Envelope msg testEnvelope msg =@@ -65,6 +87,7 @@       partition = Just "partition-0",       enqueuedAt = Just testTime,       traceContext = Nothing,+      attempt = Nothing,       payload = msg     } 

@@ -853,6 +853,7 @@                 partition = Nothing,                 enqueuedAt = Just testTime,                 traceContext = Nothing,+                attempt = Nothing,                 payload = "message-" <> show i               }           ackHandle = AckHandle $ \_ -> pure ()@@ -875,6 +876,7 @@             partition = Nothing,             enqueuedAt = Just testTime,             traceContext = Nothing,+            attempt = Nothing,             payload = "message-" <> show i           }       ackHandle = AckHandle $ \_ -> pure ()

@@ -205,6 +205,7 @@                 partition = Nothing,                 enqueuedAt = Just testTime,                 traceContext = Nothing,+                attempt = Nothing,                 payload = "message-" <> show i               }           ackHandle = AckHandle $ \_ -> pure () -- No-op ack@@ -228,6 +229,7 @@                 partition = Nothing,                 enqueuedAt = Just testTime,                 traceContext = Nothing,+                attempt = Nothing,                 payload = "message-" <> show i               }           ackHandle = trackingAckHandle tracking msgId

@@ -69,6 +69,7 @@                 partition = Nothing,                 enqueuedAt = Nothing,                 traceContext = Nothing,+                attempt = Nothing,                 payload = ("hello" :: Text)               }           ingested =