修复跨节点短信冷却与微信状态膨胀

发码前通过占位验证码与投影 CAS 原子占用冷却窗口限制活动微信 OAuth state 数量并在创建前清理过期状态补充认证架构、运维决策和回归测试
2026-08-27 19:34:14 +08:00
parent 6e9948ac9e
commit 874aa09b9a
5 changed files with 161 additions and 12 deletions
@@ -26,7 +26,7 @@
 ## 2026-08-27 短期认证状态进入共享 typed projection

 - 背景：短信验证码和微信 OAuth state 仍只存在 API 进程内 HashMap，多节点请求或 API 重启会直接丢失，无法满足无粘性会话的鉴权恢复要求。
- 决策：`AuthStoreProjectionView` 增加 `phone_codes` 与 `wechat_states` typed 字段，由 `auth_store_projection_meta` 以 JSON 投影持久化；启动恢复、CAS 同步和失败后的权威刷新都覆盖这两类短期状态。验证码哈希使用部署级稳定盐（当前复用 `GENARRATIVE_JWT_SECRET`），各 API 节点必须一致；认证 handler 在发码、消费验证码、创建/消费微信 state 后都要完成 projection sync，失败即返回服务错误；短期状态首次未命中时允许从正式投影做一次受 CAS 保护的只读刷新，再重试读取，不能依赖粘性会话。短期状态仍由 `module-auth` 内存工作集执行领域校验，但不再把本机 HashMap 当作持久化或跨节点真相。
+- 决策：`AuthStoreProjectionView` 增加 `phone_codes` 与 `wechat_states` typed 字段，由 `auth_store_projection_meta` 以 JSON 投影持久化；启动恢复、CAS 同步和失败后的权威刷新都覆盖这两类短期状态。验证码哈希使用部署级稳定盐（当前复用 `GENARRATIVE_JWT_SECRET`），各 API 节点必须一致；发码前先刷新权威投影并用占位验证码记录做一次 projection CAS，只有占用成功才调用短信 provider，避免跨节点冷却竞态；认证 handler 在发码、消费验证码、创建/消费微信 state 后都要完成 projection sync，失败即返回服务错误；短期状态首次未命中时允许从正式投影做一次受 CAS 保护的只读刷新，再重试读取，不能依赖粘性会话。微信 OAuth state 设置有界活动数量，避免单个 JSON 投影无界膨胀。短期状态仍由 `module-auth` 内存工作集执行领域校验，但不再把本机 HashMap 当作持久化或跨节点真相。
 - 影响范围：`module-auth` projection、`spacetime-module` auth schema/procedure、`spacetime-client` bindings/facade、api-server 手机号 / 微信 handler、认证架构与运维文档。
 - 验证方式：运行 module-auth projection roundtrip（验证码可跨恢复校验、微信 state 可跨恢复消费）、SpacetimeDB schema/runtime/DDD 门禁、api-server 定向测试、编码和 diff 检查。

@@ -418,6 +418,8 @@ Responses 的终态载荷既是工具调用的恢复源，也是正文的恢复

 `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` 为准。`api-server` 多节点必须使用相同的部署级验证码哈希盐（当前复用 `GENARRATIVE_JWT_SECRET`）；轮换该 secret 会使尚未消费的短信验证码失效，但不会改变已持久化账号或 session。

+短期状态的并发保护：发短信前先从正式投影刷新工作集，再写入不可消费的占位验证码并通过 `sync_auth_store_projection` 的基线 CAS 占用手机号 / 场景冷却窗口；只有占用成功后才调用外部短信 provider，provider 成功后再同步真实验证码哈希。微信 OAuth state 在 `module-auth` 工作集内限制活动数量，超过上限直接拒绝创建，避免单行 JSON 投影无界增长；过期 state 仍由投影导出时清理。
+
 ### `bark_battle_draft_config`

 - Rust 结构体：`BarkBattleDraftConfigRow`
