From f5344e8697dfc57880bbbc996bf54c637d7cac5b Mon Sep 17 00:00:00 2001 From: os-warren Date: Fri, 21 Aug 2026 09:21:41 +0000 Subject: [PATCH 1/2] fix(messaging): classify the delivery dispatchers' updateMany sweeps as global (#10673) --- ...ery-claim-tenant-audit.integration.test.ts | 222 ++++++++++++++++++ .../src/outbox-dispatcher-scope.ts | 57 +++++ .../service-messaging/src/sql-http-outbox.ts | 18 +- .../service-messaging/src/sql-outbox.ts | 17 +- 4 files changed, 302 insertions(+), 12 deletions(-) create mode 100644 packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts create mode 100644 packages/services/service-messaging/src/outbox-dispatcher-scope.ts diff --git a/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts b/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts new file mode 100644 index 0000000000..baf3b7c13d --- /dev/null +++ b/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts @@ -0,0 +1,222 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #10673 — the delivery dispatchers' predicate writes are classified, not + * silenced. + * + * ## What was measured + * On a walled deployment (`OS_TENANCY_POSTURE=isolated`) the SQL driver's + * `auditMissingTenant` gate printed, for both delivery objects: + * + * [tenant-audit] updateMany on tenant-scoped object "sys_http_delivery" + * without options.tenantId — writes will not be tenant-isolated. + * [tenant-audit] updateMany on tenant-scoped object "sys_notification_delivery" + * without options.tenantId — writes will not be tenant-isolated. + * + * The audit is right that the writes are environment-wide. The fix is the + * classification it demands, not the quiet: each `multi: true` write on the + * claim path is declared a global dispatcher sweep (`bypassTenantAudit`), with + * the warrant in `outbox-dispatcher-scope.ts`. See that file for why a + * `tenantId` is not merely absent but unavailable and unwanted here. + * + * ## Why this harness rather than the composed boot + * The card's repro is an EE image booted under docker compose, which this + * checkout cannot run. What stands in its place is the instrument's OWN + * criterion, exercised end to end: a real `SqlDriver` on better-sqlite3, real + * `syncSchemas()` (so `organization_id` is really provisioned and + * `resolveTenantField` really answers), the real `OS_TENANCY_POSTURE` read, + * the production `SqlHttpOutbox` / `SqlNotificationOutbox`, and the driver's + * own logger as the assertion surface — the same substitution + * `sql-driver-tenant-audit-posture.test.ts` makes. + * + * ## The vacuity traps closed here, explicitly + * 1. **A green that means "the audit was never armed".** Every test that + * asserts silence first asserts the gate's own preconditions are live — + * `resolveTenantField(object) === 'organization_id'` — and then performs a + * deliberately unscoped `multi: true` write on the SAME object through the + * SAME driver and requires the warning to appear. Without that positive + * control an object that stopped being tenant-scoped, a posture that + * stopped resolving, or a typo in the matcher would all read as a fix. + * 2. **A "fix" that touches nothing.** Silence is cheap for an implementation + * that claims no rows. Every claim assertion pins the ROWS: both + * organizations' rows move, and their `organization_id` survives the write + * — the cross-organization reach is the operation's semantics, so a + * regression to per-organization scoping must go red here. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import { SqlHttpOutbox } from './sql-http-outbox.js'; +import { SqlNotificationOutbox, DELIVERY_OBJECT } from './sql-outbox.js'; +import { HttpDelivery, SYS_HTTP_DELIVERY } from './objects/http-delivery.object.js'; +import { NotificationDelivery } from './objects/notification-delivery.object.js'; + +const OLD_POSTURE = process.env.OS_TENANCY_POSTURE; +const OLD_AUDIT = process.env.OS_TENANT_AUDIT; + +let engine: ObjectQL; +let driver: SqlDriver; +let warns: Array<{ msg: string; meta: any }>; + +/** The audit line the card quotes, matched on object + op. */ +const auditedUpdateMany = (object: string): boolean => + warns.some((w) => w.msg.includes(`[tenant-audit] updateMany on tenant-scoped object "${object}"`)); + +beforeEach(async () => { + // The posture is read LIVE by `isMultiTenantMode()` (#5262), so setting it + // here really does arm the gate for the writes below. + process.env.OS_TENANCY_POSTURE = 'isolated'; + delete process.env.OS_TENANT_AUDIT; + + driver = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + }); + warns = []; + (driver as any).logger = { warn: (msg: string, meta: any) => warns.push({ msg, meta }) }; + + engine = new ObjectQL(); + engine.registerDriver(driver, true); + await engine.init(); + engine.registry.registerObject(HttpDelivery as any, '@objectstack/service-messaging'); + engine.registry.registerObject(NotificationDelivery as any, '@objectstack/service-messaging'); + await engine.syncSchemas(); +}); + +afterEach(async () => { + try { await engine?.destroy(); } catch { /* noop */ } + if (OLD_POSTURE === undefined) delete process.env.OS_TENANCY_POSTURE; + else process.env.OS_TENANCY_POSTURE = OLD_POSTURE; + if (OLD_AUDIT === undefined) delete process.env.OS_TENANT_AUDIT; + else process.env.OS_TENANT_AUDIT = OLD_AUDIT; +}); + +/** + * The positive control. An unscoped predicate write on `object`, issued + * directly through the engine with no `bypassTenantAudit`, MUST produce the + * audit line — otherwise a silent run proves nothing about the code under + * test. Deliberately run AFTER the assertion it guards: the gate throttles one + * warning per `${object}:${op}`, so this only fires if the production path + * consumed no `updateMany` warning of its own. + */ +async function controlUnscopedUpdateMany(object: string): Promise { + // A PREDICATE write (no `id`), so it routes through `driver.updateMany` + // exactly as the production path does; matching zero rows is fine — the + // audit fires before the statement runs. + await engine.update(object, { attempts: 99 }, { where: { status: '__control_no_such_status__' }, multi: true } as any); + expect( + auditedUpdateMany(object), + `positive control failed: an unscoped multi:true write on ${object} produced no [tenant-audit] ` + + 'line, so this file cannot distinguish "classified" from "audit not armed"', + ).toBe(true); +} + +async function seedHttpRow(id: string, org: string, over: Record = {}): Promise { + const now = new Date(); + await engine.insert(SYS_HTTP_DELIVERY, { + id, + source: 'test', + ref_id: id, + dedup_key: id, + url: 'https://receiver.example/hook', + method: 'POST', + payload_json: '{}', + partition_key: 0, + status: 'pending', + attempts: 0, + organization_id: org, + created_at: now, + updated_at: now, + ...over, + } as any); +} + +// ─────────────────────────────────────────────────────────────────────────── +describe('sys_http_delivery — the dispatcher claim path is a classified global sweep', () => { + it('claims across organizations without a tenant-audit finding', async () => { + // The gate's own precondition: this object really is tenant-scoped. + expect((driver as any).resolveTenantField(SYS_HTTP_DELIVERY)).toBe('organization_id'); + + await seedHttpRow('h_a', 'org_a'); + await seedHttpRow('h_b', 'org_b'); + + const outbox = new SqlHttpOutbox(engine as any, { partitionCount: 1 }); + const claimed = await outbox.claim({ nodeId: 'node1', limit: 10, claimTtlMs: 60_000 }); + + // ① the classified write is silent… + expect(auditedUpdateMany(SYS_HTTP_DELIVERY)).toBe(false); + // ② …and it still reaches every organization, which is the point. + expect(claimed.map((c) => c.id).sort()).toEqual(['h_a', 'h_b']); + const rows = (await engine.find(SYS_HTTP_DELIVERY, { where: {} })) as any[]; + expect(rows.map((r) => `${r.id}:${r.organization_id}:${r.status}`).sort()).toEqual([ + 'h_a:org_a:in_flight', + 'h_b:org_b:in_flight', + ]); + + await controlUnscopedUpdateMany(SYS_HTTP_DELIVERY); + }); + + it('reaps a crashed node\'s in_flight rows in every organization, without a finding', async () => { + const stale = Date.now() - 10 * 60_000; + await seedHttpRow('h_a', 'org_a', { status: 'in_flight', claimed_by: 'dead_node', claimed_at: stale }); + await seedHttpRow('h_b', 'org_b', { status: 'in_flight', claimed_by: 'dead_node', claimed_at: stale }); + + const outbox = new SqlHttpOutbox(engine as any, { partitionCount: 1 }); + const claimed = await outbox.claim({ nodeId: 'node2', limit: 10, claimTtlMs: 60_000 }); + + expect(auditedUpdateMany(SYS_HTTP_DELIVERY)).toBe(false); + // Both organizations' abandoned rows were recovered AND re-claimed by + // the live node — a per-organization reap would have stranded one. + expect(claimed.map((c) => c.id).sort()).toEqual(['h_a', 'h_b']); + const rows = (await engine.find(SYS_HTTP_DELIVERY, { where: {} })) as any[]; + expect(rows.every((r) => r.claimed_by === 'node2')).toBe(true); + + await controlUnscopedUpdateMany(SYS_HTTP_DELIVERY); + }); +}); + +// ─────────────────────────────────────────────────────────────────────────── +describe('sys_notification_delivery — the dispatcher claim path is a classified global sweep', () => { + it('claims across organizations without a tenant-audit finding', async () => { + expect((driver as any).resolveTenantField(DELIVERY_OBJECT)).toBe('organization_id'); + + const outbox = new SqlNotificationOutbox(engine as any, { partitionCount: 1 }); + const idA = await outbox.enqueue({ + notificationId: 'n_a', recipientId: 'u_a', channel: 'inbox', organizationId: 'org_a', payload: {}, + } as any); + const idB = await outbox.enqueue({ + notificationId: 'n_b', recipientId: 'u_b', channel: 'inbox', organizationId: 'org_b', payload: {}, + } as any); + + const claimed = await outbox.claim({ nodeId: 'node1', limit: 10, claimTtlMs: 60_000 }); + + expect(auditedUpdateMany(DELIVERY_OBJECT)).toBe(false); + expect(claimed.map((c) => c.id).sort()).toEqual([idA, idB].sort()); + // The organization stamped at enqueue survives the sweep untouched — + // the sweep moves `status`, never a row's tenant. + expect(claimed.map((c) => c.organizationId).sort()).toEqual(['org_a', 'org_b']); + + await controlUnscopedUpdateMany(DELIVERY_OBJECT); + }); + + it('collapses a digest window across organizations without a finding', async () => { + const outbox = new SqlNotificationOutbox(engine as any, { partitionCount: 1 }); + await outbox.enqueue({ + notificationId: 'n_a', recipientId: 'u_a', channel: 'inbox', organizationId: 'org_a', + payload: {}, digestKey: 'u_a|inbox|w1', + } as any); + await outbox.enqueue({ + notificationId: 'n_b', recipientId: 'u_b', channel: 'inbox', organizationId: 'org_b', + payload: {}, digestKey: 'u_b|inbox|w1', + } as any); + + const claimed = await outbox.claimDigest({ nodeId: 'node1', limit: 10, claimTtlMs: 60_000 }); + + expect(auditedUpdateMany(DELIVERY_OBJECT)).toBe(false); + expect(claimed.map((c) => c.organizationId).sort()).toEqual(['org_a', 'org_b']); + + await controlUnscopedUpdateMany(DELIVERY_OBJECT); + }); +}); diff --git a/packages/services/service-messaging/src/outbox-dispatcher-scope.ts b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts new file mode 100644 index 0000000000..e8e59054e8 --- /dev/null +++ b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts @@ -0,0 +1,57 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +import type { EngineUpdateOptions } from '@objectstack/spec/data'; + +/** + * The write options for a delivery-outbox **dispatcher sweep** — every + * predicate write (`multi: true`, i.e. `driver.updateMany`) that + * {@link SqlNotificationOutbox} and {@link SqlHttpOutbox} issue against + * `sys_notification_delivery` / `sys_http_delivery` from the claim path. + * + * ## Why these writes carry `bypassTenantAudit` instead of a `tenantId` + * + * Both objects are tenant-scoped: the kernel provisions `organization_id` on + * them, so `SqlDriver.resolveTenantField()` answers `organization_id` and the + * driver's `auditMissingTenant` gate treats every unscoped write to them as a + * finding on a walled deployment (`OS_TENANCY_POSTURE=isolated|group`). That + * gate is right to ask, and the two legal answers are "thread the caller's + * tenant" or "declare this write global, and say why". These sweeps are the + * second, and the warrant is structural rather than aesthetic: + * + * 1. **No request context exists to thread.** The only callers are + * `NotificationDispatcher` and `HttpDispatcher`, whose `runPartition()` + * runs off a `setInterval` tick under a cluster lock keyed + * `notify.dispatcher.partition.` / `http.dispatcher.partition.`. + * There is no HTTP request, no session and no active organization on that + * path — the tick is a platform actor, not a tenant's. + * 2. **The outbox contract has no tenant to thread even in principle.** + * `ClaimOptions` / `HttpClaimOptions` are `{ nodeId, limit, partition, + * claimTtlMs, now }`. Partitioning is `hash(refId | notificationId | + * digestKey) mod N` — a load-spreading key, deliberately *not* an + * organization key — so a partition holds rows from every organization by + * construction. + * 3. **Scoping them would break delivery, not isolate it.** One outbox and + * one dispatcher pair are constructed per ENVIRONMENT + * (`messaging-service-plugin.ts`), and they drain the whole environment's + * queue. An `organization_id = ` predicate on the claim would + * strand every other organization's pending notifications and callouts + * forever, and one on the visibility-timeout reap would leave rows a + * crashed node abandoned for other organizations permanently `in_flight`. + * Crossing organizations is the operation's *semantics*, not an oversight. + * + * ⚠️ This is a **diagnostics** flag and nothing else: per its spec + * (`DriverOptionsSchema.bypassTenantAudit`) it "never changes what the write + * touches". It silences a warning about a write that was already, and + * correctly, environment-wide. It must never be reached for to quiet a write + * that a request context could have scoped — that is the failure mode the + * audit exists to prevent, and the row-level writes on these same objects + * (`ack`, `redeliver`) are single-record `multi: false` writes that do **not** + * use this helper. + * + * @param where Predicate identifying the rows this sweep claims or reaps. + */ +export function dispatcherSweepOptions( + where: Record, +): EngineUpdateOptions & { multi: true; bypassTenantAudit: true } { + return { where, multi: true, bypassTenantAudit: true }; +} diff --git a/packages/services/service-messaging/src/sql-http-outbox.ts b/packages/services/service-messaging/src/sql-http-outbox.ts index e9c70137fc..73068d9507 100644 --- a/packages/services/service-messaging/src/sql-http-outbox.ts +++ b/packages/services/service-messaging/src/sql-http-outbox.ts @@ -4,6 +4,7 @@ import { randomUUID } from 'node:crypto'; import type { IDataEngine } from '@objectstack/spec/contracts'; import { hashPartition } from './backoff.js'; import { toEpochMs } from './audit-timestamp.js'; +import { dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; import { deliveryBody, signBody } from './http-sender.js'; import { HttpRedeliverError, @@ -193,13 +194,12 @@ export class SqlHttpOutbox implements IHttpOutbox { await this.engine.update( this.objectName, { status: 'pending', claimed_by: null, claimed_at: null }, - { - where: { - status: 'in_flight', - claimed_at: { $lt: now - opts.claimTtlMs }, - }, - multi: true, - }, + // Environment-wide by design: recovers rows a crashed node abandoned, + // for every organization. Warrant in `outbox-dispatcher-scope.ts`. + dispatcherSweepOptions({ + status: 'in_flight', + claimed_at: { $lt: now - opts.claimTtlMs }, + }), ); // 2. Pick candidate ids. @@ -221,7 +221,9 @@ export class SqlHttpOutbox implements IHttpOutbox { await this.engine.update( this.objectName, { status: 'in_flight', claimed_by: opts.nodeId, claimed_at: now }, - { where: { id: { $in: ids }, status: 'pending' }, multi: true }, + // Environment-wide by design: the dispatcher drains every + // organization's queue. Warrant in `outbox-dispatcher-scope.ts`. + dispatcherSweepOptions({ id: { $in: ids }, status: 'pending' }), ); // 4. Read back the rows we actually own. diff --git a/packages/services/service-messaging/src/sql-outbox.ts b/packages/services/service-messaging/src/sql-outbox.ts index a066e51167..5ff09c982f 100644 --- a/packages/services/service-messaging/src/sql-outbox.ts +++ b/packages/services/service-messaging/src/sql-outbox.ts @@ -12,6 +12,7 @@ import type { } from './outbox.js'; import { hashPartition } from './backoff.js'; import { toEpochMs } from './audit-timestamp.js'; +import { dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; export const DELIVERY_OBJECT = 'sys_notification_delivery'; @@ -127,7 +128,9 @@ export class SqlNotificationOutbox implements INotificationOutbox { await this.engine.update( this.objectName, { status: 'pending', claimed_by: null, claimed_at: null }, - { where: { status: 'in_flight', claimed_at: { $lt: now - opts.claimTtlMs } }, multi: true } as any, + // Environment-wide by design: recovers rows a crashed node abandoned, + // for every organization. Warrant in `outbox-dispatcher-scope.ts`. + dispatcherSweepOptions({ status: 'in_flight', claimed_at: { $lt: now - opts.claimTtlMs } }), ); // 2. Candidate ids: ready pending rows in our partition. Batched (digest) @@ -150,7 +153,9 @@ export class SqlNotificationOutbox implements INotificationOutbox { await this.engine.update( this.objectName, { status: 'in_flight', claimed_by: opts.nodeId, claimed_at: now }, - { where: { id: { $in: ids }, status: 'pending' }, multi: true } as any, + // Environment-wide by design: the dispatcher drains every + // organization's queue. Warrant in `outbox-dispatcher-scope.ts`. + dispatcherSweepOptions({ id: { $in: ids }, status: 'pending' }), ); // 4. Read back only the rows we own. @@ -167,7 +172,9 @@ export class SqlNotificationOutbox implements INotificationOutbox { await this.engine.update( this.objectName, { status: 'pending', claimed_by: null, claimed_at: null }, - { where: { status: 'in_flight', claimed_at: { $lt: now - opts.claimTtlMs } }, multi: true } as any, + // Environment-wide by design: recovers rows a crashed node abandoned, + // for every organization. Warrant in `outbox-dispatcher-scope.ts`. + dispatcherSweepOptions({ status: 'in_flight', claimed_at: { $lt: now - opts.claimTtlMs } }), ); // 2. All DUE batched rows in our partition — a window is claimed whole, so @@ -190,7 +197,9 @@ export class SqlNotificationOutbox implements INotificationOutbox { await this.engine.update( this.objectName, { status: 'in_flight', claimed_by: opts.nodeId, claimed_at: now }, - { where: { id: { $in: ids }, status: 'pending' }, multi: true } as any, + // Environment-wide by design: the dispatcher drains every + // organization's queue. Warrant in `outbox-dispatcher-scope.ts`. + dispatcherSweepOptions({ id: { $in: ids }, status: 'pending' }), ); // 4. Read back the rows we own. From 72607c032fcfddec9eac469e984ed01964455e11 Mon Sep 17 00:00:00 2001 From: os-warren Date: Fri, 21 Aug 2026 09:24:21 +0000 Subject: [PATCH 2/2] chore: changeset for delivery dispatcher sweep classification (#10673) --- ...-dispatcher-sweep-tenant-classification.md | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) create mode 100644 .changeset/delivery-dispatcher-sweep-tenant-classification.md diff --git a/.changeset/delivery-dispatcher-sweep-tenant-classification.md b/.changeset/delivery-dispatcher-sweep-tenant-classification.md new file mode 100644 index 0000000000..5f988c1867 --- /dev/null +++ b/.changeset/delivery-dispatcher-sweep-tenant-classification.md @@ -0,0 +1,20 @@ +--- +"@objectstack/service-messaging": patch +--- + +Classify the delivery dispatchers' predicate writes on `sys_http_delivery` and +`sys_notification_delivery` as global environment sweeps (#10673). On a walled +deployment (`OS_TENANCY_POSTURE=isolated|group`) the SQL driver's tenant-audit +gate reported every `updateMany` these outboxes issue from the claim path as an +un-isolated write. The audit was right to ask: both objects are tenant-scoped +via `organization_id`. The answer is that these six writes — the +visibility-timeout reap and the atomic claim in `SqlHttpOutbox.claim`, +`SqlNotificationOutbox.claim` and `SqlNotificationOutbox.claimDigest` — are +issued by a `setInterval` dispatcher tick under a cluster lock, with no request +context and no tenant anywhere in the `ClaimOptions` contract, and they must +cross organizations: one outbox drains the whole environment's queue, so a +per-organization predicate would strand every other organization's deliveries. +They now pass `bypassTenantAudit` through a single documented helper that +carries that warrant. Diagnostics only — per its spec the flag never changes +what a write touches, and the row-level `ack` / `redeliver` writes are +unaffected.