keiro 0.13.0.0 → 0.14.0.0

raw patch · 8 files changed

+571/−60 lines, 8 filesdep ~keiro-corePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: keiro-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +23/−0
• keiro.cabal +4/−3
• src/Keiro/Outbox.hs +50/−21
• src/Keiro/Outbox/Rejection.hs +62/−0
• src/Keiro/Outbox/Schema.hs +96/−33
• src/Keiro/Outbox/Types.hs +23/−1
• src/Keiro/Telemetry.hs +11/−0
• test/Main.hs +302/−2

@@ -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

@@ -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

@@ -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

@@ -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 == '-'

@@ -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         }

@@ -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)

@@ -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

@@ -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)     }