fix(service-messaging): enforce ack()'s claimed-row precondition in both outbox implementations

.changeset/outbox-ack-status-precondition.md

@@ -0,0 +1,53 @@
+---
+'@objectstack/service-messaging': minor
+---
+
+`INotificationOutbox.ack()` enforces its declared precondition — the row must be claimed — in both implementations, and `attempts` moves only for a real dispatch attempt
+
+`ack()` is the dispatcher's completion callback for a row it CLAIMED, and
+neither implementation checked that. `MemoryNotificationOutbox.ack` looked the
+row up by id and mutated it; `SqlNotificationOutbox.ack` read only `attempts`
+by id. So `ack(id, { success: false, suppressed: true })` on an unclaimed
+`pending` row succeeded, flipped the row to terminal `suppressed`, and
+incremented `attempts` — which made `ack` read like the cancellation primitive
+this interface deliberately does not have.
+
+That was a trap in two directions. It **raced the dispatcher**: between a
+caller's `list()` and its `ack()`, `claim()` could take the row — `claim` is
+atomic by contract and `ack` was never part of that atom — so a suppression
+could land on a delivery already on the wire, or a dispatcher's real outcome
+could be overwritten by a caller that thought it was cancelling. And it
+**corrupted `attempts`**: the counter feeds the retry schedule
+(`classifyDeliveryAttempt(result, errorClass, row.attempts, …)`), so a row
+"cancelled" this way arrived at its next real attempt with the backoff already
+advanced by an attempt that never went out.
+
+Both implementations now refuse an ack on a row that is not `in_flight`,
+throwing `NotificationAckError` with this package's already-registered
+ADR-0112 code `DELIVERY_NOT_ELIGIBLE` — the same refusal
+`SqlHttpOutbox.redeliver` raises when its own compare-and-set misses. A refused
+ack writes **nothing**: status, `attempts` and `error` are left exactly as they
+were, so the row stays claimable and its backoff position stays honest. An id
+matching no row remains a silent no-op — an absent row has no state to corrupt
+and no claim to lose.
+
+`SqlNotificationOutbox` does it as an **atomic conditional update** rather than
+a read-then-write, because a read cannot hold a row still and a read-then-write
+is the same defect wearing a different hat. The precondition is re-stated in
+the write (`where: { id, status: 'in_flight' }`), which — per #11009 — must
+ride the predicate path: on the by-id path the driver binds only the primary
+key and the extra predicate is silently discarded. `attempts` is incremented
+inside that condition and nowhere else, so the counter can only move for a row
+that was genuinely claimed. A conditional write that matches nothing is
+reported rather than passed off as success.
+
+`NotificationDispatcher` absorbs exactly one refusal — `DELIVERY_NOT_ELIGIBLE`
+— logs it and continues with the rest of the batch, because a send slower than
+`claimTtlMs` legitimately loses its claim to the visibility-timeout reap, and
+letting that unwind the partition loop would strand every still-valid row in
+the batch `in_flight` until its own timeout expired. Any other error still
+propagates.
+
+The sibling HTTP outbox is deliberately untouched: `assertHttpRedeliverable`
+depends on `IHttpOutbox.ack` incrementing `attempts` unconditionally, so that
+`attempts === 0` on a terminal row still means "parked, never sent".

packages/services/service-messaging/src/delivery-update-tenant-audit.integration.test.ts

