diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## 0.8.0.1 — 2026-07-04 + +### Other Changes + +- Reduced per-message allocation on the `Async` and `Ahead` concurrency paths + by widening the streamly dispatch buffer to `2 * n` (the thread bound stays + at `n`). A buffer equal to the thread count throttled worker dispatch and + caused dispatch churn that allocated up to +90% on the `Async` + concurrency-levels benchmarks (EP-30). No API or behavior change. +- Documented the accepted ~126 bytes/message cost of the separate handler and + finalizer exception frames that guarantee "always finalize" (EP-31). + ## 0.8.0.0 — 2026-07-04 ### Breaking Changes diff --git a/shibuya-core.cabal b/shibuya-core.cabal --- a/shibuya-core.cabal +++ b/shibuya-core.cabal @@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-core -version: 0.8.0.0 +version: 0.8.0.1 synopsis: Supervised queue processing framework for Haskell description: A supervised queue processing framework inspired by Broadway (Elixir). diff --git a/src/Shibuya/Internal/Runner/Finalize.hs b/src/Shibuya/Internal/Runner/Finalize.hs --- a/src/Shibuya/Internal/Runner/Finalize.hs +++ b/src/Shibuya/Internal/Runner/Finalize.hs @@ -29,6 +29,12 @@ -- | Call a message finalizer until it succeeds or the bounded retry budget is -- exhausted. The resolved decision is never recomputed between attempts. +-- +-- The 'catchAny' here is a per-message exception frame distinct from the +-- handler catch in 'processOne'; together they cost ~126 bytes/message more than +-- 0.7.1.0's single combined catch (measured, EP-31 S3). This cost is accepted as +-- the price of bounded finalizer retry + the always-finalize guarantee. See +-- docs/plans/31-investigate-and-reduce-the-shared-per-message-metrics-tracing-finalize-allocation-regression.md. finalizeWithRetry :: (IOE :> es, Tracing :> es) => OTel.Span -> diff --git a/src/Shibuya/Internal/Runner/Supervised.hs b/src/Shibuya/Internal/Runner/Supervised.hs --- a/src/Shibuya/Internal/Runner/Supervised.hs +++ b/src/Shibuya/Internal/Runner/Supervised.hs @@ -518,12 +518,21 @@ partitioned n (PartitionedInOrder, Async n) -> partitioned n + -- 'maxThreads n' is the hard concurrency bound (at most n handlers run at + -- once); do not remove it. The output buffer is set to '2 * n', not 'n': + -- a buffer equal to the thread count throttles streamly's worker dispatch + -- (a worker is only forked when both the thread AND buffer checks pass), + -- causing dispatch churn that allocated up to +90% on the Async + -- concurrency-levels benchmarks. Enlarging the buffer to 2n relieves the + -- churn while keeping the n-thread bound, and matches the pending-item + -- limit the partitioned path already uses (see 'partitioned' above). See + -- docs/plans/30-investigate-and-reduce-the-async-ahead-concurrency-allocation-regression.md. (_, Ahead n) -> Stream.fold Fold.drain $ - StreamP.parMapM (StreamP.maxThreads n . StreamP.maxBuffer n . StreamP.ordered True) processAction inboxStream + StreamP.parMapM (StreamP.maxThreads n . StreamP.maxBuffer (2 * n) . StreamP.ordered True) processAction inboxStream (_, Async n) -> Stream.fold Fold.drain $ - StreamP.parMapM (StreamP.maxThreads n . StreamP.maxBuffer n) processAction inboxStream + StreamP.parMapM (StreamP.maxThreads n . StreamP.maxBuffer (2 * n)) processAction inboxStream -- After draining, check if we halted maybeHalt <- readIORef haltRef @@ -581,6 +590,15 @@ -- Call handler and finalizer separately. A handler exception is -- substituted with immediate retry so the adapter always observes a -- finalization decision for an ingested message. + -- + -- This is a *separate* per-message 'catchAny' from the one inside + -- 'finalizeWithRetry' below. 0.7.1.0 used a single combined catch around + -- handler+finalize, which skipped finalization when the handler threw; + -- splitting them is what makes "always finalize" hold. The extra + -- per-message exception frame costs ~126 bytes/message (measured, EP-31 + -- S3: docs/plans/31-…-shared-per-message-…-allocation-regression.md) and is + -- accepted as the price of the always-finalize guarantee — do not merge the + -- two catches back together to reclaim it. handlerResult <- catchAny (Right <$> handler (toMessage ingested)) diff --git a/test/Shibuya/RunnerSpec.hs b/test/Shibuya/RunnerSpec.hs --- a/test/Shibuya/RunnerSpec.hs +++ b/test/Shibuya/RunnerSpec.hs @@ -8,7 +8,7 @@ import Effectful (Eff, IOE, liftIO, runEff, (:>)) import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Mock (TrackingAck (..), newTrackingAck, trackingAckHandle) -import Shibuya.App (AppConfig (..), AppError (..), QueueProcessor (..), defaultAppConfig, mkProcessor, runApp, waitApp) +import Shibuya.App (AppConfig (..), AppError (..), QueueProcessor (..), defaultAppConfig, mkProcessor, runApp, stopApp, waitApp) import Shibuya.Core.Ack (AckDecision (..)) import Shibuya.Core.AckHandle (AckHandle (..)) import Shibuya.Core.Error (ConfigError (..), PolicyError (..)) @@ -70,6 +70,12 @@ Left err -> pure $ Left err Right appHandle -> do waitApp appHandle + -- Stop the app to cancel the (always-linked) master coordinator. + -- Without this the idle master blocks forever on its mailbox and the + -- RTS eventually raises BlockedIndefinitelyOnSTM, which propagates + -- through the link as a flaky ExceptionInLinkedThread landing on + -- whichever test happens to be running when GC fires. + stopApp appHandle pure $ Right () -- Verify result @@ -101,6 +107,9 @@ pure (decs, Left err) Right appHandle -> do waitApp appHandle + -- See note in "processes messages from mock adapter": stop the app so + -- the linked master does not deadlock and flake a later test. + stopApp appHandle decs <- liftIO $ readIORef tracking.trackedDecisions pure (decs, Right ()) @@ -132,6 +141,12 @@ Left err -> pure $ Left err Right appHandle -> do waitApp appHandle + -- Stop the app to cancel the (always-linked) master coordinator. + -- Without this the idle master blocks forever on its mailbox and the + -- RTS eventually raises BlockedIndefinitelyOnSTM, which propagates + -- through the link as a flaky ExceptionInLinkedThread landing on + -- whichever test happens to be running when GC fires. + stopApp appHandle pure $ Right () result `shouldBe` Right () @@ -185,6 +200,12 @@ Left err -> pure $ Left err Right appHandle -> do waitApp appHandle + -- Stop the app to cancel the (always-linked) master coordinator. + -- Without this the idle master blocks forever on its mailbox and the + -- RTS eventually raises BlockedIndefinitelyOnSTM, which propagates + -- through the link as a flaky ExceptionInLinkedThread landing on + -- whichever test happens to be running when GC fires. + stopApp appHandle pure $ Right () result `shouldBe` Right ()