Skip to content

Commit 1a47a53

Browse files
os-samclaude
andauthored
fix(service-messaging): enforce ack()'s claimed-row precondition in both outbox implementations (#11453) (#11858)
`ack()` is the dispatcher's completion callback for a row it CLAIMED, and neither implementation checked that, so `ack(id, { success: false, suppressed: true })` on an unclaimed `pending` row succeeded — flipping the row terminal and recording an attempt that never went on the wire. That made `ack` read like the cancellation primitive this interface deliberately does not have, and it raced `claim()` (atomic by contract; `ack` was never part of that atom). Both implementations now refuse a row that is not `in_flight`, with `NotificationAckError` / `DELIVERY_NOT_ELIGIBLE` — this package's already registered ADR-0112 code, the same refusal `SqlHttpOutbox.redeliver` raises when its own compare-and-set misses. A refused ack writes nothing. `SqlNotificationOutbox` does it as an atomic conditional update, not a read-then-write: the precondition is re-stated in the write, which per #11009 must ride the predicate path (the by-id path silently discards it). `attempts` increments inside that condition and nowhere else, so it can only move for a row that was genuinely claimed. The sibling HTTP outbox is untouched: `assertHttpRedeliverable` depends on `IHttpOutbox.ack` incrementing unconditionally, so `attempts === 0` on a terminal row still means "parked, never sent". Claude-Session: https://claude.ai/code/session_01APWX2AwT3a4xDcjPCe8bk4 Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
1 parent ce56937 commit 1a47a53

10 files changed

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

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