@@ -18,6 +18,16 @@
  * tenant-classification contract this file pins (#10740) is unchanged —
  * threaded `tenantId`, never `bypassTenantAudit`.
  *
+ * [#11453] `SqlNotificationOutbox.ack` has since made the SAME move for the
+ * same reason: its new status precondition ("this row must still be
+ * `in_flight`") is a compare-and-set, and a predicate on the by-id path is
+ * silently discarded, so it rides `multi: true` too. Its audit op is
+ * `updateMany` and its spy below records `SqlDriver.updateMany`. Its
+ * CLASSIFICATION is unchanged — declared global, now via
+ * `dispatcherAckCasOptions` — which is the point of pinning the two
+ * separately: the op moved, the warrant did not. Of the three sites only
+ * `SqlHttpOutbox.ack` still writes by id.
+ *
  * The `ack` pair is declared global (`dispatcherAckOptions`, warrant in
  * `outbox-dispatcher-scope.ts`). `redeliver` is NOT: it is served to any
  * authenticated user, so it threads the caller's tenant instead. ⛔ A
@@ -74,14 +84,14 @@ let driver: SqlDriver;
 let warns: Array<{ msg: string; meta: any }>;
 /** Every `options` bag that reached `SqlDriver.update` — the `update` op only. */
 let driverUpdates: Array<{ object: string; id: unknown; options: any }>;
-/** Every `options` bag that reached `SqlDriver.updateMany` — `redeliver`'s op since #11009. */
+/** Every `options` bag that reached `SqlDriver.updateMany` — `redeliver`'s op since #11009, and the notification `ack`'s since #11453. */
 let driverUpdateManys: Array<{ object: string; where: unknown; options: any }>;
 
 /** The audit line for the SINGLE-RECORD op, matched on object + op. */
 const auditedUpdate = (object: string): boolean =>
  warns.some((w) => w.msg.includes(`[tenant-audit] update on tenant-scoped object "${object}"`));
 
-/** The audit line for the PREDICATE op — `redeliver`'s write since #11009. */
+/** The audit line for the PREDICATE op — `redeliver`'s write since #11009, the notification `ack`'s since #11453. */
 const auditedUpdateMany = (object: string): boolean =>
  warns.some((w) => w.msg.includes(`[tenant-audit] updateMany on tenant-scoped object "${object}"`));
 
@@ -211,7 +221,7 @@ async function seedDeadRow(id: string, org: string): Promise<void> {
 }
 
 // ───────────────────────────────────────────────────────────────────────────
-describe('ack — the two dispatcher sites are a classified global sweep (update op)', () => {
+describe('ack — the two dispatcher sites are a classified global sweep (update + updateMany ops)', () => {
  it('SqlHttpOutbox.ack records a REAL delivery in every organization, without a finding', async () => {
  // The gate's own precondition: this object really is tenant-scoped.
  expect((driver as any).resolveTenantField(SYS_HTTP_DELIVERY)).toBe('organization_id');
@@ -279,12 +289,23 @@ describe('ack — the two dispatcher sites are a classified global sweep (update
  'org_b:success:1',
  ]);
  // ② Declared global, for both organizations' rows.
- const ackWrites = driverUpdates.filter((u) => u.object === DELIVERY_OBJECT);
+ //
+ // [#11453] The ack's op is `updateMany` now, so the reading moves to
+ // that spy. The claim path writes there too (its reap and its atomic
+ // claim), so the filter names what an ACK write looks like — and that
+ // predicate is not incidental: `{ id: <scalar>, status: 'in_flight' }`
+ // IS the compare-and-set, so matching on it pins that the ack reached
+ // the driver CONDITIONAL rather than as a blind by-id write.
+ const ackWrites = driverUpdateManys.filter(
+ (u) => u.object === DELIVERY_OBJECT
+ && typeof (u.where as any)?.id === 'string'
+ && (u.where as any)?.status === 'in_flight',
+ );
  expect(ackWrites).toHaveLength(2);
  expect(ackWrites.every((u) => u.options?.bypassTenantAudit === true)).toBe(true);
- expect(auditedUpdate(DELIVERY_OBJECT)).toBe(false);
+ expect(auditedUpdateMany(DELIVERY_OBJECT)).toBe(false);
 
- await controlUnscopedUpdate(DELIVERY_OBJECT, idA);
+ await controlUnscopedUpdateMany(DELIVERY_OBJECT, idA);
  });
 });
 

packages/services/service-messaging/src/dispatcher.ts

@@ -1,7 +1,7 @@
 // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
 
 import type { MessagingChannel, MessagingChannelContext, Notification, SendResult } from './channel.js';
-import type { INotificationOutbox, NotificationDeliveryRecord } from './outbox.js';
+import type { AckResult, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js';
 import { classifyDeliveryAttempt } from './backoff.js';
 import { renderDigest } from './digest-render.js';
 
@@ -207,7 +207,7 @@ export class NotificationDispatcher {
  const channel = this.opts.channels.getChannel(channelName);
  if (!channel) {
  for (const row of rows) {
- await this.opts.outbox.ack(row.id, { success: false, error: `channel '${channelName}' not registered`, dead: true });
+ await this.ackAttempt(row, { success: false, error: `channel '${channelName}' not registered`, dead: true });
  this.opts.onAttempt?.(row, false);
  }
  return;
@@ -237,7 +237,7 @@ export class NotificationDispatcher {
  const now = this.opts.now?.() ?? Date.now();
  for (const row of rows) {
  const ack = classifyDeliveryAttempt(result, errorClass, row.attempts, now, this.opts.rng);
- await this.opts.outbox.ack(row.id, ack);
+ await this.ackAttempt(row, ack);
  this.opts.onAttempt?.(row, result.ok);
  }
  }
@@ -246,7 +246,7 @@ export class NotificationDispatcher {
  const channel = this.opts.channels.getChannel(row.channel);
  if (!channel) {
  // No transport for this channel → terminal, observable on the row.
- await this.opts.outbox.ack(row.id, {
+ await this.ackAttempt(row, {
  success: false,
  error: `channel '${row.channel}' not registered`,
  dead: true,
@@ -283,9 +283,42 @@ export class NotificationDispatcher {
  const errorClass = !result.ok && channel.classifyError ? channel.classifyError(result.error) : undefined;
  const now = this.opts.now?.() ?? Date.now();
  const ack = classifyDeliveryAttempt(result, errorClass, row.attempts, now, this.opts.rng);
- await this.opts.outbox.ack(row.id, ack);
+ await this.ackAttempt(row, ack);
  this.opts.onAttempt?.(row, result.ok);
  }
+
+ /**
+ * [#11453] Record one attempt's outcome, tolerating the ONE refusal a
+ * correct dispatcher can legitimately provoke.
+ *
+ * `ack()` now refuses a row that is not `in_flight`, and this loop can meet
+ * that honestly: a send slower than `claimTtlMs` lets the visibility-timeout
+ * reap return the row to `pending` for another node, so by the time we ack,
+ * the row is not ours. That is a race we are ALLOWED to lose — the delivery
+ * is re-driven by whoever holds the row now, which is what at-least-once
+ * means — and the refusal is the outbox correctly declining to overwrite
+ * someone else's state.
+ *
+ * ⛔ What it must not do is abort the tick. The rows still validly claimed
+ * by this node are processed after this one; letting a lost race unwind the
+ * partition loop would strand every one of them `in_flight` until their own
+ * timeouts expire, turning one lost race into a batch-wide delay.
+ *
+ * Only `DELIVERY_NOT_ELIGIBLE` is absorbed. A store fault is not a lost
+ * race and still propagates to `runTick`'s handler.
+ */
+ private async ackAttempt(row: NotificationDeliveryRecord, result: AckResult): Promise<void> {
+ try {
+ await this.opts.outbox.ack(row.id, result);
+ } catch (err) {
+ if ((err as { code?: string })?.code !== 'DELIVERY_NOT_ELIGIBLE') throw err;
+ this.opts.logger?.warn?.('notification-dispatcher: ack refused, claim no longer held', {
+ nodeId: this.opts.nodeId,
+ deliveryId: row.id,
+ error: (err as Error)?.message ?? String(err),
+ });
+ }
+ }
 }
 
 /** Group claimed digest rows by their `digestKey` (insertion order preserved). */

packages/services/service-messaging/src/index.ts

@@ -97,6 +97,9 @@ export type {
  ClaimOptions,
  AckResult,
 } from './outbox.js';
+// [#11453] `ack()`'s status precondition refuses with this, so a caller that
+// wants to distinguish "I lost the claim" from a transport fault can catch it.
+export { NotificationAckError } from './outbox.js';
 export { SqlNotificationOutbox, DELIVERY_OBJECT } from './sql-outbox.js';
 export type { SqlNotificationOutboxOptions } from './sql-outbox.js';
 export { MemoryNotificationOutbox } from './memory-outbox.js';

packages/services/service-messaging/src/memory-outbox.ts

@@ -9,6 +9,7 @@ import type {
  INotificationOutbox,
  NotificationDeliveryRecord,
 } from './outbox.js';
+import { NotificationAckError, notificationAckNotClaimedMessage } from './outbox.js';
 import { hashPartition } from './backoff.js';
 
 /**
@@ -117,8 +118,26 @@ export class MemoryNotificationOutbox implements INotificationOutbox {
 
  async ack(id: string, result: AckResult): Promise<void> {
  const r = this.rows.get(id);
+ // An id matching no row is not a contract violation: there is no state
+ // to corrupt and no claim to lose. Unchanged, and declared on the
+ // interface so the two backends agree about it.
  if (!r) return;
+ // [#11453] The status precondition. `ack` completes a delivery this
+ // caller CLAIMED; an unclaimed `pending` row (the ack-as-cancel trap)
+ // or an already-terminal one is refused, and nothing below runs — so a
+ // refused ack leaves status, attempts and error exactly as they were.
+ // Single-threaded, so this test and the mutation are already one atomic
+ // step; `SqlNotificationOutbox` spells the same guard as a conditional
+ // UPDATE because it is not.
+ if (r.status !== 'in_flight') {
+ throw new NotificationAckError(
+ notificationAckNotClaimedMessage(id, r.status),
+ 'DELIVERY_NOT_ELIGIBLE',
+ );
+ }
  const now = this.clock();
+ // Reached only for a genuinely claimed row, so this counts a real
+ // dispatch attempt and nothing else (#11453).
  r.attempts += 1;
  r.lastAttemptedAt = now;
  r.claimedBy = undefined;