diff --git a/.changeset/great-jars-sleep.md b/.changeset/great-jars-sleep.md new file mode 100644 index 0000000000..aa1ecfa67b --- /dev/null +++ b/.changeset/great-jars-sleep.md @@ -0,0 +1,23 @@ +--- +'@objectstack/service-messaging': patch +--- + +fix(service-messaging): the durable fan-out refuses a channel nobody registered instead of writing a delivery row for it + +`MessagingService.emit()` on the reliable-delivery (outbox) path wrote one +`sys_notification_delivery` row per recipient for a channel the composition had +never registered, and the dispatcher dead-lettered every one of them on attempt +one. The inline path had always refused this case; only the durable path wrote +the rows, so a deployment whose flows notify on `['inbox','email']` without an +email plugin accumulated guaranteed-dead rows in the hot delivery table. + +The durable path now reports the same failed delivery outcome the inline path +reports — `ok: false`, `error: "channel '' not registered"`, counted in +`EmitResult.failed` — and writes no row. The refusal is logged once per channel +per emit with the number of rows it refused, not once per recipient. + +The refusal is deliberately **not** recorded in +`sys_notification.suppressed_channels`: that key answers "why can this tenant not +send on this channel", and an unregistered channel is a composition fact, +identical for every tenant in the process. The event row's column set is +unchanged. 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 index b7408afbc3..df34119444 100644 --- 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 @@ -28,8 +28,24 @@ import { registerNotifyNode } from './notify-node.js'; * 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. + * At the time #7747 landed, the first test failed with `acted: 1` — the notify + * node counted `EmitResult.delivered`, which in outbox mode is an ENQUEUED count. + * + * ## Amended by #18050 — the first case's durable record is now EMPTY + * + * #7747's repro boots without `push` registered, and back then the durable + * fan-out enqueued a row for it anyway that the dispatcher could only + * dead-letter. #18050 fixed that at the producer: `enqueueDeliveries` refuses an + * unregistered channel before it writes, reporting the same failed + * `DeliveryOutcome` the inline path always did. So the first test's scenario + * moved buckets — from "an effect I cannot count YET" (`unmeasured: 1`, the + * dispatcher decides later) to "an effect I have counted and it is zero" + * (`acted: 0, unmeasured: 0`, refused synchronously). + * + * ⛔ That is not this file's invariant weakening. #7747's invariant is "the + * summary must not out-count what the durable record shows was delivered", and + * it is asserted below against a bound that went from 0-non-dead-rows to + * 0-rows-at-all. What changed is the producer, not what is demanded of it. */ function silentLogger(): any { @@ -107,7 +123,7 @@ function notifyFlow(channels: string[]) { } 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 () => { + it('reports a MEASURED zero — not a countable act — for an unregistered channel on the durable path', async () => { // 1) Boot without the `push` channel registered. const { outbox, dispatcher, engine } = bootOutboxStack([recordingChannel('inbox').channel]); @@ -115,29 +131,44 @@ describe('notify run summary vs. the durable delivery record (#7747)', () => { 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. + // 3a) The durable record: NOTHING — and that is the #18050 change. + // This assertion used to read `toHaveLength(1)` + `status: 'dead'`: + // the durable fan-out enqueued a row for a channel with no transport + // and the dispatcher dead-lettered it on attempt ONE. That row was + // itself the defect #18050 fixed, so `enqueueDeliveries` now refuses + // the channel up front and writes no row at all. The tick is kept + // deliberately: it proves nothing APPEARS later either, which is a + // strictly stronger statement than the old "a row exists and is dead". 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"); + expect(rows).toHaveLength(0); // 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). + // did everything it can do synchronously. What must not survive is + // the claim that it DELIVERED. + // + // ⚠️ `unmeasured` moved 1 -> 0 here, and that is the POINT, not a + // relaxation. `unmeasuredEffect` means "the count is unknown because + // the dispatcher decides later". Since #18050 there is no later: the + // refusal is synchronous, so the count is KNOWN and it is zero — + // exactly the reading `notify-node.ts` demands ("this count is known + // and it is zero; claiming otherwise would take the run OUT of the + // broken-sweep filter ... on precisely the run that should be inside + // it"). `selected: 1, acted: 0, unmeasured: 0` puts this run INSIDE + // the `selected > 0 AND acted = 0 AND unmeasured = 0` alert, which is + // where a notify that reached nobody and never will belongs. + // + // It is also what makes the two fan-out paths agree: the inline case + // four tests down asserts this same triple and calls it "correctly + // eligible for the broken-sweep alert". The durable path is not a + // duplicate of it — it is the other side of the seam this file + // exists for, and it is the side that used to disagree. expect(run.success).toBe(true); - expect(run.summary).toMatchObject({ acted: 0, unmeasured: 1 }); + expect(run.summary).toMatchObject({ selected: 1, acted: 0, unmeasured: 0 }); - // The finding itself, as one assertion: the summary must not out-count - // what the durable record shows was actually delivered (here: nothing). + // The #7747 finding itself, unchanged in force: the summary must not + // out-count what the durable record shows was actually delivered. With + // no row at all the bound is 0, so this is tighter than it was before. const notDead = rows.filter((r) => r.status !== 'dead').length; expect(run.summary!.acted).toBeLessThanOrEqual(notDead); }); diff --git a/packages/services/service-messaging/src/channel-availability.test.ts b/packages/services/service-messaging/src/channel-availability.test.ts index c09afd1b21..2d8400f687 100644 --- a/packages/services/service-messaging/src/channel-availability.test.ts +++ b/packages/services/service-messaging/src/channel-availability.test.ts @@ -339,8 +339,14 @@ describe('channel availability at fan-out (#17732)', () => { it('leaves an UNREGISTERED channel on its existing path — ⛔ not folded into suppression', async () => { // Out of the ruling's scope on purpose: an unregistered channel has no - // implementation to ask, so it keeps today's behaviour exactly. Pinned - // so the boundary is deliberate rather than accidental. + // implementation to ask, so it is not a SUPPRESSION. Pinned so the + // boundary is deliberate rather than accidental. + // + // ⚠️ This is the INLINE path, and it is unchanged. #18050 later gave the + // DURABLE path the same answer this one already gave — a failed + // `DeliveryOutcome` and no row — so "keeps today's behaviour exactly" is + // no longer true of the outbox half; `unregistered-channel.test.ts` + // pins that half, including that it still records no suppression. const data = capturingEngine(); const service = new MessagingService({ logger: silentLogger(), getData: () => data.engine }); service.registerChannel(channelDouble('inbox').channel); diff --git a/packages/services/service-messaging/src/messaging-service.ts b/packages/services/service-messaging/src/messaging-service.ts index 92991ec310..27c346e821 100644 --- a/packages/services/service-messaging/src/messaging-service.ts +++ b/packages/services/service-messaging/src/messaging-service.ts @@ -879,8 +879,9 @@ export class MessagingService { * The single notification ingress. Writes the L2 event, resolves the * audience, and fans the result out to its channels. An unregistered * channel, or a channel that throws, is reported as a failed delivery — it - * never aborts the rest of the fan-out. A `dedupKey` that matches an - * existing event short-circuits: the event id is returned and no new + * never aborts the rest of the fan-out, and on the durable path it costs no + * `sys_notification_delivery` row either (#18050). A `dedupKey` that matches + * an existing event short-circuits: the event id is returned and no new * deliveries are produced. * * A channel that answers `isAvailable: { available: false }` for the tenant @@ -1036,11 +1037,14 @@ export class MessagingService { * default would mute every channel that has not been updated, which is a * far worse failure than the workless rows this exists to stop. * `channel-availability.test.ts` pins it from both sides. - * 2. **An UNREGISTERED channel is left alone.** It has no implementation to - * ask, so it keeps today's path exactly: the inline fan-out reports it as - * a failed delivery, the outbox enqueues a row the dispatcher - * dead-letters. That is a real, separate defect — it is filed, ⛔ not - * widened into this ruling. + * 2. **An UNREGISTERED channel is not answered here.** It has no + * implementation to ask, so this consult cannot reach it and its absence + * is ⛔ NOT a suppression: `suppressed_channels` answers "why can this + * TENANT not send", and an unregistered channel is a composition fact. + * Both fan-out paths refuse it instead, reporting one failed + * {@link DeliveryOutcome} per `(recipient × channel)` and writing no + * delivery row at all — inline in {@link MessagingService.fanOut}, on the + * durable path in {@link MessagingService.enqueueDeliveries} (#18050). * 3. **A throw is AVAILABLE.** Fail-open, matching the preference filter one * step down: a broken probe must degrade into today's behaviour, never * into a silent notification outage. Logged at `warn` — the degradation @@ -1085,6 +1089,28 @@ export class MessagingService { * dispatcher does the actual send + retry; here `ok` means "accepted for * delivery" (enqueued), not yet delivered — progress is observable on the * `sys_notification_delivery` row. + * + * ## An UNREGISTERED channel is refused here, not enqueued (#18050) + * + * A channel nobody registered has no transport to reach, so a row written + * for it is a row the dispatcher can only dead-letter on attempt ONE — + * `processRow` / `processDigestGroup` both ack `dead: true` the moment + * `getChannel()` answers nothing. Writing it costs an insert, a claim, an + * update and a retained terminal row per recipient, to record a fact known + * before the first write. + * + * The refusal is reported as the SAME failed {@link DeliveryOutcome} the + * inline path already produces for this case, so "nothing was sent and here + * is why" has one shape on both paths and the caller's `failed` count keeps + * its meaning. + * + * ⛔ NOT folded into `sys_notification.suppressed_channels`. That vocabulary + * answers "why can this TENANT not send on this channel" — a per-tenant + * configuration fact an operator filters and reports on. An unregistered + * channel is a COMPOSITION fact: identical for every tenant in the process, + * and fixed by mounting the channel, not by configuring the tenant. + * Recording it there would make a per-tenant report assert a + * deployment-wide misconfiguration. */ private async enqueueDeliveries( outbox: INotificationOutbox, @@ -1107,8 +1133,23 @@ export class MessagingService { actorId: input.actorId, }; const deliveries: DeliveryOutcome[] = []; + // [#18050] Counted, not logged in place: an emit to a 500-recipient + // audience would otherwise print 500 identical lines for one missing + // channel. Said ONCE per channel below, with the volume it refused — + // which is the number an operator needs to size the misconfiguration. + const refused = new Map(); for (const { recipient, channels, notBefore, digest } of targets) { for (const channel of channels) { + if (!this.channels.has(channel)) { + refused.set(channel, (refused.get(channel) ?? 0) + 1); + deliveries.push({ + channel, + recipient, + ok: false, + error: `channel '${channel}' not registered`, + }); + continue; + } try { const id = await outbox.enqueue({ notificationId, @@ -1130,6 +1171,13 @@ export class MessagingService { } } } + for (const [channel, count] of refused) { + this.ctx.logger.warn( + `[messaging] emit: channel '${channel}' is not registered; refused ${count} ` + + `delivery row(s) the dispatcher could only dead-letter. Register the channel ` + + `(or drop it from this notify's channel list) — nothing was sent on it.`, + ); + } return deliveries; } diff --git a/packages/services/service-messaging/src/unregistered-channel.test.ts b/packages/services/service-messaging/src/unregistered-channel.test.ts new file mode 100644 index 0000000000..1624c74578 --- /dev/null +++ b/packages/services/service-messaging/src/unregistered-channel.test.ts @@ -0,0 +1,254 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// [#18050] The durable (outbox) fan-out refuses a channel nobody registered +// instead of writing a `sys_notification_delivery` row for it. +// +// ## What was wrong +// +// `enqueueDeliveries` had no registration check at all: a `notify` naming a +// channel the composition never mounted produced ONE row per recipient, and the +// dispatcher dead-lettered every one of them on attempt ONE (`processRow` and +// `processDigestGroup` both ack `dead: true` the moment `getChannel()` answers +// nothing). The inline P0 path had checked this since forever; only the durable +// P1 path wrote the rows. That gap is the symptom #17732 opened with, and the +// ruling on #17732 (`isAvailable`) structurally cannot reach it — an +// unregistered channel has no implementation to ask. +// +// ## Why the facts below are pinned on ONE pass +// +// Each alone is satisfied by an implementation broken in a different direction: +// +// * "the unregistered channel got no row" — also true of a fan-out that +// enqueued NOTHING AT ALL. +// * "the registered channel got its rows" — also true of the old code. +// * "the caller was told" — also true of an implementation +// that reports the failure AND writes the row anyway. +// +// Only the conjunction, in one emit over one outbox, says the change +// DISCRIMINATES between the two channels. +// +// ## And the two boundaries that matter more than the feature +// +// 1. ⛔ It is NOT a suppression. `sys_notification.suppressed_channels` +// answers "why can this TENANT not send on this channel" — a per-tenant +// configuration fact. An unregistered channel is a COMPOSITION fact, +// identical for every tenant in the process. Folding it in would also +// re-open what #18041 settled: the key is written ONLY when something was +// actually suppressed, so the common path's column set never moves. +// 2. The refusal keeps the shape the INLINE path already used, so "nothing +// was sent and here is why" reads the same on both paths. Pinned by +// running the same emit through both and comparing the outcomes. + +import { describe, it, expect } from 'vitest'; +import { MessagingService } from './messaging-service.js'; +import { MemoryNotificationOutbox } from './memory-outbox.js'; +import type { Delivery, MessagingChannel } from './channel.js'; + +function silentLogger() { + return { info: () => {}, warn: () => {}, error: () => {} }; +} + +/** + * A data engine double that captures the `sys_notification` insert. + * + * Deliberately implements only what this path reaches — `insert` for the L2 + * event, `find` for the preference lookup. `emit()` without a `dedupKey` and + * with plain user-id recipients performs no `findOne`, and no `update` or + * `delete` at all, so the double carries none. + */ +function capturingEngine() { + const inserts: Array<{ object: string; row: Record }> = []; + return { + inserts, + engine: { + async insert(object: string, row: Record) { + inserts.push({ object, row }); + return { id: `evt_${inserts.length}`, ...row }; + }, + async find() { + return []; + }, + } as never, + }; +} + +function channelDouble(id: string): { channel: MessagingChannel; sent: Delivery[] } { + const sent: Delivery[] = []; + return { + sent, + channel: { + id, + async send(_ctx, delivery) { + sent.push(delivery); + return { ok: true }; + }, + }, + }; +} + +describe('an unregistered channel on the durable fan-out (#18050)', () => { + it('writes no delivery row for it, still enqueues the registered one, and tells the caller — one pass', async () => { + const outbox = new MemoryNotificationOutbox(1); + const data = capturingEngine(); + const service = new MessagingService({ + logger: silentLogger(), + outbox, + getData: () => data.engine, + }); + service.registerChannel(channelDouble('inbox').channel); + // `email` is NEVER registered — the reported composition: a deployment + // with no email plugin whose flows still notify on ['inbox','email']. + + const result = await service.emit({ + topic: 'deal.won', + audience: ['user_1', 'user_2'], + channels: ['inbox', 'email'], + organizationId: 'org_1', + payload: { title: 'Deal closed' }, + }); + + const rows = await outbox.list(); + + // (1) The unregistered channel got NO row — the whole point. Before this + // change there were two, both dead on attempt one. + expect(rows.filter((r) => r.channel === 'email')).toHaveLength(0); + // (2) … while the registered channel in the SAME fan-out got one per + // recipient. Without this, (1) is also satisfied by writing nothing. + expect(rows.filter((r) => r.channel === 'inbox').map((r) => r.recipientId).sort()) + .toEqual(['user_1', 'user_2']); + expect(rows).toHaveLength(2); + + // (3) The caller is told, per (recipient × channel), in the failed + // `DeliveryOutcome` shape — a two-channel emit reporting two rows is + // never indistinguishable from a fan-out bug. + expect(result.enqueued).toBe(2); + expect(result.failed).toBe(2); + expect( + result.deliveries + .filter((d) => d.channel === 'email') + .map((d) => ({ recipient: d.recipient, ok: d.ok, error: d.error })) + .sort((a, b) => a.recipient.localeCompare(b.recipient)), + ).toEqual([ + { recipient: 'user_1', ok: false, error: "channel 'email' not registered" }, + { recipient: 'user_2', ok: false, error: "channel 'email' not registered" }, + ]); + }); + + it('⛔ records NO suppression for it — the event row keeps the column set it had', async () => { + // The #18041 fence, from both sides: `suppressed_channels` is a + // per-tenant availability answer, and it is written ONLY when something + // was actually suppressed. An unregistered channel is neither, so the + // insert must not even NAME the column — a stack whose `sys_notification` + // predates it would answer INVALID_FIELD and lose the notification. + const outbox = new MemoryNotificationOutbox(1); + const data = capturingEngine(); + const service = new MessagingService({ + logger: silentLogger(), + outbox, + getData: () => data.engine, + }); + service.registerChannel(channelDouble('inbox').channel); + + const result = await service.emit({ + topic: 'deal.won', + audience: ['user_1'], + channels: ['inbox', 'nowhere'], + payload: { title: 'Deal closed' }, + }); + + expect(result.suppressed).toEqual([]); + expect(data.inserts).toHaveLength(1); + expect(Object.prototype.hasOwnProperty.call(data.inserts[0].row, 'suppressed_channels')).toBe(false); + expect(Object.keys(data.inserts[0].row).sort()).toEqual([ + 'actor_id', 'created_at', 'dedup_key', 'organization_id', + 'payload', 'severity', 'source_id', 'source_object', 'topic', + ]); + }); + + it('answers IDENTICALLY on the inline and the durable path — one shape for "nothing was sent and why"', async () => { + // The cross-path equality, executable. The two paths are different code + // (`fanOut` vs `enqueueDeliveries`), so only comparing their output + // proves an operator reading a failure report does not have to know + // which one produced it. + const emit = { topic: 'deal.won', audience: ['user_1', 'user_2'], channels: ['nowhere'], payload: { title: 'x' } }; + + const inlineData = capturingEngine(); + const inline = new MessagingService({ logger: silentLogger(), getData: () => inlineData.engine }); + inline.registerChannel(channelDouble('inbox').channel); + const inlineResult = await inline.emit(emit); + + const durableData = capturingEngine(); + const durable = new MessagingService({ + logger: silentLogger(), + outbox: new MemoryNotificationOutbox(1), + getData: () => durableData.engine, + }); + durable.registerChannel(channelDouble('inbox').channel); + const durableResult = await durable.emit(emit); + + const shape = (r: { deliveries: readonly { channel: string; recipient: string; ok: boolean; error?: string }[] }) => + r.deliveries + .map((d) => ({ channel: d.channel, recipient: d.recipient, ok: d.ok, error: d.error })) + .sort((a, b) => a.recipient.localeCompare(b.recipient)); + + expect(shape(durableResult)).toEqual(shape(inlineResult)); + expect(shape(inlineResult)).toEqual([ + { channel: 'nowhere', recipient: 'user_1', ok: false, error: "channel 'nowhere' not registered" }, + { channel: 'nowhere', recipient: 'user_2', ok: false, error: "channel 'nowhere' not registered" }, + ]); + // `failed` counts them on both paths; neither reports them as enqueued + // or delivered, so no summary can claim work that never existed. + expect([durableResult.failed, durableResult.enqueued, durableResult.delivered]).toEqual([2, 0, 0]); + expect([inlineResult.failed, inlineResult.enqueued, inlineResult.delivered]).toEqual([2, 0, 0]); + }); + + it('says it ONCE per channel per emit, with the volume — not once per recipient', async () => { + // The durable path is the high-volume one: a 500-recipient audience on + // one missing channel must not print 500 identical lines. The count is + // what sizes the misconfiguration, so it is part of the message. + const warnings: string[] = []; + const data = capturingEngine(); + const service = new MessagingService({ + logger: { ...silentLogger(), warn: (...a: unknown[]) => warnings.push(String(a[0])) }, + outbox: new MemoryNotificationOutbox(1), + getData: () => data.engine, + }); + service.registerChannel(channelDouble('inbox').channel); + + await service.emit({ + topic: 'deal.won', + audience: ['user_1', 'user_2', 'user_3', 'user_4'], + channels: ['inbox', 'email'], + payload: { title: 'x' }, + }); + + const lines = warnings.filter((w) => w.includes("channel 'email' is not registered")); + expect(lines).toHaveLength(1); + expect(lines[0]).toContain('refused 4 delivery row(s)'); + // Absence must be loud AND actionable: the line names the remedy. + expect(lines[0]).toContain('Register the channel'); + }); + + it('refuses every unregistered channel in one emit, independently', async () => { + const outbox = new MemoryNotificationOutbox(1); + const data = capturingEngine(); + const warnings: string[] = []; + const service = new MessagingService({ + logger: { ...silentLogger(), warn: (...a: unknown[]) => warnings.push(String(a[0])) }, + outbox, + getData: () => data.engine, + }); + service.registerChannel(channelDouble('inbox').channel); + + const result = await service.emit({ + topic: 'deal.won', + audience: ['user_1'], + channels: ['inbox', 'email', 'sms'], + payload: { title: 'x' }, + }); + + expect((await outbox.list()).map((r) => r.channel)).toEqual(['inbox']); + expect(result.deliveries.filter((d) => !d.ok).map((d) => d.channel).sort()).toEqual(['email', 'sms']); + expect(warnings.filter((w) => w.includes('is not registered'))).toHaveLength(2); + }); +});