fix(db-server,sdk): 新脚本实例启动时回收旧实例遗留的事务会话

.changeset/tx-reap-on-start.md

@@ -0,0 +1,6 @@
+---
+"@sfmc-bds/db-server": patch
+"@sfmc-bds/sdk": patch
+---
+
+修复 `/reload` 后模块初始化偶发「事务繁忙」：旧脚本实例刚开启、来不及提交或回滚的交互式事务会话会占住 db-server 全局事务槽直到空闲超时（15 秒），新实例模块 init 的读写因此失败。db-server 新增 `POST /api/sfmc/db/tx/reap`（只中止鉴权身份所属模块的会话，返回 `aborted` 数量）；SDK 宿主在 `ModuleRegistry.bootAll()` 之前对各模块调用一次回收，失败只记日志、不阻断启动（兼容未升级的 db-server）。

modules/sdk/@sfmc-sdk/src/module-loader/install.ts

@@ -168,6 +168,8 @@ export function installHostBootstrap(options: InstallOptions = {}): HostBackend
       const db = getDbClient(d.id);
       setServiceModuleContext(d.id, token, () => db.isTxRecording());
     }
+    // db-server 全局只有一个事务槽，任一模块的遗留会话都会挡住其他模块，必须在全部模块 boot 前回收
+    await reapStaleTxSessions(ModuleRegistry.list().map((d) => d.id));
     await ModuleRegistry.bootAll();
     announceLoaded();
     // 若环境无 worldLoad 事件（如 1.18 稳定版）或事件在 ConfigManager 就绪前已触发，立即执行 bootAfterWorldLoad
@@ -222,6 +224,27 @@ export function installHostBootstrap(options: InstallOptions = {}): HostBackend
   return _bootstrapBackend();
 }
 
