diff --git a/deploy/systemd/genarrative-database-backup.service b/deploy/systemd/genarrative-database-backup.service index 276b9f9ab..4ccc00922 100644 --- a/deploy/systemd/genarrative-database-backup.service +++ b/deploy/systemd/genarrative-database-backup.service @@ -11,6 +11,12 @@ WorkingDirectory=/opt/genarrative/current EnvironmentFile=/etc/genarrative/api-server.env ExecStart=/usr/bin/node -- /opt/genarrative/current/scripts/database-backup-to-oss.mjs --env-file /etc/genarrative/api-server.env --stop-service spacetimedb.service --restart-service-after genarrative-api.service --restart-service-after genarrative-external-generation-worker@1.service --restart-service-after genarrative-external-generation-controller.service +# 备份脚本必须受独立内存上限保护，不能因目录扫描异常拖垮整台 release 主机。 +Environment=NODE_OPTIONS=--max-old-space-size=768 +MemoryHigh=768M +MemoryMax=1G +OOMPolicy=stop + # 备份需要停止 / 启动 spacetimedb.service，并读取 /stdb、写入 /var/lib/genarrative/database-backups。 # 停止 SpacetimeDB 会连带停止 Requires 它的 API / worker / controller，冷备份后必须显式拉起。 PrivateTmp=true diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index 0cb6906fd..2870fe214 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -1309,8 +1309,8 @@ CI 上 `background_agent_runtime_recovers_stale_running_before_pending_task` 在 - OSS 固定恢复入口为 `//latest.json`。CAS 文件和 full/history catalog 保持不可变；latest pointer 只保存最新 full catalog 与已发布 history catalog 的 object key、长度和 SHA，不包含主机绝对路径或文件内容。每次 state 变化先验真全部引用 catalog，再覆盖上传并 HEAD 验真 latest pointer，成功后才落本地 state；history 还必须在 pointer 成功后才允许删除源文件。全新机器可仅凭 bucket、database、prefix 与 OSS 凭据自动下载 pointer 和 full catalog。 - dev 带宽不足时，允许把已冻结的 dev 基线经 `10.2.0.10 -> 10.2.4.16` 内网 rsync 到 release 独立 staging，再用 release 出口上传 dev bucket；staging 不得指向 release `/stdb`，不得停止或修改 release 服务，传输凭据必须临时创建并在演练后移除。catalog 不记录 staging 绝对路径，files state 可回传 dev 继续 history。 - 恢复边界：恢复时默认从 OSS `latest.json` 自动定位 full catalog，创建目录并按相对路径下载每个对象、逐文件校验长度与 SHA；本地 state 只用于备份续跑，不再是异机恢复前置条件。远程 dev 已完成真实 OSS、清理、重启和异机隔离恢复演练；release timer 与 publish 前备份继续保持原行为。 -- systemd 接线：主 service 保持 `archive-full`。Server-Provision 新增默认值为 `archive-full` 的 `DATABASE_BACKUP_PROFILE`；dev 或 release 显式选择 `files-history` 时，必须为各自主机指定独立 work-dir，并先用 current release 脚本执行 history dry-run，确认已有 full state 后才安装仓库托管 drop-in，并删除现场手写旧 drop-in。切回默认 profile 必须删除所有 history 覆盖。 -- 影响范围：`scripts/database-backup-to-oss.mjs`、备份门禁、生产 env 示例、systemd 模板、Server-Provision、SpacetimeDB 运维与恢复流程；release timer 可在独立 baseline 验证后显式选择 profile，publish 前备份是否切换仍需单独决策。 +- systemd 接线：主 service 保持 `archive-full`。Server-Provision 新增默认值为 `archive-full` 的 `DATABASE_BACKUP_PROFILE`；development 可显式选择 `files-history`，必须指定独立 work-dir 并先用 current release 脚本执行 history dry-run，确认已有 full state 后才安装仓库托管 drop-in；release 拒绝 `files-history`，直到流式 catalog 改造完成，以免大目录扫描再次触发 Node 内存峰值。切回默认 profile 必须删除所有 history 覆盖；备份 unit 同时设置 Node heap 与 systemd memory 上限，避免备份异常拖垮业务主机。 +- 影响范围：`scripts/database-backup-to-oss.mjs`、备份门禁、生产 env 示例、systemd 模板、Server-Provision、SpacetimeDB 运维与恢复流程；release timer 固定使用 archive-full，publish 前备份是否切换仍需单独决策。 - 验证方式：`npm run check:database-backup`、`npm run check:production-ops`、`npm run check:encoding`、`git diff --check`；dev 现场必须完成逐文件 full catalog、重复 full 零 PUT、history dry-run、上传后清理、STDB 重启和按 catalog 隔离恢复 roundtrip。 - 关联：。 diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index df3653c50..e0fb2ab47 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -329,6 +329,7 @@ Responses 的终态载荷既是工具调用的恢复源，也是正文的恢复 - Rust 结构体：`AiTask` - 源码：`server-rs/crates/spacetime-module/src/ai/tasks.rs` +- `module-ai` 的进程内热状态不是持久化真相：文本增量按阶段有序聚合并受单阶段 512 KiB 上限约束；terminal task 立即释放增量明细，内存工作集最多保留 1024 个任务。需要长期查询时必须读取 SpacetimeDB 的 `ai_task` / `ai_task_stage` 投影，不得依赖进程重启后仍存在的内存快照。 ### `ai_task_event` @@ -413,7 +414,7 @@ Responses 的终态载荷既是工具调用的恢复源，也是正文的恢复 - Rust 结构体：`AuthStoreProjectionMeta` - 源码：`server-rs/crates/spacetime-module/src/auth/tables.rs` -认证恢复策略：`api-server` 启动时只从 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）导出 typed `AuthStoreProjectionView`，再恢复 `module-auth` 的进程内认证工作集；运行中 Bearer `sid` 或 refresh cookie 在本进程工作集内未命中时直接按失效处理，不再从 SpacetimeDB 导出整包认证状态刷新内存，避免旧投影把重复手机号或旧会话重新灌回进程。`module-auth` 只保留内存工作集和 projection 导入 / 导出能力，不再保留 JSON 快照导入 / 导出能力，也不写本地持久化文件；`auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 不再是兼容恢复源。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作必须在返回客户端前通过 `sync_auth_store_projection` 成功同步 SpacetimeDB 正式认证表；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号或会话当成成功结果。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。若启动恢复阶段 SpacetimeDB 不可连接或超时，`api-server` 会按固定间隔持续重试认证工作集恢复，恢复成功后才开始监听 HTTP，避免一次短超时让进程永久停留在依赖不可用状态。 +认证恢复策略：`api-server` 启动时只从 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）导出 typed `AuthStoreProjectionView`，再恢复 `module-auth` 的进程内认证工作集；运行中 Bearer `sid` 或 refresh cookie 在本进程工作集内未命中时直接按失效处理，不再从 SpacetimeDB 导出整包认证状态刷新内存，避免旧投影把重复手机号或旧会话重新灌回进程。`module-auth` 只保留内存工作集和 projection 导入 / 导出能力，不再保留 JSON 快照导入 / 导出能力，也不写本地持久化文件；`auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 不再是兼容恢复源。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作必须在返回客户端前通过 `sync_auth_store_projection` 成功同步 SpacetimeDB 正式认证表；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号或会话当成成功结果。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。refresh session 工作集最多保留 8192 条，短信验证码和微信登录 state 分别最多保留 4096 条；写入前清理过期项，达到上限时拒绝新增而不继续膨胀。若启动恢复阶段 SpacetimeDB 不可连接或超时，`api-server` 会按固定间隔持续重试认证工作集恢复，恢复成功后才开始监听 HTTP，避免一次短超时让进程永久停留在依赖不可用状态。 `auth_store_snapshot` 表和旧 `import_auth_store_snapshot_json` / `export_auth_store_snapshot_from_tables` procedure 已删除。认证投影同步只读写 `user_account`、`auth_identity`、`refresh_session` 和 `auth_store_projection_meta`；`auth_identity` 不再写 `phone_e164`、`display_name`、`avatar_url`，这些账号资料只以 `user_account` 为准。 @@ -1232,6 +1233,7 @@ RPG 创作入口的配置 ID 是 `rpg`，当前 `visible=true`、`open=true`； - Rust 结构体：`RefreshSession` - 源码：`server-rs/crates/spacetime-module/src/auth/tables.rs` +- 认证工作集只保留 active 会话以及最近 24 小时内的 revoked / expired 会话；超过宽限期的失效会话在 refresh session 写路径和 projection 导出前从内存索引移除，并随下一次 typed projection 同步从正式表清理。该清理不改变 active 多端登录、单端登出或全端登出语义。 ### `runtime_setting` diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index 34f10d64f..25e6132dd 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -99,7 +99,7 @@ HTTP 角色的 `GENARRATIVE_SPACETIME_POOL_SIZE` 只表示 procedure / reducer 生产拆分角色时，`external-generation-worker` 和 `external-generation-controller` 的专属 env 示例会把 `GENARRATIVE_SPACETIME_POOL_SIZE` 覆盖为 `1`；非 HTTP 角色不创建 API 缓存读连接，只保留 `external_generation_job` 队列窄订阅作为响应式唤醒信号，实际抢占和扩缩容判断仍走 SpacetimeDB procedure。worker / controller 不执行模型定价 seed，启动时先调用受 runtime writer 鉴权的 queue-stats procedure 做只读预检，身份不匹配时 fail-fast；当前正式 systemd unit 通过共同加载 API env 继承同一 `GENARRATIVE_SPACETIME_TOKEN`，默认路径为 `/etc/genarrative/api-server.env`，自定义部署由 provision 和 API deploy 按实际参数渲染，专属角色 env 示例不重复配置该 token。`GENARRATIVE_EXTERNAL_GENERATION_WORKER_POLL_INTERVAL_MS` 与 controller poll interval 只作为订阅失效、漏事件和 lease 过期这类时间条件的兜底，不作为正常领取任务的主路径。 -生产 worker 默认 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LEASE_SECONDS=600`，只覆盖 worker 心跳抖动和短暂断连窗口，不再把 lease 当成完整任务时长；默认 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_JOB_TIMEOUT_SECONDS=900`。`editor_image_generation`、`editor_image_edit`、`editor_icon_spritesheet_generation`、`editor_ui_design_asset_extraction` 四类 VectorEngine 图片任务与角色动画 / 视频类长任务使用 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LONG_JOB_TIMEOUT_SECONDS=1800`，手动去背景、音效和背景音乐继续使用普通预算。worker 在单次尝试超过执行预算后会停止续租并释放 worker 槽位，但不会取消已启动的业务 future 或主动写入失败 / 重试状态；在途执行由 lease fencing 仲裁，有效租约内写回仍可完成，租约过期后任务才可重新领取，attempt 耗尽时由认领事务标记失败并结算退款。生产部署和 provision 脚本会给 `/etc/genarrative/api-server.env` 与 `/etc/genarrative/external-generation-worker.env` 补齐这些变量；已有自定义值不覆盖，只会把历史旧默认 `3600` 迁移为 `600`。 +生产 worker 默认 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LEASE_SECONDS=600`，只覆盖 worker 心跳抖动和短暂断连窗口，不再把 lease 当成完整任务时长；默认 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_JOB_TIMEOUT_SECONDS=900`。`editor_image_generation`、`editor_image_edit`、`editor_icon_spritesheet_generation`、`editor_ui_design_asset_extraction` 四类 VectorEngine 图片任务与角色动画 / 视频类长任务使用 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LONG_JOB_TIMEOUT_SECONDS=1800`，手动去背景、音效和背景音乐继续使用普通预算。worker 在单次尝试超过执行预算后会停止续租，但不会取消已启动的业务 future 或主动写入失败 / 重试状态；执行许可会一直绑定到 active 或 detached work 真正结束（或超过租约仲裁窗口被取消），避免超时任务脱管后立即补进新的高内存任务。在途执行由 lease fencing 仲裁，有效租约内写回仍可完成，租约过期后任务才可重新领取，attempt 耗尽时由认领事务标记失败并结算退款。生产部署和 provision 脚本会给 `/etc/genarrative/api-server.env` 与 `/etc/genarrative/external-generation-worker.env` 补齐这些变量；已有自定义值不覆盖，只会把历史旧默认 `3600` 迁移为 `600`。 lease 过期后不代表任务一定再次执行：claim transaction 只有在 `attempt < max_attempts` 时才会递增 attempt 并返回 worker；如果过期的是最终 attempt，则直接把 job 收口为 `failed`、清理 lease，并按入队冻结价格为当前 attempt 原子退款或写 cancellation intent。该终态任务不会再次进入 provider executor，迟到 consume 会被 settlement intent 拒绝。 @@ -401,7 +401,7 @@ UI 相关修改要重点验证： ### SpacetimeDB 数据目录 OSS 备份 -数据库备份不放进 `spacetime-module` reducer / procedure：备份属于文件系统与 OSS 外部副作用，必须由运维脚本在 SpacetimeDB 宿主外执行。当前统一脚本为 `scripts/database-backup-to-oss.mjs`（npm 命令 `npm run database:backup:oss`）。默认 `--storage-format archive --mode full` 保持原有全量压缩包冷备行为；`--storage-format files` 不生成 tar.gz，而是把目录树映射成逐文件 CAS 对象与 catalog，full 重跑只上传新增或内容变化的文件，history 只处理已被最新 snapshot 完全覆盖的历史 commitlog 与旧 snapshot。`Genarrative-Server-Provision` 的 `DATABASE_BACKUP_PROFILE` 默认是 `archive-full`，继续安装每天 `03:20` 左右执行的全量冷备主 service；development 和 release 都可以显式选择 `files-history`，但指定 work-dir 必须已经有与本机 database/bucket 匹配且已发布的 full baseline state： +数据库备份不放进 `spacetime-module` reducer / procedure：备份属于文件系统与 OSS 外部副作用，必须由运维脚本在 SpacetimeDB 宿主外执行。当前统一脚本为 `scripts/database-backup-to-oss.mjs`（npm 命令 `npm run database:backup:oss`）。默认 `--storage-format archive --mode full` 保持原有全量压缩包冷备行为；`--storage-format files` 不生成 tar.gz，而是把目录树映射成逐文件 CAS 对象与 catalog，full 重跑只上传新增或内容变化的文件，history 只处理已被最新 snapshot 完全覆盖的历史 commitlog 与旧 snapshot。`Genarrative-Server-Provision` 的 `DATABASE_BACKUP_PROFILE` 默认是 `archive-full`，继续安装每天 `03:20` 左右执行的全量冷备主 service；当前 release 只允许 `archive-full`，避免 `files-history` 在大目录上构造全量 catalog 导致 Node 内存峰值；development 才可以显式选择 `files-history`，且指定 work-dir 必须已经有与本机 database/bucket 匹配且已发布的 full baseline state： ```bash npm run database:backup:oss -- --data-dir /stdb --stop-service spacetimedb.service --restart-service-after genarrative-api.service --restart-service-after genarrative-external-generation-worker@1.service --restart-service-after genarrative-external-generation-controller.service @@ -437,7 +437,7 @@ GENARRATIVE_DATABASE_BACKUP_OSS_ACCESS_KEY_SECRET= `GENARRATIVE_DATABASE_BACKUP_OSS_BUCKET` 为空时会回退 `ALIYUN_OSS_BUCKET`；AccessKey 默认复用 `ALIYUN_OSS_ACCESS_KEY_ID` / `ALIYUN_OSS_ACCESS_KEY_SECRET`，也可用 `GENARRATIVE_DATABASE_BACKUP_OSS_ACCESS_KEY_ID` / `GENARRATIVE_DATABASE_BACKUP_OSS_ACCESS_KEY_SECRET` 为备份 bucket 单独配置最小权限账号。冷备脚本会在停止 SpacetimeDB 前检查 `GENARRATIVE_DATABASE_BACKUP_WORK_DIR` 所在文件系统剩余空间；未设置 `GENARRATIVE_DATABASE_BACKUP_MIN_FREE_BYTES` 时，按数据目录大小加安全余量估算，空间不足会在停库前失败，避免写满根分区。即使打包或上传前步骤失败，只要脚本已经停过 SpacetimeDB，也会先恢复 SpacetimeDB 并执行 `--restart-service-after` 指定的 API / worker / controller，再带着原始备份错误退出。`Genarrative-Server-Provision` 会创建 `/var/lib/genarrative/database-backups` 并归属 `genarrative:genarrative`，同时安装并启用 `genarrative-database-backup.timer`。手动检查定时器：`systemctl list-timers genarrative-database-backup.timer`；手动触发一次：`systemctl start genarrative-database-backup.service`。如果 timer 显示 `enabled` 但 `inactive/dead` 且 `NEXT` / `Trigger` 为空，先写入当前 stamp 避免 `Persistent=true` 在白天立刻补跑冷备份：`touch /var/lib/systemd/timers/stamp-genarrative-database-backup.timer && systemctl daemon-reload && systemctl start genarrative-database-backup.timer`，随后确认下一次触发时间约为次日 `03:20`。 -`files-history` 使用仓库模板 `deploy/systemd/genarrative-database-backup-files-history.conf` 覆盖主 service 的 `ExecStart`，从 `/etc/genarrative/api-server.env` 读取 data-dir、database、bucket、prefix 与 OSS 凭据，不在 unit 写死环境目标，也不传 `--stop-service`。Server-Provision 在改动 drop-in 前，先用 current release 的同一脚本、同一 env 和 `DATABASE_BACKUP_FILES_HISTORY_WORK_DIR` 执行一次 history `--dry-run`；缺少已发布 full catalog 的 files state、current 脚本过旧或配置不匹配都会在安装 drop-in 和 `daemon-reload` 前失败。选择 `archive-full` 会主动删除仓库托管的 `10-files-history.conf` 与 dev 试点遗留的 `10-dev-files.conf`，防止 systemd 继续合并旧覆盖。dev 可继续指定已有 `/var/lib/genarrative/database-backups/dev-files`，release 建议先在 `/var/lib/genarrative/database-backups/release-files` 建立自己的 full baseline；两台机器不得复用或互传本地 state 目录冒充本机基线。启用时通过 Server-Provision Job 选择目标、`DATABASE_BACKUP_PROFILE=files-history` 和对应 work-dir，先保持 `DRY_RUN=true` 核对，再以同参数正式 provision。不要直接在 `/etc/systemd/system` 手写第二份 drop-in。 +`files-history` 使用仓库模板 `deploy/systemd/genarrative-database-backup-files-history.conf` 覆盖主 service 的 `ExecStart`，从 `/etc/genarrative/api-server.env` 读取 data-dir、database、bucket、prefix 与 OSS 凭据，不在 unit 写死环境目标，也不传 `--stop-service`。Server-Provision 在 development 改动 drop-in 前，先用 current release 的同一脚本、同一 env 和 `DATABASE_BACKUP_FILES_HISTORY_WORK_DIR` 执行一次 history `--dry-run`；缺少已发布 full catalog 的 files state、current 脚本过旧或配置不匹配都会在安装 drop-in 和 `daemon-reload` 前失败。选择 `archive-full` 会主动删除仓库托管的 `10-files-history.conf` 与 dev 试点遗留的 `10-dev-files.conf`，防止 systemd 继续合并旧覆盖。`genarrative-database-backup.service` 还通过 `NODE_OPTIONS=--max-old-space-size=768`、`MemoryHigh=768M`、`MemoryMax=1G` 和 `OOMPolicy=stop` 给备份进程设置独立护栏；release 若现场残留 files-history drop-in，必须先按 archive-full 重新 provision 并确认 drop-in 已删除，再恢复定时器。dev 可继续指定已有 `/var/lib/genarrative/database-backups/dev-files`；两台机器不得复用或互传本地 state 目录冒充本机基线。启用时通过 Server-Provision Job 选择目标、`DATABASE_BACKUP_PROFILE=files-history` 和对应 work-dir，先保持 `DRY_RUN=true` 核对，再以同参数正式 provision。不要直接在 `/etc/systemd/system` 手写第二份 drop-in。 files full 会递归扫描 data-dir，保留空目录、每个普通文件的相对路径，以及目标仍位于 data-dir 内部的相对符号链接；绝对链接或解析后越界的链接直接拒绝。文件按 SHA-256 上传到不可变对象 key，catalog 记录目录、路径、长度、SHA、对象 key 和相对链接目标，不写 staging 主机的绝对路径。相同 catalog 重跑不重复 PUT；新增或变化文件先 HEAD CAS 对象，存在且长度/SHA 元数据一致就复用，否则上传。16 MiB 及以下对象使用单次 PUT 后 HEAD 验真，大对象继续使用 multipart；对象操作默认以 16 路并行执行，可用 `GENARRATIVE_DATABASE_BACKUP_FILES_CONCURRENCY=1..64` 调整。需要给线上入口留带宽时设置 `GENARRATIVE_DATABASE_BACKUP_UPLOAD_MAX_BYTES_PER_SECOND=`，该共享限速器只包裹备份上传流，空值或 `0` 表示不限速，不修改主机全局 qdisc。并发、限速和单次 PUT 都不改变“全部对象、catalog 与 latest pointer 成功后才推进 state/清理”的顺序。full 基线必须来自停库后的 data-dir 或已通过恢复验证的冻结副本；源文件上传前后 stat 虽会复核，但在线扫描不能保证大量文件属于同一跨文件一致时点。catalog 验真后，脚本把最新 full/history 引用发布到固定 `//latest.json`，全新机器不需要本地 state 即可自动发现恢复入口。 @@ -478,7 +478,7 @@ node -- scripts/database-backup-to-oss.mjs \ dev 出口过慢时，可以把冻结基线经内网 rsync 到 release 独立 staging，再由 release 上传 dev bucket。staging 必须位于 `/var/lib/genarrative/dev-database-backup-staging/` 一类隔离目录，命令显式传 staging `--data-dir`、独立 `--work-dir`、dev `--bucket`，且不得传 `--stop-service`；禁止指向或修改 release `/stdb`。中转 key 只为本次传输临时授权，结束后从 dev 私钥和 release `authorized_keys` 同时移除。上传完成后把整个 files work-dir/state 回传 dev，history 才能延续同一 baseline catalog。 -完整恢复默认从 OSS 固定 `latest.json` 读取最新 full catalog：先创建 `directories`，再把每个 `files[].objectKey` 下载到 `/` 并逐项核对 `sizeBytes` / `sha256`；history catalog 用于证明已清理历史仍有 OSS 对象，不需要把已被 full baseline 覆盖的旧文件叠回当前恢复目录。本地 state 仍可作为兼容入口，并同时支持旧 v1 JSON 与 v2 gzip，但不再是异机恢复的前置条件。随后用隔离 data-dir 启动同版本 standalone，验证 `/v1/ping`、日志中的 snapshot restore / commitlog replay / module launch、代表性 SQL 和 reducer。dev 已完成这轮 OSS-only 异机恢复与重启演练；release 已使用独立 `/var/lib/genarrative/database-backups/release-files` full baseline 和 `files-history` profile，现场最终 `ExecStart`、timer 状态与最近备份结果仍须在变更时重新核对。 +完整恢复默认从 OSS 固定 `latest.json` 读取最新 full catalog：先创建 `directories`，再把每个 `files[].objectKey` 下载到 `/` 并逐项核对 `sizeBytes` / `sha256`；history catalog 用于证明已清理历史仍有 OSS 对象，不需要把已被 full baseline 覆盖的旧文件叠回当前恢复目录。本地 state 仍可作为兼容入口，并同时支持旧 v1 JSON 与 v2 gzip，但不再是异机恢复的前置条件。随后用隔离 data-dir 启动同版本 standalone，验证 `/v1/ping`、日志中的 snapshot restore / commitlog replay / module launch、代表性 SQL 和 reducer。dev 已完成这轮 OSS-only 异机恢复与重启演练；release 使用 archive-full 时，现场最终 `ExecStart`、timer 状态与最近备份结果仍须在变更时重新核对。 ```bash node -- scripts/database-backup-to-oss.mjs \ diff --git a/jenkins/Jenkinsfile.production-server-provision b/jenkins/Jenkinsfile.production-server-provision index 69ea81c22..2587c339f 100644 --- a/jenkins/Jenkinsfile.production-server-provision +++ b/jenkins/Jenkinsfile.production-server-provision @@ -34,8 +34,8 @@ pipeline { string(name: 'WEB_LINK', defaultValue: '/srv/genarrative/web', description: 'Nginx 静态站点目录或软链接') string(name: 'API_ENV_FILE', defaultValue: '/etc/genarrative/api-server.env', description: 'api-server 环境文件') string(name: 'API_PORT', defaultValue: '8082', description: 'api-server 本机监听端口') - choice(name: 'DATABASE_BACKUP_PROFILE', choices: ['archive-full', 'files-history'], description: '数据库定时备份 profile；默认 archive-full，files-history 仅在指定 work-dir 已有完整 full baseline 后启用') - string(name: 'DATABASE_BACKUP_FILES_HISTORY_WORK_DIR', defaultValue: '/var/lib/genarrative/database-backups/files-history', description: 'files-history 的本地 state/catalog 目录；dev/release 必须使用各自已建立 full baseline 的独立目录') + choice(name: 'DATABASE_BACKUP_PROFILE', choices: ['archive-full', 'files-history'], description: '数据库定时备份 profile；release 仅允许 archive-full，files-history 仅供 development 在指定 work-dir 已有完整 full baseline 后启用') + string(name: 'DATABASE_BACKUP_FILES_HISTORY_WORK_DIR', defaultValue: '/var/lib/genarrative/database-backups/files-history', description: 'development files-history 的本地 state/catalog 目录；必须使用已建立 full baseline 的独立目录') choice(name: 'NGINX_CONFIG_MODE', choices: ['none', 'production-https', 'development-http'], description: 'Nginx 配置模式；开发服无域名时选 development-http，release 正式入口选 production-https') booleanParam(name: 'ENABLE_SERVICES', defaultValue: true, description: '启用并启动 spacetimedb 与 api-server systemd 服务') booleanParam(name: 'ENABLE_OTELCOL', defaultValue: true, description: '安装并启用本机 OpenTelemetry Collector；api-server 模板默认开启 OTLP，如需关闭请在 API_ENV_FILE 中将 GENARRATIVE_OTEL_ENABLED 改为 false') @@ -113,6 +113,9 @@ pipeline { if (!(databaseBackupProfile in ['archive-full', 'files-history'])) { error("DATABASE_BACKUP_PROFILE 只能是 archive-full 或 files-history，当前值: ${params.DATABASE_BACKUP_PROFILE}") } + if (params.DEPLOY_TARGET == 'release' && databaseBackupProfile == 'files-history') { + error('release 仅允许 archive-full；files-history 会把整棵历史目录加载到 Node 内存，需先完成流式 catalog 改造后才能重新启用。') + } def databaseBackupFilesHistoryWorkDir = params.DATABASE_BACKUP_FILES_HISTORY_WORK_DIR?.trim() if (!(databaseBackupFilesHistoryWorkDir ==~ /^\/var\/lib\/genarrative\/database-backups\/[A-Za-z0-9._\/-]+$/) || databaseBackupFilesHistoryWorkDir.contains('..')) { error("DATABASE_BACKUP_FILES_HISTORY_WORK_DIR 必须是 /var/lib/genarrative/database-backups/ 下不含连续点号的绝对路径，当前值: ${params.DATABASE_BACKUP_FILES_HISTORY_WORK_DIR}") diff --git a/scripts/check-production-ops-guardrails.mjs b/scripts/check-production-ops-guardrails.mjs index c567e8232..11ecd0462 100644 --- a/scripts/check-production-ops-guardrails.mjs +++ b/scripts/check-production-ops-guardrails.mjs @@ -821,6 +821,18 @@ const checks = [ reason: '生产冷备份 service 必须用 node -- 分隔脚本参数，避免 Node 22 抢占业务 --env-file。', }, + { + file: 'deploy/systemd/genarrative-database-backup.service', + includes: 'Environment=NODE_OPTIONS=--max-old-space-size=768', + reason: + '备份 Node 进程必须设置独立 heap 上限，避免目录扫描异常拖垮 release 主机。', + }, + { + file: 'deploy/systemd/genarrative-database-backup.service', + includes: 'MemoryMax=1G', + reason: + '备份 service 必须设置 systemd 内存硬上限，避免异常进程消耗整机内存。', + }, { file: 'deploy/systemd/genarrative-database-backup.service', excludes: '--storage-format files', @@ -901,10 +913,10 @@ const checks = [ }, { file: 'jenkins/Jenkinsfile.production-server-provision', - excludes: + includes: "params.DEPLOY_TARGET == 'release' && databaseBackupProfile == 'files-history'", reason: - 'release 必须能在显式选择 profile 且 baseline 预检通过后启用 files-history。', + 'release 必须拒绝 files-history，避免逐文件 catalog 扫描再次触发生产内存峰值。', }, { file: 'scripts/database-backup-to-oss.mjs', diff --git a/scripts/database-backup-to-oss.mjs b/scripts/database-backup-to-oss.mjs index d9a1ff9eb..86057f07a 100644 --- a/scripts/database-backup-to-oss.mjs +++ b/scripts/database-backup-to-oss.mjs @@ -555,7 +555,11 @@ function assertSafeRelativePath(dataDir, absolutePath) { } function statFingerprint(absolutePath, rootPath = absolutePath) { - const entries = []; + // 候选 snapshot 可能包含数十万条目录项；增量更新摘要，避免把每条 + // fingerprint 字符串同时保存在 entries[] 后再 join，造成一次性内存峰值。 + const fingerprintHash = createHash('sha256'); + let isFirstEntry = true; + let entryCount = 0; let totalSize = 0n; const visit = (currentPath) => { const stat = lstatSync(currentPath, {bigint: true}); @@ -567,7 +571,7 @@ function statFingerprint(absolutePath, rootPath = absolutePath) { if (kind === 'other') { throw new Error(`history 候选只允许普通文件或目录: ${currentPath}`); } - entries.push([ + const entry = [ entryPath, kind, stat.dev.toString(), @@ -575,7 +579,13 @@ function statFingerprint(absolutePath, rootPath = absolutePath) { stat.mode.toString(), stat.size.toString(), stat.mtimeNs.toString(), - ].join('\0')); + ].join('\0'); + if (!isFirstEntry) { + fingerprintHash.update('\n'); + } + fingerprintHash.update(entry); + isFirstEntry = false; + entryCount += 1; if (stat.isFile()) { totalSize += stat.size; } else { @@ -586,9 +596,9 @@ function statFingerprint(absolutePath, rootPath = absolutePath) { }; visit(rootPath); return { - fingerprint: sha256Hex(entries.join('\n')), + fingerprint: fingerprintHash.digest('hex'), sizeBytes: totalSize.toString(), - entryCount: entries.length, + entryCount, }; } @@ -1285,14 +1295,17 @@ export async function collectDirectFileEntries({dataDir, candidates = null, obje throw new Error(`files 扫描期间源文件发生变化: ${relativePath}`); } const basePrefix = normalizeObjectPrefix(objectPrefix, database); - files.set(relativePath, { + const file = { path: relativePath, sizeBytes: Number(after.size), sha256, mode: after.mode, objectKey: `${basePrefix}/files/sha256/${sha256.slice(0, 2)}/${sha256}`, - sourceStat: after, - }); + }; + // 上传前后的 inode/stat 仍用于防止在线扫描漂移，但设为不可枚举，避免 + // 把仅供本地校验的副本再次写入 catalog 或 result JSON。 + Object.defineProperty(file, 'sourceStat', {value: after, enumerable: false}); + files.set(relativePath, file); }; for (const root of roots.sort((left, right) => left.relativePath.localeCompare(right.relativePath))) { @@ -1309,14 +1322,40 @@ export async function collectDirectFileEntries({dataDir, candidates = null, obje } function directCatalogIdentity({mode, baselineCatalogId, rootName, directories, files, symlinks}) { - return sha256Hex(JSON.stringify({ - mode, - baselineCatalogId: baselineCatalogId || '', - rootName, - directories, - files: files.map(({path, sizeBytes, sha256, mode, objectKey}) => ({path, sizeBytes, sha256, mode, objectKey})), - symlinks, + // 不把数十万条文件元数据先拼成一个巨型 JSON 字符串；分段写入 hash + // 保持与 JSON.stringify 同样的字段顺序和转义结果，同时把峰值降到单条记录。 + const hash = createHash('sha256'); + hash.update('{"mode":'); + hash.update(JSON.stringify(mode)); + hash.update(',"baselineCatalogId":'); + hash.update(JSON.stringify(baselineCatalogId || '')); + hash.update(',"rootName":'); + hash.update(JSON.stringify(rootName)); + hash.update(',"directories":'); + updateJsonArrayHash(hash, directories, (directory) => JSON.stringify(directory)); + hash.update(',"files":'); + updateJsonArrayHash(hash, files, (file) => JSON.stringify({ + path: file.path, + sizeBytes: file.sizeBytes, + sha256: file.sha256, + mode: file.mode, + objectKey: file.objectKey, })); + hash.update(',"symlinks":'); + updateJsonArrayHash(hash, symlinks, (symlink) => JSON.stringify(symlink)); + hash.update('}'); + return hash.digest('hex'); +} + +function updateJsonArrayHash(hash, values, serialize) { + hash.update('['); + values.forEach((value, index) => { + if (index > 0) { + hash.update(','); + } + hash.update(serialize(value)); + }); + hash.update(']'); } function readDirectFilesState(statePath, {database, bucket}) { @@ -1621,7 +1660,7 @@ export async function runDirectFilesBackup({ baselineCatalogId, rootName, directories: collected.directories, - files: collected.files.map(({sourceStat: _sourceStat, ...file}) => file), + files: collected.files, symlinks: collected.symlinks, }; writeManifest({manifestPath: catalogPath, payload: catalog}); diff --git a/scripts/jenkins-server-provision.sh b/scripts/jenkins-server-provision.sh index 894458fd2..259b6f6c8 100755 --- a/scripts/jenkins-server-provision.sh +++ b/scripts/jenkins-server-provision.sh @@ -78,6 +78,10 @@ validate_database_backup_profile() { exit 1 ;; esac + if [[ "${DEPLOY_TARGET}" == "release" && "${DATABASE_BACKUP_PROFILE}" == "files-history" ]]; then + echo "[server-provision] release 仅允许 archive-full；files-history 会把整棵历史目录加载到 Node 内存，需先完成流式 catalog 改造后才能重新启用。" >&2 + exit 1 + fi if [[ ! "${DATABASE_BACKUP_FILES_HISTORY_WORK_DIR}" =~ ^/var/lib/genarrative/database-backups/[A-Za-z0-9._/-]+$ || "${DATABASE_BACKUP_FILES_HISTORY_WORK_DIR}" == *..* ]]; then echo "[server-provision] DATABASE_BACKUP_FILES_HISTORY_WORK_DIR 必须是 /var/lib/genarrative/database-backups/ 下不含连续点号的绝对路径，当前值: ${DATABASE_BACKUP_FILES_HISTORY_WORK_DIR}" >&2 exit 1 diff --git a/server-rs/crates/api-server/src/external_generation_worker.rs b/server-rs/crates/api-server/src/external_generation_worker.rs index b27804a9d..76223aefc 100644 --- a/server-rs/crates/api-server/src/external_generation_worker.rs +++ b/server-rs/crates/api-server/src/external_generation_worker.rs @@ -14,6 +14,7 @@ use spacetime_client::{ ExternalGenerationJobRenewLeaseRecordInput, ExternalGenerationQueueWakeSubscription, }; use tokio::{ + sync::{OwnedSemaphorePermit, Semaphore}, task::{JoinHandle, JoinSet}, time::sleep, }; @@ -92,6 +93,10 @@ pub(crate) async fn run_external_generation_worker(state: AppState) -> Result<() let concurrency = state.config.external_generation_worker_concurrency.max(1); let poll_interval = state.config.external_generation_worker_poll_interval; let lease = state.config.external_generation_worker_lease; + // 超时任务不能立即取消（在途 procedure 仍可能写回），因此执行容量必须同时 + // 约束 active 与 detached work；否则每次超时都会释放 tasks 槽位，实际内存占用 + // 会超过配置并发。 + let work_slots = std::sync::Arc::new(Semaphore::new(concurrency)); let mut tasks = JoinSet::new(); let mut shutdown = external_generation_worker_shutdown_signal(); let mut queue_wake = None; @@ -115,14 +120,24 @@ pub(crate) async fn run_external_generation_worker(state: AppState) -> Result<() loop { ensure_external_generation_queue_wake_subscription(&state, &mut queue_wake).await; - while tasks.len() >= concurrency { - if await_worker_task_or_shutdown(&mut tasks, &mut shutdown).await { - drain_external_generation_worker_tasks(&mut tasks).await; - return Ok(()); + while work_slots.available_permits() == 0 { + tokio::select! { + _ = shutdown.as_mut() => { + drain_external_generation_worker_tasks(&mut tasks).await; + return Ok(()); + } + permit = work_slots.clone().acquire_owned() => { + if permit.is_err() { + drain_external_generation_worker_tasks(&mut tasks).await; + return Ok(()); + } + // 只用 acquire 作为容量变化唤醒信号，许可立即归还；真正领取任务 + // 时在下方按返回的 job 数量逐个 try_acquire。 + } } } - let available = concurrency.saturating_sub(tasks.len()).max(1); + let available = work_slots.available_permits().max(1); let now_micros = current_utc_micros(); let lease_expires_at_micros = now_micros.saturating_add(duration_micros_i64(lease)); @@ -178,9 +193,13 @@ pub(crate) async fn run_external_generation_worker(state: AppState) -> Result<() for job in jobs { let state = state.clone(); let worker_id = worker_id.clone(); + let permit = work_slots + .clone() + .try_acquire_owned() + .expect("claimed job must have an execution capacity permit"); tasks.spawn(async move { if let Err(error) = - process_external_generation_job(state, worker_id, lease, job).await + process_external_generation_job(state, worker_id, lease, job, permit).await { error!(error = %error, "external generation worker 执行任务失败"); } @@ -255,16 +274,6 @@ async fn await_worker_task(tasks: &mut JoinSet<()>) { } } -async fn await_worker_task_or_shutdown( - tasks: &mut JoinSet<()>, - shutdown: &mut ExternalGenerationShutdownSignal, -) -> bool { - tokio::select! { - _ = shutdown.as_mut() => true, - _ = await_worker_task(tasks) => false, - } -} - async fn await_one_task_or_queue_wake_or_sleep_or_shutdown( tasks: &mut JoinSet<()>, queue_wake: &mut Option, @@ -326,6 +335,7 @@ async fn process_external_generation_job( worker_id: String, lease: Duration, job: ExternalGenerationJobRecord, + permit: OwnedSemaphorePermit, ) -> Result<(), String> { let heartbeat_interval = external_generation_worker_heartbeat_interval(lease); let job_timeout = external_generation_worker_job_timeout(&state.config, job.job_kind.as_str()); @@ -384,6 +394,7 @@ async fn process_external_generation_job( &job, lease, "任务超过执行预算", + Some(permit), ); Err(message) } @@ -393,6 +404,7 @@ async fn process_external_generation_job( &job, lease, "任务租约续期失败", + Some(permit), ); Err(error) } @@ -417,6 +429,7 @@ fn detach_external_generation_work_until_lease_expiry( job: &ExternalGenerationJobRecord, lease: Duration, reason: &'static str, + permit: Option, ) { let job_id = job.job_id.clone(); let job_kind = job.job_kind.clone(); @@ -454,6 +467,9 @@ fn detach_external_generation_work_until_lease_expiry( ); } } + // 保持执行许可直到 work 真正结束或被取消，避免超时任务脱管后继续 + // 累积图片/音频响应占用。 + drop(permit); }); } @@ -1972,6 +1988,7 @@ mod tests { &job, Duration::from_millis(200), "任务超过执行预算", + None, ); tokio::time::sleep(Duration::from_millis(100)).await; @@ -1981,6 +1998,39 @@ mod tests { ); } + #[tokio::test] + async fn worker_detached_work_keeps_execution_slot_until_finished() { + let slots = std::sync::Arc::new(tokio::sync::Semaphore::new(1)); + let permit = slots + .clone() + .acquire_owned() + .await + .expect("the only execution slot should be available"); + let work_handle = tokio::spawn(async { + tokio::time::sleep(Duration::from_millis(20)).await; + Ok(()) + }); + let job = external_generation_job_record_fixture(Some("lease-1")); + + detach_external_generation_work_until_lease_expiry( + work_handle, + &job, + Duration::from_millis(200), + "任务超过执行预算", + Some(permit), + ); + + assert!( + slots.try_acquire().is_err(), + "脱管 work 完成前不得重新领取执行容量" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + assert!( + slots.try_acquire().is_ok(), + "脱管 work 完成后应归还执行容量" + ); + } + #[tokio::test] async fn worker_detached_work_is_aborted_after_lease_arbitration_window() { let connection = std::sync::Arc::new(tokio::sync::Semaphore::new(1)); @@ -1999,6 +2049,7 @@ mod tests { &job, Duration::from_millis(10), "任务超过执行预算", + None, ); let reacquired = diff --git a/server-rs/crates/module-ai/src/application/store.rs b/server-rs/crates/module-ai/src/application/store.rs index f2d0c1175..4ff835d1a 100644 --- a/server-rs/crates/module-ai/src/application/store.rs +++ b/server-rs/crates/module-ai/src/application/store.rs @@ -1,5 +1,5 @@ use std::{ - collections::HashMap, + collections::{BTreeMap, HashMap}, sync::{Arc, Mutex}, }; @@ -9,6 +9,10 @@ use crate::{ use super::ensure_task_is_not_terminal; +const MAX_RETAINED_TASKS: usize = 1024; +const MAX_TASK_TEXT_OUTPUT_BYTES: usize = 512 * 1024; +const MAX_RETAINED_TASK_TEXT_BYTES: usize = 64 * 1024 * 1024; + #[derive(Clone, Debug, Default)] pub struct InMemoryAiTaskStore { inner: Arc>, @@ -17,7 +21,9 @@ pub struct InMemoryAiTaskStore { #[derive(Debug, Default)] struct InMemoryAiTaskStoreState { tasks: HashMap, - text_chunks: HashMap>, + // Keep only the ordered deltas needed to handle an out-of-order chunk. + // Completed tasks drop this map immediately; it is not a second durable log. + text_chunks: HashMap>>, } impl InMemoryAiTaskStore { @@ -34,7 +40,25 @@ impl InMemoryAiTaskStore { return Err(AiTaskServiceError::TaskAlreadyExists); } - state.text_chunks.insert(task.task_id.clone(), Vec::new()); + if state.tasks.len() >= MAX_RETAINED_TASKS { + let oldest_terminal = state + .tasks + .values() + .filter(|value| value.status.is_terminal()) + .min_by_key(|value| value.completed_at_micros.or(Some(value.updated_at_micros))) + .map(|value| value.task_id.clone()); + let Some(task_id) = oldest_terminal else { + return Err(AiTaskServiceError::Store( + "AI 任务仓储已达到内存容量上限".to_string(), + )); + }; + state.tasks.remove(&task_id); + state.text_chunks.remove(&task_id); + } + + state + .text_chunks + .insert(task.task_id.clone(), HashMap::new()); state.tasks.insert(task.task_id.clone(), task.clone()); Ok(task) } @@ -56,7 +80,11 @@ impl InMemoryAiTaskStore { .get_mut(task_id.trim()) .ok_or(AiTaskServiceError::TaskNotFound)?; apply(task)?; - Ok(task.clone()) + let snapshot = task.clone(); + if snapshot.status.is_terminal() { + state.text_chunks.remove(task_id.trim()); + } + Ok(snapshot) } pub(super) fn append_text_chunk( @@ -67,41 +95,62 @@ impl InMemoryAiTaskStore { .inner .lock() .map_err(|_| AiTaskServiceError::Store("AI 任务仓储锁已中毒".to_string()))?; - { + if chunk.delta_text.len() > MAX_TASK_TEXT_OUTPUT_BYTES { + return Err(AiTaskServiceError::Store( + "AI 任务文本输出超过内存上限".to_string(), + )); + } + let (previous_stage_output_bytes, previous_latest_output_bytes) = { let task = state .tasks - .get_mut(&chunk.task_id) + .get(&chunk.task_id) .ok_or(AiTaskServiceError::TaskNotFound)?; ensure_task_is_not_terminal(task.status)?; - let stage = task .stages - .iter_mut() + .iter() .find(|stage| stage.stage_kind == chunk.stage_kind) .ok_or(AiTaskServiceError::StageNotFound)?; - if stage.status == AiTaskStageStatus::Pending { - stage.status = AiTaskStageStatus::Running; - stage.started_at_micros = Some(chunk.created_at_micros); - } + ( + stage.text_output.as_ref().map_or(0, String::len), + task.latest_text_output.as_ref().map_or(0, String::len), + ) + }; - task.status = AiTaskStatus::Running; - task.started_at_micros - .get_or_insert(chunk.created_at_micros); + let (previous_chunk, aggregated_bytes, aggregated_text) = { + let chunks = state + .text_chunks + .get_mut(&chunk.task_id) + .ok_or(AiTaskServiceError::TaskNotFound)?; + let stage_chunks = chunks.entry(chunk.stage_kind).or_default(); + let previous_chunk = stage_chunks.insert(chunk.sequence, chunk.delta_text.clone()); + let aggregated_bytes = stage_chunks + .values() + .fold(0_usize, |total, delta| total.saturating_add(delta.len())); + let mut aggregated_text = String::with_capacity(aggregated_bytes); + for delta in stage_chunks.values() { + aggregated_text.push_str(delta); + } + (previous_chunk, aggregated_bytes, aggregated_text) + }; + if aggregated_bytes > MAX_TASK_TEXT_OUTPUT_BYTES { + rollback_text_chunk(&mut state, &chunk, previous_chunk); + return Err(AiTaskServiceError::Store( + "AI 任务文本输出超过内存上限".to_string(), + )); } - let chunks = state - .text_chunks - .get_mut(&chunk.task_id) - .ok_or(AiTaskServiceError::TaskNotFound)?; - chunks.push(chunk.clone()); - chunks.sort_by_key(|value| value.sequence); + let retained_text_bytes = retained_text_bytes(&state) + .saturating_sub(previous_stage_output_bytes) + .saturating_sub(previous_latest_output_bytes) + .saturating_add(aggregated_bytes.saturating_mul(2)); + if retained_text_bytes > MAX_RETAINED_TASK_TEXT_BYTES { + rollback_text_chunk(&mut state, &chunk, previous_chunk); + return Err(AiTaskServiceError::Store( + "AI 任务仓储文本工作集超过内存上限".to_string(), + )); + } - let aggregated_text = chunks - .iter() - .filter(|value| value.stage_kind == chunk.stage_kind) - .map(|value| value.delta_text.as_str()) - .collect::>() - .join(""); let normalized_output = if aggregated_text.trim().is_empty() { None } else { @@ -112,11 +161,19 @@ impl InMemoryAiTaskStore { .tasks .get_mut(&chunk.task_id) .ok_or(AiTaskServiceError::TaskNotFound)?; + ensure_task_is_not_terminal(task.status)?; let stage = task .stages .iter_mut() .find(|stage| stage.stage_kind == chunk.stage_kind) .ok_or(AiTaskServiceError::StageNotFound)?; + if stage.status == AiTaskStageStatus::Pending { + stage.status = AiTaskStageStatus::Running; + stage.started_at_micros = Some(chunk.created_at_micros); + } + task.status = AiTaskStatus::Running; + task.started_at_micros + .get_or_insert(chunk.created_at_micros); stage.text_output = normalized_output.clone(); task.latest_text_output = normalized_output; task.updated_at_micros = chunk.created_at_micros; @@ -136,3 +193,41 @@ impl InMemoryAiTaskStore { .ok_or(AiTaskServiceError::TaskNotFound) } } + +fn rollback_text_chunk( + state: &mut InMemoryAiTaskStoreState, + chunk: &AiTextChunkSnapshot, + previous_chunk: Option, +) { + if let Some(stage_chunks) = state + .text_chunks + .get_mut(&chunk.task_id) + .and_then(|chunks| chunks.get_mut(&chunk.stage_kind)) + { + if let Some(previous_chunk) = previous_chunk { + stage_chunks.insert(chunk.sequence, previous_chunk); + } else { + stage_chunks.remove(&chunk.sequence); + } + } +} + +fn retained_text_bytes(state: &InMemoryAiTaskStoreState) -> usize { + let snapshot_bytes = state.tasks.values().fold(0_usize, |total, task| { + let latest = task.latest_text_output.as_ref().map_or(0, String::len); + let stages = task.stages.iter().fold(0_usize, |stage_total, stage| { + stage_total.saturating_add(stage.text_output.as_ref().map_or(0, String::len)) + }); + total.saturating_add(latest).saturating_add(stages) + }); + state + .text_chunks + .values() + .fold(snapshot_bytes, |total, stages| { + stages.values().fold(total, |stage_total, chunks| { + chunks.values().fold(stage_total, |chunk_total, delta| { + chunk_total.saturating_add(delta.len()) + }) + }) + }) +} diff --git a/server-rs/crates/module-ai/src/tests.rs b/server-rs/crates/module-ai/src/tests.rs index 766320035..513f27a66 100644 --- a/server-rs/crates/module-ai/src/tests.rs +++ b/server-rs/crates/module-ai/src/tests.rs @@ -112,6 +112,47 @@ fn append_text_chunk_aggregates_stream_output_by_stage() { assert_eq!(second_chunk.sequence, 2); } +#[test] +fn append_text_chunk_rejects_output_over_stage_memory_limit_without_mutating_task() { + let service = build_service(); + let task = service + .create_task(build_create_input(AiTaskKind::CharacterChat)) + .expect("task should create"); + let max_output = "a".repeat(512 * 1024); + + let (updated, _) = service + .append_text_chunk( + &task.task_id, + AiTaskStageKind::RequestModel, + 1, + max_output.clone(), + task.created_at_micros + 1, + ) + .expect("the stage limit itself should be accepted"); + assert_eq!( + updated.latest_text_output.as_deref().map(str::len), + Some(max_output.len()) + ); + + let error = service + .append_text_chunk( + &task.task_id, + AiTaskStageKind::RequestModel, + 2, + "b".to_string(), + task.created_at_micros + 2, + ) + .expect_err("output beyond the stage limit should fail"); + assert!(matches!(error, AiTaskServiceError::Store(_))); + let after_rejection = service + .get_task(&task.task_id) + .expect("task should remain readable"); + assert_eq!( + after_rejection.latest_text_output.as_deref().map(str::len), + Some(max_output.len()) + ); +} + #[test] fn complete_stage_updates_latest_outputs() { let service = build_service(); diff --git a/server-rs/crates/module-auth/src/lib.rs b/server-rs/crates/module-auth/src/lib.rs index 6336495d0..58309b45a 100644 --- a/server-rs/crates/module-auth/src/lib.rs +++ b/server-rs/crates/module-auth/src/lib.rs @@ -28,6 +28,11 @@ use shared_kernel::{ use time::{Duration, OffsetDateTime}; use tracing::{info, warn}; +const REFRESH_SESSION_STALE_RETENTION: Duration = Duration::days(1); +const MAX_REFRESH_SESSIONS: usize = 8_192; +const MAX_PHONE_CODES: usize = 4_096; +const MAX_WECHAT_STATES: usize = 4_096; + #[derive(Clone, Debug)] pub struct InMemoryAuthStore { inner: Arc>, @@ -364,6 +369,7 @@ impl RefreshSessionService { input: CreateRefreshSessionInput, now: OffsetDateTime, ) -> Result { + self.store.prune_stale_sessions(now)?; self.store .find_by_user_id(&input.user_id) .map_err(map_password_store_error)? @@ -400,6 +406,7 @@ impl RefreshSessionService { input: RotateRefreshSessionInput, now: OffsetDateTime, ) -> Result { + self.store.prune_stale_sessions(now)?; let Some(refresh_token_hash) = normalize_required_string(&input.refresh_token_hash) else { return Err(RefreshSessionError::MissingToken); }; @@ -454,6 +461,7 @@ impl RefreshSessionService { user_id: &str, now: OffsetDateTime, ) -> Result { + self.store.prune_stale_sessions(now)?; self.store .find_by_user_id(user_id) .map_err(map_password_store_error)? @@ -468,6 +476,7 @@ impl RefreshSessionService { input: RevokeRefreshSessionByUserInput, now: OffsetDateTime, ) -> Result { + self.store.prune_stale_sessions(now)?; self.store .find_by_user_id(&input.user_id) .map_err(map_password_store_error)? @@ -492,6 +501,7 @@ impl RefreshSessionService { session_id: &str, now: OffsetDateTime, ) -> Result { + self.store.prune_stale_sessions(now)?; self.store .is_session_active_for_user(user_id, session_id.trim(), now) } @@ -511,6 +521,7 @@ impl PhoneAuthService { input: SendPhoneCodeInput, now: OffsetDateTime, ) -> Result { + self.store.prune_expired_phone_codes(now)?; let scene = input.scene.clone(); validate_mainland_china_country_code(input.country_code.as_deref())?; let normalized_phone = normalize_mainland_china_phone_number(&input.pure_phone_number)?; @@ -525,6 +536,8 @@ impl PhoneAuthService { ); self.store .ensure_phone_code_not_cooling_down(&normalized_phone.e164, &scene, now)?; + self.store + .ensure_phone_code_capacity(&normalized_phone.e164, &scene)?; let expires_at = now .checked_add(Duration::minutes(SMS_CODE_TTL_MINUTES)) .ok_or_else(|| PhoneAuthError::Store("短信验证码过期时间计算溢出".to_string()))?; @@ -787,6 +800,7 @@ impl WechatAuthStateService { input: CreateWechatAuthStateInput, now: OffsetDateTime, ) -> Result { + self.store.prune_wechat_states(now)?; let created_at = format_rfc3339(now).map_err(|message| { WechatAuthError::Store(format!("微信 state 时间格式化失败：{message}")) })?; @@ -1063,10 +1077,18 @@ impl InMemoryAuthStoreState { } } + let now = OffsetDateTime::now_utc(); for session in view.refresh_sessions { if !existing_user_ids.contains(&session.user_id) { continue; } + if should_prune_refresh_session_fields( + &session.expires_at, + session.revoked_at.as_deref(), + now, + ) { + continue; + } let client_info = serde_json::from_str::(&session.client_info_json) .map_err(|error| format!("解析 refresh session 客户端信息失败：{error}"))?; @@ -1185,6 +1207,8 @@ impl InMemoryAuthStore { &self, updated_at_micros: i64, ) -> Result { + self.prune_stale_sessions(OffsetDateTime::now_utc()) + .map_err(|error| error.to_string())?; let state = self .inner .lock() @@ -1262,6 +1286,38 @@ impl InMemoryAuthStore { }) } + fn prune_stale_sessions(&self, now: OffsetDateTime) -> Result<(), RefreshSessionError> { + let mut state = self + .inner + .lock() + .map_err(|_| RefreshSessionError::Store("会话仓储锁已中毒".to_string()))?; + let stale_session_ids = state + .sessions_by_id + .iter() + .filter(|(_, stored)| should_prune_refresh_session(&stored.session, now)) + .map(|(session_id, _)| session_id.clone()) + .collect::>(); + if stale_session_ids.is_empty() { + return Ok(()); + } + + for session_id in stale_session_ids { + let Some(stored) = state.sessions_by_id.remove(&session_id) else { + continue; + }; + if state + .session_id_by_refresh_token_hash + .get(&stored.session.refresh_token_hash) + .is_some_and(|mapped_id| mapped_id == &session_id) + { + state + .session_id_by_refresh_token_hash + .remove(&stored.session.refresh_token_hash); + } + } + self.persist_refresh_state(&state) + } + fn persist_state(&self, state: &InMemoryAuthStoreState) -> Result<(), String> { let _ = state; Ok(()) @@ -1894,6 +1950,11 @@ impl InMemoryAuthStore { "refresh token hash 已存在，无法重复创建会话".to_string(), )); } + if state.sessions_by_id.len() >= MAX_REFRESH_SESSIONS { + return Err(RefreshSessionError::Store( + "refresh session 内存容量已达到上限".to_string(), + )); + } state.session_id_by_refresh_token_hash.insert( session.refresh_token_hash.clone(), @@ -1918,10 +1979,41 @@ impl InMemoryAuthStore { .map_err(|_| PhoneAuthError::Store("短信验证码仓储锁已中毒".to_string()))?; // 手机号和业务场景共同决定同一份验证码快照，重复发送时直接覆盖旧值。 let key = build_phone_code_key(&code.phone_number, &code.scene); + if !state.phone_codes_by_key.contains_key(&key) + && state.phone_codes_by_key.len() >= MAX_PHONE_CODES + { + return Err(PhoneAuthError::Store( + "短信验证码内存容量已达到上限，请稍后重试".to_string(), + )); + } state.phone_codes_by_key.insert(key, code); Ok(()) } + fn prune_expired_phone_codes(&self, now: OffsetDateTime) -> Result<(), PhoneAuthError> { + let mut state = self + .inner + .lock() + .map_err(|_| PhoneAuthError::Store("短信验证码仓储锁已中毒".to_string()))?; + let expired_keys = state + .phone_codes_by_key + .iter() + .filter_map(|(key, stored)| { + OffsetDateTime::parse( + &stored.expires_at, + &time::format_description::well_known::Rfc3339, + ) + .ok() + .filter(|expires_at| *expires_at <= now) + .map(|_| key.clone()) + }) + .collect::>(); + for key in expired_keys { + state.phone_codes_by_key.remove(&key); + } + Ok(()) + } + fn ensure_phone_code_not_cooling_down( &self, phone_number: &str, @@ -1961,6 +2053,26 @@ impl InMemoryAuthStore { }) } + fn ensure_phone_code_capacity( + &self, + phone_number: &str, + scene: &PhoneAuthScene, + ) -> Result<(), PhoneAuthError> { + let state = self + .inner + .lock() + .map_err(|_| PhoneAuthError::Store("短信验证码仓储锁已中毒".to_string()))?; + let key = build_phone_code_key(phone_number, scene); + if state.phone_codes_by_key.contains_key(&key) + || state.phone_codes_by_key.len() < MAX_PHONE_CODES + { + return Ok(()); + } + Err(PhoneAuthError::Store( + "短信验证码内存容量已达到上限，请稍后重试".to_string(), + )) + } + fn get_active_phone_code( &self, phone_number: &str, @@ -2041,6 +2153,11 @@ impl InMemoryAuthStore { { return Err(WechatAuthError::Store("微信 state 已存在".to_string())); } + if state.wechat_states_by_token.len() >= MAX_WECHAT_STATES { + return Err(WechatAuthError::Store( + "微信登录 state 内存容量已达到上限，请稍后重试".to_string(), + )); + } state.wechat_states_by_token.insert( state_record.state_token.clone(), StoredWechatAuthState { @@ -2405,6 +2522,33 @@ impl InMemoryAuthStore { Ok(()) } + fn prune_wechat_states(&self, now: OffsetDateTime) -> Result<(), WechatAuthError> { + let mut state = self + .inner + .lock() + .map_err(|_| WechatAuthError::Store("微信 state 仓储锁已中毒".to_string()))?; + let stale_tokens = state + .wechat_states_by_token + .iter() + .filter_map(|(token, stored)| { + if stored.state.consumed_at.is_some() { + return Some(token.clone()); + } + OffsetDateTime::parse( + &stored.state.expires_at, + &time::format_description::well_known::Rfc3339, + ) + .ok() + .filter(|expires_at| *expires_at <= now) + .map(|_| token.clone()) + }) + .collect::>(); + for token in stale_tokens { + state.wechat_states_by_token.remove(&token); + } + Ok(()) + } + fn revoke_session_by_user_and_session_id( &self, user_id: &str, @@ -2568,6 +2712,25 @@ impl InMemoryAuthStore { } } +fn should_prune_refresh_session(session: &RefreshSessionRecord, now: OffsetDateTime) -> bool { + should_prune_refresh_session_fields(&session.expires_at, session.revoked_at.as_deref(), now) +} + +fn should_prune_refresh_session_fields( + expires_at: &str, + revoked_at: Option<&str>, + now: OffsetDateTime, +) -> bool { + let stale_before = now.saturating_sub(REFRESH_SESSION_STALE_RETENTION); + if let Some(revoked_at) = revoked_at { + return OffsetDateTime::parse(revoked_at, &time::format_description::well_known::Rfc3339) + .is_ok_and(|timestamp| timestamp <= stale_before); + } + + OffsetDateTime::parse(expires_at, &time::format_description::well_known::Rfc3339) + .is_ok_and(|timestamp| timestamp <= stale_before) +} + fn map_sms_provider_error_to_phone_error(error: SmsProviderError) -> PhoneAuthError { match error { SmsProviderError::InvalidVerifyCode => PhoneAuthError::InvalidVerifyCode, @@ -4012,6 +4175,70 @@ mod tests { ); } + #[tokio::test] + async fn stale_refresh_sessions_are_pruned_from_both_indexes() { + let store = build_store(); + let refresh_service = build_refresh_service(store.clone()); + let user = create_phone_login_user(store.clone(), "13800138008").await; + let now = OffsetDateTime::now_utc(); + + refresh_service + .create_session( + CreateRefreshSessionInput { + user_id: user.id.clone(), + refresh_token_hash: hash_refresh_session_token("stale-revoked"), + issued_by_provider: AuthLoginMethod::Password, + client_info: build_client_info(), + }, + now - Duration::days(2), + ) + .expect("stale session should create"); + store + .revoke_session_by_refresh_token_hash( + &hash_refresh_session_token("stale-revoked"), + now - Duration::days(2), + ) + .expect("stale session should revoke"); + + refresh_service + .create_session( + CreateRefreshSessionInput { + user_id: user.id.clone(), + refresh_token_hash: hash_refresh_session_token("recent-revoked"), + issued_by_provider: AuthLoginMethod::Password, + client_info: build_client_info(), + }, + now, + ) + .expect("recent session should create"); + store + .revoke_session_by_refresh_token_hash( + &hash_refresh_session_token("recent-revoked"), + now, + ) + .expect("recent session should revoke"); + + let projection = store + .export_projection_view(now.unix_timestamp()) + .expect("projection export should prune stale sessions"); + assert_eq!(projection.refresh_sessions.len(), 1); + assert_eq!( + projection.refresh_sessions[0].refresh_token_hash, + hash_refresh_session_token("recent-revoked") + ); + + let stale_error = refresh_service + .rotate_session( + RotateRefreshSessionInput { + refresh_token_hash: hash_refresh_session_token("stale-revoked"), + next_refresh_token_hash: hash_refresh_session_token("stale-next"), + }, + now, + ) + .expect_err("pruned session should no longer be indexed"); + assert_eq!(stale_error, RefreshSessionError::SessionNotFound); + } + #[tokio::test] async fn wechat_login_hits_existing_user_by_union_id_before_openid() { let store = build_store();