Finalize.hs

-- | __Internal module.__ Exposed for the test suite and benchmarks only.
-- No PVP guarantees: anything here may change or disappear in any release.
-- Application authors should import "Shibuya" instead.
--
-- Shared resilient finalization for framework-owned ack paths.
--
-- Both single-message and batch processing call adapter-provided
-- 'AckHandle.finalize' through this module. A transient finalizer exception is
-- retried with the fixed schedule below; each retry uses the same already
-- resolved 'AckDecision'.
module Shibuya.Internal.Runner.Finalize
  ( finalizeRetryDelaysMicros,
    finalizeWithRetry,
  )
where

import Effectful (Eff, IOE, liftIO, (:>))
import OpenTelemetry.Trace.Core qualified as OTel
import Shibuya.Core.Ack (AckDecision)
import Shibuya.Core.AckHandle (AckHandle (..))
import Shibuya.Core.Ingested (Ingested (..))
import Shibuya.Telemetry.Effect (Tracing, recordException)
import UnliftIO (SomeException, catchAny)
import UnliftIO.Concurrent (threadDelay)

-- | Retry delays, in microseconds, used after failed finalization attempts.
finalizeRetryDelaysMicros :: [Int]
finalizeRetryDelaysMicros = [10_000, 50_000, 250_000]

-- | 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 ->
  Ingested es msg ->
  AckDecision ->
  Eff es (Either SomeException ())
finalizeWithRetry traceSpan ingested decision = go finalizeRetryDelaysMicros
  where
    go delays =
      catchAny
        (Right <$> ingested.ack.finalize decision)
        ( \ex -> do
            recordException traceSpan ex
            case delays of
              [] -> pure (Left ex)
              delay : rest -> do
                liftIO $ threadDelay delay
                go rest
        )