|
| 1 | +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. |
| 2 | + |
| 3 | +/** |
| 4 | + * #4765 — the SQL outboxes must never put `updated_at` in an UPDATE payload. |
| 5 | + * |
| 6 | + * ObjectQL's builtin `sys_stamp_audit_update` hook owns that column on every |
| 7 | + * update, and the column is `readonly`, so a caller-supplied value is stripped |
| 8 | + * by `stripReadonlyFields` (#2948) — which WARNs once per call. `claim()` and |
| 9 | + * `claimDigest()` open with an unconditional "reap stale in_flight" UPDATE that |
| 10 | + * runs on every dispatcher tick whether or not a row is stale, so the write was |
| 11 | + * both a no-op (stripped, then re-stamped) and, at 3 claim paths × 8 partitions |
| 12 | + * × a 500 ms tick, 48 identical warnings a second on an idle `pnpm dev`. |
| 13 | + * |
| 14 | + * INSERT is a different path (no strip, and `created_at` IS caller-owned): it |
| 15 | + * keeps writing both audit columns, as `Date`s — a native TIMESTAMP column |
| 16 | + * rejects a bare epoch-ms number on Postgres (see `audit-timestamp.ts`). |
| 17 | + */ |
| 18 | + |
| 19 | +import { describe, it, expect } from 'vitest'; |
| 20 | +import type { IDataEngine } from '@objectstack/spec/contracts'; |
| 21 | +import { SqlNotificationOutbox } from './sql-outbox.js'; |
| 22 | +import { SqlHttpOutbox } from './sql-http-outbox.js'; |
| 23 | + |
| 24 | +interface RecordedUpdate { |
| 25 | + object: string; |
| 26 | + data: Record<string, unknown>; |
| 27 | +} |
| 28 | + |
| 29 | +/** |
| 30 | + * Minimal recording `IDataEngine`. `find` replays a scripted queue of result |
| 31 | + * sets so a claim can be driven all the way through candidate-select → |
| 32 | + * atomic-claim → read-back; everything else is inert. |
| 33 | + */ |
| 34 | +function makeEngine(findResults: Array<Array<Record<string, unknown>>> = []) { |
| 35 | + const updates: RecordedUpdate[] = []; |
| 36 | + const inserts: Array<{ object: string; data: Record<string, unknown> }> = []; |
| 37 | + let findCall = 0; |
| 38 | + |
| 39 | + const engine = { |
| 40 | + async insert(object: string, data: Record<string, unknown>) { |
| 41 | + inserts.push({ object, data }); |
| 42 | + return data; |
| 43 | + }, |
| 44 | + async update(object: string, data: Record<string, unknown>) { |
| 45 | + updates.push({ object, data }); |
| 46 | + return { matched: 0, modified: 0 }; |
| 47 | + }, |
| 48 | + async find(_object: string) { |
| 49 | + return findResults[findCall++] ?? []; |
| 50 | + }, |
| 51 | + async findOne(_object: string, opts?: { fields?: string[] }) { |
| 52 | + // `ack()` reads the current attempt count; `enqueue()` probes for a |
| 53 | + // dedup winner and must miss so the insert path runs. |
| 54 | + if (opts?.fields?.includes('attempts')) return { attempts: 2 }; |
| 55 | + return null; |
| 56 | + }, |
| 57 | + async delete() { return { matched: 0, modified: 0 }; }, |
| 58 | + } as unknown as IDataEngine; |
| 59 | + |
| 60 | + return { engine, updates, inserts }; |
| 61 | +} |
| 62 | + |
| 63 | +/** Every audit column an UPDATE payload must leave to the platform. */ |
| 64 | +const PLATFORM_OWNED_ON_UPDATE = ['updated_at']; |
| 65 | + |
| 66 | +function expectNoPlatformAuditColumns(updates: RecordedUpdate[]) { |
| 67 | + expect(updates.length).toBeGreaterThan(0); |
| 68 | + for (const u of updates) { |
| 69 | + for (const column of PLATFORM_OWNED_ON_UPDATE) { |
| 70 | + expect( |
| 71 | + Object.prototype.hasOwnProperty.call(u.data, column), |
| 72 | + `UPDATE on ${u.object} must not write '${column}' — keys: ${Object.keys(u.data).join(', ')}`, |
| 73 | + ).toBe(false); |
| 74 | + } |
| 75 | + } |
| 76 | +} |
| 77 | + |
| 78 | +describe('SqlNotificationOutbox — audit columns on UPDATE (#4765)', () => { |
| 79 | + it('claim() writes no updated_at (reap + atomic claim)', async () => { |
| 80 | + // find #1 → candidate ids; find #2 → read-back of the rows we own. |
| 81 | + const { engine, updates } = makeEngine([[{ id: 'd1' }], []]); |
| 82 | + const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); |
| 83 | + |
| 84 | + await outbox.claim({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 }); |
| 85 | + |
| 86 | + // Both the unconditional reap and the atomic claim ran. |
| 87 | + expect(updates).toHaveLength(2); |
| 88 | + expectNoPlatformAuditColumns(updates); |
| 89 | + // The claim still stamps its OWN columns — this is not a blanket strip. |
| 90 | + expect(updates[1].data).toMatchObject({ status: 'in_flight', claimed_by: 'n1' }); |
| 91 | + expect(updates[1].data.claimed_at).toEqual(expect.any(Number)); |
| 92 | + }); |
| 93 | + |
| 94 | + it('claim() writes no updated_at even when nothing is claimable', async () => { |
| 95 | + // The reap UPDATE fires on every tick regardless — that is the firehose. |
| 96 | + const { engine, updates } = makeEngine([[]]); |
| 97 | + const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); |
| 98 | + |
| 99 | + await outbox.claim({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 }); |
| 100 | + |
| 101 | + expect(updates).toHaveLength(1); |
| 102 | + expectNoPlatformAuditColumns(updates); |
| 103 | + }); |
| 104 | + |
| 105 | + it('claimDigest() writes no updated_at', async () => { |
| 106 | + const { engine, updates } = makeEngine([[{ id: 'd1' }], []]); |
| 107 | + const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); |
| 108 | + |
| 109 | + await outbox.claimDigest({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 }); |
| 110 | + |
| 111 | + expect(updates).toHaveLength(2); |
| 112 | + expectNoPlatformAuditColumns(updates); |
| 113 | + }); |
| 114 | + |
| 115 | + it('ack() writes no updated_at but still bumps its own columns', async () => { |
| 116 | + const { engine, updates } = makeEngine(); |
| 117 | + const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); |
| 118 | + |
| 119 | + await outbox.ack('d1', { success: false, error: 'boom', nextAttemptAt: 123 }); |
| 120 | + |
| 121 | + expectNoPlatformAuditColumns(updates); |
| 122 | + expect(updates[0].data).toMatchObject({ |
| 123 | + status: 'pending', |
| 124 | + attempts: 3, // 2 (read back) + 1 |
| 125 | + error: 'boom', |
| 126 | + next_attempt_at: 123, |
| 127 | + }); |
| 128 | + }); |
| 129 | + |
| 130 | + it('enqueue() still writes both audit columns as Dates on INSERT', async () => { |
| 131 | + const { engine, inserts } = makeEngine(); |
| 132 | + const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); |
| 133 | + |
| 134 | + await outbox.enqueue({ notificationId: 'n1', recipientId: 'u1', channel: 'email', payload: {} }); |
| 135 | + |
| 136 | + expect(inserts).toHaveLength(1); |
| 137 | + expect(inserts[0].data.created_at).toBeInstanceOf(Date); |
| 138 | + expect(inserts[0].data.updated_at).toBeInstanceOf(Date); |
| 139 | + }); |
| 140 | +}); |
| 141 | + |
| 142 | +describe('SqlHttpOutbox — audit columns on UPDATE (#4765)', () => { |
| 143 | + const enqueueInput = { |
| 144 | + source: 'flow', |
| 145 | + refId: 'r1', |
| 146 | + dedupKey: 'd1', |
| 147 | + url: 'https://example.test/hook', |
| 148 | + payload: { hello: 'world' }, |
| 149 | + }; |
| 150 | + |
| 151 | + it('claim() writes no updated_at (reap + atomic claim)', async () => { |
| 152 | + const { engine, updates } = makeEngine([[{ id: 'h1' }], []]); |
| 153 | + const outbox = new SqlHttpOutbox(engine, { partitionCount: 8 }); |
| 154 | + |
| 155 | + await outbox.claim({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 }); |
| 156 | + |
| 157 | + expect(updates).toHaveLength(2); |
| 158 | + expectNoPlatformAuditColumns(updates); |
| 159 | + expect(updates[1].data).toMatchObject({ status: 'in_flight', claimed_by: 'n1' }); |
| 160 | + }); |
| 161 | + |
| 162 | + it('claim() writes no updated_at even when nothing is claimable', async () => { |
| 163 | + const { engine, updates } = makeEngine([[]]); |
| 164 | + const outbox = new SqlHttpOutbox(engine, { partitionCount: 8 }); |
| 165 | + |
| 166 | + await outbox.claim({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 }); |
| 167 | + |
| 168 | + expect(updates).toHaveLength(1); |
| 169 | + expectNoPlatformAuditColumns(updates); |
| 170 | + }); |
| 171 | + |
| 172 | + it('ack() writes no updated_at but still bumps its own columns', async () => { |
| 173 | + const { engine, updates } = makeEngine(); |
| 174 | + const outbox = new SqlHttpOutbox(engine, { partitionCount: 8 }); |
| 175 | + |
| 176 | + await outbox.ack('h1', { |
| 177 | + success: false, |
| 178 | + error: 'boom', |
| 179 | + httpStatus: 503, |
| 180 | + nextRetryAt: 456, |
| 181 | + durationMs: 12, |
| 182 | + }); |
| 183 | + |
| 184 | + expectNoPlatformAuditColumns(updates); |
| 185 | + expect(updates[0].data).toMatchObject({ |
| 186 | + status: 'pending', |
| 187 | + attempts: 3, |
| 188 | + error: 'boom', |
| 189 | + response_code: 503, |
| 190 | + next_retry_at: 456, |
| 191 | + }); |
| 192 | + }); |
| 193 | + |
| 194 | + it('enqueue() still writes both audit columns as Dates on INSERT', async () => { |
| 195 | + const { engine, inserts } = makeEngine(); |
| 196 | + const outbox = new SqlHttpOutbox(engine, { partitionCount: 8 }); |
| 197 | + |
| 198 | + await outbox.enqueue(enqueueInput); |
| 199 | + |
| 200 | + expect(inserts).toHaveLength(1); |
| 201 | + expect(inserts[0].data.created_at).toBeInstanceOf(Date); |
| 202 | + expect(inserts[0].data.updated_at).toBeInstanceOf(Date); |
| 203 | + }); |
| 204 | +}); |
0 commit comments