From 42d917d0c24805fa96f6fec700f2b64e69dfecaa Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 12 Aug 2026 01:40:28 +0000 Subject: [PATCH] fix(services): notify's run summary stops claiming a delivery that dead-lettered (#7747) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stack booted without the `push` channel registered, running a flow whose notify node targets `['push']`, produced two operator-facing records that contradicted each other: `sys_notification_delivery` held `status: 'dead'`, `error: "channel 'push' not registered"`, while the flow-run summary reported `status: 'success', acted: 1`. The seam is `EmitResult.delivered`. With the durable outbox in play (ADR-0030 P1), `emit()` returns as soon as the `(recipient x channel)` rows are enqueued and the dispatcher decides the outcome afterwards — but `delivered` counted those enqueued rows under a name that says they arrived, and `notify` fed the number straight into `acted`. A count minted before any send attempt then survived the dead-letter unrevised; nothing ever revisits it. - `EmitResult` separates the two counts. `delivered` now means a channel ACCEPTED the delivery — terminal and observed, which only the inline (P0) fan-out can report. New `enqueued` carries the outbox path's accepted rows: durable, unsent, outcome pending on `sys_notification_delivery`. - The notify node counts only delivered toward `acted`, and reports `unmeasuredEffect` when deliveries are merely enqueued — the qualifier a `connector_action` already uses for an effect the platform cannot count, and deliberately not a bare `acted: 0`, which would claim the run did nothing. The broken-sweep alert is `selected > 0 AND acted = 0 AND unmeasured = 0`, so a pending delivery suppresses the alert without asserting success. Node output gains `enqueued` next to `delivered` and `notificationId`. The run still reports `success`: the flow did everything it can do synchronously, and failing it would let a channel registered a moment later retroactively break the flow. Notify must not block on a downstream channel, so "delivered" is not a claim it is ever positioned to make — it simply stops making it. Tests wire the REAL MessagingService + NotificationDispatcher behind the notify node and assert on the two durable records (folded run summary, outbox row), not on call counts — the finding is that those records disagree. Reverse- verified: on origin/main the durable assertions pass and the summary asserts `acted: 1, unmeasured: 0`. Pin updated deliberately: `messaging-service.test.ts` asserted `delivered: 2 // 2 enqueued (accepted)` — the conflation written down — now `enqueued: 2, delivered: 0`. `connector-nodes.test.ts:292` is unaffected (it pins connector, not notify, accounting) and stays green. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LGwDLmaML1LtLmQ4F4Aq7z --- .changeset/notify-acted-dead-letter.md | 42 ++++ .../services/service-automation/package.json | 1 + ...otify-delivery-outcome.integration.test.ts | 194 ++++++++++++++++++ .../src/builtin/notify-node.ts | 50 ++++- .../src/messaging-service.test.ts | 13 +- .../src/messaging-service.ts | 53 ++++- pnpm-lock.yaml | 3 + 7 files changed, 345 insertions(+), 11 deletions(-) create mode 100644 .changeset/notify-acted-dead-letter.md create mode 100644 packages/services/service-automation/src/builtin/notify-delivery-outcome.integration.test.ts diff --git a/.changeset/notify-acted-dead-letter.md b/.changeset/notify-acted-dead-letter.md new file mode 100644 index 0000000000..e14f2ac224 --- /dev/null +++ b/.changeset/notify-acted-dead-letter.md @@ -0,0 +1,42 @@ +--- +"@objectstack/service-messaging": patch +"@objectstack/service-automation": patch +--- + +fix(services): a notify flow-run summary no longer reports a delivery the delivery record dead-lettered (#7747) + +Boot a stack without the `push` channel registered, fire a flow whose `notify` +node targets `['push']`, and the two records an operator can read **contradicted +each other**: `sys_notification_delivery` held `status: 'dead'`, +`error: "channel 'push' not registered"`, while the flow-run summary said +`status: 'success', acted: 1`. Nothing was delivered, and the surface built to +answer "did this sweep actually do anything" (#4354) said it had. + +The seam is `EmitResult.delivered`. With the durable outbox in play (ADR-0030 +P1), `emit()` returns as soon as the `(recipient × channel)` rows are enqueued — +the dispatcher sends and decides the outcome afterwards — but `delivered` +counted those *enqueued* rows anyway, under a name that says they arrived. The +`notify` node then fed that number straight into `acted`, so a count minted +before any send attempt survived unrevised through the dead-letter. It was never +a "stale by a moment" number either: nothing ever revisits it. + +- `EmitResult` now separates the two. `delivered` means a channel **accepted** + the delivery — a terminal, observed outcome, which only the inline (P0) + fan-out can report. New `enqueued` carries the outbox path's accepted rows: + durable, unsent, outcome pending on `sys_notification_delivery`. +- The `notify` node counts only what was delivered toward `acted`. When + deliveries are merely enqueued it reports `unmeasuredEffect` instead — the + qualifier a `connector_action` already uses for an effect the platform cannot + count, and deliberately **not** a bare `acted: 0`, which would claim the run + did nothing. The broken-sweep alert is + `selected > 0 AND acted = 0 AND unmeasured = 0`, so a pending delivery + suppresses the alert without asserting success. The node's output gains + `enqueued` alongside `delivered` and `notificationId`. + +The run still reports `success`: the flow did everything it can do +synchronously, and failing it would let a channel registered a moment later +retroactively break the flow. Notify does not block a flow on a downstream +channel, so "delivered" is not a claim it is ever in a position to make — what +changes is that it no longer makes it. Inline (P0) fan-out is untouched: it has +the channel's answer by the time `emit()` returns, so `acted` stays a real +measurement there, including the measured zero for an unregistered channel. diff --git a/packages/services/service-automation/package.json b/packages/services/service-automation/package.json index 359e5913a4..2bf0642ddd 100644 --- a/packages/services/service-automation/package.json +++ b/packages/services/service-automation/package.json @@ -28,6 +28,7 @@ "@objectstack/objectql": "workspace:*", "@objectstack/plugin-security": "workspace:*", "@objectstack/service-job": "workspace:*", + "@objectstack/service-messaging": "workspace:*", "@types/node": "^26.1.2", "typescript": "^6.0.3", "vitest": "^4.1.10" diff --git a/packages/services/service-automation/src/builtin/notify-delivery-outcome.integration.test.ts b/packages/services/service-automation/src/builtin/notify-delivery-outcome.integration.test.ts new file mode 100644 index 0000000000..b7408afbc3 --- /dev/null +++ b/packages/services/service-automation/src/builtin/notify-delivery-outcome.integration.test.ts @@ -0,0 +1,194 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +import { describe, it, expect } from 'vitest'; +import { + MessagingService, + MemoryNotificationOutbox, + NotificationDispatcher, +} from '@objectstack/service-messaging'; +import type { MessagingChannel } from '@objectstack/service-messaging'; +import { AutomationEngine } from '../engine.js'; +import { registerNotifyNode } from './notify-node.js'; + +/** + * #7747 — the run summary an operator reads must not claim a delivery that + * `sys_notification_delivery` records as dead. + * + * The QA repro verbatim: boot WITHOUT the `push` channel registered, fire a + * flow whose notify node targets `['push']`, then read the run summary and the + * delivery record. This wires the REAL `MessagingService` (outbox-backed, P1) + * and the REAL `NotificationDispatcher` behind the notify node rather than a + * fake, because the defect lives in the seam BETWEEN them: `emit()` returns + * once the row is enqueued and the dispatcher decides the outcome afterwards, + * so a fake that answers `emit()` in one shot cannot express the disagreement + * at all. + * + * The assertions are deliberately on the two DURABLE operator-facing records — + * the folded run summary and the outbox row — not on how many times anything + * was called: the finding is precisely that those two records contradict each + * other, so an internal call-count assertion would pass while the defect stands. + * + * On `origin/main` the first test fails with `acted: 1` — the notify node + * counts `EmitResult.delivered`, which in outbox mode is an ENQUEUED count. + */ + +function silentLogger(): any { + const l: any = { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} }; + l.child = () => l; + return l; +} + +/** A channel that records what it was handed, so a real send is distinguishable. */ +function recordingChannel(id: string): { channel: MessagingChannel; sent: unknown[] } { + const sent: unknown[] = []; + return { + sent, + channel: { + id, + async send(_ctx, delivery) { + sent.push(delivery); + return { ok: true }; + }, + }, + }; +} + +/** Wire the notify node against a given messaging service. */ +function engineWith(messaging: MessagingService): AutomationEngine { + const engine = new AutomationEngine(silentLogger()); + registerNotifyNode(engine, { + logger: silentLogger(), + getService: (name: string) => (name === 'messaging' ? messaging : undefined), + } as any); + return engine; +} + +/** + * A stack booted the way the repro describes: messaging present and + * outbox-backed (P1), with only the channels named here registered. + */ +function bootOutboxStack(registered: MessagingChannel[]) { + const outbox = new MemoryNotificationOutbox(1); + const messaging = new MessagingService({ logger: silentLogger(), outbox }); + for (const c of registered) messaging.registerChannel(c); + + const dispatcher = new NotificationDispatcher({ + nodeId: 'node-test', + outbox, + channels: messaging, + channelContext: { logger: silentLogger() }, + partitionCount: 1, + intervalMs: 10_000, // ticks are driven manually + }); + + return { outbox, messaging, dispatcher, engine: engineWith(messaging) }; +} + +function notifyFlow(channels: string[]) { + return { + name: 'nudge', + label: 'Nudge', + type: 'autolaunched' as const, + nodes: [ + { id: 'start', type: 'start' as const, label: 'Start' }, + { + id: 'notify', + type: 'notify' as const, + label: 'Notify', + config: { recipients: ['user_1'], title: 'Renewal due', message: 'Ping', channels }, + }, + { id: 'end', type: 'end' as const, label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'notify' }, + { id: 'e2', source: 'notify', target: 'end' }, + ], + }; +} + +describe('notify run summary vs. the durable delivery record (#7747)', () => { + it('does not report a countable act for a delivery that dead-letters on an unregistered channel', async () => { + // 1) Boot without the `push` channel registered. + const { outbox, dispatcher, engine } = bootOutboxStack([recordingChannel('inbox').channel]); + + // 2) Fire a flow whose notify node targets ['push']. + engine.registerFlow('nudge', notifyFlow(['push'])); + const run = await engine.execute('nudge'); + + // 3a) The durable record: the dispatcher dead-letters the row, because + // no transport for `push` exists. + await dispatcher.tick(); + const rows = await outbox.list(); + expect(rows).toHaveLength(1); + expect(rows[0].channel).toBe('push'); + expect(rows[0].status).toBe('dead'); + expect(rows[0].error).toContain("channel 'push' not registered"); + + // 3b) The record an operator reads. The run still SUCCEEDS — the flow + // did everything it can do synchronously, and failing it would make + // a channel that registers a moment later retroactively break the + // flow. What must not survive is the claim that it DELIVERED: + // `acted` is the count the broken-sweep alert trusts, and the honest + // answer at the moment the run settles is "an effect I cannot count + // yet" — which the platform already spells `unmeasured`, and which + // is not the same as `acted: 0` alone (that would claim the run did + // nothing, and trip the alert on every healthy notify). + expect(run.success).toBe(true); + expect(run.summary).toMatchObject({ acted: 0, unmeasured: 1 }); + + // The finding itself, as one assertion: the summary must not out-count + // what the durable record shows was actually delivered (here: nothing). + const notDead = rows.filter((r) => r.status !== 'dead').length; + expect(run.summary!.acted).toBeLessThanOrEqual(notDead); + }); + + it('reports the same uncountable effect for a channel that IS registered — the outcome is simply not known yet', async () => { + // The counterpart that stops the fix from degenerating into "unregistered + // channels are special": at the moment the run settles, a healthy + // outbox-backed delivery is equally unsent. What separates the two cases + // is the outbox row — which is exactly where `unmeasured` points. + const inbox = recordingChannel('inbox'); + const { outbox, dispatcher, engine } = bootOutboxStack([inbox.channel]); + + engine.registerFlow('nudge', notifyFlow(['inbox'])); + const run = await engine.execute('nudge'); + + expect(run.success).toBe(true); + expect(run.summary).toMatchObject({ acted: 0, unmeasured: 1 }); + // Nothing had been sent when the run settled… + expect(inbox.sent).toHaveLength(0); + // …and the delivery lands afterwards, on the record that owns the truth. + await dispatcher.tick(); + expect(inbox.sent).toHaveLength(1); + expect((await outbox.list())[0].status).toBe('success'); + }); + + it('still reports a countable act when the messaging stack delivers inline (no outbox)', async () => { + // The inline (P0) path really does know the outcome by the time `emit()` + // returns, so `acted` stays a measurement there — the fix narrows what + // `acted` may claim, it does not blanket every notify as unmeasurable. + const inbox = recordingChannel('inbox'); + const messaging = new MessagingService({ logger: silentLogger() }); + messaging.registerChannel(inbox.channel); + const engine = engineWith(messaging); + + engine.registerFlow('nudge', notifyFlow(['inbox'])); + const run = await engine.execute('nudge'); + + expect(inbox.sent).toHaveLength(1); + expect(run.summary).toMatchObject({ acted: 1, unmeasured: 0 }); + }); + + it('an inline send to an unregistered channel is a measured zero, not an unmeasured shrug', async () => { + // Inline fan-out DOES observe "channel not registered" synchronously, so + // that run is correctly eligible for the broken-sweep alert. + const messaging = new MessagingService({ logger: silentLogger() }); + messaging.registerChannel(recordingChannel('inbox').channel); + const engine = engineWith(messaging); + + engine.registerFlow('nudge', notifyFlow(['push'])); + const run = await engine.execute('nudge'); + + expect(run.summary).toMatchObject({ acted: 0, unmeasured: 0 }); + }); +}); diff --git a/packages/services/service-automation/src/builtin/notify-node.ts b/packages/services/service-automation/src/builtin/notify-node.ts index 6f435bea65..c7d47a83f3 100644 --- a/packages/services/service-automation/src/builtin/notify-node.ts +++ b/packages/services/service-automation/src/builtin/notify-node.ts @@ -26,7 +26,23 @@ export interface MessagingServiceSurface { source?: { object: string; id: string }; actorId?: string; channels?: string[]; - }): Promise<{ notificationId: string; delivered: number; failed: number }>; + }): Promise<{ + notificationId: string; + /** Deliveries a channel ACCEPTED — a terminal, observed outcome. */ + delivered: number; + failed: number; + /** + * Deliveries durably accepted into the messaging outbox but NOT yet + * attempted; their real outcome lands on `sys_notification_delivery` + * afterwards (#7747). + * + * Optional because this is a STRUCTURAL mirror of a service resolved at + * runtime: an older or third-party messaging implementation may not + * report it, and absent reads as "nothing is in flight" — which is the + * only answer such a stack could honestly give. + */ + enqueued?: number; + }>; } /** @@ -279,18 +295,46 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext) actorId, channels: channels.length ? channels : undefined, }); + const delivered = Number(result.delivered) || 0; + const enqueued = Number(result.enqueued) || 0; return { success: true, output: { notificationId: result.notificationId, - delivered: result.delivered, + delivered, + // Surfaced so a flow author templating the outcome can + // tell "sent" from "handed to the outbox", and so the + // notification id above has a stated reason to be + // followed into `sys_notification_delivery`. + enqueued, failed: result.failed, }, // A notification IS the action for a nudge/alert sweep, so it // counts toward `acted` (#4354) — otherwise the flow whose // whole job is to notify would report acting on nothing, and // the broken-sweep detector would fire on every healthy run. - metrics: { acted: Number(result.delivered) || 0 }, + // + // But only a delivery a channel ACCEPTED is countable. With + // the outbox in play (ADR-0030 P1) `emit()` returns once the + // rows are durable and the dispatcher decides the outcome + // afterwards — including dead-lettering an unregistered + // channel — so counting the enqueue as `acted` made the run + // summary assert a delivery that `sys_notification_delivery` + // recorded as `dead` (#7747). The honest answer at the moment + // the run settles is "an effect I cannot count yet", which is + // exactly `unmeasuredEffect` — the same qualifier a + // `connector_action` uses, and pointedly NOT a bare + // `acted: 0`, which would claim the run did nothing and trip + // the broken-sweep alert on every healthy notify. The alert + // is `selected > 0 AND acted = 0 AND unmeasured = 0`, so a + // pending delivery correctly suppresses it while refusing to + // claim success. + // + // Waiting for the real outcome is not on the table: a notify + // node must not block a flow on a downstream channel. + metrics: enqueued > 0 + ? { ...(delivered > 0 ? { acted: delivered } : {}), unmeasuredEffect: true } + : { acted: delivered }, }; } catch (err) { return { success: false, error: `notify failed: ${(err as Error).message}` }; diff --git a/packages/services/service-messaging/src/messaging-service.test.ts b/packages/services/service-messaging/src/messaging-service.test.ts index a99be6fb18..d78de5d174 100644 --- a/packages/services/service-messaging/src/messaging-service.test.ts +++ b/packages/services/service-messaging/src/messaging-service.test.ts @@ -101,6 +101,10 @@ describe('MessagingService', () => { expect(inbox.seen[0].notification.title).toBe('Deal closed'); expect(result.delivered).toBe(2); expect(result.failed).toBe(0); + // Inline fan-out leaves nothing in flight — the counterpart to the + // outbox pin below, and what keeps `delivered` a terminal count on + // BOTH paths rather than a name two things share (#7747). + expect(result.enqueued).toBe(0); expect(result.notificationId).toMatch(/^evt_/); // synthesized w/o data layer expect(result.deliveries[0]).toMatchObject({ channel: 'inbox', recipient: 'user_1', ok: true, externalId: 'row_1' }); }); @@ -308,7 +312,14 @@ describe('MessagingService', () => { // Nothing sent inline — the dispatcher owns the send. expect(inbox.seen).toHaveLength(0); - expect(result.delivered).toBe(2); // 2 enqueued (accepted) + // …so nothing is DELIVERED yet, and the result says so (#7747). This + // pin used to read `delivered: 2` with the comment "2 enqueued + // (accepted)" — the conflation itself, written down: callers were + // handed an enqueue count under the name `delivered`, and it stayed + // put when the dispatcher later dead-lettered the row. + expect(result.enqueued).toBe(2); + expect(result.delivered).toBe(0); + expect(result.failed).toBe(0); const rows = await outbox.list(); expect(rows).toHaveLength(2); expect(rows.every((r) => r.status === 'pending')).toBe(true); diff --git a/packages/services/service-messaging/src/messaging-service.ts b/packages/services/service-messaging/src/messaging-service.ts index bd6d1ee615..24112ca477 100644 --- a/packages/services/service-messaging/src/messaging-service.ts +++ b/packages/services/service-messaging/src/messaging-service.ts @@ -90,7 +90,33 @@ export interface EmitResult { /** True when `dedupKey` matched an existing event and fan-out was skipped. */ readonly deduped: boolean; readonly deliveries: DeliveryOutcome[]; + /** + * Deliveries a channel ACCEPTED — a terminal, observed outcome. + * + * Only the inline (P0) fan-out can produce a non-zero count here, because + * only it has a channel's answer by the time `emit()` returns. The + * outbox-backed (P1) path reports its accepted rows as {@link enqueued} + * instead: it has handed the work to the dispatcher and genuinely does not + * yet know whether it will land. + * + * This field used to carry the enqueued count too (#7747), which read as + * "1 delivered" to every caller while the row was still pending — and + * stayed 1 after the dispatcher dead-lettered it. A caller that reports + * `delivered` to an operator was therefore contradicting + * `sys_notification_delivery`; keeping the two counts apart is what makes + * "delivered" mean delivered. + */ readonly delivered: number; + /** + * Deliveries durably accepted into the outbox but NOT yet attempted — the + * P1 path's success count. Their real outcome (`success` / `failed` / + * `dead` / `suppressed`) lands on the `sys_notification_delivery` row + * afterwards, which is the record that owns that truth. + * + * Non-zero means "this much is in flight, ask the delivery record how it + * went" — never "this much arrived". + */ + readonly enqueued: number; readonly failed: number; } @@ -575,7 +601,7 @@ export class MessagingService { this.ctx.logger.info( `[messaging] emit: dedupKey '${input.dedupKey}' already emitted (${existing}); skipping`, ); - return { notificationId: existing, deduped: true, deliveries: [], delivered: 0, failed: 0 }; + return { notificationId: existing, deduped: true, deliveries: [], delivered: 0, enqueued: 0, failed: 0 }; } } @@ -598,7 +624,7 @@ export class MessagingService { this.ctx.logger.info( `[messaging] emit: dedupKey '${input.dedupKey}' raced; converged to ${winner}`, ); - return { notificationId: winner, deduped: true, deliveries: [], delivered: 0, failed: 0 }; + return { notificationId: winner, deduped: true, deliveries: [], delivered: 0, enqueued: 0, failed: 0 }; } } throw err; @@ -611,7 +637,7 @@ export class MessagingService { }); if (recipients.length === 0) { this.ctx.logger.warn(`[messaging] emit: topic '${input.topic}' resolved to 0 recipients`); - return { notificationId, deduped: false, deliveries: [], delivered: 0, failed: 0 }; + return { notificationId, deduped: false, deliveries: [], delivered: 0, enqueued: 0, failed: 0 }; } // 3b) Preference filter (ADR-0030 P2): drop the (recipient × channel) @@ -625,14 +651,25 @@ export class MessagingService { }); if (targets.length === 0) { this.ctx.logger.info(`[messaging] emit: topic '${input.topic}' suppressed for all recipients by preference`); - return { notificationId, deduped: false, deliveries: [], delivered: 0, failed: 0 }; + return { notificationId, deduped: false, deliveries: [], delivered: 0, enqueued: 0, failed: 0 }; } // 4) Either enqueue durable deliveries (P1 outbox) or fan out inline (P0). if (this.outbox) { const deliveries = await this.enqueueDeliveries(this.outbox, notificationId, targets, input, payload); - const delivered = deliveries.filter((d) => d.ok).length; - return { notificationId, deduped: false, deliveries, delivered, failed: deliveries.length - delivered }; + // An accepted row here is ENQUEUED, not delivered — the dispatcher + // decides that later and records it on `sys_notification_delivery`. + // Reporting it as `delivered` is what let a run summary claim a + // delivery the delivery record went on to mark `dead` (#7747), so + // the two counts stay apart: `delivered` is 0 on this path by + // construction, and there is no moment at which it becomes non-zero + // retroactively. `failed` keeps its meaning — an enqueue that threw + // never reached the outbox at all. + const enqueued = deliveries.filter((d) => d.ok).length; + return { + notificationId, deduped: false, deliveries, + delivered: 0, enqueued, failed: deliveries.length - enqueued, + }; } const notification: Notification = { @@ -648,8 +685,10 @@ export class MessagingService { payload: input.payload, }; + // Inline (P0): every channel has already answered, so `delivered` is a + // real terminal count and nothing is left in flight. const { deliveries, delivered, failed } = await this.fanOut(notification, targets); - return { notificationId, deduped: false, deliveries, delivered, failed }; + return { notificationId, deduped: false, deliveries, delivered, enqueued: 0, failed }; } /** diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f805b00004..e00b45736e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2143,6 +2143,9 @@ importers: '@objectstack/service-job': specifier: workspace:* version: link:../service-job + '@objectstack/service-messaging': + specifier: workspace:* + version: link:../service-messaging '@types/node': specifier: ^26.1.2 version: 26.1.2