+/**
+ * 回收各模块旧脚本实例遗留的交互式事务会话。
+ * 使用场景：runStartup 在 ModuleRegistry.bootAll() 之前调用。/reload 时旧实例刚开启的事务
+ * 来不及提交或回滚，会占住 db-server 的全局事务槽直到空闲超时，新实例的模块 init 读写因此返回「事务繁忙」。
+ * 逐个模块串行请求；没有 token 的模块跳过，单个失败（如旧版 db-server 无此接口）只记日志，不阻断启动。
+ */
+async function reapStaleTxSessions(moduleIds: string[]): Promise<void> {
+  let aborted = 0;
+  const failed: string[] = [];
+  for (const id of moduleIds) {
+    if (!ConfigManager.getModuleToken(id)) continue;
+    try {
+      aborted += await getDbClient(id).reapOrphanSessions();
+    } catch (e) {
+      failed.push(`${id}: ${(e as Error).message}`);
+    }
+  }
+  if (aborted > 0) debug.w("HOST", `已回收旧脚本实例遗留的 ${aborted} 个事务会话`);
+  if (failed.length > 0) debug.w("HOST", `回收遗留事务会话失败: ${failed.join("; ")}`);
+}
+
 function _bootstrapBackend(): HostBackend {
   return {
     bindDataAdapter: (adapter: DataAdapter) => {

modules/sdk/@sfmc-sdk/src/sapi/db/client.ts

@@ -92,6 +92,12 @@ export interface DbClient {
   tx<T>(fn: (tx: TxContext) => Promise<T>): Promise<T>;
   audit(table: string, rowId: string | number, action: string, data?: Record<string, unknown>): Promise<void>;
   idempotent<T>(action: string, key: string, fn: () => Promise<T>): Promise<T>;
+  /**
+   * 让 db-server 中止本模块名下仍未结束的交互式事务会话，返回被中止的数量。
+   * 使用场景：宿主 install 在新脚本实例启动、模块 boot 之前调用，回收 /reload 前旧实例遗留的会话；
+   * 业务模块不应调用（会中止自己正在进行的事务）。
+   */
+  reapOrphanSessions(): Promise<number>;
 }
 
 type StepOk = { ok: true; result: TxStepResult };
@@ -297,6 +303,13 @@ export function createDbClient(moduleId: string, token: string): DbClient {
       await post("/api/sfmc/db/idempotent/commit", { action, key });
       return result;
     },
+
+    async reapOrphanSessions() {
+      requireModuleContext("reapOrphanSessions");
+      if (state.currentTxId) throw new DbError("reapOrphanSessions 不可在事务内调用", "forbidden", 0);
+      const res = await post<{ aborted?: number }>("/api/sfmc/db/tx/reap", {});
+      return typeof res?.aborted === "number" ? res.aborted : 0;
+    },
   };
 
   return client;
@@ -391,7 +404,7 @@ export function isDbTxRecording(): boolean {
 }
 
 /** SAPI 侧数据库门面（兼容单例）：转发到当前激活的作用域客户端。 */
-export const db: Omit<DbClient, "moduleId" | "setAuthToken" | "isTxRecording"> = {
+export const db: Omit<DbClient, "moduleId" | "setAuthToken" | "isTxRecording" | "reapOrphanSessions"> = {
   defineTable(name, columns, opts) {
     return activeClient().defineTable(name, columns, opts);
   },

packages/db-server/src/routes/db-routes.ts

@@ -8,6 +8,7 @@
  *   POST /api/sfmc/db/tx/step           body {txId, step}                → {result}
  *   POST /api/sfmc/db/tx/commit         body {txId}                      → TxResponse
  *   POST /api/sfmc/db/tx/rollback       body {txId}                      → {ok}
+ *   POST /api/sfmc/db/tx/reap           body {}                          → {ok, aborted}（回收本模块遗留会话）
  *   POST /api/sfmc/db/query             body {table, opts?}              → {rows}
  *   ...
  */
@@ -120,6 +121,11 @@ export function createDbRoutes(depsIn: Partial<DbRoutesDeps>) {
         requireTxId: true,
         run: ({ txId }) => deps.txRunner!.rollbackSession(txId, moduleId) as SessionReply,
       },
+      {
+        // 新脚本实例启动时回收本模块旧实例遗留的会话；只作用于鉴权身份自己的会话
+        path: "/api/sfmc/db/tx/reap",
+        run: () => ({ ok: true, aborted: deps.txRunner!.abortModuleSessions(moduleId) }),
+      },
     ];
     for (const op of sessionOps) {
       if (path !== op.path) continue;

packages/db-server/src/tx-runner.ts

@@ -386,6 +386,26 @@ export class TxRunner {
     return { ok: true };
   }
 
+  /**
+   * 中止某模块名下的全部交互式事务会话。
+   * 使用场景：BDS 执行 /reload 或重启后，新脚本实例在任何模块初始化之前经
+   * POST /api/sfmc/db/tx/reap 调用，回收旧实例来不及提交/回滚而遗留的会话，
+   * 避免它们占住全局事务槽直到空闲超时，导致其他模块的读写返回「事务繁忙」。
+   *
+   * @param moduleId 会话所属模块（由路由按鉴权身份传入，模块只能回收自己的会话）。
+   * @returns 被中止的会话数量。
+   */
+  abortModuleSessions(moduleId: string): number {
+    const stale = [...this.sessions.entries()]
+      .filter(([, session]) => session.moduleId === moduleId)
+      .map(([txId]) => txId);
+    for (const txId of stale) {
+      log.warn(`[tx-session ${txId.slice(0, 8)}] reaped module=${moduleId}（脚本实例已重载）`);
+      this.abortSession(txId);
+    }
+    return stale.length;
+  }
+
   private abortSession(txId: string): void {
     const session = this.sessions.get(txId);
     if (!session) return;

packages/db-server/test/tx-reap.test.ts

@@ -0,0 +1,68 @@
+/**
+ * tx-reap.test.ts — 交互式事务会话回收（abortModuleSessions / POST /api/sfmc/db/tx/reap）
+ *
+ * 场景：BDS /reload 时旧脚本实例刚开启的会话来不及提交或回滚，会占住全局事务槽直到空闲超时。
+ * 覆盖：回收只作用于指定模块、回收后事务槽立即可用、无遗留会话时返回 0。
+ */
+import assert from "node:assert/strict";
+import { DatabaseSync } from "node:sqlite";
+import { test } from "node:test";
+import { createQuery } from "../src/lib/sqlite.js";
+import type { ModuleManifestV2 } from "../src/manifest-loader.js";
+import type { SchemaRegistry } from "../src/schema-registry.js";
+import type { ServiceRegistry } from "../src/service-registry.js";
+import { TxRunner } from "../src/tx-runner.js";
+
+/** 只用到 begin/rollback/abort 的 TxRunner：schema 与 service 不会被调用，manifest 只需存在。 */
+function createRunner(moduleIds: string[]) {
+  const db = new DatabaseSync(":memory:");
+  const enabled = new Map(moduleIds.map((id) => [id, {} as ModuleManifestV2]));
+  const runner = new TxRunner(
+    {
+      db,
+      query: createQuery(db),
+      schema: {} as SchemaRegistry,
+      serviceRegistry: {} as ServiceRegistry,
+      enabled,
+    },
+    { slotWaitTimeoutMs: 50, sessionIdleTimeoutMs: 60_000 }
+  );
+  return { db, runner };
+}
+
+test("abortModuleSessions only reaps the given module and frees the transaction slot", async () => {
+  const { db, runner } = createRunner(["economy", "land"]);
+  try {
+    const stale = await runner.beginSession("economy");
+    assert.equal(stale.ok, true);
+
+    // 遗留会话占住事务槽：其他模块开事务会等待超时并返回 transaction_busy
+    const blocked = await runner.beginSession("land");
+    assert.equal(blocked.ok, false);
+    assert.equal(!blocked.ok && blocked.code, "transaction_busy");
+
+    // 回收别的模块不会动 economy 的会话
+    assert.equal(runner.abortModuleSessions("land"), 0);
+    assert.equal((await runner.beginSession("land")).ok, false);
+
+    // 回收 economy 后事务槽立即可用
+    assert.equal(runner.abortModuleSessions("economy"), 1);
+    const fresh = await runner.beginSession("land");
+    assert.equal(fresh.ok, true);
+    if (fresh.ok) assert.deepEqual(runner.rollbackSession(fresh.txId, "land"), { ok: true });
+
+    // 被回收的会话已结束：再回滚按幂等处理
+    if (stale.ok) assert.deepEqual(runner.rollbackSession(stale.txId, "economy"), { ok: true });
+  } finally {
+    db.close();
+  }
+});
+
+test("abortModuleSessions returns 0 when the module has no open session", () => {
+  const { db, runner } = createRunner(["economy"]);
+  try {
+    assert.equal(runner.abortModuleSessions("economy"), 0);
+  } finally {
+    db.close();
+  }
+});