From 72d7214646ea59ce55530c4c6cb48915d95b3348 Mon Sep 17 00:00:00 2001 From: kdletters Date: Thu, 27 Aug 2026 13:36:28 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B6=E5=8F=A3=E8=B7=A8=E8=8A=82=E7=82=B9?= =?UTF-8?q?=E9=89=B4=E6=9D=83=E8=AF=BB=E5=8F=96=E4=B8=8E=E8=8A=82=E7=82=B9?= =?UTF-8?q?=20outbox=20=E6=81=A2=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 生产 Bearer 鉴权改用 SpacetimeDB 会话校验 procedure 为认证整包投影增加单调水位保护 新增鉴权客户端 facade 与生成 bindings 补齐 tracking 与钱包退款 outbox 启动恢复 同步后端架构、运维和项目决策文档 --- .../shared-memory/decision-log.md | 6 + ...】server-rs与SpacetimeDB数据契约-2026-05-15.md | 2 +- ...发运维】本地开发验证与生产运维-2026-05-15.md | 4 +- server-rs/Cargo.lock | 1 + server-rs/crates/api-server/src/auth.rs | 100 ++++--- .../crates/api-server/src/tracking_outbox.rs | 43 +++ .../api-server/src/wallet_refund_outbox.rs | 277 +++++++++++++++++- .../spacetime-client/src/active/mapper.rs | 5 +- .../src/active/mapper/auth.rs | 17 ++ server-rs/crates/spacetime-client/src/auth.rs | 24 ++ .../spacetime-client/src/module_bindings.rs | 6 + .../auth_session_validation_input_type.rs | 17 ++ ...ession_validation_procedure_result_type.rs | 16 + .../validate_auth_session_procedure.rs | 59 ++++ server-rs/crates/spacetime-module/Cargo.toml | 1 + .../spacetime-module/src/auth/procedures.rs | 101 +++++++ 16 files changed, 634 insertions(+), 45 deletions(-) create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/auth_session_validation_input_type.rs create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/auth_session_validation_procedure_result_type.rs create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/validate_auth_session_procedure.rs diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index 4ffd04577..19936f239 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -7747,3 +7747,9 @@ CI 上 `background_agent_runtime_recovers_stale_running_before_pending_task` 在 - DirectProject 的 Codex cwd 固定为真实 `game/` 目录时，原生文件工具和 patch 必须使用 cwd 相对路径（`index.html`、`style.css`、`game.js`）；`game/...` 仅用于 AGC manifest、回执和客户端投影，不能作为 cwd 内原生 patch 路径，以避免 `writing outside of the project`。 - 直连 Runtime 已取得 Developer Key 时，资源编辑的 `remote_credentials` 是该操作的完整身份边界；其中冻结平台快照为空表示 Developer 模式，禁止再从进程全局 GUI 登录态补回账号快照。平台账号模式仍只使用同一组凭据捕获的快照。 - 回归覆盖 Direct 系统提示路径合同和 Developer Key / GUI 快照隔离；未触碰用户项目 `.agent` 锁、账本或凭据。 + +## 2026-08-27 API 鉴权读取与节点本地 outbox 恢复切片 + +- 生产 Bearer 中间件的会话有效性改由 SpacetimeDB typed `validate_auth_session` procedure 在事务内校验 `user_account.token_version`、会话归属、撤销时间和过期时间；`InMemoryAuthStore` 仅保留启动恢复工作集及测试夹具，不作为生产请求鉴权读取源。该切片不等于登录、刷新、验证码和微信 state 的全量迁移，后续写路径仍需继续收口。 +- `sync_auth_store_projection` 增加 `auth_store_projection_meta.updated_at` 单调水位检查，早到的整包快照失败关闭；正式认证表仍是权威源，水位只是迁移期跨 API 实例的延迟快照保护。 +- tracking outbox 与 wallet refund outbox worker 启动即执行恢复；退款 outbox 会恢复崩溃遗留 `tmp-*` 文件并隔离损坏 / 冲突文件。两者仍是节点本地 durable fallback，不能用粘性会话替代共享持久化；生产节点必须保留对应目录并纳入恢复演练。 diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index df3653c50..f61169570 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -413,7 +413,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 中间件不再从 `InMemoryAuthStore` 读取用户或会话，而是每次通过 typed `validate_auth_session` procedure 在 SpacetimeDB 事务内校验 `token_version`、会话归属、撤销时间和过期时间，SpacetimeDB 不可用时 fail closed 返回服务错误。测试构建仍可使用显式的内存测试夹具。运行中 refresh cookie 在本进程工作集内未命中时直接按失效处理，不再从 SpacetimeDB 导出整包认证状态刷新内存，避免旧投影把重复手机号或旧会话重新灌回进程。`module-auth` 只保留内存工作集和 projection 导入 / 导出能力，不再保留 JSON 快照导入 / 导出能力，也不写本地持久化文件；`auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 不再是兼容恢复源。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作仍必须在返回客户端前通过 `sync_auth_store_projection` 成功同步 SpacetimeDB 正式认证表；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号或会话当成成功结果。投影同步使用 `auth_store_projection_meta.updated_at` 作为单调水位，拒绝早于当前水位的整包快照，避免两个 API 实例的延迟快照互相覆盖；这只是迁移期并发保护，不改变正式认证表的权威地位。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。若启动恢复阶段 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` 为准。 diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index 7de599e9c..bfc5e04f8 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -853,7 +853,9 @@ 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` 是每个 outbox 实例的磁盘保护阈值，不是 flush 阈值；超过后低价值 route tracking 和 BgFilter provider 失败审计可以被丢弃并记录日志 / 指标，关键同步事件不进入该丢弃路径。api-server 使用配置目录本身，BgFilter worker 固定使用其 `bgfilter-worker/` 子目录，两个进程不得操作同一个 active 文件。sealed 文件若出现无法解析的坏行，会重命名为 `corrupt-*` 隔离并记录 `genarrative.tracking_outbox.files.corrupt` 指标，避免一个坏文件阻塞后续批量入库。进程收到退出信号后会在 `GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS` 窗口内封存各自 active 文件并尽力 flush sealed 文件，超时或 SpacetimeDB 暂不可用时保留本地文件给下次同角色启动继续投递。该机制对已 enqueue 记录提供至少一次投递语义，依赖 `tracking_event.event_id` 幂等跳过重复事件；BgFilter 尚未 enqueue 或因硬上限 / 保护阈值被丢弃的审计不在该保证内。 +outbox 采用 NDJSON 文件保存原始事件。达到 `BATCH_SIZE` 时会立刻把当前 active 文件原子封存为 sealed 文件，并马上切到新的 active 继续写入；后台 worker 异步 flush sealed 文件，HTTP 请求线程不等待 SpacetimeDB。worker 启动时会先封存并 flush 已存在的 active / sealed 文件，恢复窗口内 SpacetimeDB 暂不可用则保留文件并按后续周期重试；`FLUSH_INTERVAL_MS` 只负责兜底封存长时间未满批的 active 文件。SpacetimeDB 批量 procedure 返回成功后删除 sealed 文件，失败则保留文件并重试。`MAX_BYTES` 是每个 outbox 实例的磁盘保护阈值，不是 flush 阈值；超过后低价值 route tracking 和 BgFilter provider 失败审计可以被丢弃并记录日志 / 指标，关键同步事件不进入该丢弃路径。api-server 使用配置目录本身，BgFilter worker 固定使用其 `bgfilter-worker/` 子目录，两个进程不得操作同一个 active 文件。sealed 文件若出现无法解析的坏行，会重命名为 `corrupt-*` 隔离并记录 `genarrative.tracking_outbox.files.corrupt` 指标，避免一个坏文件阻塞后续批量入库。进程收到退出信号后会在 `GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS` 窗口内封存各自 active 文件并尽力 flush sealed 文件，超时或 SpacetimeDB 暂不可用时保留本地文件给下次同角色启动继续投递。该机制对已 enqueue 记录提供至少一次投递语义，依赖 `tracking_event.event_id` 幂等跳过重复事件；BgFilter 尚未 enqueue 或因硬上限 / 保护阈值被丢弃的审计不在该保证内。 + +钱包退款 outbox 同样在 worker 启动时立即扫描 pending 文件；如果进程在“临时文件写完但尚未改名”阶段崩溃，启动恢复会校验 `tmp-*` 内容并原子提升为按 ledger id 命名的 pending 文件，损坏或冲突文件移入 `corrupt-*` 隔离目录，不会静默丢弃关键退款。该目录仍是节点本地 durable fallback，不能替代 SpacetimeDB 钱包账本；发布和主机替换必须保留 `/var/lib/genarrative/wallet-refund-outbox` 并纳入节点恢复 / 备份演练。 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/server-rs/Cargo.lock b/server-rs/Cargo.lock index 3689d1b5b..bb51bc777 100644 --- a/server-rs/Cargo.lock +++ b/server-rs/Cargo.lock @@ -5526,6 +5526,7 @@ dependencies = [ "shared-contracts", "spacetimedb", "spacetimedb-lib", + "time", ] [[package]] diff --git a/server-rs/crates/api-server/src/auth.rs b/server-rs/crates/api-server/src/auth.rs index cac3b201d..e1c9639d0 100644 --- a/server-rs/crates/api-server/src/auth.rs +++ b/server-rs/crates/api-server/src/auth.rs @@ -19,6 +19,7 @@ use serde_json::{Value, json}; use shared_contracts::auth::RuntimeGuestTokenResponse; #[cfg(any())] use shared_kernel::{format_rfc3339, new_uuid_simple_string}; +#[cfg(test)] use time::OffsetDateTime; use tracing::warn; @@ -236,54 +237,77 @@ async fn authenticate_request( ); AppError::from_status(StatusCode::UNAUTHORIZED) })?; - let current_user = state - .auth_user_service() - .get_user_by_id(claims.user_id()) - .map_err(|error| { - warn!( - %request_id, - error = %error, - "Bearer JWT 用户快照读取失败" - ); - AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - })?; - let Some(current_user) = current_user else { - warn!( - %request_id, - user_id = %claims.user_id(), - "Bearer JWT 对应用户不存在" - ); - return Err(AppError::from_status(StatusCode::UNAUTHORIZED)); - }; - if current_user.token_version != claims.token_version() { - warn!( - %request_id, - user_id = %claims.user_id(), - token_version = claims.token_version(), - current_token_version = current_user.token_version, - "Bearer JWT 版本已失效" - ); - return Err(AppError::from_status(StatusCode::UNAUTHORIZED) - .with_message("当前登录态已失效，请重新登录")); - } - + #[cfg(not(test))] let session_is_active = state - .refresh_session_service() - .is_session_active_for_user( - claims.user_id(), - claims.session_id(), - OffsetDateTime::now_utc(), - ) + .spacetime_client() + .validate_auth_session(spacetime_client::AuthSessionValidationRecordInput { + user_id: claims.user_id().to_string(), + session_id: claims.session_id().to_string(), + token_version: claims.token_version(), + }) + .await .map_err(|error| { warn!( %request_id, user_id = %claims.user_id(), session_id = %claims.session_id(), error = %error, - "Bearer JWT refresh session 状态读取失败" + "Bearer JWT SpacetimeDB 会话状态读取失败" ); AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) })?; + + #[cfg(test)] + let session_is_active = { + let current_user = state + .auth_user_service() + .get_user_by_id(claims.user_id()) + .map_err(|error| { + warn!( + %request_id, + error = %error, + "Bearer JWT 用户快照读取失败" + ); + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + })?; + let Some(current_user) = current_user else { + warn!( + %request_id, + user_id = %claims.user_id(), + "Bearer JWT 对应用户不存在" + ); + return Err(AppError::from_status(StatusCode::UNAUTHORIZED)); + }; + if current_user.token_version != claims.token_version() { + warn!( + %request_id, + user_id = %claims.user_id(), + token_version = claims.token_version(), + current_token_version = current_user.token_version, + "Bearer JWT 版本已失效" + ); + return Err(AppError::from_status(StatusCode::UNAUTHORIZED) + .with_message("当前登录态已失效，请重新登录")); + } + + state + .refresh_session_service() + .is_session_active_for_user( + claims.user_id(), + claims.session_id(), + OffsetDateTime::now_utc(), + ) + .map_err(|error| { + warn!( + %request_id, + user_id = %claims.user_id(), + session_id = %claims.session_id(), + error = %error, + "Bearer JWT refresh session 状态读取失败" + ); + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + })? + }; if !session_is_active { warn!( %request_id, diff --git a/server-rs/crates/api-server/src/tracking_outbox.rs b/server-rs/crates/api-server/src/tracking_outbox.rs index 0f85a68c9..272f08df4 100644 --- a/server-rs/crates/api-server/src/tracking_outbox.rs +++ b/server-rs/crates/api-server/src/tracking_outbox.rs @@ -140,6 +140,10 @@ impl TrackingOutbox { pub fn spawn_worker(self: Arc) { tokio::spawn(async move { + if let Err(error) = self.flush_sealed_files_once().await { + warn!(error = %error, "tracking outbox 启动恢复写入 SpacetimeDB 失败，将保留文件等待重试"); + } + loop { tokio::select! { _ = sleep(self.flush_interval) => { @@ -657,6 +661,45 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + #[tokio::test] + async fn worker_flushes_existing_active_file_immediately_on_startup() { + let dir = test_dir("worker-startup"); + std::fs::create_dir_all(&dir).unwrap(); + let active_path = dir.join(ACTIVE_FILE_NAME); + let record = TrackingOutboxRecord { + event: sample_event("startup-event"), + }; + std::fs::write(&active_path, serde_json::to_vec(&record).unwrap()).unwrap(); + + let outbox = test_outbox(dir.clone(), 500, 1024 * 1024); + outbox.spawn_worker(); + + for _ in 0..100 { + if !active_path.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + + assert!( + !active_path.exists(), + "worker should recover active file without waiting for interval" + ); + 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"); diff --git a/server-rs/crates/api-server/src/wallet_refund_outbox.rs b/server-rs/crates/api-server/src/wallet_refund_outbox.rs index 7c27580f3..a97769086 100644 --- a/server-rs/crates/api-server/src/wallet_refund_outbox.rs +++ b/server-rs/crates/api-server/src/wallet_refund_outbox.rs @@ -84,7 +84,10 @@ impl WalletRefundOutbox { fs::create_dir_all(&self.dir).await?; let pending_path = self.pending_path_for_ledger(&record.ledger_id); - if fs::metadata(&pending_path).await.is_ok() { + if self + .reuse_existing_pending_file(&pending_path, &record.ledger_id) + .await? + { self.flush_notify.notify_one(); return Ok(WalletRefundOutboxEnqueueOutcome::Enqueued); } @@ -108,7 +111,10 @@ impl WalletRefundOutbox { file.flush().await?; file.sync_data().await?; drop(file); - if fs::metadata(&pending_path).await.is_ok() { + if self + .reuse_existing_pending_file(&pending_path, &record.ledger_id) + .await? + { let _ = fs::remove_file(&temp_path).await; self.flush_notify.notify_one(); return Ok(WalletRefundOutboxEnqueueOutcome::Enqueued); @@ -119,8 +125,54 @@ impl WalletRefundOutbox { Ok(WalletRefundOutboxEnqueueOutcome::Enqueued) } + async fn reuse_existing_pending_file( + &self, + pending_path: &Path, + ledger_id: &str, + ) -> Result { + let metadata = match fs::metadata(pending_path).await { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(error.into()), + }; + if !metadata.is_file() { + return Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + format!( + "refund pending path is not a regular file: {}", + pending_path.display() + ), + ) + .into()); + } + + match read_refund_record(pending_path).await { + Ok(existing) if existing.ledger_id == ledger_id => Ok(true), + Ok(_) => Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + format!("refund ledger hash collision at {}", pending_path.display()), + ) + .into()), + Err(error) if error.is_data_corruption() => { + // Preserve the malformed durable file for inspection, then allow this + // enqueue to install a valid file for the same ledger id. + self.quarantine_file(pending_path).await?; + warn!( + source = %pending_path.display(), + "wallet refund outbox 已隔离损坏 pending 文件，继续写入新的幂等退款记录" + ); + Ok(false) + } + Err(error) => Err(error), + } + } + pub fn spawn_worker(self: Arc) { tokio::spawn(async move { + if let Err(error) = self.flush_pending_files_once().await { + warn!(error = %error, "wallet refund outbox 启动恢复退款失败，将保留文件等待重试"); + } + loop { tokio::select! { _ = sleep(self.flush_interval) => { @@ -144,6 +196,7 @@ impl WalletRefundOutbox { async fn flush_pending_files_once(&self) -> Result<(), WalletRefundOutboxError> { fs::create_dir_all(&self.dir).await?; + self.recover_temporary_files().await?; let pending_files = self.list_pending_files().await?; for path in pending_files.into_iter().take(self.batch_size) { let record = match read_refund_record(&path).await { @@ -197,6 +250,117 @@ impl WalletRefundOutbox { Ok(()) } + async fn recover_temporary_files(&self) -> Result<(), WalletRefundOutboxError> { + let _guard = self.enqueue_lock.lock().await; + let temporary_files = self.list_temporary_files().await?; + for path in temporary_files { + let record = match read_refund_record(&path).await { + Ok(record) => record, + Err(error) if error.is_data_corruption() => { + self.quarantine_file(&path).await?; + warn!( + error = %error, + source = %path.display(), + "wallet refund outbox 崩溃遗留临时文件无法解析，已隔离" + ); + continue; + } + Err(error) => return Err(error), + }; + + let pending_path = self.pending_path_for_ledger(&record.ledger_id); + match fs::metadata(&pending_path).await { + Ok(metadata) if metadata.is_file() => { + match read_refund_record(&pending_path).await { + Ok(existing) if existing.ledger_id == record.ledger_id => { + remove_file_and_sync(&path, &self.dir).await?; + debug!( + ledger_id = %record.ledger_id, + source = %path.display(), + target = %pending_path.display(), + "wallet refund outbox 临时文件与已有幂等文件重复，已删除临时副本" + ); + continue; + } + Err(error) if error.is_data_corruption() => { + self.quarantine_file(&pending_path).await?; + } + Ok(_) => { + self.quarantine_file(&path).await?; + warn!( + ledger_id = %record.ledger_id, + source = %path.display(), + target = %pending_path.display(), + "wallet refund outbox 临时文件命名冲突，已隔离临时文件" + ); + continue; + } + Err(error) => return Err(error), + } + } + Ok(_) => { + return Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + format!( + "refund pending path is not a regular file: {}", + pending_path.display() + ), + ) + .into()); + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } + + match fs::hard_link(&path, &pending_path).await { + Ok(()) => { + sync_directory_metadata(&self.dir).await?; + remove_file_and_sync(&path, &self.dir).await?; + debug!( + ledger_id = %record.ledger_id, + source = %path.display(), + target = %pending_path.display(), + "wallet refund outbox 崩溃遗留临时文件已恢复为幂等退款文件" + ); + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + // Another writer won the same ledger id. Keep the first durable file and + // remove only this duplicate temporary link. + remove_file_and_sync(&path, &self.dir).await?; + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + // The file may have been completed by another process after the scan. + continue; + } + Err(error) => return Err(error.into()), + } + } + Ok(()) + } + + async fn list_temporary_files(&self) -> Result, WalletRefundOutboxError> { + let mut entries = fs::read_dir(&self.dir).await?; + let mut files = Vec::new(); + while let Some(entry) = entries.next_entry().await? { + let path = entry.path(); + let Some(name) = path.file_name().and_then(|value| value.to_str()) else { + continue; + }; + if name.starts_with(TEMP_FILE_PREFIX) && name.ends_with(OUTBOX_FILE_EXTENSION) { + files.push(path); + } + } + files.sort(); + Ok(files) + } + + async fn quarantine_file(&self, path: &Path) -> Result<(), WalletRefundOutboxError> { + let corrupt_path = self.corrupt_path_for(path); + fs::rename(path, &corrupt_path).await?; + sync_directory_metadata(&self.dir).await?; + Ok(()) + } + async fn list_pending_files(&self) -> Result, WalletRefundOutboxError> { let mut entries = fs::read_dir(&self.dir).await?; let mut files = Vec::new(); @@ -335,10 +499,19 @@ fn ledger_id_hash(ledger_id: &str) -> String { fn is_pending_outbox_file_name(name: &std::ffi::OsStr) -> bool { name.to_str().is_some_and(|value| { - value.starts_with(PENDING_FILE_PREFIX) && value.ends_with(OUTBOX_FILE_EXTENSION) + (value.starts_with(PENDING_FILE_PREFIX) || value.starts_with(TEMP_FILE_PREFIX)) + && value.ends_with(OUTBOX_FILE_EXTENSION) }) } +async fn remove_file_and_sync(path: &Path, dir: &Path) -> Result<(), WalletRefundOutboxError> { + match fs::remove_file(path).await { + Ok(()) => sync_directory_metadata(dir).await, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error.into()), + } +} + async fn sync_directory_metadata(path: &Path) -> Result<(), WalletRefundOutboxError> { let path = path.to_path_buf(); tokio::task::spawn_blocking(move || { @@ -459,6 +632,36 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + #[tokio::test] + async fn enqueue_does_not_silently_accept_corrupt_pending_file() { + let dir = test_dir("corrupt-pending-enqueue"); + std::fs::create_dir_all(&dir).unwrap(); + let outbox = test_outbox(dir.clone(), 1024 * 1024); + let record = sample_record("ledger-corrupt-pending"); + let pending_path = outbox.pending_path_for_ledger(&record.ledger_id); + std::fs::write(&pending_path, b"{not-json}").unwrap(); + + outbox.enqueue(record.clone()).await.unwrap(); + + assert_eq!( + read_refund_record(&pending_path).await.unwrap().ledger_id, + record.ledger_id + ); + let corrupt_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(CORRUPT_FILE_PREFIX)) + }) + .count(); + assert_eq!(corrupt_count, 1); + + let _ = std::fs::remove_dir_all(dir); + } + #[tokio::test] async fn shutdown_flush_keeps_file_when_spacetime_is_unavailable() { let dir = test_dir("shutdown"); @@ -480,4 +683,72 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + + #[tokio::test] + async fn flush_recovers_valid_crash_left_temp_file() { + let dir = test_dir("recover-temp"); + std::fs::create_dir_all(&dir).unwrap(); + let outbox = test_outbox(dir.clone(), 1024 * 1024); + let record = sample_record("ledger-temp"); + let temp_path = outbox.temp_path(); + std::fs::write(&temp_path, serde_json::to_vec(&record).unwrap()).unwrap(); + + let result = outbox.flush_pending_files_once().await; + + assert!(matches!(result, Err(WalletRefundOutboxError::Spacetime(_)))); + assert!(!temp_path.exists()); + assert!(outbox.pending_path_for_ledger(&record.ledger_id).exists()); + + let _ = std::fs::remove_dir_all(dir); + } + + #[tokio::test] + async fn flush_quarantines_corrupt_crash_left_temp_file() { + let dir = test_dir("recover-corrupt-temp"); + std::fs::create_dir_all(&dir).unwrap(); + let outbox = test_outbox(dir.clone(), 1024 * 1024); + let temp_path = outbox.temp_path(); + std::fs::write(&temp_path, b"{not-json}").unwrap(); + + outbox.flush_pending_files_once().await.unwrap(); + + assert!(!temp_path.exists()); + let corrupt_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(CORRUPT_FILE_PREFIX)) + }) + .count(); + assert_eq!(corrupt_count, 1); + + let _ = std::fs::remove_dir_all(dir); + } + + #[tokio::test] + async fn worker_recovers_temp_file_immediately_on_startup() { + let dir = test_dir("worker-startup"); + std::fs::create_dir_all(&dir).unwrap(); + let outbox = test_outbox(dir.clone(), 1024 * 1024); + let record = sample_record("ledger-worker-startup"); + let temp_path = outbox.temp_path(); + std::fs::write(&temp_path, serde_json::to_vec(&record).unwrap()).unwrap(); + + outbox.clone().spawn_worker(); + + for _ in 0..100 { + if !temp_path.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + + assert!(!temp_path.exists()); + assert!(outbox.pending_path_for_ledger(&record.ledger_id).exists()); + + let _ = std::fs::remove_dir_all(dir); + } } diff --git a/server-rs/crates/spacetime-client/src/active/mapper.rs b/server-rs/crates/spacetime-client/src/active/mapper.rs index 2d925be39..66845286e 100644 --- a/server-rs/crates/spacetime-client/src/active/mapper.rs +++ b/server-rs/crates/spacetime-client/src/active/mapper.rs @@ -38,6 +38,7 @@ pub use self::ai::{ AiResultReferenceRecord, AiTaskMutationRecord, AiTaskRecord, AiTaskStageRecord, AiTextChunkRecord, }; +pub use self::auth::AuthSessionValidationRecordInput; pub use self::editor_agent::{ EditorAgentConversationCreateRecordInput, EditorAgentConversationDeleteRecordInput, EditorAgentConversationRecord, EditorAgentConversationTouchRecordInput, @@ -93,8 +94,8 @@ pub(crate) use self::assets::{ map_optional_asset_object_procedure_result, map_procedure_result, }; pub(crate) use self::auth::{ - map_auth_store_projection_procedure_result, map_auth_store_projection_sync_procedure_result, - map_auth_store_projection_view_input, + map_auth_session_validation_result, map_auth_store_projection_procedure_result, + map_auth_store_projection_sync_procedure_result, map_auth_store_projection_view_input, }; pub(crate) use self::editor_agent::{ map_editor_agent_conversation_list_procedure_result, diff --git a/server-rs/crates/spacetime-client/src/active/mapper/auth.rs b/server-rs/crates/spacetime-client/src/active/mapper/auth.rs index e0ba264bf..c6210fffd 100644 --- a/server-rs/crates/spacetime-client/src/active/mapper/auth.rs +++ b/server-rs/crates/spacetime-client/src/active/mapper/auth.rs @@ -1,5 +1,22 @@ use super::*; +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AuthSessionValidationRecordInput { + pub user_id: String, + pub session_id: String, + pub token_version: u64, +} + +pub(crate) fn map_auth_session_validation_result( + result: crate::module_bindings::AuthSessionValidationProcedureResult, +) -> Result { + if result.error_message.is_some() { + return Err(SpacetimeClientError::procedure_failed(result.error_message)); + } + + Ok(result.active) +} + pub(crate) fn map_auth_store_projection_procedure_result( result: crate::module_bindings::AuthStoreProjectionProcedureResult, ) -> Result { diff --git a/server-rs/crates/spacetime-client/src/auth.rs b/server-rs/crates/spacetime-client/src/auth.rs index 89f14e61a..9d55dac1e 100644 --- a/server-rs/crates/spacetime-client/src/auth.rs +++ b/server-rs/crates/spacetime-client/src/auth.rs @@ -1,6 +1,30 @@ use super::*; impl SpacetimeClient { + pub async fn validate_auth_session( + &self, + input: AuthSessionValidationRecordInput, + ) -> Result { + let procedure_input = crate::module_bindings::AuthSessionValidationInput { + user_id: input.user_id, + session_id: input.session_id, + token_version: input.token_version, + }; + + self.call_after_connect("validate_auth_session", move |connection, sender| { + connection.procedures().validate_auth_session_then( + procedure_input, + move |_, result| { + let mapped = result + .map_err(SpacetimeClientError::from_sdk_error) + .and_then(map_auth_session_validation_result); + send_once(&sender, mapped); + }, + ); + }) + .await + } + pub async fn export_auth_store_projection_from_tables( &self, ) -> Result { diff --git a/server-rs/crates/spacetime-client/src/module_bindings.rs b/server-rs/crates/spacetime-client/src/module_bindings.rs index b9e6e2dad..f808362ee 100644 --- a/server-rs/crates/spacetime-client/src/module_bindings.rs +++ b/server-rs/crates/spacetime-client/src/module_bindings.rs @@ -115,6 +115,8 @@ pub mod asset_operation_wallet_settlement_type; pub mod attach_ai_result_reference_and_return_procedure; pub mod auth_identity_table; pub mod auth_identity_type; +pub mod auth_session_validation_input_type; +pub mod auth_session_validation_procedure_result_type; pub mod auth_store_projection_identity_type; pub mod auth_store_projection_meta_table; pub mod auth_store_projection_meta_type; @@ -844,6 +846,7 @@ pub mod user_account_table; pub mod user_account_type; pub mod user_browse_history_table; pub mod user_browse_history_type; +pub mod validate_auth_session_procedure; pub mod visual_novel_agent_message_row_type; pub mod visual_novel_agent_message_table; pub mod visual_novel_agent_session_row_type; @@ -974,6 +977,8 @@ pub use asset_operation_wallet_settlement_type::AssetOperationWalletSettlement; pub use attach_ai_result_reference_and_return_procedure::attach_ai_result_reference_and_return; pub use auth_identity_table::*; pub use auth_identity_type::AuthIdentity; +pub use auth_session_validation_input_type::AuthSessionValidationInput; +pub use auth_session_validation_procedure_result_type::AuthSessionValidationProcedureResult; pub use auth_store_projection_identity_type::AuthStoreProjectionIdentity; pub use auth_store_projection_meta_table::*; pub use auth_store_projection_meta_type::AuthStoreProjectionMeta; @@ -1703,6 +1708,7 @@ pub use user_account_table::*; pub use user_account_type::UserAccount; pub use user_browse_history_table::*; pub use user_browse_history_type::UserBrowseHistory; +pub use validate_auth_session_procedure::validate_auth_session; pub use visual_novel_agent_message_row_type::VisualNovelAgentMessageRow; pub use visual_novel_agent_message_table::*; pub use visual_novel_agent_session_row_type::VisualNovelAgentSessionRow; diff --git a/server-rs/crates/spacetime-client/src/module_bindings/auth_session_validation_input_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/auth_session_validation_input_type.rs new file mode 100644 index 000000000..2d1479970 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/auth_session_validation_input_type.rs @@ -0,0 +1,17 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +pub struct AuthSessionValidationInput { + pub user_id: String, + pub session_id: String, + pub token_version: u64, +} + +impl __sdk::InModule for AuthSessionValidationInput { + type Module = super::RemoteModule; +} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/auth_session_validation_procedure_result_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/auth_session_validation_procedure_result_type.rs new file mode 100644 index 000000000..5e9afc3d5 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/auth_session_validation_procedure_result_type.rs @@ -0,0 +1,16 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +pub struct AuthSessionValidationProcedureResult { + pub active: bool, + pub error_message: Option, +} + +impl __sdk::InModule for AuthSessionValidationProcedureResult { + type Module = super::RemoteModule; +} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/validate_auth_session_procedure.rs b/server-rs/crates/spacetime-client/src/module_bindings/validate_auth_session_procedure.rs new file mode 100644 index 000000000..e96904972 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/validate_auth_session_procedure.rs @@ -0,0 +1,59 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +use super::auth_session_validation_input_type::AuthSessionValidationInput; +use super::auth_session_validation_procedure_result_type::AuthSessionValidationProcedureResult; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +struct ValidateAuthSessionArgs { + pub input: AuthSessionValidationInput, +} + +impl __sdk::InModule for ValidateAuthSessionArgs { + type Module = super::RemoteModule; +} + +#[allow(non_camel_case_types)] +/// Extension trait for access to the procedure `validate_auth_session`. +/// +/// Implemented for [`super::RemoteProcedures`]. +pub trait validate_auth_session { + fn validate_auth_session(&self, input: AuthSessionValidationInput) { + self.validate_auth_session_then(input, |_, _| {}); + } + + fn validate_auth_session_then( + &self, + input: AuthSessionValidationInput, + + __callback: impl FnOnce( + &super::ProcedureEventContext, + Result, + ) + Send + + 'static, + ); +} + +impl validate_auth_session for super::RemoteProcedures { + fn validate_auth_session_then( + &self, + input: AuthSessionValidationInput, + + __callback: impl FnOnce( + &super::ProcedureEventContext, + Result, + ) + Send + + 'static, + ) { + self.imp + .invoke_procedure_with_callback::<_, AuthSessionValidationProcedureResult>( + "validate_auth_session", + ValidateAuthSessionArgs { input }, + __callback, + ); + } +} diff --git a/server-rs/crates/spacetime-module/Cargo.toml b/server-rs/crates/spacetime-module/Cargo.toml index 9f264dd64..d2a516e1b 100644 --- a/server-rs/crates/spacetime-module/Cargo.toml +++ b/server-rs/crates/spacetime-module/Cargo.toml @@ -20,3 +20,4 @@ module-runtime = { workspace = true, features = ["spacetime-types"] } sha2 = { workspace = true } spacetimedb = { workspace = true, features = ["unstable"] } spacetimedb-lib = { workspace = true, features = ["serde"] } +time = { workspace = true, features = ["parsing"] } diff --git a/server-rs/crates/spacetime-module/src/auth/procedures.rs b/server-rs/crates/spacetime-module/src/auth/procedures.rs index 5823f5450..8d5ff8eec 100644 --- a/server-rs/crates/spacetime-module/src/auth/procedures.rs +++ b/server-rs/crates/spacetime-module/src/auth/procedures.rs @@ -1,4 +1,6 @@ use crate::{ProcedureContext, ReducerContext, SpacetimeType, Table, Timestamp}; +use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; use super::tables::{ AuthIdentity, AuthStoreProjectionMeta, RefreshSession, UserAccount, auth_identity, @@ -77,6 +79,68 @@ pub struct AuthStoreProjectionSyncProcedureResult { pub error_message: Option, } +#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)] +pub struct AuthSessionValidationInput { + pub user_id: String, + pub session_id: String, + pub token_version: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)] +pub struct AuthSessionValidationProcedureResult { + pub active: bool, + pub error_message: Option, +} + +#[spacetimedb::procedure] +pub fn validate_auth_session( + ctx: &mut ProcedureContext, + input: AuthSessionValidationInput, +) -> AuthSessionValidationProcedureResult { + match ctx.try_with_tx(|tx| validate_auth_session_tx(tx, input.clone())) { + Ok(active) => AuthSessionValidationProcedureResult { + active, + error_message: None, + }, + Err(message) => AuthSessionValidationProcedureResult { + active: false, + error_message: Some(message), + }, + } +} + +fn validate_auth_session_tx( + ctx: &ReducerContext, + input: AuthSessionValidationInput, +) -> Result { + let Some(user) = ctx.db.user_account().user_id().find(&input.user_id) else { + return Ok(false); + }; + if user.token_version != input.token_version { + return Ok(false); + } + + let Some(session) = ctx + .db + .refresh_session() + .session_id() + .find(&input.session_id) + else { + return Ok(false); + }; + if session.user_id != input.user_id || session.revoked_at.is_some() { + return Ok(false); + } + + let expires_at = OffsetDateTime::parse(&session.expires_at, &Rfc3339) + .map_err(|_| "refresh session 过期时间格式非法".to_string())?; + let now = OffsetDateTime::from_unix_timestamp_nanos( + i128::from(ctx.timestamp.to_micros_since_unix_epoch()) * 1_000, + ) + .map_err(|_| "SpacetimeDB 当前时间超出认证时间范围".to_string())?; + Ok(expires_at > now) +} + #[spacetimedb::procedure] pub fn sync_auth_store_projection( ctx: &mut ProcedureContext, @@ -118,6 +182,14 @@ fn sync_auth_store_projection_tx( ctx: &ReducerContext, input: AuthStoreProjectionView, ) -> Result { + let current_updated_at_micros = ctx + .db + .auth_store_projection_meta() + .meta_id() + .find(&AUTH_STORE_PROJECTION_META_ID.to_string()) + .map(|row| row.updated_at.to_micros_since_unix_epoch()); + ensure_newer_auth_projection_version(current_updated_at_micros, input.updated_at_micros)?; + let user_ids = input .users .iter() @@ -255,6 +327,27 @@ fn sync_auth_store_projection_tx( }) } +/// Full projections are emitted by API-local auth worksets. The metadata row +/// is the monotonic watermark that keeps a delayed snapshot from deleting or +/// replacing data written by a newer snapshot on another API instance. The +/// check and the subsequent writes run in the same procedure transaction, so +/// concurrent calls are serialized by SpacetimeDB. Replaying the same +/// version remains idempotent. +fn ensure_newer_auth_projection_version( + current_updated_at_micros: Option, + incoming_updated_at_micros: i64, +) -> Result<(), String> { + if let Some(current_updated_at_micros) = current_updated_at_micros { + if incoming_updated_at_micros < current_updated_at_micros { + return Err(format!( + "认证投影版本冲突：请求版本 {incoming_updated_at_micros} 早于当前版本 {current_updated_at_micros}" + )); + } + } + + Ok(()) +} + fn export_auth_store_projection_from_tables_tx( ctx: &ReducerContext, ) -> Result { @@ -465,4 +558,12 @@ mod tests { ); assert_eq!(created_at, "2026-07-01T00:00:00Z"); } + + #[test] + fn auth_projection_version_must_advance_monotonically() { + assert!(ensure_newer_auth_projection_version(None, 1).is_ok()); + assert!(ensure_newer_auth_projection_version(Some(10), 11).is_ok()); + assert!(ensure_newer_auth_projection_version(Some(10), 10).is_ok()); + assert!(ensure_newer_auth_projection_version(Some(10), 9).is_err()); + } }