diff --git a/.changeset/tx-reap-on-start.md b/.changeset/tx-reap-on-start.md new file mode 100644 index 00000000..ada372e3 --- /dev/null +++ b/.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)。 diff --git a/modules/sdk/@sfmc-sdk/src/module-loader/install.ts b/modules/sdk/@sfmc-sdk/src/module-loader/install.ts index fdd44230..5402acf3 100644 --- a/modules/sdk/@sfmc-sdk/src/module-loader/install.ts +++ b/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 { + 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) => { diff --git a/modules/sdk/@sfmc-sdk/src/sapi/db/client.ts b/modules/sdk/@sfmc-sdk/src/sapi/db/client.ts index f610b823..8dd7e017 100644 --- a/modules/sdk/@sfmc-sdk/src/sapi/db/client.ts +++ b/modules/sdk/@sfmc-sdk/src/sapi/db/client.ts @@ -92,6 +92,12 @@ export interface DbClient { tx(fn: (tx: TxContext) => Promise): Promise; audit(table: string, rowId: string | number, action: string, data?: Record): Promise; idempotent(action: string, key: string, fn: () => Promise): Promise; + /** + * 让 db-server 中止本模块名下仍未结束的交互式事务会话,返回被中止的数量。 + * 使用场景:宿主 install 在新脚本实例启动、模块 boot 之前调用,回收 /reload 前旧实例遗留的会话; + * 业务模块不应调用(会中止自己正在进行的事务)。 + */ + reapOrphanSessions(): Promise; } 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 = { +export const db: Omit = { defineTable(name, columns, opts) { return activeClient().defineTable(name, columns, opts); }, diff --git a/packages/db-server/src/routes/db-routes.ts b/packages/db-server/src/routes/db-routes.ts index efae9519..3404034b 100644 --- a/packages/db-server/src/routes/db-routes.ts +++ b/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) { 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; diff --git a/packages/db-server/src/tx-runner.ts b/packages/db-server/src/tx-runner.ts index fb51d187..187af920 100644 --- a/packages/db-server/src/tx-runner.ts +++ b/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; diff --git a/packages/db-server/test/tx-reap.test.ts b/packages/db-server/test/tx-reap.test.ts new file mode 100644 index 00000000..a83e5b7a --- /dev/null +++ b/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(); + } +});