From 2b8d44b369f877a59277aa119bca0270d9636a4b Mon Sep 17 00:00:00 2001 From: kdletters Date: Thu, 27 Aug 2026 20:48:49 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E9=89=B4=E6=9D=83=E8=B7=A8?= =?UTF-8?q?=E8=8A=82=E7=82=B9=E6=8A=95=E5=BD=B1=E6=81=A2=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 刷新登录、会话、密码与微信认证主链路的正式投影 补齐支付查单、过期补偿和订阅消息的跨节点身份读取 修复同步失败后的 pending 状态恢复并更新鉴权运维文档 --- .../shared-memory/decision-log.md | 2 +- ...】server-rs与SpacetimeDB数据契约-2026-05-15.md | 2 +- ...发运维】本地开发验证与生产运维-2026-05-15.md | 4 +- server-rs/crates/api-server/src/auth_me.rs | 9 ++++ .../crates/api-server/src/auth_public_user.rs | 15 +++++++ .../crates/api-server/src/auth_sessions.rs | 15 +++++++ server-rs/crates/api-server/src/logout.rs | 7 ++++ server-rs/crates/api-server/src/logout_all.rs | 7 ++++ .../crates/api-server/src/password_entry.rs | 7 ++++ .../api-server/src/password_management.rs | 17 ++++++++ server-rs/crates/api-server/src/phone_auth.rs | 7 ++++ .../crates/api-server/src/profile_identity.rs | 7 ++++ .../profile_recharge_expiration_listener.rs | 8 ++++ .../crates/api-server/src/refresh_session.rs | 10 +++++ .../crates/api-server/src/runtime_profile.rs | 7 ++++ server-rs/crates/api-server/src/state.rs | 31 +++++++------- .../crates/api-server/src/wechat/auth.rs | 42 +++++++++++++++++++ server-rs/crates/api-server/src/wechat/pay.rs | 4 ++ .../src/wechat/subscribe_message.rs | 7 ++++ 19 files changed, 190 insertions(+), 18 deletions(-) diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index 5a692a20c..d502bc2af 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -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 节点必须一致；发码前先刷新权威投影并用占位验证码记录做一次 projection CAS，只有占用成功才调用短信 provider，避免跨节点冷却竞态；认证 handler 在发码、消费验证码、创建/消费微信 state 后都要完成 projection sync，失败即返回服务错误；短期状态首次未命中时允许从正式投影做一次受 CAS 保护的只读刷新，再重试读取，不能依赖粘性会话。微信 OAuth state 设置有界活动数量，避免单个 JSON 投影无界膨胀。短期状态仍由 `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，失败即返回服务错误；所有会读取或变更本机认证工作集的认证主链路（登录、刷新、`/me`、会话管理、密码、绑定和微信 state）在领域操作前先从正式投影做一次受 CAS 保护的只读刷新，刷新失败时 fail closed，不能依赖粘性会话；同步遇到 CAS 冲突时，若本次尝试期间没有新的本地变更则恢复正式快照，若仍有待同步 revision 则由后续认证请求重试，避免节点永久卡在 pending。微信 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 检查。 diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index fff89e2ce..189cb3cf9 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -414,7 +414,7 @@ Responses 的终态载荷既是工具调用的恢复源，也是正文的恢复 - 源码：`server-rs/crates/spacetime-module/src/auth/tables.rs` - 职责：保存 typed 认证投影的单调版本，以及短期手机号验证码和微信 OAuth state 的序列化投影；`phone_codes_json` / `wechat_states_json` 只承载短期认证状态，不替代 `user_account`、`auth_identity` 或 `refresh_session` 的正式表语义。 -认证恢复策略：`api-server` 启动时从 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）以及 `auth_store_projection_meta` 中的短期状态投影导出 typed `AuthStoreProjectionView`，再恢复 `module-auth` 的进程内认证工作集；生产 Bearer 中间件不再从 `InMemoryAuthStore` 读取用户或会话，而是每次通过 typed `validate_auth_session` procedure 在 SpacetimeDB 事务内校验 `token_version`、会话归属、撤销时间和过期时间，SpacetimeDB 不可用时 fail closed 返回服务错误。`validate_auth_session`、投影导出和投影同步均从 `ctx.sender()` 派生调用方，并复用现役 runtime service identity 白名单；启动恢复先完成该服务身份初始化，普通 SpacetimeDB identity 不能读取或改写私有认证表。测试构建仍可使用显式的内存测试夹具。短期认证请求在本进程工作集内未命中且工作集没有待同步变更时，先对正式投影做一次受 CAS 保护的只读刷新，再决定验证码 / OAuth state 是否失效，不依赖粘性会话；refresh cookie 仍只按正式 `refresh_session` 校验，其他认证数据也不得绕过正式同步。`module-auth` 只保留内存工作集和 projection 导入 / 导出能力，不再保留 JSON 快照导入 / 导出能力，也不写本地持久化文件；`auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 不再是兼容恢复源。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作仍必须在返回客户端前通过 `sync_auth_store_projection` 成功同步 SpacetimeDB 正式认证表；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号、会话、短信验证码或微信 state 当成成功结果。每个 API 工作集绑定启动恢复或上次成功同步得到的 `auth_store_projection_meta.updated_at` 版本作为 `base_updated_at_micros`，SpacetimeDB 在同一事务内执行基线 CAS，并要求新的 `updated_at_micros` 严格递增；基线不一致或版本不晚于当前值时整包写入失败，冲突节点在确认本次同步前没有遗留未确认本地变更且同步期间没有其它本地认证变更后，才可丢弃本地工作集并从正式表恢复，不能用陈旧工作集删除、恢复或覆盖另一节点的新状态；若此前已有待重试 revision 或同期还有本地变更则保留旧基线并继续失败关闭，不用自动恢复覆盖未提交变更；同步成功但期间又出现新本地变更时最多连续补同步三轮，仍未稳定则失败关闭。这只是迁移期并发保护，不改变正式认证表的权威地位。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。若启动恢复阶段 SpacetimeDB 不可连接或超时，`api-server` 会按固定间隔持续重试认证工作集恢复，恢复成功后才开始监听 HTTP，避免一次短超时让进程永久停留在依赖不可用状态。 +认证恢复策略：`api-server` 启动时从 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）以及 `auth_store_projection_meta` 中的短期状态投影导出 typed `AuthStoreProjectionView`，再恢复 `module-auth` 的进程内认证工作集；生产 Bearer 中间件不再从 `InMemoryAuthStore` 读取用户或会话，而是每次通过 typed `validate_auth_session` procedure 在 SpacetimeDB 事务内校验 `token_version`、会话归属、撤销时间和过期时间，SpacetimeDB 不可用时 fail closed 返回服务错误。`validate_auth_session`、投影导出和投影同步均从 `ctx.sender()` 派生调用方，并复用现役 runtime service identity 白名单；启动恢复先完成该服务身份初始化，普通 SpacetimeDB identity 不能读取或改写私有认证表。测试构建仍可使用显式的内存测试夹具。所有会读取或变更本机认证工作集的认证主链路（登录、刷新、`/me`、会话管理、密码、绑定和微信 state）在领域操作前先从正式投影做一次受 CAS 保护的只读刷新，刷新失败时 fail closed；refresh cookie 仍只按正式 `refresh_session` 校验，其他认证数据也不得绕过正式同步。`module-auth` 只保留内存工作集和 projection 导入 / 导出能力，不再保留 JSON 快照导入 / 导出能力，也不写本地持久化文件；`auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 不再是兼容恢复源。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作仍必须在返回客户端前通过 `sync_auth_store_projection` 成功同步 SpacetimeDB 正式认证表；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号、会话、短信验证码或微信 state 当成成功结果。每个 API 工作集绑定启动恢复或上次成功同步得到的 `auth_store_projection_meta.updated_at` 版本作为 `base_updated_at_micros`，SpacetimeDB 在同一事务内执行基线 CAS，并要求新的 `updated_at_micros` 严格递增；基线不一致或版本不晚于当前值时整包写入失败，冲突节点只有在确认本次同步尝试期间没有新的本地认证变更后，才可丢弃失败工作集并从正式表恢复，不能用陈旧工作集删除、恢复或覆盖另一节点的新状态；若同期仍有本地变更则保留 pending revision，并由后续认证请求先重试同步，不把临时数据库故障变成永久卡死；同步成功但期间又出现新本地变更时最多连续补同步三轮，仍未稳定则失败关闭。这只是迁移期并发保护，不改变正式认证表的权威地位。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。若启动恢复阶段 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` 为准。`api-server` 多节点必须使用相同的部署级验证码哈希盐（当前复用 `GENARRATIVE_JWT_SECRET`）；轮换该 secret 会使尚未消费的短信验证码失效，但不会改变已持久化账号或 session。 diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index 0641a1ddc..062547649 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -777,7 +777,7 @@ node scripts/test-ve-llm.mjs ### 手机验证码短信 -手机验证码发送走阿里云普通短信 `SendSms`，验证码由 `module-auth` 在当前 `api-server` 进程内生成、哈希存储和校验，不再调用阿里云托管验证码的 `SendSmsVerifyCode` / `CheckSmsVerifyCode`。因此 `api-server` 重启后，已发送但未校验的验证码会失效。 +手机验证码发送走阿里云普通短信 `SendSms`，验证码由 `module-auth` 在当前 `api-server` 进程内生成并哈希，短期验证码投影随 `auth_store_projection_meta` 同步到 SpacetimeDB 后由任一 API 节点恢复和校验；不再调用阿里云托管验证码的 `SendSmsVerifyCode` / `CheckSmsVerifyCode`。因此只要 SpacetimeDB 正常，`api-server` 重启不会使已发送但未过期的验证码失效。 生产默认短信配置： @@ -866,7 +866,7 @@ systemctl restart genarrative-api.service journalctl -u genarrative-api.service --since '30 seconds ago' --no-pager | grep -E 'tracking outbox|Permission denied|os error 13' ``` -`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，以避免用空本地状态或旧快照覆盖认证表。 +`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 投影恢复；所有会读取或变更本机认证工作集的认证主链路在领域操作前都会从正式投影做一次只读刷新，刷新失败即 fail closed，不依赖粘性会话。不要再配置或依赖 `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 字段来绕过该保护。 diff --git a/server-rs/crates/api-server/src/auth_me.rs b/server-rs/crates/api-server/src/auth_me.rs index bab8c4349..74321a901 100644 --- a/server-rs/crates/api-server/src/auth_me.rs +++ b/server-rs/crates/api-server/src/auth_me.rs @@ -17,6 +17,15 @@ pub async fn auth_me( Extension(authenticated): Extension, ) -> Result, AppError> { let user_id = authenticated.claims().user_id().to_string(); + // Bearer 已由 SpacetimeDB 校验；/me 返回的资料也必须先从同一正式投影恢复， + // 否则节点间会因为本机工作集滞后把有效用户返回为 401 或旧资料。 + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; let user = state .password_entry_service() .get_user_by_id(&user_id) diff --git a/server-rs/crates/api-server/src/auth_public_user.rs b/server-rs/crates/api-server/src/auth_public_user.rs index e07b4f6ab..d7f676dfb 100644 --- a/server-rs/crates/api-server/src/auth_public_user.rs +++ b/server-rs/crates/api-server/src/auth_public_user.rs @@ -15,6 +15,13 @@ pub async fn get_public_user_by_code( Extension(request_context): Extension, Path(code): Path, ) -> Result, AppError> { + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; let user = state .password_entry_service() .get_user_by_public_user_code(&code) @@ -41,6 +48,14 @@ pub async fn get_public_user_by_id( return Err(AppError::from_status(StatusCode::BAD_REQUEST).with_message("用户 ID 不能为空")); } + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; + let user = state .auth_user_service() .get_user_by_id(user_id) diff --git a/server-rs/crates/api-server/src/auth_sessions.rs b/server-rs/crates/api-server/src/auth_sessions.rs index c9ba471d3..0ef62c01a 100644 --- a/server-rs/crates/api-server/src/auth_sessions.rs +++ b/server-rs/crates/api-server/src/auth_sessions.rs @@ -29,6 +29,13 @@ pub async fn auth_sessions( ) -> Result, AppError> { // 当前设备识别仍然依赖 refresh cookie 命中的原始 token，对旧前端行为保持兼容。 let user_id = authenticated.claims().user_id().to_string(); + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; let current_refresh_token_hash = maybe_refresh_token.and_then(|token| { let token = token.0.token().trim(); if token.is_empty() { @@ -78,6 +85,14 @@ pub async fn revoke_auth_session( ); } + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; + let revoke_result = state .refresh_session_service() .revoke_session_by_user_and_session( diff --git a/server-rs/crates/api-server/src/logout.rs b/server-rs/crates/api-server/src/logout.rs index 4de4f4efb..101b533e9 100644 --- a/server-rs/crates/api-server/src/logout.rs +++ b/server-rs/crates/api-server/src/logout.rs @@ -25,6 +25,13 @@ pub async fn logout( Extension(authenticated): Extension, maybe_refresh_token: Option>, ) -> Result { + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; let refresh_token_hash = maybe_refresh_token.and_then(|token| { let token = token.0.token().trim().to_string(); if token.is_empty() { diff --git a/server-rs/crates/api-server/src/logout_all.rs b/server-rs/crates/api-server/src/logout_all.rs index e56e1b775..5bc545e1e 100644 --- a/server-rs/crates/api-server/src/logout_all.rs +++ b/server-rs/crates/api-server/src/logout_all.rs @@ -23,6 +23,13 @@ pub async fn logout_all( Extension(request_context): Extension, Extension(authenticated): Extension, ) -> Result { + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; state .auth_user_service() .logout_all_sessions( diff --git a/server-rs/crates/api-server/src/password_entry.rs b/server-rs/crates/api-server/src/password_entry.rs index f2e82660e..d1acc699c 100644 --- a/server-rs/crates/api-server/src/password_entry.rs +++ b/server-rs/crates/api-server/src/password_entry.rs @@ -27,6 +27,13 @@ pub async fn password_entry( headers: HeaderMap, Json(payload): Json, ) -> Result { + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; let input = PasswordEntryInput { country_code: payload.country_code, pure_phone_number: payload.pure_phone_number, diff --git a/server-rs/crates/api-server/src/password_management.rs b/server-rs/crates/api-server/src/password_management.rs index 38a0e40ab..61b92fe98 100644 --- a/server-rs/crates/api-server/src/password_management.rs +++ b/server-rs/crates/api-server/src/password_management.rs @@ -33,6 +33,13 @@ pub async fn change_password( Extension(authenticated): Extension, Json(payload): Json, ) -> Result { + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; let result = state .password_entry_service() .change_password_and_revoke_all_sessions( @@ -82,6 +89,16 @@ pub async fn reset_password( ); } + // reset_password 消费的是跨节点共享的短期验证码；先恢复正式投影， + // 避免发码节点与消费节点的本机工作集不一致。 + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新短信验证码状态失败：{error}")) + })?; + let result = match state .phone_auth_service() .reset_password( diff --git a/server-rs/crates/api-server/src/phone_auth.rs b/server-rs/crates/api-server/src/phone_auth.rs index 1614d0c04..37e77acec 100644 --- a/server-rs/crates/api-server/src/phone_auth.rs +++ b/server-rs/crates/api-server/src/phone_auth.rs @@ -139,6 +139,13 @@ pub async fn phone_login( AppError::from_status(StatusCode::BAD_REQUEST).with_message("手机号登录暂未启用") ); } + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新短信验证码状态失败：{error}")) + })?; let invite_code = payload.invite_code.clone(); let login_input = PhoneLoginInput { country_code: payload.country_code, diff --git a/server-rs/crates/api-server/src/profile_identity.rs b/server-rs/crates/api-server/src/profile_identity.rs index fd91b8eb6..dac4f110c 100644 --- a/server-rs/crates/api-server/src/profile_identity.rs +++ b/server-rs/crates/api-server/src/profile_identity.rs @@ -23,6 +23,13 @@ pub async fn update_profile_identity( Extension(authenticated): Extension, Json(payload): Json, ) -> Result, AppError> { + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; if let Some(avatar_data_url) = payload.avatar_data_url.as_deref() { validate_avatar_data_url(avatar_data_url)?; } diff --git a/server-rs/crates/api-server/src/profile_recharge_expiration_listener.rs b/server-rs/crates/api-server/src/profile_recharge_expiration_listener.rs index 5804203c3..070ac5752 100644 --- a/server-rs/crates/api-server/src/profile_recharge_expiration_listener.rs +++ b/server-rs/crates/api-server/src/profile_recharge_expiration_listener.rs @@ -265,6 +265,14 @@ async fn process_expired_virtual_payment_order( state: &AppState, order: &RuntimeProfileRechargeOrderRecord, ) -> Result<(), ExpirationCompensationError> { + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + ExpirationCompensationError::Runtime(format!( + "failed to refresh auth projection for virtual payment query: {error}" + )) + })?; let identity = state .wechat_auth_service() .get_identity_by_user_id(&order.user_id) diff --git a/server-rs/crates/api-server/src/refresh_session.rs b/server-rs/crates/api-server/src/refresh_session.rs index 4cdd87bb8..c6c783f67 100644 --- a/server-rs/crates/api-server/src/refresh_session.rs +++ b/server-rs/crates/api-server/src/refresh_session.rs @@ -39,6 +39,16 @@ pub async fn refresh_session( let next_refresh_token = platform_auth::create_refresh_session_token(); let next_refresh_token_hash = hash_refresh_session_token(&next_refresh_token); + // refresh_session 是跨节点的正式认证入口；先加载最新投影，再在本机工作集执行领域轮换， + // 避免请求落到旧节点时把合法 refresh cookie 误判为不存在。 + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(axum::http::StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新认证状态失败：{error}")) + })?; + let rotated = match state.refresh_session_service().rotate_session( RotateRefreshSessionInput { refresh_token_hash: refresh_token_hash.clone(), diff --git a/server-rs/crates/api-server/src/runtime_profile.rs b/server-rs/crates/api-server/src/runtime_profile.rs index 290e5b12e..67625908d 100644 --- a/server-rs/crates/api-server/src/runtime_profile.rs +++ b/server-rs/crates/api-server/src/runtime_profile.rs @@ -1650,6 +1650,13 @@ async fn resolve_wechat_identity_for_payment( state: &AppState, user_id: &str, ) -> Result { + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新微信认证状态失败：{error}")) + })?; if let Some(identity) = state .wechat_auth_service() .get_identity_by_user_id(user_id) diff --git a/server-rs/crates/api-server/src/state.rs b/server-rs/crates/api-server/src/state.rs index cf9792f80..284b437cb 100644 --- a/server-rs/crates/api-server/src/state.rs +++ b/server-rs/crates/api-server/src/state.rs @@ -618,6 +618,10 @@ impl AppState { let editor_oss_http_client = build_editor_oss_http_client()?; let http_request_permit_pools = HttpRequestPermitPools::from_config(&config); let (profile_recharge_order_updates, _) = broadcast::channel(128); + // `ensure_orphan_work_owner_user` 只为公开作品作者回退提供进程内占位账号， + // 不属于正式认证投影；将当前工作集 revision 作为已同步起点，首次认证请求会 + // 先按正式投影刷新并自然丢弃该占位账号，避免把它误当成待提交认证变更。 + let initial_auth_store_revision = auth_store.revision(); Ok(Self(Arc::new(AppStateInner { config, @@ -646,7 +650,7 @@ impl AppState { oss_client, auth_store, auth_projection_version: AtomicI64::new(auth_projection_version), - auth_projection_synced_revision: AtomicU64::new(0), + auth_projection_synced_revision: AtomicU64::new(initial_auth_store_revision), auth_projection_sync_lock: AsyncMutex::new(()), password_entry_service, refresh_session_service, @@ -1310,11 +1314,6 @@ impl AppState { #[cfg(not(test))] let _sync_guard = self.auth_projection_sync_lock.lock().await; #[cfg(not(test))] - let sync_start_revision = self.auth_store.revision(); - #[cfg(not(test))] - let mut local_changed_during_sync = - self.auth_projection_synced_revision.load(Ordering::Acquire) != sync_start_revision; - #[cfg(not(test))] for attempt in 0..3 { let base_updated_at_micros = self.auth_projection_version.load(Ordering::Acquire); let now_updated_at_micros = @@ -1332,9 +1331,6 @@ impl AppState { .auth_store .export_projection_view_with_revision(updated_at_micros) .map_err(SpacetimeClientError::Runtime)?; - if attempted_revision != sync_start_revision { - local_changed_during_sync = true; - } projection.base_updated_at_micros = base_updated_at_micros; // 当前仍由 module-auth 的进程内工作集执行业务规则；这里只用 typed projection 同步正式认证表。 @@ -1351,7 +1347,6 @@ impl AppState { .store(attempted_revision, Ordering::Release); return Ok(()); } - local_changed_during_sync = true; warn!( attempt, "认证投影同步期间工作集发生变化，将继续同步最新工作集" @@ -1363,8 +1358,9 @@ impl AppState { error = %error, "认证投影同步 SpacetimeDB 正式表失败，当前认证流程中止" ); - if local_changed_during_sync || self.auth_store.revision() != attempted_revision - { + // 当前请求已经失败；只要同步尝试期间没有新的本地变更，恢复为 + // 数据库快照，避免一次 CAS 冲突把本节点永久留在“待同步”状态。 + if self.auth_store.revision() != attempted_revision { warn!( "认证投影同步失败期间工作集发生并发变化，跳过自动恢复以避免覆盖未提交变更" ); @@ -1406,14 +1402,21 @@ impl AppState { )) } - /// 在只读短期认证状态未命中时，从正式投影刷新一次本地工作集，避免 OAuth / 短信 - /// 请求落到另一节点后必须依赖粘性会话才能成功。 + /// 在认证主链路执行前，从正式投影刷新一次本地工作集，避免请求落到另一节点后 + /// 因本机工作集滞后而必须依赖粘性会话才能成功。 pub async fn refresh_auth_store_from_spacetime(&self) -> Result<(), SpacetimeClientError> { #[cfg(test)] return Ok(()); #[cfg(not(test))] { + // 上一次业务操作可能已经改了工作集，但在返回响应前遇到数据库暂时不可用。 + // 先重试提交这份待同步变更，避免只读请求把节点永久卡在 pending 状态。 + if self.auth_projection_synced_revision.load(Ordering::Acquire) + != self.auth_store.revision() + { + self.sync_auth_store_tables_to_spacetime().await?; + } let _sync_guard = self.auth_projection_sync_lock.lock().await; let expected_revision = self.auth_store.revision(); if self.auth_projection_synced_revision.load(Ordering::Acquire) != expected_revision { diff --git a/server-rs/crates/api-server/src/wechat/auth.rs b/server-rs/crates/api-server/src/wechat/auth.rs index d4f26dcd2..092c05e59 100644 --- a/server-rs/crates/api-server/src/wechat/auth.rs +++ b/server-rs/crates/api-server/src/wechat/auth.rs @@ -43,6 +43,13 @@ pub async fn start_wechat_login( if !state.config.wechat_auth_enabled { return Err(AppError::from_status(StatusCode::BAD_REQUEST).with_message("微信登录暂未启用")); } + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新微信认证状态失败：{error}")) + })?; let user_agent = headers .get("user-agent") .and_then(|value| value.to_str().ok()) @@ -95,6 +102,13 @@ pub async fn start_wechat_bind( if !state.config.wechat_auth_enabled { return Err(AppError::from_status(StatusCode::BAD_REQUEST).with_message("微信登录暂未启用")); } + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新微信认证状态失败：{error}")) + })?; let user_agent = headers .get("user-agent") .and_then(|value| value.to_str().ok()) @@ -164,6 +178,20 @@ pub async fn handle_wechat_callback( .into_response()); } + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + warn!( + request_id = request_context.request_id(), + operation = request_context.operation(), + error = %error, + "微信回调前刷新认证投影失败" + ); + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message("刷新微信认证状态失败") + })?; + let consume_result = state .wechat_auth_state_service() .consume_state(&state_token, OffsetDateTime::now_utc()); @@ -320,6 +348,13 @@ pub async fn bind_wechat_phone( if !state.config.wechat_auth_enabled { return Err(AppError::from_status(StatusCode::BAD_REQUEST).with_message("微信登录暂未启用")); } + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新微信认证状态失败：{error}")) + })?; let result = if let Some(wechat_phone_code) = payload .wechat_phone_code .as_deref() @@ -440,6 +475,13 @@ pub async fn login_wechat_mini_program( if !state.config.wechat_auth_enabled { return Err(AppError::from_status(StatusCode::BAD_REQUEST).with_message("微信登录暂未启用")); } + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新微信认证状态失败：{error}")) + })?; let code = payload.code.trim(); if code.is_empty() { return Err( diff --git a/server-rs/crates/api-server/src/wechat/pay.rs b/server-rs/crates/api-server/src/wechat/pay.rs index 556c8f00e..3cc839884 100644 --- a/server-rs/crates/api-server/src/wechat/pay.rs +++ b/server-rs/crates/api-server/src/wechat/pay.rs @@ -239,6 +239,10 @@ async fn confirm_virtual_payment_recharge_order( )); } + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| WechatPayError::Upstream(format!("刷新认证状态失败：{error}")))?; let identity = state .wechat_auth_service() .get_identity_by_user_id(&order.user_id) diff --git a/server-rs/crates/api-server/src/wechat/subscribe_message.rs b/server-rs/crates/api-server/src/wechat/subscribe_message.rs index 8946afaf1..863c0185b 100644 --- a/server-rs/crates/api-server/src/wechat/subscribe_message.rs +++ b/server-rs/crates/api-server/src/wechat/subscribe_message.rs @@ -56,6 +56,13 @@ async fn send_generation_result_subscribe_message( AppError::from_status(StatusCode::SERVICE_UNAVAILABLE) .with_message("微信订阅消息模板 ID 未配置") })?; + state + .refresh_auth_store_from_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("刷新微信认证状态失败：{error}")) + })?; let user = state .auth_user_service() .get_user_by_id(&message.owner_user_id)