diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index b583c8932..5291b460e 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -16,6 +16,14 @@ --- +## 2026-08-27 外部生成历史采用受控保留清理 + +- 背景：`external_generation_job`、`external_generation_job_summary` 与 `external_generation_job_event` 都是持久化表；摘要和 payload 边界收紧后，已确认的终态历史仍会继续占用 SpacetimeDB 常驻内存，且事件审计链会随任务数量增长。 +- 决策：新增仅 migration operator 可调用的 `prune_external_generation_job_history_and_return`。默认按 `source_module=editor-canvas`、30 天保留期和 `job_id` 游标分批运行；只删除主任务与摘要状态一致、属于 completed / failed / cancelled、摘要已有 `notification_acknowledged_at` 且终态时间达到 cutoff 的任务，并在同一事务内删除该任务的全部事件、摘要和主任务。默认 dry-run，必须固定 dry-run 返回的 cutoff 后再 apply；pending / running、未确认通知、摘要缺失或状态不一致的数据永不删除。其他 source module 必须显式指定并单独评估；资产对象和钱包流水不随任务历史删除；不新增自动定时器或 runtime 清理权限。 +- 影响范围：`server-rs/crates/spacetime-module/src/external_generation.rs`、外部生成事件 job_id 单列索引、SpacetimeDB 生成 bindings、`scripts/spacetime-maintain-external-generation-jobs.mjs`、架构与生产运维文档。 +- 验证方式：覆盖终态 / 活跃态 / 已确认与未确认摘要、状态或身份不一致、cutoff 边界测试；运行 SpacetimeDB module tests/check、bindings 生成、schema/encoding/diff 门禁，并在维护窗口先 dry-run 再 apply。 +- 关联文档：`docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md`、`docs/【开发运维】本地开发验证与生产运维-2026-05-15.md`、PR #203。 + ## 2026-08-27 SpacetimeDB 工具链统一升级到 2.8.3 - 背景：SpacetimeDB 2.8.0 引入 TypeScript submodule 与调度延迟观测，2.8.1 修复 v1 WebSocket 订阅移除死锁、TypeScript SDK `array` 读缓存别名和 Rust string 默认值支持，2.8.2 修复 table accessor 改名自动迁移，2.8.3 修复 scheduled function 从实际执行时间重排导致的长期漂移。仓库若继续锁定 2.7.0，会保留这些已知运行时与 SDK 问题。 diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index 1579cc82d..a707563c8 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -362,13 +362,14 @@ Responses 的终态载荷既是工具调用的恢复源，也是正文的恢复 - 源码：`server-rs/crates/spacetime-module/src/external_generation.rs` - 用途：外部生成正式任务列表的轻量投影，按 `job_id` 保存 owner、来源、状态、可选 `phase`、价格、有界错误摘要、通知确认时间、各阶段时间和入队时提取的 `request_prompt`，不包含 request/result payload、worker lease 或 dedupe 内部字段。错误摘要统一拒绝内联媒体并限制为 2048 字符；列表在单次 owner 扫描中同时计数并只保留请求 limit 的固定大小 top-N，不得先收集全量历史再截断。enqueue、claim、renew、phase update、complete、fail 事务同步投影；acknowledge 只更新该轻量表并写审计事件，后续主任务同步必须保留已有确认时间，禁止为了写确认时间加载 / 重写大 payload 行。BFF 的列表、状态和确认只调用 summary procedure；`running + processing` 映射为“正在处理”，其它 running（含旧行 `phase=None`）映射为“正在生成”。历史终态任务由迁移操作员的游标分批 maintenance procedure 在压缩 payload 时同步回填摘要，正式列表不得为兼容旧数据回扫完整主表。 - 非阻断告警：摘要字段 `warning_message` 是展示投影，由完成任务的轻量 `result_payload_json.warning.reason` 原样提取，不等同于公开 inline / external v1 的原始结构化诊断字段。complete 和历史 backfill 共用同一构建路径；历史任务按其结果载荷中已写入的 `reason` 快照投影，不为格式升级重写或补前缀。单 job 状态和任务列表 BFF 以 `warning: string` 返回该可直接展示的完整文案，不再返回结构化 code，Web 不得再次补前缀或按字符串推断告警类型。错误与告警摘要都不复制内联媒体并限制为 2048 字符。`phase` 与 `warning_message` 分别表示当前执行阶段和成功降级提示，不得混用；worker / BFF / Web 必须同版本协调发布，不保证滚动混部或旧 Web 缓存下的字符串语义兼容。 -- 正式读取 procedure 为 `get_external_generation_job_summary_and_return`、`list_external_generation_job_summaries_and_return` 和 `acknowledge_external_generation_job_summaries_and_return`。历史维护 procedure 为 `compact_external_generation_job_payloads_and_return` 与 `backfill_external_generation_job_summaries_and_return`，仅 migration operator 可调用；运维入口统一使用 `npm run spacetime:external-generation:maintain -- ...`，默认 dry-run、单批最多 25 条。B-tree cursor 选择阶段最多反序列化 `limit + 1` 行，apply 再按主键逐条读取选中行；怀疑存在单行异常巨型 JSON 时必须先使用 `--limit 1`。payload 压缩额外固定使用 `source_module = editor-canvas` 的复合 cursor 索引，不得静默改写其它玩法历史任务。 +- 正式读取 procedure 为 `get_external_generation_job_summary_and_return`、`list_external_generation_job_summaries_and_return` 和 `acknowledge_external_generation_job_summaries_and_return`。历史维护 procedure 为 `compact_external_generation_job_payloads_and_return`、`backfill_external_generation_job_summaries_and_return` 与 `prune_external_generation_job_history_and_return`，仅 migration operator 可调用；运维入口统一使用 `npm run spacetime:external-generation:maintain -- ...`，默认 dry-run、单批最多 25 条。B-tree cursor 选择阶段最多反序列化 `limit + 1` 行，apply 再按主键逐条读取选中行；怀疑存在单行异常巨型 JSON 时必须先使用 `--limit 1`。payload 压缩额外固定使用 `source_module = editor-canvas` 的复合 cursor 索引，不得静默改写其它玩法历史任务。历史清理默认使用 `--prune-history`、`source_module = editor-canvas` 和 30 天保留期；只有主任务与摘要状态一致且属于 completed / failed / cancelled、摘要已有 `notification_acknowledged_at`、终态时间不晚于 cutoff 的记录才是候选。apply 在同一事务内按事件 → 摘要 → 主任务顺序删除，事件不得独立清理；pending / running、未确认通知、摘要缺失或状态不一致的记录永不删除。清理不触碰资产对象或钱包流水，其他 source module 必须显式指定并单独评估。 ### `external_generation_job_event` - Rust 结构体：`ExternalGenerationJobEvent` - 源码：`server-rs/crates/spacetime-module/src/external_generation.rs` - 用途：外部生成任务审计事件表，按 `job_id` 和 `owner_user_id` 记录 `enqueued`、`claimed`、`lease_renewed`、`completed`、`failed`、`acknowledged` 等状态转换事实。状态转换只能由 SpacetimeDB procedure 写入，不由前端或 worker 直接改表；该表用于追溯任务生命周期和排障，不替代 `external_generation_job` 当前状态。 +- 保留策略：事件只会随已确认通知的终态任务由 `prune_external_generation_job_history_and_return` 原子删除，不支持按事件单独清理，以保持任务、摘要和审计链一致。 ### `ai_text_chunk` diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index a344a4425..05ed13b33 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -115,7 +115,7 @@ BgFilter 对已经落入私有 OSS 的生成原图、动作抽取帧和手动去 图片编辑器任务侧栏与生成提交工作流只读取 BFF 队列接口：`GET /api/runtime/external-generation/jobs` 列出当前用户任务，`GET /api/runtime/external-generation/jobs/{jobId}` 查看单 job 状态，概览场景可使用 `GET /api/runtime/external-generation/queue-overview`。前端不直接查询 `external_generation_job` private table，也不展示 worker 内部 payload；完成态以编辑器项目和资源接口返回的正式数据为准。 -外部生成任务摘要投影与历史 payload 维护使用 `npm run spacetime:external-generation:maintain -- ...`，且只能由已授权 migration operator 的 SpacetimeDB CLI 登录态执行。脚本默认 dry-run、每次只处理一批，绝不自动循环全表；`--apply` 才写入。先发布包含 `external_generation_job_summary` 与 cursor 索引的 SpacetimeDB 模块，在维护模式内对事故时间以前的编辑器终态任务执行小批 dry-run，例如 `npm run spacetime:external-generation:maintain -- --database --server-url --limit 5 --completed-before-micros `；核对 `matched_count`、`before_bytes`、`after_bytes` 和 `inline_media_count` 后，保持本批输入 cursor 不变并追加 `--apply` 重跑同一批，即使最后一批 `has_more = false`，只要 dry-run 仍有 `matched_count` / `selected_count` 也必须 apply；只有 apply 成功后才使用它返回的 `next_cursor_job_id` 继续。B-tree cursor 的选择阶段最多反序列化 `limit + 1` 行，apply 会再按主键逐条读取选中行但不会同时保留整批 payload；如怀疑存在单行异常巨型历史 JSON，先用 `--limit 1`。payload 压缩硬限制 `source_module = editor-canvas`；终态压缩完成后，用 `--backfill-summaries` 先 dry-run、再 `--apply` 分批补齐仍缺失的活动任务或无内联媒体历史任务摘要，直到 `has_more = false`，最后再切换使用 summary procedure 的 api-server。Stdb 构建 artifact 和完整 release 包都必须包含 `scripts/spacetime-maintain-external-generation-jobs.mjs` 与 `scripts/spacetime-migration-common.mjs`。首次上线不得让 Full Build 从 Stdb 自动直落 API：`STDB_API_ROLLOUT_MODE` 默认 fail-closed 为 `pause-after-stdb`，必须填写受限的 `STDB_API_ROLLOUT_APPROVERS`；Stdb Publish 通过 `KEEP_MAINTENANCE_MODE` 保持维护文件并停止旧 API/controller/worker，暂停点最多等待 4 小时，完成上述维护并确认无后续批次后才由指定审批人放行 API。定时构建缺少审批人时必须在发布前失败，不能静默退回 `normal`；也可分开运行 Stdb publish、维护、API deploy 三个受控 Job。任一批次都不得处理 pending / running payload；不要用 runtime writer、bootstrap secret 或匿名 identity 代替 migration operator，也不要在未核对 dry-run 时直接 apply。 +外部生成任务摘要投影与历史 payload / history 维护使用 `npm run spacetime:external-generation:maintain -- ...`，且只能由已授权 migration operator 的 SpacetimeDB CLI 登录态执行。脚本默认 dry-run、每次只处理一批，绝不自动循环全表；`--apply` 才写入。先发布包含 `external_generation_job_summary` 与 cursor 索引的 SpacetimeDB 模块，在维护模式内对事故时间以前的编辑器终态任务执行小批 dry-run，例如 `npm run spacetime:external-generation:maintain -- --database --server-url --limit 5 --completed-before-micros `；核对 `matched_count`、`before_bytes`、`after_bytes` 和 `inline_media_count` 后，保持本批输入 cursor 不变并追加 `--apply` 重跑同一批，即使最后一批 `has_more = false`，只要 dry-run 仍有 `matched_count` / `selected_count` 也必须 apply；只有 apply 成功后才使用它返回的 `next_cursor_job_id` 继续。B-tree cursor 的选择阶段最多反序列化 `limit + 1` 行，apply 会再按主键逐条读取选中行但不会同时保留整批 payload；如怀疑存在单行异常巨型历史 JSON，先用 `--limit 1`。payload 压缩硬限制 `source_module = editor-canvas`；终态压缩完成后，用 `--backfill-summaries` 先 dry-run、再 `--apply` 分批补齐仍缺失的活动任务或无内联媒体历史任务摘要，直到 `has_more = false`，最后再切换使用 summary procedure 的 api-server。历史清理使用 `--prune-history`，默认 `source_module=editor-canvas`、30 天保留期；候选必须是 completed / failed / cancelled 终态、主任务与摘要状态一致、摘要存在 `notification_acknowledged_at` 且终态时间不晚于 `completed_before_micros`，否则永不删除。先 dry-run，记下输出的 `completed_before_micros`，再保持相同 `--cursor-job-id` 与 cutoff 追加 `--apply`；apply 在一个事务中删除该 job 的所有 event、summary 和主任务，资产对象与钱包流水保留。需要清理其它 source module 时必须显式 `--source-module` 并先完成业务评估；这不是自动 systemd 任务，不得授予 runtime writer 清理权限。Stdb 构建 artifact 和完整 release 包都必须包含 `scripts/spacetime-maintain-external-generation-jobs.mjs` 与 `scripts/spacetime-migration-common.mjs`。首次上线不得让 Full Build 从 Stdb 自动直落 API：`STDB_API_ROLLOUT_MODE` 默认 fail-closed 为 `pause-after-stdb`，必须填写受限的 `STDB_API_ROLLOUT_APPROVERS`；Stdb Publish 通过 `KEEP_MAINTENANCE_MODE` 保持维护文件并停止旧 API/controller/worker，暂停点最多等待 4 小时，完成上述维护并确认无后续批次后才由指定审批人放行 API。定时构建缺少审批人时必须在发布前失败，不能静默退回 `normal`；也可分开运行 Stdb publish、维护、API deploy 三个受控 Job。任一批次都不得处理 pending / running payload；不要用 runtime writer、bootstrap secret 或匿名 identity 代替 migration operator，也不要在未核对 dry-run 时直接 apply。 角色动作正式字段收口使用 `node scripts/spacetime-normalize-editor-character-actions.mjs --database --server-url `，且同样只能由已授权 migration operator 执行。必须先发布包含 normalization cursor 索引和 `normalize_editor_character_animation_metadata_and_return` 的 SpacetimeDB 模块，在 API / worker 仍处于维护模式时先运行默认全量 dry-run；脚本固定按 `asset → project-resource → showcase → canvas` 扫描，普通 scope 每批最多 25 行，canvas 每批最多 5 行。全量 dry-run 会在不写库的情况下把 asset 计划结果投影给同 owner / task / 首帧对象精确匹配的 project-resource，再把前置 scope 的计划结果投影给 canvas 检查；因此同 task 的误标预览 MP4 会先按权威视频对象排除，最终图片序列会逐帧核对并补齐精确 `asset_object` 身份。canvas 中仍引用误标 preview resource 的普通 video layer 会按 project-resource 计划态 `video` 跳过，只有 layout 明确声明动作却指向视频，或资源规划本身失败时才形成 blocker。apply 时仍要求前置 scope 已按顺序物理完成，不能跳过 asset 直接让 project-resource 借未落库结果。历史 canvas 复制的 `sourceResourceId` 不是迁移证据，不要因它仍指向原角色而手工改库，补建资源会采用最终账号素材的 DB 血缘。出现 blocker 时脚本会打印 ID、原因、owner、project、task、对象身份和来源资源；先据此区分最终候选为零 / 多个、正式与旧版冲突、帧对象不匹配或缺失资源，不得跳过 scope。确认 dry-run 后追加 `--apply`，脚本会对每批重新 dry-run、携带该批 SHA-256 apply，并在最后从头要求四个 scope 均为零匹配、零 blocker。只有该复核通过后才发布移除 action fallback 的 API / Web。Stdb build artifact 和完整 release 包必须同时包含 `scripts/spacetime-normalize-editor-character-actions.mjs` 与 `scripts/spacetime-migration-common.mjs`。本地切换分支时若要避免 dev publish 因 schema 冲突使用 `-c=on-conflict` 清库，启动命令必须追加 `--preserve-database`，让冲突直接失败。动作视频抽帧临时目录固定使用 `/var/lib/genarrative/character-animation-tmp`，该路径已由生产 API / worker unit 放行；不要让动作抽帧重新依赖 `PrivateTmp` 下的 `/tmp`。 diff --git a/scripts/spacetime-maintain-external-generation-jobs.mjs b/scripts/spacetime-maintain-external-generation-jobs.mjs index 69ead2f3c..49a6c8625 100644 --- a/scripts/spacetime-maintain-external-generation-jobs.mjs +++ b/scripts/spacetime-maintain-external-generation-jobs.mjs @@ -8,23 +8,29 @@ import { } from './spacetime-migration-common.mjs'; const MAX_BATCH_SIZE = 25; +const DEFAULT_RETENTION_DAYS = 30; +const MICROS_PER_DAY = 86_400_000_000; function usage() { return `用法： node scripts/spacetime-maintain-external-generation-jobs.mjs --database [选项] 默认只 dry-run 一批历史终态任务 payload 压缩，不修改数据库。 +使用 --prune-history 时改为清理已确认通知且超过保留期的历史任务、摘要与事件。 公共选项： --database 目标数据库（必填，也可用 GENARRATIVE_SPACETIME_DATABASE） --server spacetime CLI server 名或 URL --server-url 显式 server URL - --limit <1-${MAX_BATCH_SIZE}> 单批任务数，默认 10 + --limit <1-${MAX_BATCH_SIZE}> 单批任务数，默认 10 --cursor-job-id 从上一批 next_cursor_job_id 继续 --apply 执行写入；省略时始终 dry-run --backfill-summaries 改为回填轻量摘要投影 + --prune-history 改为清理已确认通知的终态历史 --owner-user-id 仅摘要回填可选，限定 owner - --completed-before-micros 仅 payload 压缩可选，限定终态完成时间 + --source-module 仅历史清理可选，默认 editor-canvas + --retention-days 仅历史清理可选，默认 ${DEFAULT_RETENTION_DAYS} 天 + --completed-before-micros 限定终态完成时间；历史清理默认按 retention-days 计算 --help 显示帮助 必须使用已授权 migration operator 的 spacetime CLI 登录态。脚本每次只处理一批； @@ -40,6 +46,9 @@ function parseOptions(argv) { database: process.env.GENARRATIVE_SPACETIME_DATABASE || '', limit: 10, ownerUserId: '', + pruneHistory: false, + retentionDays: DEFAULT_RETENTION_DAYS, + sourceModule: 'editor-canvas', passthrough: [], server: process.env.GENARRATIVE_SPACETIME_SERVER || '', serverUrl: process.env.GENARRATIVE_SPACETIME_SERVER_URL || '', @@ -82,6 +91,15 @@ function parseOptions(argv) { options.apply = true; } else if (arg === '--backfill-summaries') { options.backfillSummaries = true; + } else if (arg === '--prune-history') { + options.pruneHistory = true; + } else if (arg === '--source-module') { + options.sourceModule = readValue(arg).trim(); + if (!options.sourceModule) { + throw new Error('--source-module 不能为空。'); + } + } else if (arg === '--retention-days') { + options.retentionDays = parsePositiveInteger(readValue(arg), arg); } else if (arg === '--help' || arg === '-h') { options.help = true; } else { @@ -95,12 +113,49 @@ function parseOptions(argv) { if (options.ownerUserId && !options.backfillSummaries) { throw new Error('--owner-user-id 只能与 --backfill-summaries 一起使用。'); } + if (options.backfillSummaries && options.pruneHistory) { + throw new Error('--backfill-summaries 与 --prune-history 不能同时使用。'); + } + if (options.sourceModule !== 'editor-canvas' && !options.pruneHistory) { + throw new Error('--source-module 只能与 --prune-history 一起使用。'); + } + if ( + options.retentionDays !== DEFAULT_RETENTION_DAYS && + !options.pruneHistory + ) { + throw new Error('--retention-days 只能与 --prune-history 一起使用。'); + } if (options.completedBeforeMicros !== null && options.backfillSummaries) { throw new Error('--completed-before-micros 不能用于摘要回填。'); } + if ( + options.completedBeforeMicros !== null && + options.pruneHistory && + options.retentionDays !== DEFAULT_RETENTION_DAYS + ) { + throw new Error( + '--completed-before-micros 与 --retention-days 不能同时使用。', + ); + } return options; } +function resolveRetentionCutoffMicros(options) { + if (!options.pruneHistory) { + return options.completedBeforeMicros; + } + if (options.completedBeforeMicros !== null) { + return options.completedBeforeMicros; + } + const cutoff = Date.now() * 1000 - options.retentionDays * MICROS_PER_DAY; + if (!Number.isSafeInteger(cutoff)) { + throw new Error( + '--retention-days 计算出的 completed_before_micros 超出安全整数范围。', + ); + } + return cutoff; +} + try { const options = parseOptions(process.argv.slice(2)); if (options.help) { @@ -113,24 +168,36 @@ try { ); } - const procedureName = options.backfillSummaries - ? 'backfill_external_generation_job_summaries_and_return' - : 'compact_external_generation_job_payloads_and_return'; - const input = options.backfillSummaries + const completedBeforeMicros = resolveRetentionCutoffMicros(options); + + const procedureName = options.pruneHistory + ? 'prune_external_generation_job_history_and_return' + : options.backfillSummaries + ? 'backfill_external_generation_job_summaries_and_return' + : 'compact_external_generation_job_payloads_and_return'; + const input = options.pruneHistory ? { - owner_user_id: encodeSpacetimeCliOption(options.ownerUserId || null), + source_module: options.sourceModule, limit: options.limit, cursor_job_id: encodeSpacetimeCliOption(options.cursorJobId || null), + completed_before_micros: completedBeforeMicros, dry_run: !options.apply, } - : { - dry_run: !options.apply, - limit: options.limit, - cursor_job_id: encodeSpacetimeCliOption(options.cursorJobId || null), - completed_before_micros: encodeSpacetimeCliOption( - options.completedBeforeMicros, - ), - }; + : options.backfillSummaries + ? { + owner_user_id: encodeSpacetimeCliOption(options.ownerUserId || null), + limit: options.limit, + cursor_job_id: encodeSpacetimeCliOption(options.cursorJobId || null), + dry_run: !options.apply, + } + : { + dry_run: !options.apply, + limit: options.limit, + cursor_job_id: encodeSpacetimeCliOption(options.cursorJobId || null), + completed_before_micros: encodeSpacetimeCliOption( + completedBeforeMicros, + ), + }; const result = await callSpacetimeProcedureViaCli( options, procedureName, @@ -138,10 +205,29 @@ try { ); ensureProcedureOk(result); - console.log(JSON.stringify({ procedure: procedureName, ...result }, null, 2)); - const pendingApplyCount = options.backfillSummaries - ? Number(result.selected_count ?? 0) - : Number(result.matched_count ?? 0); + console.log( + JSON.stringify( + { + procedure: procedureName, + ...(options.pruneHistory + ? { + source_module: options.sourceModule, + completed_before_micros: completedBeforeMicros, + ...(options.completedBeforeMicros === null + ? { retention_days: options.retentionDays } + : {}), + } + : {}), + ...result, + }, + null, + 2, + ), + ); + const pendingApplyCount = + options.pruneHistory || options.backfillSummaries + ? Number(result.selected_count ?? 0) + : Number(result.matched_count ?? 0); if (result.has_more && options.apply) { console.log( `仍有后续批次；下一次追加 --cursor-job-id ${result.next_cursor_job_id ?? ''}。`, @@ -150,8 +236,11 @@ try { const currentCursor = options.cursorJobId ? `保留 --cursor-job-id ${options.cursorJobId}` : '仍从首批开始'; + const cutoffHint = options.pruneHistory + ? `并固定 --completed-before-micros ${completedBeforeMicros}` + : ''; console.log( - `当前仅 dry-run；请${currentCursor}并追加 --apply 重跑同一批。apply 成功后再使用其 next_cursor_job_id 进入下一批。`, + `当前仅 dry-run；请${currentCursor}${cutoffHint}并追加 --apply 重跑同一批。apply 成功后再使用其 next_cursor_job_id 进入下一批。`, ); } } catch (error) { diff --git a/server-rs/crates/spacetime-client/src/module_bindings.rs b/server-rs/crates/spacetime-client/src/module_bindings.rs index 1b24a5403..7b45258ea 100644 --- a/server-rs/crates/spacetime-client/src/module_bindings.rs +++ b/server-rs/crates/spacetime-client/src/module_bindings.rs @@ -409,6 +409,8 @@ pub mod external_generation_job_procedure_result_type; pub mod external_generation_job_renew_lease_input_type; pub mod external_generation_job_result_procedure_result_type; pub mod external_generation_job_result_snapshot_type; +pub mod external_generation_job_retention_input_type; +pub mod external_generation_job_retention_procedure_result_type; pub mod external_generation_job_snapshot_type; pub mod external_generation_job_summary_backfill_input_type; pub mod external_generation_job_summary_backfill_procedure_result_type; @@ -570,6 +572,7 @@ pub mod profile_wallet_ledger_table; pub mod profile_wallet_ledger_type; pub mod profile_wallet_manual_restriction_table; pub mod profile_wallet_manual_restriction_type; +pub mod prune_external_generation_job_history_and_return_procedure; pub mod public_work_like_table; pub mod public_work_like_type; pub mod public_work_play_daily_stat_table; @@ -1268,6 +1271,8 @@ pub use external_generation_job_procedure_result_type::ExternalGenerationJobProc pub use external_generation_job_renew_lease_input_type::ExternalGenerationJobRenewLeaseInput; pub use external_generation_job_result_procedure_result_type::ExternalGenerationJobResultProcedureResult; pub use external_generation_job_result_snapshot_type::ExternalGenerationJobResultSnapshot; +pub use external_generation_job_retention_input_type::ExternalGenerationJobRetentionInput; +pub use external_generation_job_retention_procedure_result_type::ExternalGenerationJobRetentionProcedureResult; pub use external_generation_job_snapshot_type::ExternalGenerationJobSnapshot; pub use external_generation_job_summary_backfill_input_type::ExternalGenerationJobSummaryBackfillInput; pub use external_generation_job_summary_backfill_procedure_result_type::ExternalGenerationJobSummaryBackfillProcedureResult; @@ -1429,6 +1434,7 @@ pub use profile_wallet_ledger_table::*; pub use profile_wallet_ledger_type::ProfileWalletLedger; pub use profile_wallet_manual_restriction_table::*; pub use profile_wallet_manual_restriction_type::ProfileWalletManualRestriction; +pub use prune_external_generation_job_history_and_return_procedure::prune_external_generation_job_history_and_return; pub use public_work_like_table::*; pub use public_work_like_type::PublicWorkLike; pub use public_work_play_daily_stat_table::*; diff --git a/server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_event_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_event_type.rs index 32c820175..4658052db 100644 --- a/server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_event_type.rs +++ b/server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_event_type.rs @@ -56,6 +56,7 @@ impl __sdk::__query_builder::HasCols for ExternalGenerationJobEvent { /// Provides typed access to indexed columns for query building. pub struct ExternalGenerationJobEventIxCols { pub event_id: __sdk::__query_builder::IxCol, + pub job_id: __sdk::__query_builder::IxCol, } impl __sdk::__query_builder::HasIxCols for ExternalGenerationJobEvent { @@ -63,6 +64,7 @@ impl __sdk::__query_builder::HasIxCols for ExternalGenerationJobEvent { fn ix_cols(table_name: &'static str) -> Self::IxCols { ExternalGenerationJobEventIxCols { event_id: __sdk::__query_builder::IxCol::new(table_name, "event_id"), + job_id: __sdk::__query_builder::IxCol::new(table_name, "job_id"), } } } diff --git a/server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_retention_input_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_retention_input_type.rs new file mode 100644 index 000000000..cffb3527a --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_retention_input_type.rs @@ -0,0 +1,19 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +pub struct ExternalGenerationJobRetentionInput { + pub source_module: String, + pub limit: u32, + pub cursor_job_id: Option, + pub completed_before_micros: i64, + pub dry_run: bool, +} + +impl __sdk::InModule for ExternalGenerationJobRetentionInput { + type Module = super::RemoteModule; +} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_retention_procedure_result_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_retention_procedure_result_type.rs new file mode 100644 index 000000000..1f3883749 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_retention_procedure_result_type.rs @@ -0,0 +1,24 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +pub struct ExternalGenerationJobRetentionProcedureResult { + pub ok: bool, + pub dry_run: bool, + pub scanned_count: u64, + pub selected_count: u32, + pub deleted_job_count: u32, + pub deleted_summary_count: u32, + pub deleted_event_count: u32, + pub next_cursor_job_id: Option, + pub has_more: bool, + pub error_message: Option, +} + +impl __sdk::InModule for ExternalGenerationJobRetentionProcedureResult { + type Module = super::RemoteModule; +} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/prune_external_generation_job_history_and_return_procedure.rs b/server-rs/crates/spacetime-client/src/module_bindings/prune_external_generation_job_history_and_return_procedure.rs new file mode 100644 index 000000000..877d06380 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/prune_external_generation_job_history_and_return_procedure.rs @@ -0,0 +1,62 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +use super::external_generation_job_retention_input_type::ExternalGenerationJobRetentionInput; +use super::external_generation_job_retention_procedure_result_type::ExternalGenerationJobRetentionProcedureResult; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +struct PruneExternalGenerationJobHistoryAndReturnArgs { + pub input: ExternalGenerationJobRetentionInput, +} + +impl __sdk::InModule for PruneExternalGenerationJobHistoryAndReturnArgs { + type Module = super::RemoteModule; +} + +#[allow(non_camel_case_types)] +/// Extension trait for access to the procedure `prune_external_generation_job_history_and_return`. +/// +/// Implemented for [`super::RemoteProcedures`]. +pub trait prune_external_generation_job_history_and_return { + fn prune_external_generation_job_history_and_return( + &self, + input: ExternalGenerationJobRetentionInput, + ) { + self.prune_external_generation_job_history_and_return_then(input, |_, _| {}); + } + + fn prune_external_generation_job_history_and_return_then( + &self, + input: ExternalGenerationJobRetentionInput, + + __callback: impl FnOnce( + &super::ProcedureEventContext, + Result, + ) + Send + + 'static, + ); +} + +impl prune_external_generation_job_history_and_return for super::RemoteProcedures { + fn prune_external_generation_job_history_and_return_then( + &self, + input: ExternalGenerationJobRetentionInput, + + __callback: impl FnOnce( + &super::ProcedureEventContext, + Result, + ) + Send + + 'static, + ) { + self.imp + .invoke_procedure_with_callback::<_, ExternalGenerationJobRetentionProcedureResult>( + "prune_external_generation_job_history_and_return", + PruneExternalGenerationJobHistoryAndReturnArgs { input }, + __callback, + ); + } +} diff --git a/server-rs/crates/spacetime-module/src/external_generation.rs b/server-rs/crates/spacetime-module/src/external_generation.rs index 9cfa4fa13..82a51a64a 100644 --- a/server-rs/crates/spacetime-module/src/external_generation.rs +++ b/server-rs/crates/spacetime-module/src/external_generation.rs @@ -97,6 +97,10 @@ pub struct ExternalGenerationJob { accessor = by_external_generation_job_event_job_id, btree(columns = [job_id, created_at]) ), + index( + accessor = by_external_generation_job_event_job_id_only, + btree(columns = [job_id]) + ), index( accessor = by_external_generation_job_event_owner, btree(columns = [owner_user_id, created_at]) @@ -378,6 +382,29 @@ pub struct ExternalGenerationJobPayloadCompactionProcedureResult { pub error_message: Option, } +#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)] +pub struct ExternalGenerationJobRetentionInput { + pub source_module: String, + pub limit: u32, + pub cursor_job_id: Option, + pub completed_before_micros: i64, + pub dry_run: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)] +pub struct ExternalGenerationJobRetentionProcedureResult { + pub ok: bool, + pub dry_run: bool, + pub scanned_count: u64, + pub selected_count: u32, + pub deleted_job_count: u32, + pub deleted_summary_count: u32, + pub deleted_event_count: u32, + pub next_cursor_job_id: Option, + pub has_more: bool, + pub error_message: Option, +} + #[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)] pub struct ExternalGenerationQueueStatsSnapshot { pub pending_count: u32, @@ -676,6 +703,21 @@ pub fn compact_external_generation_job_payloads_and_return( } } +#[spacetimedb::procedure] +pub fn prune_external_generation_job_history_and_return( + ctx: &mut ProcedureContext, + input: ExternalGenerationJobRetentionInput, +) -> ExternalGenerationJobRetentionProcedureResult { + let caller = ctx.sender(); + match ctx.try_with_tx(|tx| { + crate::migration::require_migration_operator(tx, caller)?; + prune_external_generation_job_history_tx(tx, input.clone()) + }) { + Ok(result) => result, + Err(message) => failed_external_generation_job_retention_result(input.dry_run, message), + } +} + #[spacetimedb::procedure] pub fn get_external_generation_queue_stats_and_return( ctx: &mut ProcedureContext, @@ -1367,6 +1409,105 @@ fn compact_external_generation_job_payloads_tx( }) } +fn prune_external_generation_job_history_tx( + ctx: &ReducerContext, + input: ExternalGenerationJobRetentionInput, +) -> Result { + let source_module = input.source_module.trim().to_string(); + validate_required("external_generation_job.source_module", &source_module)?; + let now_micros = ctx.timestamp.to_micros_since_unix_epoch(); + if input.completed_before_micros > now_micros { + return Err( + "external_generation_job.completed_before_micros 不能晚于数据库当前时间".to_string(), + ); + } + + let cursor_job_id = input + .cursor_job_id + .as_deref() + .and_then(normalize_optional_text); + let limit = input + .limit + .clamp(1, MAX_EXTERNAL_GENERATION_MAINTENANCE_BATCH_SIZE) as usize; + let cursor_range = external_generation_job_maintenance_cursor_range(cursor_job_id.as_deref()); + let cursor_to_skip = cursor_job_id.clone(); + let rows = ctx + .db + .external_generation_job() + .by_external_generation_job_source_cursor() + .filter((source_module.as_str(), cursor_range)) + .filter(move |row| { + cursor_to_skip + .as_deref() + .is_none_or(|cursor| row.job_id != cursor) + }); + let (job_ids, next_cursor_job_id, has_more, scanned_count) = + select_external_generation_job_ids_for_maintenance(rows, limit, |row| { + ctx.db + .external_generation_job_summary() + .job_id() + .find(&row.job_id) + .is_some_and(|summary| { + is_external_generation_job_retention_candidate( + row, + &summary, + &source_module, + input.completed_before_micros, + ) + }) + }); + + let mut deleted_job_count = 0u32; + let mut deleted_summary_count = 0u32; + let mut deleted_event_count = 0u32; + if !input.dry_run { + for job_id in &job_ids { + let Some(row) = ctx.db.external_generation_job().job_id().find(job_id) else { + continue; + }; + let Some(summary) = ctx + .db + .external_generation_job_summary() + .job_id() + .find(job_id) + else { + continue; + }; + if !is_external_generation_job_retention_candidate( + &row, + &summary, + &source_module, + input.completed_before_micros, + ) { + continue; + } + + deleted_event_count = deleted_event_count + .saturating_add(delete_external_generation_job_events_for_job(ctx, job_id)); + ctx.db + .external_generation_job_summary() + .job_id() + .delete(job_id); + deleted_summary_count = deleted_summary_count.saturating_add(1); + ctx.db.external_generation_job().job_id().delete(job_id); + deleted_job_count = deleted_job_count.saturating_add(1); + } + } + + Ok(ExternalGenerationJobRetentionProcedureResult { + ok: true, + dry_run: input.dry_run, + scanned_count, + selected_count: job_ids.len() as u32, + deleted_job_count, + deleted_summary_count, + deleted_event_count, + next_cursor_job_id, + has_more, + error_message: None, + }) +} + fn renew_external_generation_job_lease_tx( ctx: &ReducerContext, input: ExternalGenerationJobRenewLeaseInput, @@ -1797,6 +1938,25 @@ fn should_compact_external_generation_job_payloads( }) } +fn is_external_generation_job_retention_candidate( + row: &ExternalGenerationJob, + summary: &ExternalGenerationJobSummary, + source_module: &str, + completed_before_micros: i64, +) -> bool { + row.source_module.trim() == source_module.trim() + && summary.job_id == row.job_id + && summary.status == row.status + && is_external_generation_job_terminal(row) + && is_external_generation_job_summary_terminal(summary) + && summary.notification_acknowledged_at.is_some() + && row + .completed_at + .unwrap_or(row.updated_at) + .to_micros_since_unix_epoch() + <= completed_before_micros +} + fn external_generation_job_maintenance_cursor_range( cursor_job_id: Option<&str>, ) -> RangeFrom<&str> { @@ -1832,6 +1992,24 @@ fn select_external_generation_job_ids_for_maintenance( ) } +fn delete_external_generation_job_events_for_job(ctx: &ReducerContext, job_id: &str) -> u32 { + let event_ids = ctx + .db + .external_generation_job_event() + .by_external_generation_job_event_job_id_only() + .filter(job_id) + .map(|event| event.event_id.clone()) + .collect::>(); + let deleted_count = event_ids.len() as u32; + for event_id in event_ids { + ctx.db + .external_generation_job_event() + .event_id() + .delete(&event_id); + } + deleted_count +} + fn count_external_generation_job_summaries_for_owner( ctx: &ReducerContext, owner_user_id: &str, @@ -2544,6 +2722,24 @@ fn failed_external_generation_job_payload_compaction_result( } } +fn failed_external_generation_job_retention_result( + dry_run: bool, + message: String, +) -> ExternalGenerationJobRetentionProcedureResult { + ExternalGenerationJobRetentionProcedureResult { + ok: false, + dry_run, + scanned_count: 0, + selected_count: 0, + deleted_job_count: 0, + deleted_summary_count: 0, + deleted_event_count: 0, + next_cursor_job_id: None, + has_more: false, + error_message: Some(message), + } +} + fn validate_required(field: &str, value: &str) -> Result<(), String> { if value.trim().is_empty() { return Err(format!("{field} 不能为空")); @@ -3434,6 +3630,86 @@ mod tests { )); } + #[test] + fn retention_only_selects_acknowledged_terminal_rows_before_cutoff() { + let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_COMPLETED); + row.source_module = EXTERNAL_GENERATION_EDITOR_SOURCE_MODULE.to_string(); + row.completed_at = Some(micros(1_000)); + row.updated_at = micros(1_000); + let mut summary = build_external_generation_job_summary_row(&row, None); + summary.notification_acknowledged_at = Some(micros(2_000)); + + assert!(is_external_generation_job_retention_candidate( + &row, + &summary, + EXTERNAL_GENERATION_EDITOR_SOURCE_MODULE, + 1_000, + )); + + summary.notification_acknowledged_at = None; + assert!(!is_external_generation_job_retention_candidate( + &row, + &summary, + EXTERNAL_GENERATION_EDITOR_SOURCE_MODULE, + 1_000, + )); + + summary.notification_acknowledged_at = Some(micros(2_000)); + row.status = EXTERNAL_GENERATION_STATUS_RUNNING.to_string(); + summary.status = EXTERNAL_GENERATION_STATUS_RUNNING.to_string(); + assert!(!is_external_generation_job_retention_candidate( + &row, + &summary, + EXTERNAL_GENERATION_EDITOR_SOURCE_MODULE, + 1_000, + )); + + row.status = EXTERNAL_GENERATION_STATUS_COMPLETED.to_string(); + summary.status = EXTERNAL_GENERATION_STATUS_COMPLETED.to_string(); + row.completed_at = Some(micros(1_001)); + assert!(!is_external_generation_job_retention_candidate( + &row, + &summary, + EXTERNAL_GENERATION_EDITOR_SOURCE_MODULE, + 1_000, + )); + + row.completed_at = Some(micros(1_000)); + row.source_module = "puzzle".to_string(); + assert!(!is_external_generation_job_retention_candidate( + &row, + &summary, + EXTERNAL_GENERATION_EDITOR_SOURCE_MODULE, + 1_000, + )); + } + + #[test] + fn retention_rejects_mismatched_summary_identity_or_status() { + let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_FAILED); + row.source_module = EXTERNAL_GENERATION_EDITOR_SOURCE_MODULE.to_string(); + row.completed_at = Some(micros(1_000)); + let mut summary = build_external_generation_job_summary_row(&row, None); + summary.notification_acknowledged_at = Some(micros(2_000)); + + summary.job_id = "different-job".to_string(); + assert!(!is_external_generation_job_retention_candidate( + &row, + &summary, + EXTERNAL_GENERATION_EDITOR_SOURCE_MODULE, + 1_000, + )); + + summary.job_id = row.job_id.clone(); + summary.status = EXTERNAL_GENERATION_STATUS_CANCELLED.to_string(); + assert!(!is_external_generation_job_retention_candidate( + &row, + &summary, + EXTERNAL_GENERATION_EDITOR_SOURCE_MODULE, + 1_000, + )); + } + #[test] fn maintenance_selector_bounds_scanned_rows_and_advances_by_last_scanned_job() { let rows = (1..=4).map(|index| {