Improve task staleness detection and fix serialization group isolation

.changeset/amsg-review-hardening.md

@@ -0,0 +1,23 @@
+---
+"@rei-standard/amsg-server": minor
+"@rei-standard/amsg-shared": patch
+---
+
+代码评审加固：存量任务订阅兜底、串行分组写偏斜收口、重试状态重置、错误分类与门禁去重
+
+**@rei-standard/amsg-server**
+
+- **升级前创建的任务不再必然投递失败。** 投递时解析订阅新增兜底：用户级 `push_subscriptions` 存储里没有订阅时，回退到旧任务 payload 里内嵌的 `pushSubscription`（存储里有则永远优先用存储的那份）。普通投递路径和 agentic 路径都生效——存量部署升级后，用户来不及打开新客户端登记订阅，旧任务照样发得出去。
+- **pg / neon 串行分组占位补上写偏斜收口。** READ COMMITTED 下两个并发 tick 各领同组「不同」行时，`NOT EXISTS` 互相看不见对方未提交的租约，同组两条任务可能并发执行。现在占位提交后再复查一次同组活租约，撞上就放掉自己刚写的租约、这一跳不跑（两边都让路也安全：行保持 pending，下一跳重试）。`claimTask` 与 `push_subscriptions` 三方法同时收拢到 `adapters/pg-shared.js`，pg / neon 共用一份 SQL，语义不再可能分歧。
+- **tick 内串行分组预占用按用户隔离。** 内存侧的占坑键带上 `user_id`，与落库侧 per-user HMAC 的隔离语义对齐——多用户部署下两个用户恰好返回同一个分组 key（如共用的默认角色名）不再互相顶掉对方的任务。
+- **`PUT /update-message` 重置重试状态。** 更新任务时 `retry_count` 归零、`retry_after` 清空（后者仅在支持 `claimTask` 的适配器上写）——刚修好 apiKey / 改好排期的任务不再背着耗尽的重试预算，下一次瞬时故障不会直接把它打成永久 failed。
+- **`POST /schedule-message` 的订阅预检改为存在性检查。** 不再解密（解出来的值本来也用不上）；查询本身失败时报可重试的 503 `PUSH_SUBSCRIPTION_LOOKUP_FAILED`，不再把瞬时 DB 故障伪装成 409 `PUSH_SUBSCRIPTION_MISSING` 引导客户端去走多余的重订阅流程。
+- **订阅类错误带稳定 `code`，投递失败按类别处置。** `resolvePushSubscription` 抛出的错误带 `err.code`（`PUSH_SUBSCRIPTION_MISSING` / `PUSH_SUBSCRIPTION_STORE_UNSUPPORTED`），消费方按 code 分支即可、不必匹配 message 文案；tick 的失败处置对这两类「重试也好不了」的错误短路退避阶梯——一次性任务直接进终审处置，循环任务直接作废本次 occurrence，不再每次白跑 3 轮重试。
+- **过期守卫两处收紧 / 放开。** 重试链上的任务（`retry_count > 0`）在排定的重试时刻（`retry_after`）本身也被拖过阈值时同样按过期处理——停摆恰好落在重试窗口里的任务不再于恢复后把几天前的旧内容推出去（`getPendingTasks` 随之在返回行里带上 `retry_after`）。阈值本身可用 `ctx.staleAfterMs` 覆盖（单用户 worker 从 config 的 `staleAfterMs` 透传），依赖「再晚也送达」语义的宿主有了官方出口。
+- **单用户 worker 的两处错误边界补齐。** `cors.origin` 回调抛错按「不放行这个 origin」处理，不再逃出 `fetch()` 变成 Cloudflare 1101 错误页；`scheduled()` 的配置构建失败改为记日志跳过这一跳，不再以未捕获异常崩掉 cron 调用。
+- **存量多租户租户自动补列。** 多租户侧每个进程首次取得适配器时补跑一遍幂等的 `initSchema`（建表 / `ADD COLUMN IF NOT EXISTS`），升级加列后第一个请求就把 schema 补齐——不再依赖 CHANGELOG 里的手工 DDL 步骤（同 tenantId 重放 `/init-tenant` 到不了 `initSchema` 就 409，此前存量租户没有任何自动迁移路径）。
+- **门禁与工具函数去重。** X-User-Id 门禁（8 个 handler 里的复制粘贴，文案已分裂成两种）收拢为 `lib/request.js` 的 `requireUserId()`，对同一错误码的 message 统一为「缺少用户标识符」；`UPDATABLE_COLUMNS` 白名单三个适配器共用 `schema.js` 一份；`isValidUrl` 改为 re-export shared 的实现；tenant/blob-store 的 base64url 改用 shared 实现；tick 的预解密 payload 直通投递侧（`processSingleMessage` 新增 `predecrypted` 参数），同一份密文不再解两遍，相关失实注释一并修正；过期跳过的循环/一次性两个近似复制的分支收拢为单一尾部。
+
+**@rei-standard/amsg-shared**
+
+- `verifyVapidJwt` 的 JWT payload 解码改用 `webcrypto-utils` 的 `utf8Decode`，兑现本模块「编码辅助只住在 webcrypto-utils」的约定（行为不变）。

