From 0414753a79756699ef6367ae5bc78226d2b8bf5a Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 3 Aug 2026 03:58:23 +0000 Subject: [PATCH] =?UTF-8?q?fix(service-messaging):=20SQL=20outbox=20?= =?UTF-8?q?=E7=9A=84=20UPDATE=20=E4=B8=8D=E5=86=8D=E5=86=99=20`updated=5Fa?= =?UTF-8?q?t`=20=E2=80=94=E2=80=94=20`pnpm=20dev`=20=E6=8E=A7=E5=88=B6?= =?UTF-8?q?=E5=8F=B0=E5=81=9C=E6=AD=A2=E5=88=B7=E5=B1=8F=20(#4765)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 空闲的 dev server 每秒稳定刷 48 行同一条 WARN，直到进程退出： WARN Field 'updated_at' is read-only — ignoring incoming change (#2948) `SqlNotificationOutbox.claim()` / `.claimDigest()` 和 `SqlHttpOutbox.claim()` 的第一步都是无条件的「reap stale in_flight」谓词 UPDATE —— visibility-timeout 回收，每个 dispatcher tick 都跑，不管有没有行真的过期 —— 而这些 payload 里都带了 `updated_at`。该列是 `readonly`，归 ObjectQL 内建的 `sys_stamp_audit_update` hook 所有，所以值先被 `stripReadonlyFields` 剥掉、再被平台盖回去：一次纯 no-op 的写，代价是一条 WARN。三条 claim 路径 × 8 个 partition × dispatcher 的 500ms tick = 48 行/秒，真正的 warn 和 error 全被冲走。 两个 outbox 的所有 UPDATE payload（`claim`、`claimDigest`、`ack`、`redeliver`） 现在都不再带 `updated_at`，交给平台 hook 盖 —— 存进去的行没有任何变化。INSERT 路径不动，照旧写两个审计列，且写成 `Date`：那里 `created_at` 是调用方拥有的，而 Postgres 的原生 `TIMESTAMP` 列会拒绝裸 epoch-ms 数字。 顺带清掉一个隐患：`enqueue()` 一直守着 `new Date()` 这条规矩，但 UPDATE 路径传的 是 epoch-ms 数字。之所以没炸，仅仅因为它在到达 driver 之前就被剥掉了 —— 一旦这些 写入哪天改走 system 上下文（system 上下文跳过剥离），这个数字就会直接落库。 回归测试断言两个 outbox 交给 engine 的 UPDATE payload 里不出现 `updated_at`， 其中包含「一行都不可 claim」的场景 —— 因为无条件 reap 正是刷屏的来源；同时锁住 INSERT 仍写 `Date`。已在真实 `pnpm dev`（showcase）上验证：120 秒 5469 行输出、 其中 5378 行是这条 WARN，修复后 90 秒 102 行，启动完成后不再有任何输出。 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_018ipMgdweHC9LFSUdLByizr --- ...essaging-outbox-no-updated-at-on-update.md | 31 +++ .../service-messaging/src/sql-http-outbox.ts | 14 +- .../src/sql-outbox-audit-columns.test.ts | 204 ++++++++++++++++++ .../service-messaging/src/sql-outbox.ts | 21 +- 4 files changed, 260 insertions(+), 10 deletions(-) create mode 100644 .changeset/messaging-outbox-no-updated-at-on-update.md create mode 100644 packages/services/service-messaging/src/sql-outbox-audit-columns.test.ts diff --git a/.changeset/messaging-outbox-no-updated-at-on-update.md b/.changeset/messaging-outbox-no-updated-at-on-update.md new file mode 100644 index 0000000000..d4844cce37 --- /dev/null +++ b/.changeset/messaging-outbox-no-updated-at-on-update.md @@ -0,0 +1,31 @@ +--- +'@objectstack/service-messaging': patch +--- + +fix(service-messaging): stop the SQL outboxes from writing `updated_at` on UPDATE — `pnpm dev` no longer floods the console + +An idle dev server printed the same warning 48 times a second, forever: + +``` +WARN Field 'updated_at' is read-only — ignoring incoming change (#2948) +``` + +`SqlNotificationOutbox.claim()` / `.claimDigest()` and `SqlHttpOutbox.claim()` +open with an unconditional "reap stale in_flight" UPDATE — visibility-timeout +recovery that runs on every dispatcher tick whether or not any row is actually +stale — and every one of those payloads carried `updated_at`. That column is +`readonly` and owned by ObjectQL's builtin `sys_stamp_audit_update` hook, so the +value was stripped by `stripReadonlyFields` and re-stamped by the platform: a +no-op write that cost one warning per call. Three claim paths × 8 partitions × +the dispatcher's 500 ms tick = 48 identical lines a second, which buried every +real warning and error in the dev log. + +`updated_at` is now gone from every UPDATE payload in both outboxes (`claim`, +`claimDigest`, `ack`, `redeliver`); the platform hook keeps stamping it, so +stored rows are unchanged. INSERT still writes both audit columns, as `Date`s — +`created_at` is caller-owned there, and a native `TIMESTAMP` column rejects a +bare epoch-ms number on Postgres. + +That last point was also a latent bug this removes: `enqueue()` correctly used +`new Date()`, but the UPDATE paths passed epoch-ms numbers. Nothing broke only +because the strip discarded them before they reached the driver. diff --git a/packages/services/service-messaging/src/sql-http-outbox.ts b/packages/services/service-messaging/src/sql-http-outbox.ts index 12b41fa9d4..a66bc835f9 100644 --- a/packages/services/service-messaging/src/sql-http-outbox.ts +++ b/packages/services/service-messaging/src/sql-http-outbox.ts @@ -63,6 +63,13 @@ interface DeliveryRow { * `UPDATE WHERE status='pending'` for the exactly-once claim; precomputed * `partition_key`; SELECT-then-INSERT dedup converging on the unique index). * Dedup uniqueness is `(source, dedup_key)`; partition affinity is on `ref_id`. + * + * **No UPDATE here writes `updated_at`** (#4765) — same rule, same reason as + * {@link SqlNotificationOutbox}: the platform's `sys_stamp_audit_update` hook + * owns that column, a caller-supplied value is stripped as `readonly` (#2948) + * with a WARN per call, and `claim()`'s unconditional reap UPDATE runs on every + * dispatcher tick — so writing it turned an idle dev server into a console + * firehose while changing nothing about the stored row. */ export class SqlHttpOutbox implements IHttpOutbox { private readonly objectName: string; @@ -127,7 +134,7 @@ export class SqlHttpOutbox implements IHttpOutbox { // 1. Reap stale in_flight rows — visibility-timeout recovery. await this.engine.update( this.objectName, - { status: 'pending', claimed_by: null, claimed_at: null, updated_at: now }, + { status: 'pending', claimed_by: null, claimed_at: null }, { where: { status: 'in_flight', @@ -155,7 +162,7 @@ export class SqlHttpOutbox implements IHttpOutbox { // 3. Atomic claim. WHERE status='pending' rejects rows another worker took. await this.engine.update( this.objectName, - { status: 'in_flight', claimed_by: opts.nodeId, claimed_at: now, updated_at: now }, + { status: 'in_flight', claimed_by: opts.nodeId, claimed_at: now }, { where: { id: { $in: ids }, status: 'pending' }, multi: true }, ); @@ -205,7 +212,6 @@ export class SqlHttpOutbox implements IHttpOutbox { response_body: result.responseBody ?? null, next_retry_at: nextRetryAt, error, - updated_at: now, }, { where: { id }, multi: false }, ); @@ -230,7 +236,6 @@ export class SqlHttpOutbox implements IHttpOutbox { 'DELIVERY_NOT_ELIGIBLE', ); } - const now = Date.now(); await this.engine.update( this.objectName, { @@ -243,7 +248,6 @@ export class SqlHttpOutbox implements IHttpOutbox { response_code: null, response_body: null, error: null, - updated_at: now, }, { where: { id, status: { $in: ['success', 'failed', 'dead'] } }, multi: false }, ); diff --git a/packages/services/service-messaging/src/sql-outbox-audit-columns.test.ts b/packages/services/service-messaging/src/sql-outbox-audit-columns.test.ts new file mode 100644 index 0000000000..8ffd92346c --- /dev/null +++ b/packages/services/service-messaging/src/sql-outbox-audit-columns.test.ts @@ -0,0 +1,204 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #4765 — the SQL outboxes must never put `updated_at` in an UPDATE payload. + * + * ObjectQL's builtin `sys_stamp_audit_update` hook owns that column on every + * update, and the column is `readonly`, so a caller-supplied value is stripped + * by `stripReadonlyFields` (#2948) — which WARNs once per call. `claim()` and + * `claimDigest()` open with an unconditional "reap stale in_flight" UPDATE that + * runs on every dispatcher tick whether or not a row is stale, so the write was + * both a no-op (stripped, then re-stamped) and, at 3 claim paths × 8 partitions + * × a 500 ms tick, 48 identical warnings a second on an idle `pnpm dev`. + * + * INSERT is a different path (no strip, and `created_at` IS caller-owned): it + * keeps writing both audit columns, as `Date`s — a native TIMESTAMP column + * rejects a bare epoch-ms number on Postgres (see `audit-timestamp.ts`). + */ + +import { describe, it, expect } from 'vitest'; +import type { IDataEngine } from '@objectstack/spec/contracts'; +import { SqlNotificationOutbox } from './sql-outbox.js'; +import { SqlHttpOutbox } from './sql-http-outbox.js'; + +interface RecordedUpdate { + object: string; + data: Record; +} + +/** + * Minimal recording `IDataEngine`. `find` replays a scripted queue of result + * sets so a claim can be driven all the way through candidate-select → + * atomic-claim → read-back; everything else is inert. + */ +function makeEngine(findResults: Array>> = []) { + const updates: RecordedUpdate[] = []; + const inserts: Array<{ object: string; data: Record }> = []; + let findCall = 0; + + const engine = { + async insert(object: string, data: Record) { + inserts.push({ object, data }); + return data; + }, + async update(object: string, data: Record) { + updates.push({ object, data }); + return { matched: 0, modified: 0 }; + }, + async find(_object: string) { + return findResults[findCall++] ?? []; + }, + async findOne(_object: string, opts?: { fields?: string[] }) { + // `ack()` reads the current attempt count; `enqueue()` probes for a + // dedup winner and must miss so the insert path runs. + if (opts?.fields?.includes('attempts')) return { attempts: 2 }; + return null; + }, + async delete() { return { matched: 0, modified: 0 }; }, + } as unknown as IDataEngine; + + return { engine, updates, inserts }; +} + +/** Every audit column an UPDATE payload must leave to the platform. */ +const PLATFORM_OWNED_ON_UPDATE = ['updated_at']; + +function expectNoPlatformAuditColumns(updates: RecordedUpdate[]) { + expect(updates.length).toBeGreaterThan(0); + for (const u of updates) { + for (const column of PLATFORM_OWNED_ON_UPDATE) { + expect( + Object.prototype.hasOwnProperty.call(u.data, column), + `UPDATE on ${u.object} must not write '${column}' — keys: ${Object.keys(u.data).join(', ')}`, + ).toBe(false); + } + } +} + +describe('SqlNotificationOutbox — audit columns on UPDATE (#4765)', () => { + it('claim() writes no updated_at (reap + atomic claim)', async () => { + // find #1 → candidate ids; find #2 → read-back of the rows we own. + const { engine, updates } = makeEngine([[{ id: 'd1' }], []]); + const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); + + await outbox.claim({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 }); + + // Both the unconditional reap and the atomic claim ran. + expect(updates).toHaveLength(2); + expectNoPlatformAuditColumns(updates); + // The claim still stamps its OWN columns — this is not a blanket strip. + expect(updates[1].data).toMatchObject({ status: 'in_flight', claimed_by: 'n1' }); + expect(updates[1].data.claimed_at).toEqual(expect.any(Number)); + }); + + it('claim() writes no updated_at even when nothing is claimable', async () => { + // The reap UPDATE fires on every tick regardless — that is the firehose. + const { engine, updates } = makeEngine([[]]); + const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); + + await outbox.claim({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 }); + + expect(updates).toHaveLength(1); + expectNoPlatformAuditColumns(updates); + }); + + it('claimDigest() writes no updated_at', async () => { + const { engine, updates } = makeEngine([[{ id: 'd1' }], []]); + const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); + + await outbox.claimDigest({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 }); + + expect(updates).toHaveLength(2); + expectNoPlatformAuditColumns(updates); + }); + + it('ack() writes no updated_at but still bumps its own columns', async () => { + const { engine, updates } = makeEngine(); + const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); + + await outbox.ack('d1', { success: false, error: 'boom', nextAttemptAt: 123 }); + + expectNoPlatformAuditColumns(updates); + expect(updates[0].data).toMatchObject({ + status: 'pending', + attempts: 3, // 2 (read back) + 1 + error: 'boom', + next_attempt_at: 123, + }); + }); + + it('enqueue() still writes both audit columns as Dates on INSERT', async () => { + const { engine, inserts } = makeEngine(); + const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); + + await outbox.enqueue({ notificationId: 'n1', recipientId: 'u1', channel: 'email', payload: {} }); + + expect(inserts).toHaveLength(1); + expect(inserts[0].data.created_at).toBeInstanceOf(Date); + expect(inserts[0].data.updated_at).toBeInstanceOf(Date); + }); +}); + +describe('SqlHttpOutbox — audit columns on UPDATE (#4765)', () => { + const enqueueInput = { + source: 'flow', + refId: 'r1', + dedupKey: 'd1', + url: 'https://example.test/hook', + payload: { hello: 'world' }, + }; + + it('claim() writes no updated_at (reap + atomic claim)', async () => { + const { engine, updates } = makeEngine([[{ id: 'h1' }], []]); + const outbox = new SqlHttpOutbox(engine, { partitionCount: 8 }); + + await outbox.claim({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 }); + + expect(updates).toHaveLength(2); + expectNoPlatformAuditColumns(updates); + expect(updates[1].data).toMatchObject({ status: 'in_flight', claimed_by: 'n1' }); + }); + + it('claim() writes no updated_at even when nothing is claimable', async () => { + const { engine, updates } = makeEngine([[]]); + const outbox = new SqlHttpOutbox(engine, { partitionCount: 8 }); + + await outbox.claim({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 }); + + expect(updates).toHaveLength(1); + expectNoPlatformAuditColumns(updates); + }); + + it('ack() writes no updated_at but still bumps its own columns', async () => { + const { engine, updates } = makeEngine(); + const outbox = new SqlHttpOutbox(engine, { partitionCount: 8 }); + + await outbox.ack('h1', { + success: false, + error: 'boom', + httpStatus: 503, + nextRetryAt: 456, + durationMs: 12, + }); + + expectNoPlatformAuditColumns(updates); + expect(updates[0].data).toMatchObject({ + status: 'pending', + attempts: 3, + error: 'boom', + response_code: 503, + next_retry_at: 456, + }); + }); + + it('enqueue() still writes both audit columns as Dates on INSERT', async () => { + const { engine, inserts } = makeEngine(); + const outbox = new SqlHttpOutbox(engine, { partitionCount: 8 }); + + await outbox.enqueue(enqueueInput); + + expect(inserts).toHaveLength(1); + expect(inserts[0].data.created_at).toBeInstanceOf(Date); + expect(inserts[0].data.updated_at).toBeInstanceOf(Date); + }); +}); diff --git a/packages/services/service-messaging/src/sql-outbox.ts b/packages/services/service-messaging/src/sql-outbox.ts index ea2c6dbbd1..a066e51167 100644 --- a/packages/services/service-messaging/src/sql-outbox.ts +++ b/packages/services/service-messaging/src/sql-outbox.ts @@ -53,6 +53,18 @@ interface DeliveryRow { * `UPDATE … WHERE status='pending'` claim. `partition_key` is precomputed on * enqueue (ObjectQL has no portable `hash()` in WHERE). Mirrors * `SqlWebhookOutbox`. + * + * **No UPDATE here writes `updated_at`** (#4765). ObjectQL's builtin + * `sys_stamp_audit_update` hook stamps it on every update unconditionally, and + * `updated_at` is `readonly`, so a caller-supplied value is stripped by + * `stripReadonlyFields` (#2948) before it reaches the driver — with a WARN per + * call. `claim()` / `claimDigest()` start with an unconditional reap UPDATE that + * runs whether or not a row is stale, so on an idle dev server the three claim + * paths × 8 partitions × a 500 ms dispatcher tick spammed 48 identical warnings + * a second and drowned the console. Writing the column was already a no-op + * (stripped, then re-stamped); passing it as epoch-ms would also have been the + * wrong shape for a native TIMESTAMP column (see `toEpochMs`) had it ever + * survived the strip. Leave it to the platform. */ export class SqlNotificationOutbox implements INotificationOutbox { private readonly objectName: string; @@ -114,7 +126,7 @@ export class SqlNotificationOutbox implements INotificationOutbox { // 1. Reap stale in_flight rows (visibility-timeout recovery). await this.engine.update( this.objectName, - { status: 'pending', claimed_by: null, claimed_at: null, updated_at: now }, + { status: 'pending', claimed_by: null, claimed_at: null }, { where: { status: 'in_flight', claimed_at: { $lt: now - opts.claimTtlMs } }, multi: true } as any, ); @@ -137,7 +149,7 @@ export class SqlNotificationOutbox implements INotificationOutbox { // 3. Atomic claim — WHERE status='pending' rejects rows another worker took. await this.engine.update( this.objectName, - { status: 'in_flight', claimed_by: opts.nodeId, claimed_at: now, updated_at: now }, + { status: 'in_flight', claimed_by: opts.nodeId, claimed_at: now }, { where: { id: { $in: ids }, status: 'pending' }, multi: true } as any, ); @@ -154,7 +166,7 @@ export class SqlNotificationOutbox implements INotificationOutbox { // 1. Reap stale in_flight (same as claim). await this.engine.update( this.objectName, - { status: 'pending', claimed_by: null, claimed_at: null, updated_at: now }, + { status: 'pending', claimed_by: null, claimed_at: null }, { where: { status: 'in_flight', claimed_at: { $lt: now - opts.claimTtlMs } }, multi: true } as any, ); @@ -177,7 +189,7 @@ export class SqlNotificationOutbox implements INotificationOutbox { // 3. Atomic claim. await this.engine.update( this.objectName, - { status: 'in_flight', claimed_by: opts.nodeId, claimed_at: now, updated_at: now }, + { status: 'in_flight', claimed_by: opts.nodeId, claimed_at: now }, { where: { id: { $in: ids }, status: 'pending' }, multi: true } as any, ); @@ -224,7 +236,6 @@ export class SqlNotificationOutbox implements INotificationOutbox { claimed_at: null, next_attempt_at: nextAttemptAt, error, - updated_at: now, }, { where: { id }, multi: false } as any, );