diff --git a/.hermes/shared-memory/decision-log.md b/.hermes/shared-memory/decision-log.md index 1b2e161b..6666ad75 100644 --- a/.hermes/shared-memory/decision-log.md +++ b/.hermes/shared-memory/decision-log.md @@ -16,6 +16,14 @@ --- +## 2026-06-05 api-server 重启先摘流再排空并持久化 outbox + +- 背景：生产部署重启 api-server 时，如果只用 `/healthz` 判断存活并直接停止进程，运行中的 HTTP 请求和本地 tracking outbox active 文件都可能被中断，容易造成用户请求失败或内存/本地缓冲数据延迟丢失。 +- 决策：`/healthz` 只表示进程存活，发布和生产接流检查统一使用 `/readyz`。api-server 收到 `SIGINT` / `SIGTERM` 后先把 readiness 标记为不可用，再交给 Axum graceful shutdown 排空已有 HTTP 请求；退出前在 `GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS` 窗口内封存 active tracking outbox 并尽力 flush sealed 文件，失败或超时则保留本地文件给下次启动重试。systemd 停机窗口统一放到 `TimeoutStopSec=90`。 +- 影响范围：`server-rs/crates/api-server`、`deploy/systemd/genarrative-api.service`、生产 API deploy 脚本、Jenkins API deploy 参数、Nginx 公网健康检查暴露策略、开发运维文档。 +- 验证方式：`cargo test -p api-server --manifest-path server-rs/Cargo.toml readyz_reports_readiness_and_draining_state`、`cargo test -p api-server --manifest-path server-rs/Cargo.toml shutdown_flush_seals_active_file_for_later_retry`、`cargo check -p api-server --manifest-path server-rs/Cargo.toml`、部署脚本 `bash -n` 与 `/readyz` 本机 smoke。 +- 关联文档：`docs/【开发运维】本地开发验证与生产运维-2026-05-15.md`。 + ## 2026-06-05 OSS 平台适配器输出结构化日志 - 背景：AI 生成资产、浏览器直传签名、私有读签名和对象确认都依赖 OSS；如果 OSS 侧只有错误字符串，排查资产写入 / 确认失败时很难按操作、对象、状态码和耗时下钻。 diff --git a/.hermes/shared-memory/development-workflow.md b/.hermes/shared-memory/development-workflow.md index 6cfb9325..9fad4640 100644 --- a/.hermes/shared-memory/development-workflow.md +++ b/.hermes/shared-memory/development-workflow.md @@ -205,7 +205,7 @@ npm run check:server-rs-ddd - 使用 `npm run dev:api-server` 重新拉起后端。 - 禁止使用 `npm run api-server:maincloud`、`npm.cmd run api-server:maincloud` 或任何 `GENARRATIVE_SPACETIME_MAINCLOUD_*` 口径；这些只属于历史残留。 -- 检查 `/healthz`。 +- 本地 smoke 检查 `/healthz`；发布后或确认实例可接生产流量时检查 `/readyz`。 - 执行对应自动测试。 - 涉及 SpacetimeDB 表、reducer、procedure、row shape 或绑定变化时，同步更新 `migration.rs`、表目录和生成绑定。 - SpacetimeDB 已有表新增字段必须放在 Rust 表结构体最后，并设置明确默认值；需要修改字段名时，先询问用户并确认迁移计划，再同步更新 `server-rs/crates/spacetime-module/src/migration.rs`、表目录和生成绑定。 @@ -224,7 +224,7 @@ npm run check:server-rs-ddd ## 生产压测与观测默认口径 - 作品列表 50 HTTP req/s 压测使用 `scripts/loadtest/README.md` 中的 K6 命令；当前脚本一次 iteration 请求两个公开列表接口，因此目标 50 HTTP req/s 对应 `PEAK_RPS=25`。 -- 生产 `api-server` 默认 backlog、worker threads、HTTP 并发背压、systemd 限制、Nginx upstream timing log 和 OTLP 开关以 `docs/【开发运维】本地开发验证与生产运维-2026-05-15.md` 为准。 +- 生产 `api-server` 默认 backlog、worker threads、HTTP 并发背压、`/readyz` 接流检查、systemd 优雅停机窗口、Nginx upstream timing log 和 OTLP 开关以 `docs/【开发运维】本地开发验证与生产运维-2026-05-15.md` 为准。 - OpenTelemetry 现阶段可选发送 traces / metrics / logs，但不会取代本地 `journalctl -u genarrative-api.service`、`logs/api-server/` 与 `/var/log/nginx/genarrative.*.log`。 - 指标 label 不写 raw URI、userId、profileId 或 request_id；request_id 只用于 trace/log 串联。 diff --git a/deploy/container/nginx.conf b/deploy/container/nginx.conf index 239b5c4c..be9dd0eb 100644 --- a/deploy/container/nginx.conf +++ b/deploy/container/nginx.conf @@ -190,7 +190,7 @@ http { proxy_set_header X-Request-Id $request_id; } - location ~ ^/(generated-|healthz) { + location ~ ^/(generated-|healthz|readyz) { return 404; } diff --git a/deploy/env/api-server.env.example b/deploy/env/api-server.env.example index 1434d8be..90b2378b 100644 --- a/deploy/env/api-server.env.example +++ b/deploy/env/api-server.env.example @@ -11,6 +11,7 @@ GENARRATIVE_API_MAX_CONCURRENT_REQUESTS=512 GENARRATIVE_API_GALLERY_MAX_CONCURRENT_REQUESTS=320 GENARRATIVE_API_DETAIL_MAX_CONCURRENT_REQUESTS=64 GENARRATIVE_API_ADMIN_MAX_CONCURRENT_REQUESTS=16 +GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS=5000 GENARRATIVE_TRACKING_OUTBOX_ENABLED=true GENARRATIVE_TRACKING_OUTBOX_DIR=/var/lib/genarrative/tracking-outbox GENARRATIVE_TRACKING_OUTBOX_BATCH_SIZE=500 diff --git a/deploy/nginx/genarrative-dev-http.conf b/deploy/nginx/genarrative-dev-http.conf index b7c0cdaa..62e87f14 100644 --- a/deploy/nginx/genarrative-dev-http.conf +++ b/deploy/nginx/genarrative-dev-http.conf @@ -215,7 +215,7 @@ server { } # 开发服仍不恢复旧生成资源代理和健康检查公网入口。 - location ~ ^/(generated-|healthz) { + location ~ ^/(generated-|healthz|readyz) { return 404; } diff --git a/deploy/nginx/genarrative.conf b/deploy/nginx/genarrative.conf index c26e9bbb..fa1a111b 100644 --- a/deploy/nginx/genarrative.conf +++ b/deploy/nginx/genarrative.conf @@ -235,7 +235,7 @@ server { } # 生产公网不再暴露旧生成资源代理和健康检查入口。 - location ~ ^/(generated-|healthz) { + location ~ ^/(generated-|healthz|readyz) { return 404; } diff --git a/deploy/systemd/genarrative-api.service b/deploy/systemd/genarrative-api.service index bba53a79..82ddd339 100644 --- a/deploy/systemd/genarrative-api.service +++ b/deploy/systemd/genarrative-api.service @@ -14,7 +14,7 @@ ExecStart=/opt/genarrative/current/api-server Restart=always RestartSec=5 KillSignal=SIGINT -TimeoutStopSec=30 +TimeoutStopSec=90 LimitNOFILE=65535 TasksMax=2048 diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index 79cc0c29..11372ad4 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -47,7 +47,7 @@ npm run dev:api-server Linux 本机多用户并发开发时，`npm run dev` 和 `npm run dev:*` 单模块命令会先在系统级端口段注册表里给当前用户分配一个端口段，再把该段映射为 `web = start`、`api = start + 1`、`spacetime = start + 2`、`admin-web = start + 3`。默认注册表目录是 `/var/tmp/genarrative-dev-port-ranges/`，其中 `registry.json` 记录各用户的活跃段，`registry.lock` 负责串行化分配；可以用 `GENARRATIVE_DEV_PORT_RANGE_REGISTRY_DIR` 覆盖目录。系统自动分配时从 `10000-10099` 开始，每次占用 100 个端口块，后续块按 `10100-10199`、`10200-10299` 递增；`GENARRATIVE_DEV_PORT_RANGE` 或 `--port-range` 只在 Linux 上生效，Windows 仍按原来的 3000 / 8082 / 3101 / 3102 端口探测与漂移逻辑运行，不读这个系统级注册表。 -后端日志默认写入 `logs/api-server/`。后端 API smoke 使用 `npm run dev:api-server` 并检查 `/healthz`；不要使用旧 `api-server:maincloud` 或任何 `GENARRATIVE_SPACETIME_MAINCLOUD_*` 口径。 +后端日志默认写入 `logs/api-server/`。后端 API smoke 使用 `npm run dev:api-server` 并检查 `/healthz`；需要确认实例可接生产流量时检查 `/readyz`。不要使用旧 `api-server:maincloud` 或任何 `GENARRATIVE_SPACETIME_MAINCLOUD_*` 口径。 开发态 `npm run dev` 与 `npm run dev:api-server` 会默认注入 `GENARRATIVE_DEV_PASSWORD_ENTRY_AUTO_REGISTER_ENABLED=true`，因此密码登录在本地开发环境可直接注册未知手机号账号；生产环境仍按 `api-server` 配置默认关闭该开关。 @@ -260,8 +260,9 @@ Jenkins 按 web / api / Spacetime module / build / deploy / publish 拆分 50 HTTP req/s 首版压测优化口径： - `api-server` 生产模板默认 `GENARRATIVE_API_LISTEN_BACKLOG=1024`、`GENARRATIVE_API_WORKER_THREADS=4`；本地未设置 worker threads 时继续使用 Tokio 默认值。 -- `GENARRATIVE_API_MAX_CONCURRENT_REQUESTS=512` 开启应用内 HTTP 并发背压；`GENARRATIVE_API_GALLERY_MAX_CONCURRENT_REQUESTS=320`、`GENARRATIVE_API_DETAIL_MAX_CONCURRENT_REQUESTS=64`、`GENARRATIVE_API_ADMIN_MAX_CONCURRENT_REQUESTS=16` 分别限制公开列表、公开详情和后台 API 热路径。超过许可时直接返回 `429 Too Many Requests` 和 `Retry-After: 1`，`/healthz` 不受该限制。这些值不是 RPS 限速；如果压测中 429 上升但内存和 p95 收敛，说明背压正在保护进程。直连 `api-server` 的极高 RPS 压测若出现 `connection refused`，通常已经打到 TCP 监听 / accept 层，应同时检查 backlog、Nginx upstream keepalive 和前置限流。 -- `genarrative-api.service` 设置 `LimitNOFILE=65535`、`TasksMax=2048`；上线后用 `systemctl show genarrative-api.service -p LimitNOFILE -p TasksMax` 和 `cat /proc/$(pidof api-server)/limits` 核对。 +- `GENARRATIVE_API_MAX_CONCURRENT_REQUESTS=512` 开启应用内 HTTP 并发背压；`GENARRATIVE_API_GALLERY_MAX_CONCURRENT_REQUESTS=320`、`GENARRATIVE_API_DETAIL_MAX_CONCURRENT_REQUESTS=64`、`GENARRATIVE_API_ADMIN_MAX_CONCURRENT_REQUESTS=16` 分别限制公开列表、公开详情和后台 API 热路径。超过许可时直接返回 `429 Too Many Requests` 和 `Retry-After: 1`，`/healthz` 与 `/readyz` 不受该限制。这些值不是 RPS 限速；如果压测中 429 上升但内存和 p95 收敛，说明背压正在保护进程。直连 `api-server` 的极高 RPS 压测若出现 `connection refused`，通常已经打到 TCP 监听 / accept 层，应同时检查 backlog、Nginx upstream keepalive 和前置限流。 +- `api-server` 正常运行时 `/healthz` 返回进程存活状态，`/readyz` 返回是否仍接收新流量；收到 `SIGINT` / `SIGTERM` 后会先把 readiness 标记为不可用，再让 Axum 停止接新连接并等待已有 HTTP 请求排空。systemd 仍以 `KillSignal=SIGINT` 停服务，`TimeoutStopSec=90` 作为长请求排空上限。 +- `genarrative-api.service` 设置 `LimitNOFILE=65535`、`TasksMax=2048`；上线后用 `systemctl show genarrative-api.service -p LimitNOFILE -p TasksMax -p TimeoutStopUSec` 和 `cat /proc/$(pidof api-server)/limits` 核对。 - Server provision 不再通过 Windows helper 下载。`Genarrative-Server-Provision` 的 `Prepare Provision Tools` 在 Linux build 节点直接准备 `spacetime-x86_64-unknown-linux-gnu.tar.gz` 和 `otelcol-contrib_0.151.0_linux_amd64.tar.gz`，再 stash `provision-tools/` 给后续发布阶段；如果 build 节点需要代理，在 `PROVISION_DOWNLOAD_PROXY` 配置 Linux 侧可访问的 HTTP 代理。后续 Linux 目标节点只消费 `provision-tools/`，不再回退到外网下载。 - `Genarrative-Stdb-Module-Build`、`Genarrative-Web-Build`、`Genarrative-Api-Build`、`Genarrative-*Deploy`、`Genarrative-Database-Import/Export`、`Genarrative-Full-Build-And-Deploy` 和 `Genarrative-Notify-Email` 的生产流水线现都以 Linux agent 为主，统一走 `http://127.0.0.1:3000/GenarrativeAI/Genarrative.git` 优先、`https://git.genarrative.world/GenarrativeAI/Genarrative.git` 备用的 checkout 口径。 - `otelcol-contrib.service` 作为可选系统服务加入 provision，默认监听 `127.0.0.1:4317/4318` 并使用 `deploy/otelcol/genarrative-debug.yaml`。api-server 是否发送 OTLP 仍由 `GENARRATIVE_OTEL_ENABLED` 控制，服务 unit 见 `deploy/systemd/otelcol-contrib.service`。 @@ -391,9 +392,10 @@ GENARRATIVE_TRACKING_OUTBOX_DIR=/var/lib/genarrative/tracking-outbox GENARRATIVE_TRACKING_OUTBOX_BATCH_SIZE=500 GENARRATIVE_TRACKING_OUTBOX_FLUSH_INTERVAL_MS=1000 GENARRATIVE_TRACKING_OUTBOX_MAX_BYTES=268435456 +GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS=5000 ``` -outbox 采用 NDJSON 文件保存原始事件。达到 `BATCH_SIZE` 时会立刻把当前 active 文件原子封存为 sealed 文件，并马上切到新的 active 继续写入；后台 worker 异步 flush sealed 文件，HTTP 请求线程不等待 SpacetimeDB。`FLUSH_INTERVAL_MS` 只负责兜底封存长时间未满批的 active 文件。SpacetimeDB 批量 procedure 返回成功后删除 sealed 文件，失败则保留文件并重试。`MAX_BYTES` 是磁盘保护阈值，不是 flush 阈值；超过后低价值 route tracking 可以被丢弃并记录日志 / 指标，关键同步事件不进入该丢弃路径。sealed 文件若出现无法解析的坏行，会重命名为 `corrupt-*` 隔离并记录 `genarrative.tracking_outbox.files.corrupt` 指标，避免一个坏文件阻塞后续批量入库。该机制提供至少一次投递语义，依赖 `tracking_event.event_id` 幂等跳过重复事件。 +outbox 采用 NDJSON 文件保存原始事件。达到 `BATCH_SIZE` 时会立刻把当前 active 文件原子封存为 sealed 文件，并马上切到新的 active 继续写入；后台 worker 异步 flush sealed 文件，HTTP 请求线程不等待 SpacetimeDB。`FLUSH_INTERVAL_MS` 只负责兜底封存长时间未满批的 active 文件。SpacetimeDB 批量 procedure 返回成功后删除 sealed 文件，失败则保留文件并重试。`MAX_BYTES` 是磁盘保护阈值，不是 flush 阈值；超过后低价值 route tracking 可以被丢弃并记录日志 / 指标，关键同步事件不进入该丢弃路径。sealed 文件若出现无法解析的坏行，会重命名为 `corrupt-*` 隔离并记录 `genarrative.tracking_outbox.files.corrupt` 指标，避免一个坏文件阻塞后续批量入库。api-server 收到退出信号后会在 `GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS` 窗口内封存 active 文件并尽力 flush sealed 文件，超时或 SpacetimeDB 暂不可用时保留本地文件给下次启动继续投递。该机制提供至少一次投递语义，依赖 `tracking_event.event_id` 幂等跳过重复事件。 release 机器如果日志每秒刷 `tracking outbox ... Permission denied (os error 13)`，先检查 `/etc/genarrative/api-server.env` 是否缺少 `GENARRATIVE_TRACKING_OUTBOX_DIR`。缺少时 `api-server` 会回退到本地开发默认相对路径 `server-rs/.data/tracking-outbox`，而 systemd 的工作目录是只读发布目录 `/opt/genarrative/releases/`，`genarrative` 用户无法在其中创建 `server-rs`。修复顺序： diff --git a/jenkins/Jenkinsfile.production-api-deploy b/jenkins/Jenkinsfile.production-api-deploy index 95432266..8d17e08e 100644 --- a/jenkins/Jenkinsfile.production-api-deploy +++ b/jenkins/Jenkinsfile.production-api-deploy @@ -24,7 +24,7 @@ pipeline { string(name: 'RELEASE_ROOT', defaultValue: '/opt/genarrative/releases', description: '生产 release 根目录') string(name: 'CURRENT_LINK', defaultValue: '/opt/genarrative/current', description: '当前版本软链接') string(name: 'SERVICE_NAME', defaultValue: 'genarrative-api.service', description: 'systemd 服务名') - string(name: 'HEALTH_URL', defaultValue: 'http://127.0.0.1:8082/healthz', description: '本机健康检查地址') + string(name: 'HEALTH_URL', defaultValue: 'http://127.0.0.1:8082/readyz', description: '本机 readiness 检查地址') string(name: 'API_ENV_FILE', defaultValue: '/etc/genarrative/api-server.env', description: 'api-server 环境文件') string(name: 'DATABASE', defaultValue: 'genarrative-prod', description: 'api-server 连接的 SpacetimeDB database') string(name: 'SPACETIME_SERVER_URL', defaultValue: 'http://127.0.0.1:3101', description: 'api-server 连接的 SpacetimeDB server URL') diff --git a/scripts/deploy/production-api-deploy.sh b/scripts/deploy/production-api-deploy.sh index 9373b375..0f861923 100644 --- a/scripts/deploy/production-api-deploy.sh +++ b/scripts/deploy/production-api-deploy.sh @@ -5,10 +5,10 @@ set -euo pipefail usage() { cat <<'EOF' 用法: - ./scripts/deploy/production-api-deploy.sh --source-dir build/ [--version ] [--release-root /opt/genarrative/releases] [--current-link /opt/genarrative/current] [--service genarrative-api.service] [--health-url http://127.0.0.1:8082/healthz] [--api-env-file /etc/genarrative/api-server.env] [--database genarrative-prod] [--spacetime-server-url http://127.0.0.1:3101] + ./scripts/deploy/production-api-deploy.sh --source-dir build/ [--version ] [--release-root /opt/genarrative/releases] [--current-link /opt/genarrative/current] [--service genarrative-api.service] [--health-url http://127.0.0.1:8082/readyz] [--api-env-file /etc/genarrative/api-server.env] [--database genarrative-prod] [--spacetime-server-url http://127.0.0.1:3101] 说明: - 进入维护模式，校验并发布 api-server 单文件，更新 current 链接，重启 systemd 服务并执行 healthz 检查。 + 进入维护模式，校验并发布 api-server 单文件，更新 current 链接，重启 systemd 服务并执行 readiness 检查。 若传入 --database，会在重启前把 GENARRATIVE_SPACETIME_DATABASE 写入 api-server 环境文件，避免服务继续读取旧库。 失败时保留维护模式。 EOF @@ -209,6 +209,7 @@ ensure_runtime_env_and_dirs() { # 旧生产环境文件会被 server-provision 保留，不一定包含新增的运行态写入路径。 # 发布前只补缺省值，不覆盖线上已经定制过的目录或开关。 + ensure_env_value "${api_env_file}" "GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS" "5000" ensure_env_value "${api_env_file}" "GENARRATIVE_TRACKING_OUTBOX_ENABLED" "true" ensure_env_value "${api_env_file}" "GENARRATIVE_TRACKING_OUTBOX_DIR" "/var/lib/genarrative/tracking-outbox" ensure_env_value "${api_env_file}" "GENARRATIVE_TRACKING_OUTBOX_BATCH_SIZE" "500" @@ -228,7 +229,7 @@ VERSION="" RELEASE_ROOT="/opt/genarrative/releases" CURRENT_LINK="/opt/genarrative/current" SERVICE_NAME="genarrative-api.service" -HEALTH_URL="http://127.0.0.1:8082/healthz" +HEALTH_URL="http://127.0.0.1:8082/readyz" API_ENV_FILE="/etc/genarrative/api-server.env" DATABASE="" SPACETIME_SERVER_URL="" @@ -362,7 +363,7 @@ ln -sfn "${RELEASE_DIR}" "${CURRENT_LINK}" echo "[production-api-deploy] 重启服务: ${SERVICE_NAME}" systemctl restart "${SERVICE_NAME}" -echo "[production-api-deploy] 等待 healthz: ${HEALTH_URL}" +echo "[production-api-deploy] 等待 readiness: ${HEALTH_URL}" for _ in {1..30}; do if curl -fsS "${HEALTH_URL}" >/dev/null; then "${SCRIPT_DIR}/maintenance-off.sh" @@ -373,5 +374,5 @@ for _ in {1..30}; do sleep 2 done -echo "[production-api-deploy] healthz 检查超时: ${HEALTH_URL}" >&2 +echo "[production-api-deploy] readiness 检查超时: ${HEALTH_URL}" >&2 exit 1 diff --git a/server-rs/crates/api-server/Cargo.toml b/server-rs/crates/api-server/Cargo.toml index c07bdcf4..28b54d93 100644 --- a/server-rs/crates/api-server/Cargo.toml +++ b/server-rs/crates/api-server/Cargo.toml @@ -54,7 +54,7 @@ shared-kernel = { workspace = true } shared-logging = { workspace = true } socket2 = { workspace = true } spacetime-client = { workspace = true } -tokio = { workspace = true, features = ["macros", "rt-multi-thread", "net", "time", "sync", "fs", "io-util"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "net", "time", "sync", "fs", "io-util", "signal"] } tokio-stream = { workspace = true } futures-util = { workspace = true } time = { workspace = true, features = ["formatting"] } diff --git a/server-rs/crates/api-server/src/app.rs b/server-rs/crates/api-server/src/app.rs index 602b608a..5fe098a3 100644 --- a/server-rs/crates/api-server/src/app.rs +++ b/server-rs/crates/api-server/src/app.rs @@ -877,6 +877,46 @@ mod tests { ); } + #[tokio::test] + async fn readyz_reports_readiness_and_draining_state() { + let state = AppState::new(AppConfig::default()).expect("state should build"); + let app = build_router(state.clone()); + + let ready_response = app + .clone() + .oneshot( + Request::builder() + .uri("/readyz") + .header("x-request-id", "req-ready") + .body(Body::empty()) + .expect("readyz request should build"), + ) + .await + .expect("readyz request should succeed"); + assert_eq!(ready_response.status(), StatusCode::OK); + let ready_body = read_json_response(ready_response).await; + assert_eq!(ready_body["ok"], Value::Bool(true)); + assert_eq!(ready_body["ready"], Value::Bool(true)); + + state.mark_not_ready(); + let draining_response = app + .oneshot( + Request::builder() + .uri("/readyz") + .header("x-request-id", "req-draining") + .body(Body::empty()) + .expect("readyz request should build"), + ) + .await + .expect("readyz request should succeed"); + assert_eq!(draining_response.status(), StatusCode::SERVICE_UNAVAILABLE); + let draining_body = read_json_response(draining_response).await; + assert_eq!( + draining_body["error"]["details"]["reason"], + "api_server_draining" + ); + } + #[tokio::test] async fn creative_agent_draft_edit_rejects_unconfirmed_template_session() { let app = build_internal_creative_agent_app(); diff --git a/server-rs/crates/api-server/src/backpressure.rs b/server-rs/crates/api-server/src/backpressure.rs index 3fc2b689..1f6baf7a 100644 --- a/server-rs/crates/api-server/src/backpressure.rs +++ b/server-rs/crates/api-server/src/backpressure.rs @@ -102,7 +102,7 @@ fn reject_overloaded_request(request: &Request) -> Response { } fn should_bypass_backpressure(request: &Request) -> bool { - request.uri().path() == "/healthz" + matches!(request.uri().path(), "/healthz" | "/readyz") } fn classify_request_permit_pool(path: &str) -> HttpRequestPermitPoolKind { @@ -200,6 +200,7 @@ mod tests { .route("/held", get(held_request)) .route("/fast", get(fast_request)) .route("/healthz", get(fast_request)) + .route("/readyz", get(fast_request)) .layer(middleware::from_fn_with_state( backpressure_state, limit_concurrent_requests, @@ -297,6 +298,13 @@ mod tests { .expect("healthz request should complete"); assert_eq!(health_response.status(), StatusCode::OK); + let ready_response = app + .clone() + .oneshot(test_request("/readyz")) + .await + .expect("readyz request should complete"); + assert_eq!(ready_response.status(), StatusCode::OK); + gate.release.notify_one(); let completed_response = held_response .await diff --git a/server-rs/crates/api-server/src/config.rs b/server-rs/crates/api-server/src/config.rs index 263c7556..3fe02061 100644 --- a/server-rs/crates/api-server/src/config.rs +++ b/server-rs/crates/api-server/src/config.rs @@ -25,6 +25,7 @@ pub struct AppConfig { pub gallery_max_concurrent_requests: Option, pub detail_max_concurrent_requests: Option, pub admin_max_concurrent_requests: Option, + pub shutdown_outbox_flush_timeout: Duration, pub tracking_outbox_enabled: bool, pub tracking_outbox_dir: PathBuf, pub tracking_outbox_batch_size: usize, @@ -169,6 +170,7 @@ impl Default for AppConfig { gallery_max_concurrent_requests: None, detail_max_concurrent_requests: None, admin_max_concurrent_requests: None, + shutdown_outbox_flush_timeout: Duration::from_millis(5_000), tracking_outbox_enabled: true, tracking_outbox_dir: PathBuf::from("server-rs/.data/tracking-outbox"), tracking_outbox_batch_size: 500, @@ -365,6 +367,11 @@ impl AppConfig { { config.admin_max_concurrent_requests = Some(max_concurrent_requests); } + if let Some(timeout_ms) = + read_first_positive_u64_env(&["GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS"]) + { + config.shutdown_outbox_flush_timeout = Duration::from_millis(timeout_ms); + } if let Some(enabled) = read_first_bool_env(&["GENARRATIVE_TRACKING_OUTBOX_ENABLED"]) { config.tracking_outbox_enabled = enabled; } @@ -1324,6 +1331,7 @@ mod tests { std::env::remove_var("GENARRATIVE_API_GALLERY_MAX_CONCURRENT_REQUESTS"); std::env::remove_var("GENARRATIVE_API_DETAIL_MAX_CONCURRENT_REQUESTS"); std::env::remove_var("GENARRATIVE_API_ADMIN_MAX_CONCURRENT_REQUESTS"); + std::env::remove_var("GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS"); std::env::remove_var("GENARRATIVE_TRACKING_OUTBOX_ENABLED"); std::env::remove_var("GENARRATIVE_TRACKING_OUTBOX_DIR"); std::env::remove_var("GENARRATIVE_TRACKING_OUTBOX_BATCH_SIZE"); @@ -1336,6 +1344,7 @@ mod tests { std::env::set_var("GENARRATIVE_API_GALLERY_MAX_CONCURRENT_REQUESTS", "64"); std::env::set_var("GENARRATIVE_API_DETAIL_MAX_CONCURRENT_REQUESTS", "32"); std::env::set_var("GENARRATIVE_API_ADMIN_MAX_CONCURRENT_REQUESTS", "16"); + std::env::set_var("GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS", "3000"); std::env::set_var("GENARRATIVE_TRACKING_OUTBOX_ENABLED", "false"); std::env::set_var( "GENARRATIVE_TRACKING_OUTBOX_DIR", @@ -1354,6 +1363,10 @@ mod tests { assert_eq!(config.gallery_max_concurrent_requests, Some(64)); assert_eq!(config.detail_max_concurrent_requests, Some(32)); assert_eq!(config.admin_max_concurrent_requests, Some(16)); + assert_eq!( + config.shutdown_outbox_flush_timeout, + std::time::Duration::from_millis(3_000) + ); assert!(!config.tracking_outbox_enabled); assert_eq!( config.tracking_outbox_dir, @@ -1374,6 +1387,7 @@ mod tests { std::env::remove_var("GENARRATIVE_API_GALLERY_MAX_CONCURRENT_REQUESTS"); std::env::remove_var("GENARRATIVE_API_DETAIL_MAX_CONCURRENT_REQUESTS"); std::env::remove_var("GENARRATIVE_API_ADMIN_MAX_CONCURRENT_REQUESTS"); + std::env::remove_var("GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS"); std::env::remove_var("GENARRATIVE_TRACKING_OUTBOX_ENABLED"); std::env::remove_var("GENARRATIVE_TRACKING_OUTBOX_DIR"); std::env::remove_var("GENARRATIVE_TRACKING_OUTBOX_BATCH_SIZE"); diff --git a/server-rs/crates/api-server/src/health.rs b/server-rs/crates/api-server/src/health.rs index ba043e02..ee83a012 100644 --- a/server-rs/crates/api-server/src/health.rs +++ b/server-rs/crates/api-server/src/health.rs @@ -1,7 +1,15 @@ -use axum::{Json, extract::Extension}; +use axum::{ + Json, + extract::{Extension, State}, + http::StatusCode, + response::{IntoResponse, Response}, +}; use serde_json::{Value, json}; -use crate::{api_response::json_success_body, request_context::RequestContext}; +use crate::{ + api_response::json_success_body, http_error::AppError, request_context::RequestContext, + state::AppState, +}; pub async fn health_check(Extension(request_context): Extension) -> Json { json_success_body( @@ -12,3 +20,28 @@ pub async fn health_check(Extension(request_context): Extension) }), ) } + +pub async fn readiness_check( + State(state): State, + Extension(request_context): Extension, +) -> Response { + if state.is_ready() { + return json_success_body( + Some(&request_context), + json!({ + "ok": true, + "ready": true, + "service": "genarrative-api-server", + }), + ) + .into_response(); + } + + AppError::from_status(StatusCode::SERVICE_UNAVAILABLE) + .with_message("api-server 正在退出，不再接收新流量") + .with_details(json!({ + "reason": "api_server_draining", + "ready": false, + })) + .into_response_with_context(Some(&request_context)) +} diff --git a/server-rs/crates/api-server/src/main.rs b/server-rs/crates/api-server/src/main.rs index 0c511311..fc9ee2e4 100644 --- a/server-rs/crates/api-server/src/main.rs +++ b/server-rs/crates/api-server/src/main.rs @@ -99,25 +99,35 @@ use shared_logging::{OtelConfig, init_tracing}; use socket2::{Domain, Protocol, Socket, Type}; use std::{ collections::HashSet, - env, fs, io, + env, fs, future, io, net::{SocketAddr, TcpListener as StdTcpListener}, - panic, thread, + panic, + sync::Arc, + thread, time::Duration, }; use tokio::net::TcpListener; use tokio::runtime::Builder as TokioRuntimeBuilder; use tokio::time::timeout; -use tracing::{error, info}; +use tracing::{error, info, warn}; use crate::{ app::{build_router, build_spacetime_unavailable_router}, config::AppConfig, state::{AppState, AppStateInitError}, + tracking_outbox::TrackingOutbox, }; const API_SERVER_STARTUP_STACK_SIZE_BYTES: usize = 32 * 1024 * 1024; const AUTH_STORE_STARTUP_RESTORE_TIMEOUT: Duration = Duration::from_secs(8); +#[derive(Clone)] +struct ShutdownContext { + app_state: Option, + tracking_outbox: Option>, + outbox_flush_timeout: Duration, +} + fn main() -> Result<(), io::Error> { // Windows 本地调试下 Axum 路由树和启动恢复链较重，显式放大启动线程栈，避免 debug 构建在进入监听前栈溢出。 let server_thread = thread::Builder::new() @@ -158,19 +168,33 @@ async fn run_server(config: AppConfig) -> Result<(), io::Error> { let listen_backlog = config.listen_backlog; let worker_threads = config.worker_threads; let otel_enabled = config.otel_enabled; + let outbox_flush_timeout = config.shutdown_outbox_flush_timeout; let listener = build_tcp_listener(bind_address, listen_backlog)?; - let router = match restore_app_state_for_startup(config).await { + let (router, shutdown_context) = match restore_app_state_for_startup(config).await { Ok(state) => { state.puzzle_gallery_cache().spawn_cleanup_task(); - if let Some(outbox) = state.tracking_outbox() { + let tracking_outbox = state.tracking_outbox(); + if let Some(outbox) = tracking_outbox.clone() { outbox.spawn_worker(); } - build_router(state) - } - Err(AppStateInitError::DependencyUnavailable(message)) => { - build_spacetime_unavailable_router(message) + ( + build_router(state.clone()), + ShutdownContext { + app_state: Some(state), + tracking_outbox, + outbox_flush_timeout, + }, + ) } + Err(AppStateInitError::DependencyUnavailable(message)) => ( + build_spacetime_unavailable_router(message), + ShutdownContext { + app_state: None, + tracking_outbox: None, + outbox_flush_timeout, + }, + ), Err(error) => { return Err(std::io::Error::other(format!( "初始化应用状态失败：{error}" @@ -186,7 +210,98 @@ async fn run_server(config: AppConfig) -> Result<(), io::Error> { "api-server 已完成 tracing 初始化并开始监听" ); - axum::serve(listener, router).await + let result = axum::serve(listener, router) + .with_graceful_shutdown(shutdown_signal(shutdown_context.clone())) + .await; + finalize_shutdown(shutdown_context).await; + result +} + +async fn shutdown_signal(context: ShutdownContext) { + let signal = wait_for_shutdown_signal().await; + if let Some(state) = context.app_state.as_ref() { + state.mark_not_ready(); + } + info!( + signal, + "api-server 收到退出信号，已标记 readiness 不可用并开始排空 HTTP 请求" + ); +} + +async fn wait_for_shutdown_signal() -> &'static str { + #[cfg(unix)] + { + tokio::select! { + signal = wait_for_ctrl_c_signal() => signal, + signal = wait_for_sigterm_signal() => signal, + } + } + + #[cfg(not(unix))] + { + wait_for_ctrl_c_signal().await + } +} + +async fn wait_for_ctrl_c_signal() -> &'static str { + if let Err(error) = tokio::signal::ctrl_c().await { + error!(error = %error, "监听 SIGINT 失败，无法通过 Ctrl-C 触发优雅退出"); + future::pending::<()>().await; + } + "sigint" +} + +#[cfg(unix)] +async fn wait_for_sigterm_signal() -> &'static str { + let mut signal = match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + { + Ok(signal) => signal, + Err(error) => { + error!(error = %error, "监听 SIGTERM 失败，无法通过 systemd terminate 触发优雅退出"); + future::pending::<()>().await; + unreachable!("pending future never returns"); + } + }; + signal.recv().await; + "sigterm" +} + +async fn finalize_shutdown(context: ShutdownContext) { + if let Some(state) = context.app_state.as_ref() { + state.mark_not_ready(); + } + + let Some(outbox) = context.tracking_outbox else { + return; + }; + + if context.outbox_flush_timeout.is_zero() { + warn!("api-server 退出时 tracking outbox flush timeout 为 0，跳过主动 flush"); + return; + } + + let timeout_ms = context + .outbox_flush_timeout + .as_millis() + .min(u128::from(u64::MAX)) as u64; + info!(timeout_ms, "api-server 退出前封存并 flush tracking outbox"); + match timeout(context.outbox_flush_timeout, outbox.flush_for_shutdown()).await { + Ok(Ok(())) => { + info!("api-server 退出前 tracking outbox flush 完成"); + } + Ok(Err(error)) => { + warn!( + error = %error, + "api-server 退出前 tracking outbox flush 未完成，已保留本地文件等待下次启动重试" + ); + } + Err(_) => { + warn!( + timeout_ms, + "api-server 退出前 tracking outbox flush 超时，已保留本地文件等待下次启动重试" + ); + } + } } fn build_tcp_listener( diff --git a/server-rs/crates/api-server/src/modules/health.rs b/server-rs/crates/api-server/src/modules/health.rs index 5e2f19ac..dd488807 100644 --- a/server-rs/crates/api-server/src/modules/health.rs +++ b/server-rs/crates/api-server/src/modules/health.rs @@ -1,7 +1,12 @@ use axum::{Router, routing::get}; -use crate::{health::health_check, state::AppState}; +use crate::{ + health::{health_check, readiness_check}, + state::AppState, +}; pub fn router(_state: AppState) -> Router { - Router::new().route("/healthz", get(health_check)) + Router::new() + .route("/healthz", get(health_check)) + .route("/readyz", get(readiness_check)) } diff --git a/server-rs/crates/api-server/src/state.rs b/server-rs/crates/api-server/src/state.rs index 4d178687..d4bbb445 100644 --- a/server-rs/crates/api-server/src/state.rs +++ b/server-rs/crates/api-server/src/state.rs @@ -2,7 +2,10 @@ use std::{ collections::HashMap, error::Error, fmt, - sync::{Arc, Mutex}, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, }; use axum::extract::FromRef; @@ -229,6 +232,7 @@ pub struct AppStateInner { // 配置会在后续中间件、路由和平台适配接入时逐步消费。 #[allow(dead_code)] pub config: AppConfig, + ready: AtomicBool, http_request_permit_pools: HttpRequestPermitPools, auth_jwt_config: JwtConfig, admin_runtime: Option, @@ -399,6 +403,7 @@ impl AppState { Ok(Self(Arc::new(AppStateInner { config, + ready: AtomicBool::new(true), http_request_permit_pools, auth_jwt_config, admin_runtime, @@ -447,6 +452,14 @@ impl AppState { self.http_request_permit_pools.clone() } + pub fn is_ready(&self) -> bool { + self.ready.load(Ordering::Acquire) + } + + pub fn mark_not_ready(&self) { + self.ready.store(false, Ordering::Release); + } + pub async fn upsert_creation_entry_type_config( &self, input: module_runtime::CreationEntryTypeAdminUpsertInput, diff --git a/server-rs/crates/api-server/src/tracking_outbox.rs b/server-rs/crates/api-server/src/tracking_outbox.rs index cf2b4a97..eb04762b 100644 --- a/server-rs/crates/api-server/src/tracking_outbox.rs +++ b/server-rs/crates/api-server/src/tracking_outbox.rs @@ -159,6 +159,16 @@ impl TrackingOutbox { }); } + pub async fn flush_for_shutdown(&self) -> Result<(), TrackingOutboxError> { + { + let mut inner = self.inner.lock().await; + self.ensure_initialized_locked(&mut inner).await?; + self.seal_active_locked(&mut inner, "shutdown").await?; + } + + self.flush_sealed_files_once().await + } + async fn seal_active_if_due(&self) -> Result<(), TrackingOutboxError> { let mut inner = self.inner.lock().await; self.ensure_initialized_locked(&mut inner).await?; @@ -176,7 +186,11 @@ impl TrackingOutbox { crate::telemetry::update_tracking_outbox_pending_files(sealed_files.len()); for path in sealed_files { let started_at = Instant::now(); - let metadata = fs::metadata(&path).await?; + let metadata = match fs::metadata(&path).await { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => return Err(error.into()), + }; let file_bytes = metadata.len(); let events = match read_outbox_events(&path).await { Ok(events) => events, @@ -203,7 +217,11 @@ impl TrackingOutbox { match self.spacetime_client.record_tracking_events(events).await { Ok(accepted_count) => { - fs::remove_file(&path).await?; + match fs::remove_file(&path).await { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } self.subtract_total_bytes(file_bytes).await; crate::telemetry::record_tracking_outbox_flush( started_at.elapsed(), @@ -596,6 +614,34 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + #[tokio::test] + async fn shutdown_flush_seals_active_file_for_later_retry() { + let dir = test_dir("shutdown"); + let outbox = test_outbox(dir.clone(), 500, 1024 * 1024); + + outbox.enqueue(sample_event("event-1")).await.unwrap(); + let result = outbox.flush_for_shutdown().await; + + assert!( + matches!(result, Err(TrackingOutboxError::Spacetime(_))), + "missing test SpacetimeDB should keep sealed file for retry" + ); + assert!(!dir.join(ACTIVE_FILE_NAME).exists()); + let sealed_count = std::fs::read_dir(&dir) + .unwrap() + .filter_map(Result::ok) + .filter(|entry| { + entry + .file_name() + .to_str() + .is_some_and(|name| name.starts_with(SEALED_FILE_PREFIX)) + }) + .count(); + assert_eq!(sealed_count, 1); + + let _ = std::fs::remove_dir_all(dir); + } + #[test] fn directory_size_excludes_quarantined_corrupt_files() { let dir = test_dir("directory-size");