packages/rei-standard-amsg/server/src/server/adapters/d1.js

@@ -15,15 +15,8 @@ import {
   CLIENT_STATE_TABLE_SQL,
   PUSH_SUBSCRIPTION_TABLE_SQL
 } from './schema.sqlite.js';
-
-// Update methods build a dynamic SET clause from object keys. Callers pass only
-// hardcoded column names today, but enforcing a whitelist keeps a future caller
-// from ever turning a caller-supplied key into interpolated SQL.
-const UPDATABLE_COLUMNS = new Set([
-  'user_id', 'uuid', 'encrypted_payload', 'message_type',
-  'next_send_at', 'lease_until', 'retry_after', 'serialize_group',
-  'status', 'retry_count', 'created_at', 'updated_at'
-]);
+// 列名不分方言：三个适配器共用 schema.js 里的这一份白名单，加列只改一处。
+import { UPDATABLE_COLUMNS } from './schema.js';
 
 // LIKE 前缀转义：用户 key 里的 % _ \ 不能变成通配符/转义符。
 function escapeLikePrefix(prefix) {
@@ -197,7 +190,7 @@ export class D1Adapter {
   async getPendingTasks(limit = 50) {
     const now = this._now();
     const res = await this._db.prepare(
-      `SELECT id, user_id, uuid, encrypted_payload, message_type, next_send_at, status, retry_count
+      `SELECT id, user_id, uuid, encrypted_payload, message_type, next_send_at, retry_after, status, retry_count
        FROM scheduled_messages
        WHERE status = 'pending' AND next_send_at <= ?
          AND (lease_until IS NULL OR lease_until <= ?)

packages/rei-standard-amsg/server/src/server/adapters/neon.js

@@ -14,6 +14,7 @@ import {
   COLUMNS_SQL,
   UPDATABLE_COLUMNS
 } from './schema.js';
+import * as pgShared from './pg-shared.js';
 
 export class NeonAdapter {
   /** @param {string} connectionString */
@@ -198,7 +199,7 @@ export class NeonAdapter {
   async getPendingTasks(limit = 50) {
     const sql = this._getSql();
     return sql.query(
-      `SELECT id, user_id, uuid, encrypted_payload, message_type, next_send_at, status, retry_count
+      `SELECT id, user_id, uuid, encrypted_payload, message_type, next_send_at, retry_after, status, retry_count
        FROM scheduled_messages
        WHERE status = 'pending' AND next_send_at <= NOW()
          AND (lease_until IS NULL OR lease_until <= NOW())
@@ -209,65 +210,11 @@ export class NeonAdapter {
     );
   }
 
-  /**
-   * 领取一条到点的任务：在 lease_until 上写下「这条归我管到什么时候」，
-   * 本次投递期间别的 tick 领不走它。
-   *
-   * 租约写在自己的列上，next_send_at 全程不动——那一列是用户设的触发时刻，
-   * 任务列表要读它、循环任务推进下一次也要拿它当基准。
-   *
-   * 两个 tick 抢同一行时只有一个改得动，另一个拿不到 RETURNING 行，据此跳
-   * 过。WHERE 里的两个条件各管一件事：
-   *   - lease_until 为空或已过期：没人正在跑这条。领了任务的 tick 中途没了
-   *     也不会把行焊死，租约到期后自然可以被接手。
-   *   - next_send_at 等于读这行时看到的值：读出来之后用户又改了排期的话，
-   *     这一跳就不该再按旧时刻发。
-   *
-   * 不加一个 'sending' 状态来表达「正在跑」：status 上有 CHECK 约束，加值
-   * 要改表。
-   *
-   * 比 next_send_at 时两边都截到毫秒：列是 timestamptz（微秒精度），驱动读
-   * 出来是 JS Date（毫秒精度），原值送回去可能因为亚毫秒差对不上。
-   *
-   * 带 serializeGroup 时多一道分组门：同一分组里已经有别的行拿着未到期的租
-   * 约，这条就领不走（同一分组同时只跑一条）。判定和写租约在同一条 UPDATE
-   * 里完成，「先查再占」的空档天然不存在。分组门只看租约，不看
-   * `retry_after`：等着重试的任务其实闲着，不该把同分组的其他任务一起堵住。
-   *
-   * @param {number} taskId
-   * @param {string|Date} expectedNextSendAt - 读这行时拿到的 next_send_at 原值
-   * @param {string|Date} leaseUntil - 租期末尾
-   * @param {string|null} [serializeGroup] - 串行分组标识；空表示不参与分组串行
-   * @returns {Promise<boolean>} true = 领到了；false = 别人正拿着租约、同分组
-   *   有任务正在跑、排期被改过、或行已不是 pending
-   */
+  // 领取一条到点的任务。SQL 与并发语义在 pg-shared.js（pg / neon 共用一份，
+  // 语义说明也在那里）。
   async claimTask(taskId, expectedNextSendAt, leaseUntil, serializeGroup = null) {
     const sql = this._getSql();
-    const grouped = typeof serializeGroup === 'string' && serializeGroup.length > 0;
-    const params = [leaseUntil, taskId, expectedNextSendAt];
-    let setClause = 'lease_until = $1, updated_at = NOW()';
-    let groupGuard = '';
-    if (grouped) {
-      params.push(serializeGroup); // $4
-      setClause = 'lease_until = $1, serialize_group = $4, updated_at = NOW()';
-      groupGuard = `
-          AND NOT EXISTS (
-            SELECT 1 FROM scheduled_messages busy
-             WHERE busy.serialize_group = $4 AND busy.id <> $2
-               AND busy.status = 'pending' AND busy.lease_until > NOW()
-          )`;
-    }
-    const rows = await sql.query(
-      `UPDATE scheduled_messages
-          SET ${setClause}
-        WHERE id = $2 AND status = 'pending'
-          AND date_trunc('milliseconds', next_send_at)
-            = date_trunc('milliseconds', $3::timestamptz)
-          AND (lease_until IS NULL OR lease_until <= NOW())${groupGuard}
-       RETURNING id`,
-      params
-    );
-    return rows.length > 0;
+    return pgShared.claimTask((text, params) => sql.query(text, params), taskId, expectedNextSendAt, leaseUntil, serializeGroup);
   }
 
   async listTasks(userId, opts = {}) {
@@ -327,60 +274,20 @@ export class NeonAdapter {
     return rows.length > 0 ? rows[0].status : null;
   }
 
-  // ── push_subscriptions (user-level Web Push subscription) ──────────────
+  // ── push_subscriptions（实现见 pg-shared.js，pg / neon 共用一份）─────────
 
-  /**
-   * 这个用户当前登记的推送订阅（密文原样返回，解密在上层）。
-   *
-   * @param {string} userId
-   * @returns {Promise<{ subscription: string, updated_at: number }|null>}
-   */
   async getPushSubscription(userId) {
     const sql = this._getSql();
-    const rows = await sql.query(
-      'SELECT subscription, updated_at FROM push_subscriptions WHERE user_id = $1 LIMIT 1',
-      [userId]
-    );
-    if (rows.length === 0) return null;
-    // BIGINT 在 pg 驱动里读出来是字符串，统一成 number 再往上给。
-    return { subscription: rows[0].subscription, updated_at: Number(rows[0].updated_at) };
+    return pgShared.getPushSubscription((text, params) => sql.query(text, params), userId);
   }
 
-  /**
-   * 覆盖写这个用户的订阅。一个用户一行，没有 last-write-wins 之类的比较——
-   * 客户端拿到的新订阅永远比旧的有效，旧的那份只会 410。
-   *
-   * @param {string} userId
-   * @param {string} encryptedSubscription
-   * @param {number} updatedAt - epoch 毫秒
-   * @returns {Promise<boolean>}
-   */
   async upsertPushSubscription(userId, encryptedSubscription, updatedAt) {
     const sql = this._getSql();
-    const rows = await sql.query(
-      `INSERT INTO push_subscriptions (user_id, subscription, updated_at)
-       VALUES ($1, $2, $3)
-       ON CONFLICT (user_id) DO UPDATE SET
-         subscription = EXCLUDED.subscription,
-         updated_at = EXCLUDED.updated_at
-       RETURNING user_id`,
-      [userId, encryptedSubscription, updatedAt]
-    );
-    return rows.length > 0;
+    return pgShared.upsertPushSubscription((text, params) => sql.query(text, params), userId, encryptedSubscription, updatedAt);
   }
 
-  /**
-   * 删掉这个用户的订阅（设置页「停止接收推送」）。
-   *
-   * @param {string} userId
-   * @returns {Promise<boolean>} true = 确实删掉了一行
-   */
   async deletePushSubscription(userId) {
     const sql = this._getSql();
-    const rows = await sql.query(
-      'DELETE FROM push_subscriptions WHERE user_id = $1 RETURNING user_id',
-      [userId]
-    );
-    return rows.length > 0;
+    return pgShared.deletePushSubscription((text, params) => sql.query(text, params), userId);
   }
 }

packages/rei-standard-amsg/server/src/server/adapters/pg-shared.js

@@ -0,0 +1,151 @@
+/**
+ * pg / neon 共用的 Postgres 查询实现。
+ *
+ * 两个适配器只在「怎么把 SQL 发出去」上不同（pg 的连接池 vs neon 的 HTTP
+ * 驱动），SQL 与并发语义必须逐字一致——各自复制一份的话，修一边漏一边就会让
+ * 两种 Postgres 部署的串行化行为静默分歧，而各自的测试还都是绿的。所以这里
+ * 按执行器参数化：适配器只递一个 `query(text, params) → rows`。
+ *
+ * D1（SQLite 方言、单写者、ISO TEXT 时间戳）不走这份实现。
+ *
+ * @typedef {(text: string, params?: any[]) => Promise<any[]>} PgQuery
+ */
+
+/**
+ * 领取一条到点的任务：在 lease_until 上写下「这条归我管到什么时候」，
+ * 本次投递期间别的 tick 领不走它。
+ *
+ * 租约写在自己的列上，next_send_at 全程不动——那一列是用户设的触发时刻，
+ * 任务列表要读它、循环任务推进下一次也要拿它当基准。
+ *
+ * 两个 tick 抢同一行时只有一个改得动，另一个拿不到 RETURNING 行，据此跳
+ * 过。WHERE 里的两个条件各管一件事：
+ *   - lease_until 为空或已过期：没人正在跑这条。领了任务的 tick 中途没了
+ *     也不会把行焊死，租约到期后自然可以被接手。
+ *   - next_send_at 等于读这行时看到的值：读出来之后用户又改了排期的话，
+ *     这一跳就不该再按旧时刻发。
+ *
+ * 不加一个 'sending' 状态来表达「正在跑」：status 上有 CHECK 约束，加值
+ * 要改表。
+ *
+ * 比 next_send_at 时两边都截到毫秒：列是 timestamptz（微秒精度），驱动读
+ * 出来是 JS Date（毫秒精度），原值送回去可能因为亚毫秒差对不上。
+ *
+ * 带 serializeGroup 时多一道分组门：同一分组里已经有别的行拿着未到期的租
+ * 约，这条就领不走（同一分组同时只跑一条）。判定和写租约在同一条 UPDATE
+ * 里完成——但那只对「抢同一行」成立；READ COMMITTED 下两个并发 tick 各领
+ * 同组的**不同**行时，各自的 NOT EXISTS 子查询都看不到对方尚未提交的租约，
+ * 也没有行锁冲突逼它重查（写偏斜）。所以占位成功后再回头查一次：真撞上了
+ * 就把自己刚写的租约放掉、这一跳不跑——两边都让也没事，行保持 pending，
+ * 下一跳重试。分组门只看租约，不看 `retry_after`：等着重试的任务其实闲着，
+ * 不该把同分组的其他任务一起堵住。
+ *
+ * @param {PgQuery} query
+ * @param {number} taskId
+ * @param {string|Date} expectedNextSendAt - 读这行时拿到的 next_send_at 原值
+ * @param {string|Date} leaseUntil - 租期末尾
+ * @param {string|null} [serializeGroup] - 串行分组标识；空表示不参与分组串行
+ * @returns {Promise<boolean>} true = 领到了；false = 已被别人领走、同分组有
+ *   任务正在跑、排期被改过、或行已不是 pending
+ */
+export async function claimTask(query, taskId, expectedNextSendAt, leaseUntil, serializeGroup = null) {
+  const grouped = typeof serializeGroup === 'string' && serializeGroup.length > 0;
+  const params = [leaseUntil, taskId, expectedNextSendAt];
+  let setClause = 'lease_until = $1, updated_at = NOW()';
+  let groupGuard = '';
+  if (grouped) {
+    params.push(serializeGroup); // $4
+    setClause = 'lease_until = $1, serialize_group = $4, updated_at = NOW()';
+    groupGuard = `
+        AND NOT EXISTS (
+          SELECT 1 FROM scheduled_messages busy
+           WHERE busy.serialize_group = $4 AND busy.id <> $2
+             AND busy.status = 'pending' AND busy.lease_until > NOW()
+        )`;
+  }
+  const rows = await query(
+    `UPDATE scheduled_messages
+        SET ${setClause}
+      WHERE id = $2 AND status = 'pending'
+        AND date_trunc('milliseconds', next_send_at)
+          = date_trunc('milliseconds', $3::timestamptz)
+        AND (lease_until IS NULL OR lease_until <= NOW())${groupGuard}
+     RETURNING id`,
+    params
+  );
+  if (rows.length === 0) return false;
+  if (!grouped) return true;
+
+  // 写偏斜的收口（见函数头注释）：自己的租约已提交，此刻再查，对方若也领
+  // 到了同组的另一行，双方至少有一方看得见冲突并让路。
+  const conflict = await query(
+    `SELECT 1 FROM scheduled_messages busy
+      WHERE busy.serialize_group = $1 AND busy.id <> $2
+        AND busy.status = 'pending' AND busy.lease_until > NOW()
+      LIMIT 1`,
+    [serializeGroup, taskId]
+  );
+  if (conflict.length === 0) return true;
+  await query(
+    'UPDATE scheduled_messages SET lease_until = NULL, updated_at = NOW() WHERE id = $1',
+    [taskId]
+  );
+  return false;
+}
+
+// ── push_subscriptions (user-level Web Push subscription) ──────────────
+
+/**
+ * 这个用户当前登记的推送订阅（密文原样返回，解密在上层）。
+ *
+ * @param {PgQuery} query
+ * @param {string} userId
+ * @returns {Promise<{ subscription: string, updated_at: number }|null>}
+ */
+export async function getPushSubscription(query, userId) {
+  const rows = await query(
+    'SELECT subscription, updated_at FROM push_subscriptions WHERE user_id = $1 LIMIT 1',
+    [userId]
+  );
+  if (rows.length === 0) return null;
+  // BIGINT 在 pg 驱动里读出来是字符串，统一成 number 再往上给。
+  return { subscription: rows[0].subscription, updated_at: Number(rows[0].updated_at) };
+}
+
+/**
+ * 覆盖写这个用户的订阅。一个用户一行，没有 last-write-wins 之类的比较——
+ * 客户端拿到的新订阅永远比旧的有效，旧的那份只会 410。
+ *
+ * @param {PgQuery} query
+ * @param {string} userId
+ * @param {string} encryptedSubscription
+ * @param {number} updatedAt - epoch 毫秒
+ * @returns {Promise<boolean>}
+ */
+export async function upsertPushSubscription(query, userId, encryptedSubscription, updatedAt) {
+  const rows = await query(
+    `INSERT INTO push_subscriptions (user_id, subscription, updated_at)
+     VALUES ($1, $2, $3)
+     ON CONFLICT (user_id) DO UPDATE SET
+       subscription = EXCLUDED.subscription,
+       updated_at = EXCLUDED.updated_at
+     RETURNING user_id`,
+    [userId, encryptedSubscription, updatedAt]
+  );
+  return rows.length > 0;
+}
+
+/**
+ * 删掉这个用户的订阅（设置页「停止接收推送」）。
+ *
+ * @param {PgQuery} query
+ * @param {string} userId
+ * @returns {Promise<boolean>} true = 确实删掉了一行
+ */
+export async function deletePushSubscription(query, userId) {
+  const rows = await query(
+    'DELETE FROM push_subscriptions WHERE user_id = $1 RETURNING user_id',
+    [userId]
+  );
+  return rows.length > 0;
+}