收口跨节点鉴权读取与节点 outbox 恢复

生产 Bearer 鉴权改用 SpacetimeDB 会话校验 procedure 为认证整包投影增加单调水位保护新增鉴权客户端 facade 与生成 bindings 补齐 tracking 与钱包退款 outbox 启动恢复同步后端架构、运维和项目决策文档
2026-08-27 13:36:28 +08:00
parent d41aaddd0e
commit 72d7214646
16 changed files with 634 additions and 45 deletions
@@ -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，不能用粘性会话替代共享持久化；生产节点必须保留对应目录并纳入恢复演练。
@@ -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` 为准。

@@ -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/<version>`，`genarrative` 用户无法在其中创建 `server-rs`。修复顺序：

@@ -5526,6 +5526,7 @@ dependencies = [
 "shared-contracts",
 "spacetimedb",
 "spacetimedb-lib",
+ "time",
 ]

 [[package]]
@@ -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,
@@ -140,6 +140,10 @@ impl TrackingOutbox {

    pub fn spawn_worker(self: Arc<Self>) {
        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");
@@ -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<bool, WalletRefundOutboxError> {
+        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<Self>) {
        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<Vec<PathBuf>, 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<Vec<PathBuf>, 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);
+    }
 }
@@ -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,
@@ -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<bool, SpacetimeClientError> {
+    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<module_auth::AuthStoreProjectionView, SpacetimeClientError> {
@@ -1,6 +1,30 @@
 use super::*;

 impl SpacetimeClient {
+    pub async fn validate_auth_session(
+        &self,
+        input: AuthSessionValidationRecordInput,
+    ) -> Result<bool, SpacetimeClientError> {
+        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<module_auth::AuthStoreProjectionView, SpacetimeClientError> {
@@ -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;
@@ -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;
+}
@@ -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<String>,
+}
+
+impl __sdk::InModule for AuthSessionValidationProcedureResult {
+    type Module = super::RemoteModule;
+}
@@ -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<AuthSessionValidationProcedureResult, __sdk::InternalError>,
+            ) + Send
+            + 'static,
+    );
+}
+
+impl validate_auth_session for super::RemoteProcedures {
+    fn validate_auth_session_then(
+        &self,
+        input: AuthSessionValidationInput,
+
+        __callback: impl FnOnce(
+                &super::ProcedureEventContext,
+                Result<AuthSessionValidationProcedureResult, __sdk::InternalError>,
+            ) + Send
+            + 'static,
+    ) {
+        self.imp
+            .invoke_procedure_with_callback::<_, AuthSessionValidationProcedureResult>(
+                "validate_auth_session",
+                ValidateAuthSessionArgs { input },
+                __callback,
+            );
+    }
+}
@@ -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"] }
@@ -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<String>,
 }

+#[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<String>,
+}
+
+#[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<bool, String> {
+    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<AuthStoreProjectionSyncRecord, String> {
+    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<i64>,
+    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<AuthStoreProjectionView, String> {
@@ -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());
+    }
 }