fix(messaging): classify the delivery dispatchers' updateMany sweeps as global environment sweeps (#10725)

* fix(messaging): classify the delivery dispatchers' updateMany sweeps as global (#10673)

* chore: changeset for delivery dispatcher sweep classification (#10673)

Author: os-warren

.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.

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<void> {
+    // 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<string, unknown> = {}): Promise<void> {
+    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);
+    });
+});

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.<n>` / `http.dispatcher.partition.<n>`.
+ *     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 = <one org>` 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<string, unknown>,
+): EngineUpdateOptions & { multi: true; bypassTenantAudit: true } {
+    return { where, multi: true, bypassTenantAudit: true };
+}

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.