diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,29 @@ ## Unreleased +## 0.14.0.0 — 2026-08-21 + +### Breaking Changes + +- `PublishOutcome` gains `PublishRejected`, `OutboxStatus` gains + `OutboxRejected`, `OutboxRow` gains `rejectedAt` and `rejection`, and + `OutboxPublishSummary` gains `rejected`. Exhaustive matches and direct record + construction must handle the new terminal outcome. +- `KeiroMetrics` gains the `outboxRejected` counter field. Direct record construction + must initialize it; `newKeiroMetrics` users are unaffected. +- Requires the lockstep `keiro-core ^>=0.14.0.0`. + +### New Features + +- `mkPublishRejection` validates a stable 1–64 character code and optional non-empty, + 1024-byte detail. `PublishRejected` commits bounded terminal audit truth, schedules + no retry, releases per-key and per-source successors, and does not stop + `StopTheLine`. +- `publishClaimedOutbox` finalizes every outcome class for a claimed batch in one + transaction and derives its summary and `keiro.outbox.rejected` counter only from + conditional updates that committed. A pre-commit failure remains recoverable and + may redeliver the callback under the documented at-least-once contract. + ## 0.13.0.0 — 2026-08-17 ### Breaking Changes diff --git a/keiro.cabal b/keiro.cabal --- a/keiro.cabal +++ b/keiro.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: keiro -version: 0.13.0.0 +version: 0.14.0.0 synopsis: Event sourcing framework and workflow engine description: A library that composes kiroku, keiki, and shibuya into an @@ -98,6 +98,7 @@ other-modules: Keiro.Command.Domain + Keiro.Outbox.Rejection Keiro.Projection.Types Keiro.ReadModel.Rebuild.Group Keiro.ReadModel.Rebuild.Runner @@ -146,7 +147,7 @@ , hs-opentelemetry-semantic-conventions >=1.40 && <2 , keiki >=0.9 && <0.10 , keiki-codec-json >=0.9 && <0.10 - , keiro-core ^>=0.13.0.0 + , keiro-core ^>=0.14.0.0 , kiroku-store >=0.8 && <0.9 , lens >=5.2 && <5.4 , mmzk-typeid >=0.7 && <0.8 @@ -235,7 +236,7 @@ , hs-opentelemetry-sdk >=1.0 && <1.1 , keiki >=0.9 && <0.10 , keiro - , keiro-core ^>=0.13.0.0 + , keiro-core ^>=0.14.0.0 , keiro-test-support , kiroku-store >=0.8 && <0.9 , shibuya-core ^>=0.9.0.0 diff --git a/src/Keiro/Outbox.hs b/src/Keiro/Outbox.hs --- a/src/Keiro/Outbox.hs +++ b/src/Keiro/Outbox.hs @@ -89,6 +89,7 @@ recordOutboxDeadlettered, recordOutboxPublished, recordOutboxReclaimed, + recordOutboxRejected, recordOutboxRetried, withProducerSpan, ) @@ -268,6 +269,8 @@ PublishSucceeded | -- | Publish failed; will be retried after the configured backoff. PublishFailed !Text + | -- | The transport intentionally and permanently refused publication. + PublishRejected !PublishRejection deriving stock (Generic, Eq, Show) -- | Drain claimed outbox rows by handing the claimed batch to @publish@ and @@ -309,13 +312,14 @@ -- Counters from the aggregated pass summary (each a no-op under 'Nothing'; -- a zero delta is harmless). recordOutboxPublished mMetrics (fromIntegral (summary ^. #published)) + recordOutboxRejected mMetrics (fromIntegral (summary ^. #rejected)) recordOutboxRetried mMetrics (fromIntegral (summary ^. #retried)) recordOutboxDeadlettered mMetrics (fromIntegral (summary ^. #dead)) pure summary where publishBatch :: [OutboxRow] -> Eff es OutboxPublishSummary publishBatch [] = - pure OutboxPublishSummary {claimed = 0, published = 0, retried = 0, dead = 0, haltedOn = Nothing} + pure OutboxPublishSummary {claimed = 0, published = 0, rejected = 0, retried = 0, dead = 0, haltedOn = Nothing} publishBatch batch = case options ^. #orderingPolicy of StopTheLine -> publishStopTheLine batch batch [] Nothing @@ -337,6 +341,7 @@ outcomes' = outcomes <> [(row ^. #outboxId, outcome)] case outcome of PublishSucceeded -> publishStopTheLine original rest outcomes' Nothing + PublishRejected _ -> publishStopTheLine original rest outcomes' Nothing PublishFailed _ -> markProcessedOutcomes StopTheLine original (Map.fromList outcomes') (Just (row ^. #outboxId)) publishRows :: [OutboxRow] -> Eff es (Map.Map OutboxId PublishOutcome) @@ -394,6 +399,7 @@ firstFailure :: [(OutboxId, PublishOutcome)] -> Maybe Text firstFailure [] = Nothing firstFailure ((_, PublishSucceeded) : rest) = firstFailure rest + firstFailure ((_, PublishRejected _) : rest) = firstFailure rest firstFailure ((_, PublishFailed errMsg) : _) = Just errMsg markProcessedOutcomes :: @@ -406,28 +412,38 @@ now <- liftIO getCurrentTime let marks = foldMap (groupMarks outcomes) (outcomeGroups policy batch) sentIds = marks ^. #sentIds + rejectedRows = marks ^. #rejectedRows failedRows = marks ^. #failedRows skippedRows = marks ^. #skippedRows - failedStatuses <- - if null failedRows && null skippedRows - then pure [] - else runTransaction $ do - statuses <- traverse (markFailed now) failedRows - traverse_ (markSkipped now) skippedRows - pure statuses - _ <- markOutboxSentBatch sentIds now - let deadCount = length [() | OutboxDead <- failedStatuses] - retriedFailures = length failedStatuses - deadCount + committed <- runTransaction $ do + publishedCount <- markOutboxSentBatchTx sentIds now + rejectionMarks <- traverse (markRejected now) rejectedRows + failureMarks <- traverse (markFailed now) failedRows + skippedMarks <- traverse (markSkipped now) skippedRows + let failedStatuses = [status | Just status <- failureMarks] + deadCount = length [() | OutboxDead <- failedStatuses] + pure + CommittedMarks + { published = publishedCount, + rejected = length [() | True <- rejectionMarks], + retried = length [() | OutboxFailed <- failedStatuses] + length [() | True <- skippedMarks], + dead = deadCount + } pure OutboxPublishSummary { claimed = length batch, - published = length sentIds, - retried = retriedFailures + length skippedRows, - dead = deadCount, + published = committed ^. #published, + rejected = committed ^. #rejected, + retried = committed ^. #retried, + dead = committed ^. #dead, haltedOn = halted } - markFailed :: UTCTime -> (OutboxRow, Text) -> Tx.Transaction OutboxStatus + markRejected :: UTCTime -> (OutboxId, PublishRejection) -> Tx.Transaction Bool + markRejected now (outboxId, rejection) = + markOutboxRejectedTx outboxId rejection now + + markFailed :: UTCTime -> (OutboxRow, Text) -> Tx.Transaction (Maybe OutboxStatus) markFailed now (row, errMsg) = markOutboxFailedTx (row ^. #outboxId) @@ -436,7 +452,7 @@ (nextDelay (options ^. #backoff) (row ^. #attemptCount)) now - markSkipped :: UTCTime -> OutboxRow -> Tx.Transaction () + markSkipped :: UTCTime -> OutboxRow -> Tx.Transaction Bool markSkipped now row = markOutboxSkippedTx (row ^. #outboxId) @@ -475,6 +491,7 @@ data OutcomeMarks = OutcomeMarks { sentIds :: ![OutboxId], + rejectedRows :: ![(OutboxId, PublishRejection)], failedRows :: ![(OutboxRow, Text)], skippedRows :: ![OutboxRow] } @@ -484,26 +501,38 @@ left <> right = OutcomeMarks { sentIds = (left ^. #sentIds) <> (right ^. #sentIds), + rejectedRows = (left ^. #rejectedRows) <> (right ^. #rejectedRows), failedRows = (left ^. #failedRows) <> (right ^. #failedRows), skippedRows = (left ^. #skippedRows) <> (right ^. #skippedRows) } instance Monoid OutcomeMarks where - mempty = OutcomeMarks {sentIds = [], failedRows = [], skippedRows = []} + mempty = OutcomeMarks {sentIds = [], rejectedRows = [], failedRows = [], skippedRows = []} groupMarks :: Map.Map OutboxId PublishOutcome -> [OutboxRow] -> OutcomeMarks -groupMarks outcomes = go [] +groupMarks outcomes = go [] [] where - go sent [] = mempty {sentIds = sent} - go sent (row : rest) = + go sent rejected [] = mempty {sentIds = sent, rejectedRows = rejected} + go sent rejected (row : rest) = case fromMaybe (PublishFailed "publisher returned no outcome") (Map.lookup (row ^. #outboxId) outcomes) of - PublishSucceeded -> go (sent <> [row ^. #outboxId]) rest + PublishSucceeded -> go (sent <> [row ^. #outboxId]) rejected rest + PublishRejected rejection -> + go sent (rejected <> [(row ^. #outboxId, rejection)]) rest PublishFailed errMsg -> OutcomeMarks { sentIds = sent, + rejectedRows = rejected, failedRows = [(row, errMsg)], skippedRows = rest } + +data CommittedMarks = CommittedMarks + { published :: !Int, + rejected :: !Int, + retried :: !Int, + dead :: !Int + } + deriving stock (Generic) data OutcomeGroupKey = BatchGroup diff --git a/src/Keiro/Outbox/Rejection.hs b/src/Keiro/Outbox/Rejection.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Outbox/Rejection.hs @@ -0,0 +1,62 @@ +-- | Validated terminal refusal data for outbox publication. +-- +-- This module is intentionally not exposed by the package. Public callers use +-- the abstract 'PublishRejection' type and smart constructor re-exported from +-- "Keiro.Outbox.Types"; database decoders inside the package can use the data +-- constructor after schema constraints have validated the stored values. +module Keiro.Outbox.Rejection + ( PublishRejection (..), + PublishRejectionError (..), + mkPublishRejection, + ) +where + +import Data.ByteString qualified as ByteString +import Data.Text qualified as Text +import Data.Text.Encoding qualified as TE +import Keiro.Prelude + +-- | A stable machine-readable refusal code and optional operator detail. +-- +-- Codes are lowercase ASCII identifiers of 1 to 64 characters. Detail is +-- non-empty when present and is bounded to 1024 UTF-8 bytes. +data PublishRejection = PublishRejection + { publishRejectionCode :: !Text, + publishRejectionDetail :: !(Maybe Text) + } + deriving stock (Generic, Eq, Show) + +-- | Why 'mkPublishRejection' refused caller-provided data. +data PublishRejectionError + = InvalidPublishRejectionCode !Text + | PublishRejectionDetailEmpty + | PublishRejectionDetailTooLong !Int + deriving stock (Generic, Eq, Show) + +-- | Validate a terminal publication refusal without normalizing caller data. +mkPublishRejection :: Text -> Maybe Text -> Either PublishRejectionError PublishRejection +mkPublishRejection code detail + | not (validCode code) = Left (InvalidPublishRejectionCode code) + | otherwise = + case detail of + Just value + | Text.null value -> Left PublishRejectionDetailEmpty + | detailBytes value > 1024 -> Left (PublishRejectionDetailTooLong (detailBytes value)) + _ -> Right PublishRejection {publishRejectionCode = code, publishRejectionDetail = detail} + where + detailBytes = ByteString.length . TE.encodeUtf8 + +validCode :: Text -> Bool +validCode code = + Text.length code <= 64 + && case Text.uncons code of + Nothing -> False + Just (first, rest) -> isLowerAscii first && Text.all isCodeTail rest + where + isLowerAscii char = char >= 'a' && char <= 'z' + isCodeTail char = + isLowerAscii char + || (char >= '0' && char <= '9') + || char == '.' + || char == '_' + || char == '-' diff --git a/src/Keiro/Outbox/Schema.hs b/src/Keiro/Outbox/Schema.hs --- a/src/Keiro/Outbox/Schema.hs +++ b/src/Keiro/Outbox/Schema.hs @@ -11,6 +11,8 @@ requeueStuckOutbox, markOutboxSent, markOutboxSentBatch, + markOutboxSentBatchTx, + markOutboxRejectedTx, markOutboxFailedTx, markOutboxSkippedTx, lookupOutbox, @@ -22,7 +24,7 @@ ) where -import Contravariant.Extras (contrazip2, contrazip3, contrazip5) +import Contravariant.Extras (contrazip2, contrazip3, contrazip4, contrazip5) import Data.ByteString (ByteString) import Data.Functor.Contravariant ((>$<)) import Data.Time.Clock (NominalDiffTime, addUTCTime) @@ -38,6 +40,7 @@ contentTypeText, parseContentType, ) +import Keiro.Outbox.Rejection (PublishRejection (..)) import Keiro.Outbox.Types import Keiro.Prelude import Kiroku.Store.Effect (Store) @@ -192,21 +195,44 @@ -- Only rows still in @publishing@ transition. Returns how many rows changed; -- callers treat a shortfall as benign because delivery is already at-least-once. markOutboxSentBatch :: (Store :> es) => [OutboxId] -> UTCTime -> Eff es Int -markOutboxSentBatch [] _ = pure 0 markOutboxSentBatch outboxIds now = + runTransaction (markOutboxSentBatchTx outboxIds now) + +-- | Transactional form of 'markOutboxSentBatch' for atomic batch finalization. +markOutboxSentBatchTx :: [OutboxId] -> UTCTime -> Tx.Transaction Int +markOutboxSentBatchTx [] _ = pure 0 +markOutboxSentBatchTx outboxIds now = fromIntegral - <$> runTransaction - ( Tx.statement - (fmap unOutboxId outboxIds, now) - markSentBatchStmt - ) + <$> Tx.statement + (fmap unOutboxId outboxIds, now) + markSentBatchStmt +-- | Permanently reject a claimed row and retain bounded audit data. +-- +-- The transition matches only @publishing@. Repeating it, losing a race to +-- recovery, or attempting to overwrite another terminal state returns 'False' +-- without changing the row. +markOutboxRejectedTx :: + OutboxId -> + PublishRejection -> + UTCTime -> + Tx.Transaction Bool +markOutboxRejectedTx outboxId rejection now = + Tx.statement + ( unOutboxId outboxId, + publishRejectionCode rejection, + publishRejectionDetail rejection, + now + ) + markRejectedStmt + -- | Mark a row as failed and decide whether it is retryable or dead. -- -- Reads the current @attempt_count@; if it is greater than or equal to -- @maxAttempts@, transitions to 'OutboxDead'. Otherwise transitions to -- 'OutboxFailed' and sets @next_attempt_at = now + delay@. Returns the --- resulting status so the worker can update its summary counters. +-- resulting status when a row changed so the worker can update its summary +-- counters from committed work. A stale or terminal row returns 'Nothing'. -- -- Runs inside the caller's transaction to keep "read attempt count → write -- status" atomic with respect to other workers. @@ -222,23 +248,26 @@ Int -> NominalDiffTime -> UTCTime -> - Tx.Transaction OutboxStatus + Tx.Transaction (Maybe OutboxStatus) markOutboxFailedTx outboxId errMsg maxAttempts delay now = do currentAttempt <- Tx.statement (unOutboxId outboxId) readAttemptCountStmt - let attempt = fromMaybe 0 currentAttempt - shouldDie = attempt >= maxAttempts - nextStatus = if shouldDie then OutboxDead else OutboxFailed - nextAttempt = addUTCTime delay now - Tx.statement - (unOutboxId outboxId, statusText nextStatus, errMsg, nextAttempt, now) - markFailedStmt - pure nextStatus + case currentAttempt of + Nothing -> pure Nothing + Just attempt -> do + let shouldDie = attempt >= maxAttempts + nextStatus = if shouldDie then OutboxDead else OutboxFailed + nextAttempt = addUTCTime delay now + changed <- + Tx.statement + (unOutboxId outboxId, statusText nextStatus, errMsg, nextAttempt, now) + markFailedStmt + pure (nextStatus <$ guard changed) -- | Return a claimed row to @failed@ without consuming an attempt. -- -- Used for rows skipped because an earlier row in the same ordered group failed -- inside the same publish batch. -markOutboxSkippedTx :: OutboxId -> Text -> UTCTime -> Tx.Transaction () +markOutboxSkippedTx :: OutboxId -> Text -> UTCTime -> Tx.Transaction Bool markOutboxSkippedTx outboxId errMsg now = Tx.statement (unOutboxId outboxId, errMsg, now) markSkippedStmt @@ -410,7 +439,7 @@ WHERE earlier.source = r.source AND earlier.message_key = r.message_key AND (earlier.created_at, earlier.outbox_id) < (r.created_at, r.outbox_id) - AND earlier.status NOT IN ('sent', 'dead') + AND earlier.status NOT IN ('sent', 'dead', 'rejected') AND NOT (earlier.status IN ('pending', 'failed') AND earlier.next_attempt_at <= $2) ) ) """, postFilter = @@ -420,7 +449,7 @@ WHERE earlier.source = c.source AND earlier.message_key = c.message_key AND (earlier.created_at, earlier.outbox_id) < (c.created_at, c.outbox_id) - AND earlier.status NOT IN ('sent', 'dead') + AND earlier.status NOT IN ('sent', 'dead', 'rejected') AND NOT EXISTS ( SELECT 1 FROM candidate c2 WHERE c2.outbox_id = earlier.outbox_id ) ) ) @@ -436,7 +465,7 @@ SELECT 1 FROM keiro.keiro_outbox earlier WHERE earlier.source = r.source AND (earlier.created_at, earlier.outbox_id) < (r.created_at, r.outbox_id) - AND earlier.status NOT IN ('sent', 'dead') + AND earlier.status NOT IN ('sent', 'dead', 'rejected') AND NOT (earlier.status IN ('pending', 'failed') AND earlier.next_attempt_at <= $2) ) """, postFilter = @@ -445,7 +474,7 @@ SELECT 1 FROM keiro.keiro_outbox earlier WHERE earlier.source = c.source AND (earlier.created_at, earlier.outbox_id) < (c.created_at, c.outbox_id) - AND earlier.status NOT IN ('sent', 'dead') + AND earlier.status NOT IN ('sent', 'dead', 'rejected') AND NOT EXISTS ( SELECT 1 FROM candidate c2 WHERE c2.outbox_id = earlier.outbox_id ) ) @@ -511,8 +540,8 @@ kt.source_event_id, kt.source_global_position, kt.causation_id, kt.correlation_id, kt.traceparent, kt.tracestate, kt.payload_bytes, kt.attributes, kt.occurred_at, kt.status, kt.attempt_count, - kt.next_attempt_at, kt.last_error, kt.published_at, kt.created_at, - kt.updated_at + kt.next_attempt_at, kt.last_error, kt.published_at, kt.rejected_at, + kt.rejection_code, kt.rejection_detail, kt.created_at, kt.updated_at """ unqualifiedRowColumns :: Text @@ -524,8 +553,8 @@ source_event_id, source_global_position, causation_id, correlation_id, traceparent, tracestate, payload_bytes, attributes, occurred_at, status, attempt_count, - next_attempt_at, last_error, published_at, created_at, - updated_at + next_attempt_at, last_error, published_at, rejected_at, + rejection_code, rejection_detail, created_at, updated_at """ claimResultDecoder :: D.Row OutboxRow @@ -555,7 +584,7 @@ readAttemptCountStmt :: Statement UUID (Maybe Int) readAttemptCountStmt = preparable - "SELECT attempt_count FROM keiro.keiro_outbox WHERE outbox_id = $1" + "SELECT attempt_count FROM keiro.keiro_outbox WHERE outbox_id = $1 AND status = 'publishing'" (E.param (E.nonNullable E.uuid)) (D.rowMaybe (fromIntegral <$> D.column (D.nonNullable D.int8))) @@ -632,7 +661,30 @@ ) D.rowsAffected -markFailedStmt :: Statement (UUID, Text, Text, UTCTime, UTCTime) () +markRejectedStmt :: Statement (UUID, Text, Maybe Text, UTCTime) Bool +markRejectedStmt = + preparable + """ + UPDATE keiro.keiro_outbox + SET status = 'rejected', + rejected_at = $4, + rejection_code = $2, + rejection_detail = $3, + last_error = NULL, + published_at = NULL, + updated_at = $4 + WHERE outbox_id = $1 + AND status = 'publishing' + """ + ( contrazip4 + (E.param (E.nonNullable E.uuid)) + (E.param (E.nonNullable E.text)) + (E.param (E.nullable E.text)) + (E.param (E.nonNullable E.timestamptz)) + ) + ((> 0) <$> D.rowsAffected) + +markFailedStmt :: Statement (UUID, Text, Text, UTCTime, UTCTime) Bool markFailedStmt = preparable """ @@ -651,9 +703,9 @@ (E.param (E.nonNullable E.timestamptz)) (E.param (E.nonNullable E.timestamptz)) ) - D.noResult + ((> 0) <$> D.rowsAffected) -markSkippedStmt :: Statement (UUID, Text, UTCTime) () +markSkippedStmt :: Statement (UUID, Text, UTCTime) Bool markSkippedStmt = preparable """ @@ -671,7 +723,7 @@ (E.param (E.nonNullable E.text)) (E.param (E.nonNullable E.timestamptz)) ) - D.noResult + ((> 0) <$> D.rowsAffected) lookupOutboxStmt :: Statement UUID (Maybe OutboxRow) lookupOutboxStmt = @@ -709,8 +761,8 @@ schema_version_ref, schema_id, schema_fingerprint, source_event_id, source_global_position, causation_id, correlation_id, traceparent, tracestate, payload_bytes, attributes, occurred_at, status, - attempt_count, next_attempt_at, last_error, published_at, created_at, - updated_at + attempt_count, next_attempt_at, last_error, published_at, rejected_at, + rejection_code, rejection_detail, created_at, updated_at FROM keiro.keiro_outbox """ @@ -745,6 +797,9 @@ nextAttemptAt :: !UTCTime, lastError :: !(Maybe Text), publishedAt :: !(Maybe UTCTime), + rejectedAt :: !(Maybe UTCTime), + rejectionCode :: !(Maybe Text), + rejectionDetail :: !(Maybe Text), createdAt :: !UTCTime, updatedAt :: !UTCTime } @@ -780,6 +835,9 @@ <*> D.column (D.nonNullable D.timestamptz) <*> D.column (D.nullable D.text) <*> D.column (D.nullable D.timestamptz) + <*> D.column (D.nullable D.timestamptz) + <*> D.column (D.nullable D.text) + <*> D.column (D.nullable D.text) <*> D.column (D.nonNullable D.timestamptz) <*> D.column (D.nonNullable D.timestamptz) @@ -832,6 +890,11 @@ nextAttemptAt = raw ^. #nextAttemptAt, lastError = raw ^. #lastError, publishedAt = raw ^. #publishedAt, + rejectedAt = raw ^. #rejectedAt, + rejection = + (\code -> PublishRejection code (raw ^. #rejectionDetail)) + <$> raw + ^. #rejectionCode, createdAt = raw ^. #createdAt, updatedAt = raw ^. #updatedAt } diff --git a/src/Keiro/Outbox/Types.hs b/src/Keiro/Outbox/Types.hs --- a/src/Keiro/Outbox/Types.hs +++ b/src/Keiro/Outbox/Types.hs @@ -18,6 +18,11 @@ OutboxPublishSummary (..), OutboxMaintenanceOptions (..), OutboxMaintenanceSummary (..), + PublishRejection, + PublishRejectionError (..), + mkPublishRejection, + publishRejectionCode, + publishRejectionDetail, defaultPublishOptions, defaultMaintenanceOptions, mkOutboxPublishOptions, @@ -31,6 +36,13 @@ import Data.UUID (UUID) import Data.UUID qualified as UUID import Keiro.Integration.Event (IntegrationEvent) +import Keiro.Outbox.Rejection + ( PublishRejection, + PublishRejectionError (..), + mkPublishRejection, + publishRejectionCode, + publishRejectionDetail, + ) import Keiro.Prelude import OpenTelemetry.Trace.Core (Tracer) @@ -56,6 +68,8 @@ -- after a worker crash are reclaimed by 'Keiro.Outbox.outboxMaintenancePass' -- after 'publishingTimeout'. -- * 'OutboxSent' — Kafka acknowledged the publish; terminal. +-- * 'OutboxRejected' — the transport intentionally and permanently refused +-- publication; terminal and retained for operator audit. -- * 'OutboxFailed' — last attempt failed; will be retried after -- 'next_attempt_at'. -- * 'OutboxDead' — terminal failure after 'maxAttempts' consecutive @@ -64,6 +78,7 @@ = OutboxPending | OutboxPublishing | OutboxSent + | OutboxRejected | OutboxFailed | OutboxDead deriving stock (Generic, Eq, Show) @@ -138,6 +153,8 @@ nextAttemptAt :: !UTCTime, lastError :: !(Maybe Text), publishedAt :: !(Maybe UTCTime), + rejectedAt :: !(Maybe UTCTime), + rejection :: !(Maybe PublishRejection), createdAt :: !UTCTime, updatedAt :: !UTCTime } @@ -173,7 +190,9 @@ -- | Aggregate result of one publisher pass. -- --- @published + retried + dead@ equals the number of rows claimed. 'retried' +-- @published + rejected + retried + dead@ is the number of rows durably +-- finalized by the pass and can be less than 'claimed' when recovery wins a +-- stale-worker race. 'retried' -- includes rows that were skipped because an earlier row in the same ordered -- publish group failed; those rows are returned to @failed@ without consuming -- an attempt. 'haltedOn' is populated only by 'StopTheLine' policy and names @@ -181,6 +200,7 @@ data OutboxPublishSummary = OutboxPublishSummary { claimed :: !Int, published :: !Int, + rejected :: !Int, retried :: !Int, dead :: !Int, haltedOn :: !(Maybe OutboxId) @@ -252,6 +272,7 @@ OutboxPending -> "pending" OutboxPublishing -> "publishing" OutboxSent -> "sent" + OutboxRejected -> "rejected" OutboxFailed -> "failed" OutboxDead -> "dead" @@ -261,6 +282,7 @@ "pending" -> Right OutboxPending "publishing" -> Right OutboxPublishing "sent" -> Right OutboxSent + "rejected" -> Right OutboxRejected "failed" -> Right OutboxFailed "dead" -> Right OutboxDead bad -> Left ("unknown keiro_outbox.status: " <> bad) diff --git a/src/Keiro/Telemetry.hs b/src/Keiro/Telemetry.hs --- a/src/Keiro/Telemetry.hs +++ b/src/Keiro/Telemetry.hs @@ -72,6 +72,7 @@ keiroInstrumentationLibrary, keiroOutboxBacklogName, keiroOutboxPublishedName, + keiroOutboxRejectedName, keiroOutboxRetriedName, keiroOutboxDeadletteredName, keiroOutboxReclaimedName, @@ -124,6 +125,7 @@ newKeiroMetrics, recordOutboxBacklog, recordOutboxPublished, + recordOutboxRejected, recordOutboxRetried, recordOutboxDeadlettered, recordOutboxReclaimed, @@ -579,6 +581,9 @@ keiroOutboxPublishedName :: Text keiroOutboxPublishedName = "keiro.outbox.published" +keiroOutboxRejectedName :: Text +keiroOutboxRejectedName = "keiro.outbox.rejected" + keiroOutboxRetriedName :: Text keiroOutboxRetriedName = "keiro.outbox.retried" @@ -736,6 +741,7 @@ data KeiroMetrics = KeiroMetrics { outboxBacklog :: Gauge Int64, outboxPublished :: Counter Int64, + outboxRejected :: Counter Int64, outboxRetried :: Counter Int64, outboxDeadlettered :: Counter Int64, outboxReclaimed :: Counter Int64, @@ -796,6 +802,7 @@ newKeiroMetrics meter = liftIO $ do outboxBacklog' <- gaugeI64 keiroOutboxBacklogName "{event}" "Outbox rows awaiting publish." outboxPublished' <- counterI64 keiroOutboxPublishedName "{event}" "Outbox events successfully published." + outboxRejected' <- counterI64 keiroOutboxRejectedName "{event}" "Outbox events intentionally and permanently rejected by the publisher." outboxRetried' <- counterI64 keiroOutboxRetriedName "{event}" "Outbox publish attempts that failed and will retry." outboxDeadlettered' <- counterI64 keiroOutboxDeadletteredName "{event}" "Outbox events parked after exhausting retries." outboxReclaimed' <- counterI64 keiroOutboxReclaimedName "{event}" "Outbox rows reclaimed from a crashed or stalled publisher." @@ -848,6 +855,7 @@ KeiroMetrics { outboxBacklog = outboxBacklog', outboxPublished = outboxPublished', + outboxRejected = outboxRejected', outboxRetried = outboxRetried', outboxDeadlettered = outboxDeadlettered', outboxReclaimed = outboxReclaimed', @@ -931,6 +939,9 @@ recordOutboxPublished :: (MonadIO m) => Maybe KeiroMetrics -> Int64 -> m () recordOutboxPublished = recordCounter outboxPublished + +recordOutboxRejected :: (MonadIO m) => Maybe KeiroMetrics -> Int64 -> m () +recordOutboxRejected = recordCounter outboxRejected recordOutboxRetried :: (MonadIO m) => Maybe KeiroMetrics -> Int64 -> m () recordOutboxRetried = recordCounter outboxRetried diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -156,6 +156,7 @@ OutboxRow (..), OutboxStatus (..), PublishOutcome (..), + PublishRejectionError (..), claimOutboxBatch, defaultMaintenanceOptions, defaultPublishOptions, @@ -168,12 +169,15 @@ mintIntegrationEvent, mkIntegrationProducer, mkOutboxPublishOptions, + mkPublishRejection, outboxMaintenancePass, publishClaimedOutbox, + publishRejectionCode, + publishRejectionDetail, sampleOutboxBacklog, ) import Keiro.Outbox.Kafka qualified as OutboxKafka -import Keiro.Outbox.Schema (markOutboxFailedTx) +import Keiro.Outbox.Schema (markOutboxFailedTx, markOutboxRejectedTx) import Keiro.Prelude import Keiro.ProcessManager import Keiro.Projection @@ -5822,6 +5826,30 @@ record ^. #key `shouldBe` Nothing describe "Keiro.Outbox" $ around (withFreshStore fixture) $ do + it "validates terminal publication rejection data at its public boundary" $ \_storeHandle -> do + let validCode64 = "a" <> Text.replicate 63 "z" + validDetail1024 = Text.replicate 1024 "x" + validUtf8Detail = Text.replicate 512 "é" + valid <- shouldBeRight (mkPublishRejection validCode64 (Just validDetail1024)) + publishRejectionCode valid `shouldBe` validCode64 + publishRejectionDetail valid `shouldBe` Just validDetail1024 + shouldBeRight_ (mkPublishRejection "authorization.denied_v2" Nothing) + shouldBeRight_ (mkPublishRejection "invalid-destination" (Just validUtf8Detail)) + mkPublishRejection "" Nothing + `shouldBeLeft` InvalidPublishRejectionCode "" + mkPublishRejection "Uppercase" Nothing + `shouldBeLeft` InvalidPublishRejectionCode "Uppercase" + mkPublishRejection "1leading-digit" Nothing + `shouldBeLeft` InvalidPublishRejectionCode "1leading-digit" + mkPublishRejection "contains/slash" Nothing + `shouldBeLeft` InvalidPublishRejectionCode "contains/slash" + mkPublishRejection ("a" <> Text.replicate 64 "z") Nothing + `shouldBeLeft` InvalidPublishRejectionCode ("a" <> Text.replicate 64 "z") + mkPublishRejection "invalid-destination" (Just "") + `shouldBeLeft` PublishRejectionDetailEmpty + mkPublishRejection "invalid-destination" (Just (Text.replicate 513 "é")) + `shouldBeLeft` PublishRejectionDetailTooLong 1026 + it "validates publisher options before startup" $ \_storeHandle -> do shouldBeRight_ (mkOutboxPublishOptions defaultPublishOptions) mkOutboxPublishOptions (defaultPublishOptions & #batchSize .~ 0) @@ -6008,6 +6036,36 @@ row ^. #publishedAt `shouldSatisfy` isJust row ^. #lastError `shouldBe` Nothing + it "finalizes rejection exactly once with durable typed audit data" $ \storeHandle -> do + let oid = OutboxId outboxUuid1 + rejection <- shouldBeRight (mkPublishRejection "authorization.denied" (Just "sink policy refused this message")) + replacement <- shouldBeRight (mkPublishRejection "invalid.destination" Nothing) + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction (enqueueIntegrationEventTx oid sampleIntegrationEnvelope) + firstClaimAt <- getCurrentTime + Right [_] <- Store.runStoreIO storeHandle (claimOutboxBatch PerKeyHeadOfLine 10 firstClaimAt) + Right (Just OutboxFailed) <- + Store.runStoreIO storeHandle $ + Store.runTransaction (markOutboxFailedTx oid "transient predecessor" 5 0 firstClaimAt) + secondClaimAt <- getCurrentTime + Right [claimed] <- Store.runStoreIO storeHandle (claimOutboxBatch PerKeyHeadOfLine 10 secondClaimAt) + rejectedAt <- getCurrentTime + Right True <- + Store.runStoreIO storeHandle $ + Store.runTransaction (markOutboxRejectedTx oid rejection rejectedAt) + Right False <- + Store.runStoreIO storeHandle $ + Store.runTransaction (markOutboxRejectedTx oid replacement (addUTCTime 60 rejectedAt)) + Right (Just row) <- Store.runStoreIO storeHandle (lookupOutbox oid) + row ^. #status `shouldBe` OutboxRejected + row ^. #attemptCount `shouldBe` 2 + row ^. #nextAttemptAt `shouldBe` claimed ^. #nextAttemptAt + row ^. #lastError `shouldBe` Nothing + row ^. #publishedAt `shouldBe` Nothing + row ^. #rejectedAt `shouldBe` Just rejectedAt + row ^. #rejection `shouldBe` Just rejection + it "reclaims a row stranded in publishing by a crashed worker through maintenance" $ \storeHandle -> do let oid = OutboxId outboxUuid1 Right () <- @@ -6195,6 +6253,171 @@ Right (Just row) <- Store.runStoreIO storeHandle (lookupOutbox oid) row ^. #status `shouldBe` OutboxSent + it "finalizes a mid-run rejection and continues the same-key suffix" $ \storeHandle -> do + let row1Id = outboxIdFromOrdinal 1 + row2Id = outboxIdFromOrdinal 2 + row3Id = outboxIdFromOrdinal 3 + rows = + [ (row1Id, sampleIntegrationEnvelope & #messageId .~ "reject-run-1" & #key .~ Just "reject-run-key"), + (row2Id, sampleIntegrationEnvelope & #messageId .~ "reject-run-2" & #key .~ Just "reject-run-key"), + (row3Id, sampleIntegrationEnvelope & #messageId .~ "reject-run-3" & #key .~ Just "reject-run-key") + ] + rejection <- shouldBeRight (mkPublishRejection "unsupported.sink" (Just "the configured sink cannot accept this event type")) + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction $ + traverse_ (uncurry enqueueIntegrationEventTx) rows + let publish claimed = + pure + [ ( row ^. #outboxId, + if row ^. #outboxId == row2Id + then PublishRejected rejection + else PublishSucceeded + ) + | row <- claimed + ] + Right summary <- Store.runStoreIO storeHandle (publishClaimedOutbox publish defaultPublishOptions Nothing) + summary ^. #claimed `shouldBe` 3 + summary ^. #published `shouldBe` 2 + summary ^. #rejected `shouldBe` 1 + summary ^. #retried `shouldBe` 0 + Right (Just row1) <- Store.runStoreIO storeHandle (lookupOutbox row1Id) + Right (Just row2) <- Store.runStoreIO storeHandle (lookupOutbox row2Id) + Right (Just row3) <- Store.runStoreIO storeHandle (lookupOutbox row3Id) + row1 ^. #status `shouldBe` OutboxSent + row2 ^. #status `shouldBe` OutboxRejected + row2 ^. #rejection `shouldBe` Just rejection + row3 ^. #status `shouldBe` OutboxSent + + it "redelivers callbacks after a pre-commit finalization failure" $ \storeHandle -> do + let sentId = outboxIdFromOrdinal 1 + rejectedId = outboxIdFromOrdinal 2 + rows = + [ (sentId, sampleIntegrationEnvelope & #messageId .~ "precommit-sent" & #key .~ Just "precommit-key"), + (rejectedId, sampleIntegrationEnvelope & #messageId .~ "precommit-rejected" & #key .~ Just "precommit-key") + ] + rejection <- shouldBeRight (mkPublishRejection "invalid.destination" Nothing) + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction $ + traverse_ (uncurry enqueueIntegrationEventTx) rows + callbackInvocations <- newIORef [] + let publish claimed = do + liftIO (modifyIORef' callbackInvocations (fmap (^. #outboxId) claimed :)) + pure + [ (row ^. #outboxId, if row ^. #outboxId == rejectedId then PublishRejected rejection else PublishSucceeded) + | row <- claimed + ] + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction $ + Tx.sql + "ALTER TABLE keiro.keiro_outbox ADD CONSTRAINT keiro_outbox_test_rejection_block CHECK (status <> 'rejected')" + first <- Store.runStoreIO storeHandle (publishClaimedOutbox publish defaultPublishOptions Nothing) + first `shouldSatisfy` \case + Left _ -> True + Right _ -> False + Right (Just stillPublishingSent) <- Store.runStoreIO storeHandle (lookupOutbox sentId) + Right (Just stillPublishingRejected) <- Store.runStoreIO storeHandle (lookupOutbox rejectedId) + stillPublishingSent ^. #status `shouldBe` OutboxPublishing + stillPublishingRejected ^. #status `shouldBe` OutboxPublishing + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction $ + Tx.sql "ALTER TABLE keiro.keiro_outbox DROP CONSTRAINT keiro_outbox_test_rejection_block" + now <- getCurrentTime + let strandedAt = addUTCTime (-3600) now + maintenanceOptions = defaultMaintenanceOptions & #publishingTimeout .~ 1 + Right () <- Store.runStoreIO storeHandle (backdateOutboxUpdatedAt sentId strandedAt) + Right () <- Store.runStoreIO storeHandle (backdateOutboxUpdatedAt rejectedId strandedAt) + Right maintenance <- Store.runStoreIO storeHandle (outboxMaintenancePass maintenanceOptions Nothing) + maintenance ^. #requeued `shouldBe` 2 + Right summary <- Store.runStoreIO storeHandle (publishClaimedOutbox publish defaultPublishOptions Nothing) + summary ^. #published `shouldBe` 1 + summary ^. #rejected `shouldBe` 1 + Right (Just sentRow) <- Store.runStoreIO storeHandle (lookupOutbox sentId) + Right (Just rejectedRow) <- Store.runStoreIO storeHandle (lookupOutbox rejectedId) + sentRow ^. #status `shouldBe` OutboxSent + rejectedRow ^. #status `shouldBe` OutboxRejected + invocations <- readIORef callbackInvocations + invocations `shouldBe` replicate 2 [sentId, rejectedId] + + it "treats rejection as terminal for per-source ordering" $ \storeHandle -> do + let row1Id = outboxIdFromOrdinal 1 + row2Id = outboxIdFromOrdinal 2 + row3Id = outboxIdFromOrdinal 3 + sourceEvent oid messageId = + (oid, sampleIntegrationEnvelope & #messageId .~ messageId & #source .~ "reject-source" & #key .~ Nothing) + rows = + [ sourceEvent row1Id "reject-source-1", + sourceEvent row2Id "reject-source-2", + sourceEvent row3Id "reject-source-3" + ] + rejection <- shouldBeRight (mkPublishRejection "authorization.denied" Nothing) + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction $ + traverse_ (uncurry enqueueIntegrationEventTx) rows + let publish claimed = + pure + [ (row ^. #outboxId, if row ^. #outboxId == row2Id then PublishRejected rejection else PublishSucceeded) + | row <- claimed + ] + opts = defaultPublishOptions & #orderingPolicy .~ PerSourceStream + Right summary <- Store.runStoreIO storeHandle (publishClaimedOutbox publish opts Nothing) + summary ^. #published `shouldBe` 2 + summary ^. #rejected `shouldBe` 1 + Right (Just successor) <- Store.runStoreIO storeHandle (lookupOutbox row3Id) + successor ^. #status `shouldBe` OutboxSent + + it "counts only a rejection finalization that wins the publishing-state race" $ \storeHandle -> do + let oid = OutboxId outboxUuid1 + rejection <- shouldBeRight (mkPublishRejection "authorization.denied" Nothing) + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction (enqueueIntegrationEventTx oid sampleIntegrationEnvelope) + now <- getCurrentTime + let strandedAt = addUTCTime (-3600) now + maintenanceOptions = defaultMaintenanceOptions & #publishingTimeout .~ 1 + publish _ = do + backdateOutboxUpdatedAt oid strandedAt + _ <- outboxMaintenancePass maintenanceOptions Nothing + pure (PublishRejected rejection) + Right summary <- + Store.runStoreIO storeHandle $ + publishClaimedOutbox (perRow publish) defaultPublishOptions Nothing + summary ^. #claimed `shouldBe` 1 + summary ^. #published `shouldBe` 0 + summary ^. #rejected `shouldBe` 0 + summary ^. #retried `shouldBe` 0 + summary ^. #dead `shouldBe` 0 + Right (Just row) <- Store.runStoreIO storeHandle (lookupOutbox oid) + row ^. #status `shouldBe` OutboxFailed + row ^. #rejection `shouldBe` Nothing + + it "excludes rejected rows from claims, maintenance, backlog, and sent garbage collection" $ \storeHandle -> do + let oid = OutboxId outboxUuid1 + rejection <- shouldBeRight (mkPublishRejection "invalid.destination" (Just "destination was removed")) + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction (enqueueIntegrationEventTx oid sampleIntegrationEnvelope) + Right summary <- + Store.runStoreIO storeHandle $ + publishClaimedOutbox (perRow (const (pure (PublishRejected rejection)))) defaultPublishOptions Nothing + summary ^. #rejected `shouldBe` 1 + now <- getCurrentTime + Right () <- Store.runStoreIO storeHandle (backdateOutboxUpdatedAt oid (addUTCTime (-3600) now)) + Right claimed <- Store.runStoreIO storeHandle (claimOutboxBatch BestEffort 10 now) + claimed `shouldBe` [] + Right maintenance <- Store.runStoreIO storeHandle (outboxMaintenancePass defaultMaintenanceOptions Nothing) + maintenance ^. #requeued `shouldBe` 0 + maintenance ^. #deadLettered `shouldBe` 0 + maintenance ^. #backlog `shouldBe` 0 + Right deleted <- Store.runStoreIO storeHandle (garbageCollectSent 0 now) + deleted `shouldBe` 0 + Right (Just retained) <- Store.runStoreIO storeHandle (lookupOutbox oid) + retained ^. #status `shouldBe` OutboxRejected + it "publishClaimedOutbox skips the same-key suffix after a mid-run failure" $ \storeHandle -> do let row1Id = outboxIdFromOrdinal 1 row2Id = outboxIdFromOrdinal 2 @@ -6291,9 +6514,10 @@ now <- getCurrentTime Right [_] <- Store.runStoreIO storeHandle (claimOutboxBatch PerKeyHeadOfLine 10 now) Right True <- Store.runStoreIO storeHandle (markOutboxSent oid now) - Right _ <- + Right lateResult <- Store.runStoreIO storeHandle $ Store.runTransaction (markOutboxFailedTx oid "late failure from a timed-out worker" 5 60 now) + lateResult `shouldBe` Nothing Right (Just row) <- Store.runStoreIO storeHandle (lookupOutbox oid) row ^. #status `shouldBe` OutboxSent row ^. #lastError `shouldBe` Nothing @@ -6373,6 +6597,46 @@ row4 ^. #status `shouldBe` OutboxFailed row4 ^. #attemptCount `shouldBe` 0 + it "StopTheLine continues after rejection and halts only on transient failure" $ \storeHandle -> do + let row1Id = outboxIdFromOrdinal 1 + row2Id = outboxIdFromOrdinal 2 + row3Id = outboxIdFromOrdinal 3 + row4Id = outboxIdFromOrdinal 4 + ids = [row1Id, row2Id, row3Id, row4Id] + rows = + [ (oid, sampleIntegrationEnvelope & #messageId .~ ("stop-reject-" <> Text.pack (show i)) & #key .~ Just "stop-reject-key") + | (i, oid) <- zip [1 .. 4 :: Int] ids + ] + rejection <- shouldBeRight (mkPublishRejection "unsupported.sink" Nothing) + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction $ + traverse_ (uncurry enqueueIntegrationEventTx) rows + seenRef <- newIORef [] + let publish claimed = do + liftIO (modifyIORef' seenRef (<> fmap (^. #outboxId) claimed)) + pure + [ ( row ^. #outboxId, + if row ^. #outboxId == row1Id + then PublishRejected rejection + else + if row ^. #outboxId == row3Id + then PublishFailed "stop after rejection" + else PublishSucceeded + ) + | row <- claimed + ] + opts = defaultPublishOptions & #orderingPolicy .~ StopTheLine & #backoff .~ ConstantBackoff 0 + Right summary <- Store.runStoreIO storeHandle (publishClaimedOutbox publish opts Nothing) + summary ^. #published `shouldBe` 1 + summary ^. #rejected `shouldBe` 1 + summary ^. #retried `shouldBe` 2 + summary ^. #haltedOn `shouldBe` Just row3Id + readIORef seenRef `shouldReturn` take 3 ids + Right (Just row4) <- Store.runStoreIO storeHandle (lookupOutbox row4Id) + row4 ^. #status `shouldBe` OutboxFailed + row4 ^. #attemptCount `shouldBe` 0 + it "publishClaimedOutbox treats a missing batch outcome as a failed row" $ \storeHandle -> do let okId = outboxIdFromOrdinal 1 missingId = outboxIdFromOrdinal 2 @@ -6643,6 +6907,31 @@ other -> expectationFailure ("expected Error \"broker unreachable\", got " <> show other) other -> expectationFailure ("expected one batch span, got " <> show (length other)) + it "does not mark a terminal rejection span as an error" $ \storeHandle -> do + (processor, spansRef) <- inMemoryListExporter + provider <- createTracerProvider [processor] emptyTracerProviderOptions + rejection <- shouldBeRight (mkPublishRejection "authorization.denied" (Just "operator policy")) + let tracer = makeTracer provider "keiro-test" tracerOptions + oid = OutboxId outboxUuid1 + opts = defaultPublishOptions & #tracer ?~ tracer + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction (enqueueIntegrationEventTx oid sampleIntegrationEnvelope) + Right summary <- + Store.runStoreIO storeHandle $ + publishClaimedOutbox (perRow (const (pure (PublishRejected rejection)))) opts Nothing + summary ^. #rejected `shouldBe` 1 + _ <- shutdownTracerProvider provider Nothing + spans <- traverse captureSpan =<< readIORef spansRef + case spans of + [batchSpan] -> do + textAttr (csAttributes batchSpan) "error.type" `shouldBe` Nothing + case csStatus batchSpan of + Unset -> pure () + Ok -> pure () + other -> expectationFailure ("expected rejection span to be Unset/Ok, got " <> show other) + other -> expectationFailure ("expected one rejection span, got " <> show (length other)) + it "publishClaimedOutbox records counters and sampleOutboxBacklog records the gauge" $ \storeHandle -> do (exporter, metricsRef) <- inMemoryMetricExporter (provider, _env) <- @@ -6653,16 +6942,23 @@ keiroMetrics <- Telemetry.newKeiroMetrics meter let okId = OutboxId outboxUuid1 failId = OutboxId outboxUuid2 + rejectId = OutboxId outboxUuid3 okEvent = sampleIntegrationEnvelope & #messageId .~ "metrics-ok" & #key .~ Nothing failEvent = sampleIntegrationEnvelope & #messageId .~ "metrics-fail" & #key .~ Nothing + rejectEvent = sampleIntegrationEnvelope & #messageId .~ "metrics-reject" & #key .~ Nothing + rejection <- shouldBeRight (mkPublishRejection "unsupported.sink" (Just "not routed")) Right () <- Store.runStoreIO storeHandle $ Store.runTransaction (enqueueIntegrationEventTx okId okEvent) Right () <- Store.runStoreIO storeHandle $ Store.runTransaction (enqueueIntegrationEventTx failId failEvent) + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction (enqueueIntegrationEventTx rejectId rejectEvent) let publish row | row ^. #outboxId == okId = pure PublishSucceeded + | row ^. #outboxId == rejectId = pure (PublishRejected rejection) | otherwise = pure (PublishFailed "broker down") retryPassOpts = defaultPublishOptions @@ -6679,6 +6975,7 @@ Right summary1 <- Store.runStoreIO storeHandle (publishClaimedOutbox (perRow publish) retryPassOpts (Just keiroMetrics)) summary1 ^. #published `shouldBe` 1 + summary1 ^. #rejected `shouldBe` 1 summary1 ^. #retried `shouldBe` 1 -- Pass 2 (maxAttempts = 1): the failed row crosses the ceiling and dies. Right summary2 <- @@ -6690,6 +6987,7 @@ let scalars = flattenScalarPoints exported -- Counters are cumulative across both passes. lookup "keiro.outbox.published" scalars `shouldBe` Just (IntNumber 1) + lookup "keiro.outbox.rejected" scalars `shouldBe` Just (IntNumber 1) lookup "keiro.outbox.retried" scalars `shouldBe` Just (IntNumber 1) lookup "keiro.outbox.deadlettered" scalars `shouldBe` Just (IntNumber 1) -- Publish passes no longer run the backlog COUNT(*) on the hot path. @@ -12907,6 +13205,8 @@ nextAttemptAt = UTCTime (ModifiedJulianDay 60000) (secondsToDiffTime 0), lastError = Nothing, publishedAt = Nothing, + rejectedAt = Nothing, + rejection = Nothing, createdAt = UTCTime (ModifiedJulianDay 60000) (secondsToDiffTime 0), updatedAt = UTCTime (ModifiedJulianDay 60000) (secondsToDiffTime 0) }