feat(service-messaging): bind the claim credential in ack()'s compare-and-set

.changeset/outbox-ack-claim-credential.md

@@ -0,0 +1,11 @@
+---
+'@objectstack/service-messaging': minor
+---
+
+**BREAKING (interface member signature):** `INotificationOutbox.ack()` now takes back the claimed record instead of a bare row id, and its compare-and-set binds the claim credential the record carries (#11859). `claim()` / `claimDigest()` declare their true return type, `ClaimedDeliveryRecord[]` — the same rows as before, with the (`claimedBy`, `claimedAt`) pair the store stamps guaranteed present — so reads of claim results do not change; the one breaking edit is at ack call sites, which hand the whole record back where they previously handed `record.id` (the caller already holds it: `ack` completes a claim, and the record is what `claim()` returned).
+
+Why: `status = 'in_flight'` could prove a claim exists but not whose. In the reachable sequence — node A claims a row, the send outruns `claimTtlMs`, another node's `claim()` reaps and re-claims the row, A finishes late — A's ack still matched and wrote its outcome over B's live attempt. With the credential in the predicate a late ack matches nothing, is refused with the existing `NotificationAckError` (`DELIVERY_NOT_ELIGIBLE`, ADR-0112), and writes nothing; the caller never needs to know its own `nodeId`, because ownership is proven by round-tripping what `claim()` returned. Both implementations (`SqlNotificationOutbox`, `MemoryNotificationOutbox`) enforce it identically.
+
+Breaking ships as `minor` per the launch-window convention (`scripts/check-changeset-no-major.mjs`).
+
+<!-- adr-0087: not-required (no-migration-prescription) The changed member is a runtime TypeScript interface method (`INotificationOutbox.ack` in `packages/services/service-messaging/src/outbox.ts`): no Zod schema, no `packages/spec` declaration, no authorable key and no stored representation changes shape — `sys_notification_delivery` rows are byte-identical before and after, so `objectstack migrate meta` has nothing to visit and there is no tombstone to mint. Every affected consumer is told by the compiler at the call site (the argument type no longer accepts a string), which is more precise than a ledger entry. The checkable `runtime-interface-only` spelling is deliberately not claimed: `packages/spec/src/api/error-code-ledger.zod.ts` mentions `INotificationOutbox` in a prose comment about the shared `DELIVERY_NOT_ELIGIBLE` code, which its step-4 scan refuses as an unresolvable mention — the honest disposition here is this catch-all, with the same argument #8277 ratified. -->

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 { AckResult, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js';
+import type { AckResult, ClaimedDeliveryRecord, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js';
 import { classifyDeliveryAttempt } from './backoff.js';
 import { renderDigest } from './digest-render.js';
 
@@ -201,7 +201,7 @@ export class NotificationDispatcher {
  * ack every row in it with that one outcome. On failure the whole group
  * re-defers together (each row keeps its own backoff via its `attempts`).
  */
- private async processDigestGroup(rows: NotificationDeliveryRecord[]): Promise<void> {
+ private async processDigestGroup(rows: ClaimedDeliveryRecord[]): Promise<void> {
  const channelName = rows[0].channel;
  const recipient = rows[0].recipientId;
  const channel = this.opts.channels.getChannel(channelName);
@@ -242,7 +242,7 @@ export class NotificationDispatcher {
  }
  }
 
- private async processRow(row: NotificationDeliveryRecord): Promise<void> {
+ private async processRow(row: ClaimedDeliveryRecord): Promise<void> {
  const channel = this.opts.channels.getChannel(row.channel);
  if (!channel) {
  // No transport for this channel → terminal, observable on the row.
@@ -307,9 +307,13 @@ export class NotificationDispatcher {
  * 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> {
+ private async ackAttempt(row: ClaimedDeliveryRecord, result: AckResult): Promise<void> {
  try {
- await this.opts.outbox.ack(row.id, result);
+ // [#11859] The record is handed back WHOLE: its (claimedBy,
+ // claimedAt) pair is the claim credential the store stamped, and
+ // the ack's compare-and-set binds it — this loop never needs to
+ // know or repeat its own nodeId.
+ await this.opts.outbox.ack(row, 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', {
@@ -322,8 +326,8 @@ export class NotificationDispatcher {
 }
 
 /** Group claimed digest rows by their `digestKey` (insertion order preserved). */
-function groupByDigestKey(rows: NotificationDeliveryRecord[]): NotificationDeliveryRecord[][] {
- const groups = new Map<string, NotificationDeliveryRecord[]>();
+function groupByDigestKey(rows: ClaimedDeliveryRecord[]): ClaimedDeliveryRecord[][] {
+ const groups = new Map<string, ClaimedDeliveryRecord[]>();
  for (const r of rows) {
  const key = r.digestKey ?? r.id; // defensive — claimDigest only returns keyed rows
  let g = groups.get(key);

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

@@ -91,6 +91,9 @@ export type {
 export type {
  INotificationOutbox,
  NotificationDeliveryRecord,
+ // [#11859] What claim()/claimDigest() hand out and ack() takes back — the
+ // record carrying the claim credential the compare-and-set binds.
+ ClaimedDeliveryRecord,
  DeliveryStatus,
  DeliveryPayload,
  EnqueueDeliveryInput,

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

@@ -3,13 +3,19 @@
 import { randomUUID } from 'node:crypto';
 import type {
  AckResult,
+ ClaimedDeliveryRecord,
  ClaimOptions,
  DeliveryStatus,
  EnqueueDeliveryInput,
  INotificationOutbox,
  NotificationDeliveryRecord,
 } from './outbox.js';
-import { NotificationAckError, notificationAckNotClaimedMessage } from './outbox.js';
+import {
+ NotificationAckError,
+ notificationAckLostClaimMessage,
+ notificationAckNoCredentialMessage,
+ notificationAckNotClaimedMessage,
+} from './outbox.js';
 import { hashPartition } from './backoff.js';
 
 /**
@@ -61,7 +67,7 @@ export class MemoryNotificationOutbox implements INotificationOutbox {
  return id;
  }
 
- async claim(opts: ClaimOptions): Promise<NotificationDeliveryRecord[]> {
+ async claim(opts: ClaimOptions): Promise<ClaimedDeliveryRecord[]> {
  const now = opts.now ?? this.clock();
  // Reap stale in_flight.
  for (const r of this.rows.values()) {
@@ -72,7 +78,7 @@ export class MemoryNotificationOutbox implements INotificationOutbox {
  r.updatedAt = now;
  }
  }
- const out: NotificationDeliveryRecord[] = [];
+ const out: ClaimedDeliveryRecord[] = [];
  for (const r of this.rows.values()) {
  if (out.length >= opts.limit) break;
  if (r.status !== 'pending') continue;
@@ -83,12 +89,14 @@ export class MemoryNotificationOutbox implements INotificationOutbox {
  r.claimedBy = opts.nodeId;
  r.claimedAt = now;
  r.updatedAt = now;
- out.push({ ...r });
+ // [#11859] The copy handed out carries the claim credential the
+ // two lines above just stamped — the record IS the credential.
+ out.push({ ...r, claimedBy: opts.nodeId, claimedAt: now });
  }
  return out;
  }
 
- async claimDigest(opts: ClaimOptions): Promise<NotificationDeliveryRecord[]> {
+ async claimDigest(opts: ClaimOptions): Promise<ClaimedDeliveryRecord[]> {
  const now = opts.now ?? this.clock();
  // Reap stale in_flight (same as claim).
  for (const r of this.rows.values()) {
@@ -101,7 +109,7 @@ export class MemoryNotificationOutbox implements INotificationOutbox {
  }
  // Claim every DUE batched row in the partition — a window must be taken
  // whole, so `limit` does not truncate a group here.
- const out: NotificationDeliveryRecord[] = [];
+ const out: ClaimedDeliveryRecord[] = [];
  for (const r of this.rows.values()) {
  if (r.status !== 'pending') continue;
  if (r.digestKey == null) continue;
@@ -111,12 +119,19 @@ export class MemoryNotificationOutbox implements INotificationOutbox {
  r.claimedBy = opts.nodeId;
  r.claimedAt = now;
  r.updatedAt = now;
- out.push({ ...r });
+ out.push({ ...r, claimedBy: opts.nodeId, claimedAt: now });
  }
  return out;
  }
 
- async ack(id: string, result: AckResult): Promise<void> {
+ async ack(claimed: ClaimedDeliveryRecord, result: AckResult): Promise<void> {
+ const id = claimed.id;
+ // [#11859] The runtime half of the ClaimedDeliveryRecord contract, for
+ // JS callers and casts: a record with no claim credential was not
+ // handed out by claim()/claimDigest() and is refused before any read.
+ if (typeof claimed.claimedBy !== 'string' || typeof claimed.claimedAt !== 'number') {
+ throw new NotificationAckError(notificationAckNoCredentialMessage(id), 'DELIVERY_NOT_ELIGIBLE');
+ }
  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
@@ -135,6 +150,18 @@ export class MemoryNotificationOutbox implements INotificationOutbox {
  'DELIVERY_NOT_ELIGIBLE',
  );
  }
+ // [#11859] Ownership: the row is claimed, but not by the claim this
+ // record came from — it was reaped and re-claimed while the send ran
+ // (possibly by this same store handing it to this same node again: the
+ // credential is the PAIR, so a later claim's `claimedAt` refuses the
+ // earlier claim's ack). Refused with nothing written, so the live
+ // attempt it would have overwritten stays intact.
+ if (r.claimedBy !== claimed.claimedBy || r.claimedAt !== claimed.claimedAt) {
+ throw new NotificationAckError(
+ notificationAckLostClaimMessage(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).