@@ -868,6 +868,8 @@ journalctl -u genarrative-api.service --since '30 seconds ago' --no-pager | grep

 `Genarrative-Server-Provision` 和 `Genarrative-Api-Deploy` 会在保留旧 `/etc/genarrative/api-server.env` 的前提下补齐缺失的 tracking outbox 运行态路径，并确保 `/var/lib/genarrative/tracking-outbox` 归属 `genarrative:genarrative`。用户认证真相源只允许从 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）和 `auth_store_projection_meta` 中的短期验证码 / 微信 state 投影恢复；请求首次未命中短期状态时会从正式投影做一次只读刷新，不依赖粘性会话。不要再配置或依赖 `GENARRATIVE_AUTH_STORE_PATH` / `auth-store.json`，`module-auth` 也不再维护本地文件持久化；`auth_store_snapshot` 不再作为备查或运行期恢复源，只在正式认证表为空时一次性转移最新旧快照并清空，且旧 `get_auth_store_snapshot` / `upsert_auth_store_snapshot` / `import_auth_store_snapshot` 入口已经删除。所有 API 节点必须使用相同的 `GENARRATIVE_JWT_SECRET`，它也作为验证码哈希盐；轮换后尚未消费的验证码会失效。如果 `api-server` 启动时连不上 SpacetimeDB，会持续重试启动恢复，直到认证工作集从 SpacetimeDB 正式表和短期投影恢复成功后才开始监听 HTTP，以避免用空本地状态或旧快照覆盖认证表。

+发短信运维门禁：handler 会先刷新正式认证投影，再通过 projection CAS 写入不可消费的占位验证码来占用跨节点冷却窗口；占用失败时不得调用短信 provider。微信 OAuth state 活动数量有上限，命中上限应返回服务错误并触发限流 / 入口告警；不要通过调大单个 `auth_store_projection_meta` JSON 字段来绕过该保护。
+
 前端登录态恢复只把 `/api/auth/refresh` 的 `401` / `403` 当成权威失效信号；服务器重启窗口里的 `502` / `503` / `504`、浏览器 `Failed to fetch` 或 refresh 响应契约异常都必须保留已有本地 access token，不触发全局 auth 变化。refresh 成功响应以共享契约 `RefreshSessionResponse { token }` 为准，前端不要额外要求业务 `ok` 字段。排查“重启后用户都掉线”时，先区分前端是否被暂时不可用清掉本地 token，再检查 SpacetimeDB 正式认证表是否缺 `user_account` / `refresh_session` 数据。

 常用检查思路：
@@ -50,16 +50,33 @@ pub async fn send_phone_code(
        phone_input_masked = phone_input_masked.as_str(),
        "收到手机号验证码发送请求"
    );
+    let send_input = SendPhoneCodeInput {
+        country_code: payload.country_code,
+        pure_phone_number: payload.pure_phone_number,
+        scene: scene.clone(),
+    };
+    let send_now = OffsetDateTime::now_utc();
+    state
+        .refresh_auth_store_from_spacetime()
+        .await
+        .map_err(|error| {
+            AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR)
+                .with_message(format!("刷新短信验证码状态失败：{error}"))
+        })?;
+    state
+        .phone_auth_service()
+        .reserve_code_send(&send_input, send_now)
+        .map_err(map_phone_auth_error)?;
+    state
+        .sync_auth_store_tables_to_spacetime()
+        .await
+        .map_err(|error| {
+            AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR)
+                .with_message(format!("占用短信验证码发送窗口失败：{error}"))
+        })?;
    let result = match state
        .phone_auth_service()