Lines changed: 27 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,16 @@
1818
* tenant-classification contract this file pins (#10740) is unchanged —
1919
* threaded `tenantId`, never `bypassTenantAudit`.
2020
*
21+
* [#11453] `SqlNotificationOutbox.ack` has since made the SAME move for the
22+
* same reason: its new status precondition ("this row must still be
23+
* `in_flight`") is a compare-and-set, and a predicate on the by-id path is
24+
* silently discarded, so it rides `multi: true` too. Its audit op is
25+
* `updateMany` and its spy below records `SqlDriver.updateMany`. Its
26+
* CLASSIFICATION is unchanged — declared global, now via
27+
* `dispatcherAckCasOptions` — which is the point of pinning the two
28+
* separately: the op moved, the warrant did not. Of the three sites only
29+
* `SqlHttpOutbox.ack` still writes by id.
30+
*
2131
* The `ack` pair is declared global (`dispatcherAckOptions`, warrant in
2232
* `outbox-dispatcher-scope.ts`). `redeliver` is NOT: it is served to any
2333
* authenticated user, so it threads the caller's tenant instead. ⛔ A
@@ -74,14 +84,14 @@ let driver: SqlDriver;
7484
let warns: Array<{ msg: string; meta: any }>;
7585
/** Every `options` bag that reached `SqlDriver.update` — the `update` op only. */
7686
let driverUpdates: Array<{ object: string; id: unknown; options: any }>;
77-
/** Every `options` bag that reached `SqlDriver.updateMany` — `redeliver`'s op since #11009. */
87+
/** Every `options` bag that reached `SqlDriver.updateMany` — `redeliver`'s op since #11009, and the notification `ack`'s since #11453. */
7888
let driverUpdateManys: Array<{ object: string; where: unknown; options: any }>;
7989

8090
/** The audit line for the SINGLE-RECORD op, matched on object + op. */
8191
const auditedUpdate = (object: string): boolean =>
8292
warns.some((w) => w.msg.includes(`[tenant-audit] update on tenant-scoped object "${object}"`));
8393

84-
/** The audit line for the PREDICATE op — `redeliver`'s write since #11009. */
94+
/** The audit line for the PREDICATE op — `redeliver`'s write since #11009, the notification `ack`'s since #11453. */
8595
const auditedUpdateMany = (object: string): boolean =>
8696
warns.some((w) => w.msg.includes(`[tenant-audit] updateMany on tenant-scoped object "${object}"`));
8797

@@ -211,7 +221,7 @@ async function seedDeadRow(id: string, org: string): Promise<void> {
211221
}
212222

213223
// ───────────────────────────────────────────────────────────────────────────
214-
describe('ack — the two dispatcher sites are a classified global sweep (update op)', () => {
224+
describe('ack — the two dispatcher sites are a classified global sweep (update + updateMany ops)', () => {
215225
it('SqlHttpOutbox.ack records a REAL delivery in every organization, without a finding', async () => {
216226
// The gate's own precondition: this object really is tenant-scoped.
217227
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
279289
'org_b:success:1',
280290
]);
281291
// ② Declared global, for both organizations' rows.
282-
const ackWrites = driverUpdates.filter((u) => u.object === DELIVERY_OBJECT);
292+
//
293+
// [#11453] The ack's op is `updateMany` now, so the reading moves to
294+
// that spy. The claim path writes there too (its reap and its atomic
295+
// claim), so the filter names what an ACK write looks like — and that
296+
// predicate is not incidental: `{ id: <scalar>, status: 'in_flight' }`
297+
// IS the compare-and-set, so matching on it pins that the ack reached
298+
// the driver CONDITIONAL rather than as a blind by-id write.
299+
const ackWrites = driverUpdateManys.filter(
300+
(u) => u.object === DELIVERY_OBJECT
301+
&& typeof (u.where as any)?.id === 'string'
302+
&& (u.where as any)?.status === 'in_flight',
303+
);
283304
expect(ackWrites).toHaveLength(2);
284305
expect(ackWrites.every((u) => u.options?.bypassTenantAudit === true)).toBe(true);
285-
expect(auditedUpdate(DELIVERY_OBJECT)).toBe(false);
306+
expect(auditedUpdateMany(DELIVERY_OBJECT)).toBe(false);
286307

287-
await controlUnscopedUpdate(DELIVERY_OBJECT, idA);
308+
await controlUnscopedUpdateMany(DELIVERY_OBJECT, idA);
288309
});
289310
});
290311

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

Lines changed: 38 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
22

33
import type { MessagingChannel, MessagingChannelContext, Notification, SendResult } from './channel.js';
4-
import type { INotificationOutbox, NotificationDeliveryRecord } from './outbox.js';
4+
import type { AckResult, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js';
55
import { classifyDeliveryAttempt } from './backoff.js';
66
import { renderDigest } from './digest-render.js';
77

@@ -207,7 +207,7 @@ export class NotificationDispatcher {
207207
const channel = this.opts.channels.getChannel(channelName);
208208
if (!channel) {
209209
for (const row of rows) {
210-
await this.opts.outbox.ack(row.id, { success: false, error: `channel '${channelName}' not registered`, dead: true });
210+
await this.ackAttempt(row, { success: false, error: `channel '${channelName}' not registered`, dead: true });
211211
this.opts.onAttempt?.(row, false);
212212
}
213213
return;
@@ -237,7 +237,7 @@ export class NotificationDispatcher {
237237
const now = this.opts.now?.() ?? Date.now();
238238
for (const row of rows) {
239239
const ack = classifyDeliveryAttempt(result, errorClass, row.attempts, now, this.opts.rng);
240-
await this.opts.outbox.ack(row.id, ack);
240+
await this.ackAttempt(row, ack);
241241
this.opts.onAttempt?.(row, result.ok);
242242
}
243243
}
@@ -246,7 +246,7 @@ export class NotificationDispatcher {
246246
const channel = this.opts.channels.getChannel(row.channel);
247247
if (!channel) {
248248
// No transport for this channel → terminal, observable on the row.
249-
await this.opts.outbox.ack(row.id, {
249+
await this.ackAttempt(row, {
250250
success: false,
251251
error: `channel '${row.channel}' not registered`,
252252
dead: true,
@@ -283,9 +283,42 @@ export class NotificationDispatcher {
283283
const errorClass = !result.ok && channel.classifyError ? channel.classifyError(result.error) : undefined;
284284
const now = this.opts.now?.() ?? Date.now();
285285
const ack = classifyDeliveryAttempt(result, errorClass, row.attempts, now, this.opts.rng);
286-
await this.opts.outbox.ack(row.id, ack);
286+
await this.ackAttempt(row, ack);
287287
this.opts.onAttempt?.(row, result.ok);
288288
}
289+
290+
/**
291+
* [#11453] Record one attempt's outcome, tolerating the ONE refusal a
292+
* correct dispatcher can legitimately provoke.
293+
*
294+
* `ack()` now refuses a row that is not `in_flight`, and this loop can meet
295+
* that honestly: a send slower than `claimTtlMs` lets the visibility-timeout
296+
* reap return the row to `pending` for another node, so by the time we ack,
297+
* the row is not ours. That is a race we are ALLOWED to lose — the delivery
298+
* is re-driven by whoever holds the row now, which is what at-least-once
299+
* means — and the refusal is the outbox correctly declining to overwrite
300+
* someone else's state.
301+
*
302+
* ⛔ What it must not do is abort the tick. The rows still validly claimed
303+
* by this node are processed after this one; letting a lost race unwind the
304+
* partition loop would strand every one of them `in_flight` until their own
305+
* timeouts expire, turning one lost race into a batch-wide delay.
306+
*
307+
* Only `DELIVERY_NOT_ELIGIBLE` is absorbed. A store fault is not a lost
308+
* race and still propagates to `runTick`'s handler.
309+
*/
310+
private async ackAttempt(row: NotificationDeliveryRecord, result: AckResult): Promise<void> {
311+
try {
312+
await this.opts.outbox.ack(row.id, result);
313+
} catch (err) {
314+
if ((err as { code?: string })?.code !== 'DELIVERY_NOT_ELIGIBLE') throw err;
315+
this.opts.logger?.warn?.('notification-dispatcher: ack refused, claim no longer held', {
316+
nodeId: this.opts.nodeId,
317+
deliveryId: row.id,
318+
error: (err as Error)?.message ?? String(err),
319+
});
320+
}
321+
}
289322
}
290323

291324
/** Group claimed digest rows by their `digestKey` (insertion order preserved). */

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,9 @@ export type {
9797
ClaimOptions,
9898
AckResult,
9999
} from './outbox.js';
100+
// [#11453] `ack()`'s status precondition refuses with this, so a caller that
101+
// wants to distinguish "I lost the claim" from a transport fault can catch it.
102+
export { NotificationAckError } from './outbox.js';
100103
export { SqlNotificationOutbox, DELIVERY_OBJECT } from './sql-outbox.js';
101104
export type { SqlNotificationOutboxOptions } from './sql-outbox.js';
102105
export { MemoryNotificationOutbox } from './memory-outbox.js';

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

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import type {
99
INotificationOutbox,
1010
NotificationDeliveryRecord,
1111
} from './outbox.js';
12+
import { NotificationAckError, notificationAckNotClaimedMessage } from './outbox.js';
1213
import { hashPartition } from './backoff.js';
1314

1415
/**
@@ -117,8 +118,26 @@ export class MemoryNotificationOutbox implements INotificationOutbox {
117118

118119
async ack(id: string, result: AckResult): Promise<void> {
119120
const r = this.rows.get(id);
121+
// An id matching no row is not a contract violation: there is no state
122+
// to corrupt and no claim to lose. Unchanged, and declared on the
123+
// interface so the two backends agree about it.
120124
if (!r) return;
125+
// [#11453] The status precondition. `ack` completes a delivery this
126+
// caller CLAIMED; an unclaimed `pending` row (the ack-as-cancel trap)
127+
// or an already-terminal one is refused, and nothing below runs — so a
128+
// refused ack leaves status, attempts and error exactly as they were.
129+
// Single-threaded, so this test and the mutation are already one atomic
130+
// step; `SqlNotificationOutbox` spells the same guard as a conditional
131+
// UPDATE because it is not.
132+
if (r.status !== 'in_flight') {
133+
throw new NotificationAckError(
134+
notificationAckNotClaimedMessage(id, r.status),
135+
'DELIVERY_NOT_ELIGIBLE',
136+
);
137+
}
121138
const now = this.clock();
139+
// Reached only for a genuinely claimed row, so this counts a real
140+
// dispatch attempt and nothing else (#11453).
122141
r.attempts += 1;
123142
r.lastAttemptedAt = now;
124143
r.claimedBy = undefined;

0 commit comments

Comments
 (0)