-        .send_code(
-            SendPhoneCodeInput {
-                country_code: payload.country_code,
-                pure_phone_number: payload.pure_phone_number,
-                scene: scene.clone(),
-            },
-            OffsetDateTime::now_utc(),
-        )
+        .send_code_after_authoritative_reservation(send_input, send_now)
        .await
    {
        Ok(result) => {
@@ -32,6 +32,8 @@ use time::{Duration, OffsetDateTime};
 use tracing::{info, warn};

 const DEFAULT_PHONE_VERIFY_CODE_SALT: &str = "genarrative-phone-verify-code-v1";
+const PHONE_CODE_RESERVATION_MARKER: &str = "__genarrative_phone_code_reservation__";
+const MAX_ACTIVE_WECHAT_AUTH_STATES: usize = 1024;

 #[derive(Clone, Debug)]
 pub struct InMemoryAuthStore {
@@ -543,6 +545,65 @@ impl PhoneAuthService {
        &self,
        input: SendPhoneCodeInput,
        now: OffsetDateTime,
+    ) -> Result<SendPhoneCodeResult, PhoneAuthError> {
+        self.send_code_inner(input, now, true).await
+    }
+
+    /// 在 provider 调用前由 api-server 先同步该占位记录，以便 SpacetimeDB 的
+    /// projection CAS 原子占用跨节点冷却窗口。
+    pub fn reserve_code_send(
+        &self,
+        input: &SendPhoneCodeInput,
+        now: OffsetDateTime,
+    ) -> Result<(), PhoneAuthError> {
+        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)?;
+        self.store
+            .ensure_phone_code_not_cooling_down(&normalized_phone.e164, &scene, now)?;
+        let expires_at = now
+            .checked_add(Duration::minutes(SMS_CODE_TTL_MINUTES))
+            .ok_or_else(|| PhoneAuthError::Store("短信验证码过期时间计算溢出".to_string()))?;
+        let expires_at = format_rfc3339(expires_at).map_err(|message| {
+            PhoneAuthError::Store(format!("短信验证码过期时间格式化失败：{message}"))
+        })?;
+        let last_sent_at = format_rfc3339(now).map_err(|message| {
+            PhoneAuthError::Store(format!("短信验证码发送时间格式化失败：{message}"))
+        })?;
+        let verify_code_hash = hash_phone_verify_code(
+            &self.verify_code_salt,
+            &normalized_phone.e164,
+            &scene,
+            PHONE_CODE_RESERVATION_MARKER,
+        );
+        self.store.upsert_phone_code(
+            StoredPhoneCode {
+                phone_number: normalized_phone.e164,
+                scene,
+                verify_code_hash,
+                expires_at,
+                last_sent_at,
+                failed_attempts: 0,
+                provider_out_id: None,
+            },
+            now,
+        )
+    }
+
+    /// 仅供完成权威占用后的 provider 调用使用；占用已由 projection CAS 校验。
+    pub async fn send_code_after_authoritative_reservation(
+        &self,
+        input: SendPhoneCodeInput,
+        now: OffsetDateTime,
+    ) -> Result<SendPhoneCodeResult, PhoneAuthError> {
+        self.send_code_inner(input, now, false).await
+    }
+
+    async fn send_code_inner(
+        &self,
+        input: SendPhoneCodeInput,
+        now: OffsetDateTime,
+        check_local_cooldown: bool,
    ) -> Result<SendPhoneCodeResult, PhoneAuthError> {
        let scene = input.scene.clone();
        validate_mainland_china_country_code(input.country_code.as_deref())?;
@@ -556,8 +617,10 @@ impl PhoneAuthService {
            phone_national_masked = normalized_phone.masked_national_number.as_str(),
            "手机号验证码发送准备调用 provider"
        );
-        self.store
-            .ensure_phone_code_not_cooling_down(&normalized_phone.e164, &scene, now)?;
+        if check_local_cooldown {
+            self.store
+                .ensure_phone_code_not_cooling_down(&normalized_phone.e164, &scene, now)?;
+        }
        let expires_at = now
            .checked_add(Duration::minutes(SMS_CODE_TTL_MINUTES))
            .ok_or_else(|| PhoneAuthError::Store("短信验证码过期时间计算溢出".to_string()))?;
@@ -841,7 +904,7 @@ impl WechatAuthStateService {
            created_at: created_at.clone(),
            updated_at: created_at,
        };
-        self.store.insert_wechat_state(state.clone())?;
+        self.store.insert_wechat_state(state.clone(), now)?;
        Ok(CreateWechatAuthStateResult { state })
    }

@@ -2209,17 +2272,24 @@ impl InMemoryAuthStore {
    fn insert_wechat_state(
        &self,
        state_record: WechatAuthStateRecord,
+        now: OffsetDateTime,
    ) -> Result<(), WechatAuthError> {
        let mut state = self
            .inner
            .lock()
            .map_err(|_| WechatAuthError::Store("微信 state 仓储锁已中毒".to_string()))?;
+        prune_expired_short_lived_state(&mut state, now);
        if state
            .wechat_states_by_token
            .contains_key(&state_record.state_token)
        {
            return Err(WechatAuthError::Store("微信 state 已存在".to_string()));
        }
+        if state.wechat_states_by_token.len() >= MAX_ACTIVE_WECHAT_AUTH_STATES {
+            return Err(WechatAuthError::Store(
+                "微信登录请求过多，请稍后重试".to_string(),
+            ));
+        }
        state.wechat_states_by_token.insert(
            state_record.state_token.clone(),
            StoredWechatAuthState {
@@ -3545,6 +3615,40 @@ mod tests {
        assert_eq!(consumed_state.state.redirect_path, "/studio");
    }

+    #[test]
+    fn wechat_state_creation_is_bounded_before_projection_sync() {
+        let store = InMemoryAuthStore::default();
+        let service = WechatAuthStateService::new(store, 5);
+        let now = OffsetDateTime::now_utc();
+
+        for index in 0..MAX_ACTIVE_WECHAT_AUTH_STATES {
+            service
+                .create_state(
+                    CreateWechatAuthStateInput {
+                        redirect_path: format!("/studio?attempt={index}"),
+                        scene: WechatAuthScene::Desktop,
+                        request_user_agent: None,
+                        bind_user_id: None,
+                    },
+                    now,
+                )
+                .expect("active wechat state should fit within the projection budget");
+        }
+
+        let error = service
+            .create_state(
+                CreateWechatAuthStateInput {
+                    redirect_path: "/studio".to_string(),
+                    scene: WechatAuthScene::Desktop,
+                    request_user_agent: None,
+                    bind_user_id: None,
+                },
+                now,
+            )
+            .expect_err("wechat state creation must reject an unbounded projection");
+        assert!(matches!(error, WechatAuthError::Store(message) if message.contains("请求过多")));
+    }
+
    #[tokio::test]
    async fn refresh_from_projection_view_merges_session_created_by_another_process() {
        let source_store = InMemoryAuthStore::default();
@@ -3736,6 +3840,30 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn authoritative_phone_code_reservation_blocks_duplicate_provider_send() {
+        let service = build_phone_service(build_store());
+        let input = SendPhoneCodeInput {
+            country_code: None,
+            pure_phone_number: "13800138001".to_string(),
+            scene: PhoneAuthScene::Login,
+        };
+        let now = OffsetDateTime::now_utc();
+
+        service
+            .reserve_code_send(&input, now)
+            .expect("authoritative reservation should be representable locally");
+        let duplicate = service
+            .reserve_code_send(&input, now + Duration::seconds(1))
+            .expect_err("a second reservation must observe the local cooldown");
+        assert!(matches!(duplicate, PhoneAuthError::SendCoolingDown { .. }));
+
+        service
+            .send_code_after_authoritative_reservation(input, now)
+            .await
+            .expect("provider send should replace the reservation with the real code");
+    }
+
    #[tokio::test]
    async fn phone_send_code_keeps_different_scenes_isolated() {
        let service = build_phone_service(build_store());