修复鉴权投影并发与退款 outbox 冲突

绑定认证工作集投影版本并在 SpacetimeDB 内执行 CAS 为认证 procedure 增加 runtime service identity 校验防止退款 outbox 跨进程覆盖并校验临时文件事实更新后端契约与项目决策记录
2026-08-27 14:47:22 +08:00
parent 72d7214646
commit 36a4d37d24
10 changed files with 441 additions and 52 deletions
@@ -7751,5 +7751,5 @@ CI 上 `background_agent_runtime_recovers_stale_running_before_pending_task` 在
 ## 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 实例的延迟快照保护。
+- `validate_auth_session`、认证投影导出和同步 procedure 都从 `ctx.sender()` 取调用方并要求现役 runtime service identity。`sync_auth_store_projection` 使用 API 工作集启动恢复或上次成功同步绑定的 `base_updated_at_micros` 做事务内 CAS，并要求 `updated_at_micros` 严格递增；基线不一致或版本不晚于当前值的整包快照失败关闭，只有确认本次同步前没有遗留未确认本地变更且同步期间无其它本地认证变更时才从正式表恢复，否则保留旧基线继续失败关闭，避免自动恢复覆盖并发未提交变更。同步成功但期间出现新本地变更时最多连续补同步三轮，仍未稳定则失败关闭，并保留待重试 revision。正式认证表仍是权威源，这只是迁移期跨 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 中间件不再从 `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，避免一次短超时让进程永久停留在依赖不可用状态。
+认证恢复策略：`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 返回服务错误。`validate_auth_session`、投影导出和投影同步均从 `ctx.sender()` 派生调用方，并复用现役 runtime service identity 白名单；启动恢复先完成该服务身份初始化，普通 SpacetimeDB identity 不能读取或改写私有认证表。测试构建仍可使用显式的内存测试夹具。运行中 refresh cookie 在本进程工作集内未命中时直接按失效处理，不再从 SpacetimeDB 导出整包认证状态刷新内存，避免旧投影把重复手机号或旧会话重新灌回进程。`module-auth` 只保留内存工作集和 projection 导入 / 导出能力，不再保留 JSON 快照导入 / 导出能力，也不写本地持久化文件；`auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 不再是兼容恢复源。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作仍必须在返回客户端前通过 `sync_auth_store_projection` 成功同步 SpacetimeDB 正式认证表；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号或会话当成成功结果。每个 API 工作集绑定启动恢复或上次成功同步得到的 `auth_store_projection_meta.updated_at` 版本作为 `base_updated_at_micros`，SpacetimeDB 在同一事务内执行基线 CAS，并要求新的 `updated_at_micros` 严格递增；基线不一致或版本不晚于当前值时整包写入失败，冲突节点在确认本次同步前没有遗留未确认本地变更且同步期间没有其它本地认证变更后，才可丢弃本地工作集并从正式表恢复，不能用陈旧工作集删除、恢复或覆盖另一节点的新状态；若此前已有待重试 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` 为准。

@@ -8,7 +8,7 @@ use std::{
    fmt,
    sync::{
        Arc,
-        atomic::{AtomicBool, Ordering},
+        atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering},
    },
 };

@@ -35,7 +35,7 @@ use spacetime_client::{
    SpacetimeClient, SpacetimeClientConfig, SpacetimeClientError, SpacetimeClientHealthSnapshot,
 };
 use time::OffsetDateTime;
-use tokio::sync::{Semaphore, broadcast};
+use tokio::sync::{Mutex as AsyncMutex, Semaphore, broadcast};
 use tracing::{info, warn};

 use crate::config::AppConfig;
@@ -273,6 +273,14 @@ pub struct AppStateInner {
    oss_client: Option<OssClient>,
    #[cfg_attr(test, allow(dead_code))]
    auth_store: InMemoryAuthStore,
+    /// 当前进程工作集所基于的正式认证投影版本；跨节点写入使用它做 CAS。
+    #[cfg_attr(test, allow(dead_code))]
+    auth_projection_version: AtomicI64,
+    /// 最近一次确认写入正式投影时对应的工作集 revision；不一致表示有待重试的本地变更。
+    #[cfg_attr(test, allow(dead_code))]
+    auth_projection_synced_revision: AtomicU64,
+    #[cfg_attr(test, allow(dead_code))]
+    auth_projection_sync_lock: AsyncMutex<()>,
    password_entry_service: PasswordEntryService,
    refresh_session_service: RefreshSessionService,
    auth_user_service: AuthUserService,
@@ -505,12 +513,13 @@ impl AppState {

    pub fn new_with_empty_auth_store(config: AppConfig) -> Result<Self, AppStateInitError> {
        // 中文注释：api-server 不再把本地 auth-store.json 当作用户认证真相源，启动恢复只允许来自 SpacetimeDB。
-        Self::new_with_auth_store(config, InMemoryAuthStore::default())
+        Self::new_with_auth_store(config, InMemoryAuthStore::default(), 0)
    }

    fn new_with_auth_store(
        config: AppConfig,
        auth_store: InMemoryAuthStore,
+        auth_projection_version: i64,
    ) -> Result<Self, AppStateInitError> {
        let auth_jwt_config = JwtConfig::new(
            config.jwt_issuer.clone(),
@@ -629,6 +638,9 @@ impl AppState {
            test_external_background_removal_enqueue: Arc::new(Mutex::new(None)),
            oss_client,
            auth_store,
+            auth_projection_version: AtomicI64::new(auth_projection_version),
+            auth_projection_synced_revision: AtomicU64::new(0),
+            auth_projection_sync_lock: AsyncMutex::new(()),
            password_entry_service,
            refresh_session_service,
            auth_user_service,
@@ -1288,30 +1300,102 @@ impl AppState {
        return Ok(());

        #[cfg(not(test))]
-        let updated_at_micros = i64::try_from(
-            OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000,
-        )
-        .map_err(|_| SpacetimeClientError::Runtime("认证状态更新时间超出 i64 范围".to_string()))?;
+        let _sync_guard = self.auth_projection_sync_lock.lock().await;
        #[cfg(not(test))]
-        let projection = self
-            .auth_store
-            .export_projection_view(updated_at_micros)
-            .map_err(SpacetimeClientError::Runtime)?;
-        // 当前仍由 module-auth 的进程内工作集执行业务规则；这里只用 typed projection 同步正式认证表。
+        let sync_start_revision = self.auth_store.revision();
        #[cfg(not(test))]
-        if let Err(error) = self
-            .spacetime_client
-            .sync_auth_store_projection(projection)
-            .await
-        {
-            warn!(
-                error = %error,
-                "认证投影同步 SpacetimeDB 正式表失败，当前认证流程中止"
-            );
-            return Err(error);
+        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 =
+                i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000).map_err(
+                    |_| SpacetimeClientError::Runtime("认证状态更新时间超出 i64 范围".to_string()),
+                )?;
+            let updated_at_micros = if now_updated_at_micros > base_updated_at_micros {
+                now_updated_at_micros
+            } else {
+                base_updated_at_micros.checked_add(1).ok_or_else(|| {
+                    SpacetimeClientError::Runtime("认证状态版本超出 i64 范围".to_string())
+                })?
+            };
+            let (mut projection, attempted_revision) = self
+                .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 同步正式认证表。
+            match self
+                .spacetime_client
+                .sync_auth_store_projection(projection)
+                .await
+            {
+                Ok(_) => {
+                    self.auth_projection_version
+                        .store(updated_at_micros, Ordering::Release);
+                    if self.auth_store.revision() == attempted_revision {
+                        self.auth_projection_synced_revision
+                            .store(attempted_revision, Ordering::Release);
+                        return Ok(());
+                    }
+                    local_changed_during_sync = true;
+                    warn!(
+                        attempt,
+                        "认证投影同步期间工作集发生变化，将继续同步最新工作集"
+                    );
+                    continue;
+                }
+                Err(error) => {
+                    warn!(
+                        error = %error,
+                        "认证投影同步 SpacetimeDB 正式表失败，当前认证流程中止"
+                    );
+                    if local_changed_during_sync || self.auth_store.revision() != attempted_revision
+                    {
+                        warn!(
+                            "认证投影同步失败期间工作集发生并发变化，跳过自动恢复以避免覆盖未提交变更"
+                        );
+                    } else if let Ok(current_projection) = self
+                        .spacetime_client
+                        .export_auth_store_projection_from_tables()
+                        .await
+                    {
+                        match self.auth_store.refresh_from_projection_view_if_revision(
+                            current_projection.clone(),
+                            attempted_revision,
+                        ) {
+                            Ok(true) => {
+                                self.auth_projection_version
+                                    .store(current_projection.updated_at_micros, Ordering::Release);
+                                self.auth_projection_synced_revision
+                                    .store(self.auth_store.revision(), Ordering::Release);
+                            }
+                            Ok(false) => {
+                                warn!(
+                                    "认证投影同步冲突期间工作集发生并发变化，跳过自动恢复以避免覆盖未提交变更"
+                                );
+                            }
+                            Err(refresh_error) => {
+                                warn!(
+                                    error = %refresh_error,
+                                    "认证投影同步冲突后恢复进程内工作集失败"
+                                );
+                            }
+                        }
+                    }
+                    return Err(error);
+                }
+            }
        }
        #[cfg(not(test))]
-        Ok(())
+        Err(SpacetimeClientError::Runtime(
+            "认证工作集在同步期间持续发生变化，未能完成投影同步".to_string(),
+        ))
    }

    pub async fn try_restore_auth_store_from_spacetime(
@@ -1319,6 +1403,11 @@ impl AppState {
    ) -> Result<Self, AppStateInitError> {
        let spacetime_client =
            SpacetimeClient::new(spacetime_client_config_for_startup_restore(&config));
+        initialize_editor_generation_runtime_service_identity_for_startup(
+            &config,
+            &spacetime_client,
+        )
+        .await?;
        let mut spacetime_restore_available = false;
        let mut restore_errors = Vec::new();

@@ -1332,7 +1421,11 @@ impl AppState {
                    projection,
                    AuthStoreRestoreSource::SpacetimeTables,
                )? {
-                    let state = Self::new_with_auth_store(config, candidate.auth_store)?;
+                    let state = Self::new_with_auth_store(
+                        config,
+                        candidate.auth_store,
+                        candidate.updated_at_micros.unwrap_or_default(),
+                    )?;
                    info!(
                        source = candidate.source.as_str(),
                        updated_at_micros = candidate.updated_at_micros,
@@ -1845,6 +1938,7 @@ fn auth_store_candidate_from_projection_view(
    if projection.users.is_empty()
        && projection.identities.is_empty()
        && projection.refresh_sessions.is_empty()
+        && projection.updated_at_micros == 0
    {
        return Ok(None);
    }
@@ -1886,6 +1980,34 @@ fn spacetime_client_config_for_startup_restore(config: &AppConfig) -> SpacetimeC
    }
 }

+async fn initialize_editor_generation_runtime_service_identity_for_startup(
+    config: &AppConfig,
+    spacetime_client: &SpacetimeClient,
+) -> Result<(), AppStateInitError> {
+    let pricing_store =
+        EditorGenerationPricingStore::load(config.editor_generation_pricing_override_path.clone())
+            .map_err(|error| AppStateInitError::DependencyUnavailable(error.to_string()))?;
+    let fallback = pricing_store
+        .snapshot()
+        .map_err(|error| AppStateInitError::DependencyUnavailable(error.to_string()))?;
+    let models = editor_generation_pricing_to_records(&fallback)
+        .map_err(|error| AppStateInitError::DependencyUnavailable(error.to_string()))?;
+    spacetime_client
+        .initialize_editor_generation_pricing_config_if_missing(
+            editor_generation_pricing_upsert_input(
+                config,
+                "system:editor-generation-pricing".to_string(),
+                models,
+                crate::editor_project::current_utc_micros(),
+            ),
+        )
+        .await
+        .map_err(|error| {
+            AppStateInitError::DependencyUnavailable(format!("初始化模型定价服务身份失败：{error}"))
+        })?;
+    Ok(())
+}
+
 impl fmt::Display for AppStateInitError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
@@ -34,7 +34,7 @@ pub struct WalletRefundOutbox {
    flush_notify: Arc<Notify>,
 }

-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
 pub(crate) struct WalletRefundOutboxRecord {
    pub owner_user_id: String,
    pub amount: u64,
@@ -85,7 +85,7 @@ impl WalletRefundOutbox {

        let pending_path = self.pending_path_for_ledger(&record.ledger_id);
        if self
-            .reuse_existing_pending_file(&pending_path, &record.ledger_id)
+            .reuse_existing_pending_file(&pending_path, &record)
            .await?
        {
            self.flush_notify.notify_one();
@@ -112,23 +112,48 @@ impl WalletRefundOutbox {
        file.sync_data().await?;
        drop(file);
        if self
-            .reuse_existing_pending_file(&pending_path, &record.ledger_id)
+            .reuse_existing_pending_file(&pending_path, &record)
            .await?
        {
            let _ = fs::remove_file(&temp_path).await;
            self.flush_notify.notify_one();
            return Ok(WalletRefundOutboxEnqueueOutcome::Enqueued);
        }
-        fs::rename(&temp_path, &pending_path).await?;
-        sync_directory_metadata(&self.dir).await?;
-        self.flush_notify.notify_one();
-        Ok(WalletRefundOutboxEnqueueOutcome::Enqueued)
+        for _ in 0..2 {
+            match fs::hard_link(&temp_path, &pending_path).await {
+                Ok(()) => {
+                    sync_directory_metadata(&self.dir).await?;
+                    remove_file_and_sync(&temp_path, &self.dir).await?;
+                    self.flush_notify.notify_one();
+                    return Ok(WalletRefundOutboxEnqueueOutcome::Enqueued);
+                }
+                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
+                    if self
+                        .reuse_existing_pending_file(&pending_path, &record)
+                        .await?
+                    {
+                        remove_file_and_sync(&temp_path, &self.dir).await?;
+                        self.flush_notify.notify_one();
+                        return Ok(WalletRefundOutboxEnqueueOutcome::Enqueued);
+                    }
+                }
+                Err(error) => return Err(error.into()),
+            }
+        }
+        Err(std::io::Error::new(
+            std::io::ErrorKind::AlreadyExists,
+            format!(
+                "refund pending path could not be installed: {}",
+                pending_path.display()
+            ),
+        )
+        .into())
    }

    async fn reuse_existing_pending_file(
        &self,
        pending_path: &Path,
-        ledger_id: &str,
+        expected: &WalletRefundOutboxRecord,
    ) -> Result<bool, WalletRefundOutboxError> {
        let metadata = match fs::metadata(pending_path).await {
            Ok(metadata) => metadata,
@@ -147,7 +172,16 @@ impl WalletRefundOutbox {
        }

        match read_refund_record(pending_path).await {
-            Ok(existing) if existing.ledger_id == ledger_id => Ok(true),
+            Ok(existing) if existing == *expected => Ok(true),
+            Ok(existing) if existing.ledger_id == expected.ledger_id => Err(std::io::Error::new(
+                std::io::ErrorKind::AlreadyExists,
+                format!(
+                    "refund ledger {} 已存在但退款事实不一致: {}",
+                    expected.ledger_id,
+                    pending_path.display()
+                ),
+            )
+            .into()),
            Ok(_) => Err(std::io::Error::new(
                std::io::ErrorKind::AlreadyExists,
                format!("refund ledger hash collision at {}", pending_path.display()),
@@ -272,7 +306,7 @@ impl WalletRefundOutbox {
            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 => {
+                        Ok(existing) if existing == record => {
                            remove_file_and_sync(&path, &self.dir).await?;
                            debug!(
                                ledger_id = %record.ledger_id,
@@ -282,6 +316,16 @@ impl WalletRefundOutbox {
                            );
                            continue;
                        }
+                        Ok(existing) if existing.ledger_id == record.ledger_id => {
+                            self.quarantine_file(&path).await?;
+                            warn!(
+                                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?;
                        }
@@ -324,9 +368,51 @@ impl WalletRefundOutbox {
                    );
                }
                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?;
+                    match self
+                        .reuse_existing_pending_file(&pending_path, &record)
+                        .await
+                    {
+                        Ok(true) => {
+                            // Another writer won the same ledger id with identical facts. Keep
+                            // the first durable file and remove only this duplicate temporary
+                            // link.
+                            remove_file_and_sync(&path, &self.dir).await?;
+                        }
+                        Ok(false) => match fs::hard_link(&path, &pending_path).await {
+                            Ok(()) => {
+                                sync_directory_metadata(&self.dir).await?;
+                                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) if error.kind() == std::io::ErrorKind::AlreadyExists => {
+                                self.quarantine_file(&path).await?;
+                                warn!(
+                                    ledger_id = %record.ledger_id,
+                                    source = %path.display(),
+                                    target = %pending_path.display(),
+                                    "wallet refund outbox 临时文件无法与现有幂等文件合并，已隔离临时文件"
+                                );
+                            }
+                            Err(error) => return Err(error.into()),
+                        },
+                        Err(WalletRefundOutboxError::Io(error))
+                            if error.kind() == std::io::ErrorKind::AlreadyExists =>
+                        {
+                            self.quarantine_file(&path).await?;
+                            warn!(
+                                ledger_id = %record.ledger_id,
+                                source = %path.display(),
+                                target = %pending_path.display(),
+                                error = %error,
+                                "wallet refund outbox 临时文件与现有幂等文件事实冲突，已隔离临时文件"
+                            );
+                        }
+                        Err(error) => return Err(error),
+                    }
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                    // The file may have been completed by another process after the scan.
@@ -588,6 +674,25 @@ mod tests {
        let _ = std::fs::remove_dir_all(dir);
    }

+    #[tokio::test]
+    async fn enqueue_rejects_conflicting_existing_ledger_file() {
+        let dir = test_dir("conflicting-ledger");
+        let outbox = test_outbox(dir.clone(), 1024 * 1024);
+        outbox.enqueue(sample_record("ledger-1")).await.unwrap();
+
+        let mut conflicting = sample_record("ledger-1");
+        conflicting.amount += 1;
+        let error = outbox
+            .enqueue(conflicting)
+            .await
+            .expect_err("conflicting refund must fail");
+        assert!(
+            matches!(error, WalletRefundOutboxError::Io(error) if error.kind() == std::io::ErrorKind::AlreadyExists)
+        );
+
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
    #[tokio::test]
    async fn enqueue_drops_when_outbox_exceeds_max_bytes() {
        let dir = test_dir("max-bytes");
@@ -702,6 +807,42 @@ mod tests {
        let _ = std::fs::remove_dir_all(dir);
    }

+    #[tokio::test]
+    async fn flush_quarantines_conflicting_crash_left_temp_file() {
+        let dir = test_dir("recover-conflicting-temp");
+        std::fs::create_dir_all(&dir).unwrap();
+        let outbox = test_outbox(dir.clone(), 1024 * 1024);
+        let record = sample_record("ledger-temp-conflict");
+        let mut conflicting = record.clone();
+        conflicting.amount += 1;
+        let pending_path = outbox.pending_path_for_ledger(&record.ledger_id);
+        let temp_path = outbox.temp_path();
+        std::fs::write(&pending_path, serde_json::to_vec(&conflicting).unwrap()).unwrap();
+        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_eq!(
+            read_refund_record(&pending_path).await.unwrap(),
+            conflicting
+        );
+        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 flush_quarantines_corrupt_crash_left_temp_file() {
        let dir = test_dir("recover-corrupt-temp");
@@ -187,6 +187,9 @@ pub struct AuthStoreProjectionView {
    pub users: Vec<AuthStoreProjectionUser>,
    pub identities: Vec<AuthStoreProjectionIdentity>,
    pub refresh_sessions: Vec<AuthStoreProjectionRefreshSession>,
+    /// 当前进程工作集所基于的正式投影版本，用于事务内 CAS。
+    #[serde(default)]
+    pub base_updated_at_micros: i64,
 }

 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -12,7 +12,10 @@ pub use events::*;

 use std::{
    collections::{HashMap, HashSet},
-    sync::{Arc, Mutex},
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicU64, Ordering},
+    },
 };

 use platform_auth::{
@@ -31,6 +34,7 @@ use tracing::{info, warn};
 #[derive(Clone, Debug)]
 pub struct InMemoryAuthStore {
    inner: Arc<Mutex<InMemoryAuthStoreState>>,
+    revision: Arc<AtomicU64>,
 }

 #[derive(Debug)]
@@ -987,6 +991,7 @@ impl Default for InMemoryAuthStore {
    fn default() -> Self {
        Self {
            inner: Arc::new(Mutex::new(InMemoryAuthStoreState::default())),
+            revision: Arc::new(AtomicU64::new(0)),
        }
    }
 }
@@ -1164,9 +1169,14 @@ impl InMemoryAuthStore {
            inner: Arc::new(Mutex::new(InMemoryAuthStoreState::from_projection_view(
                view,
            )?)),
+            revision: Arc::new(AtomicU64::new(0)),
        })
    }

+    pub fn revision(&self) -> u64 {
+        self.revision.load(Ordering::Acquire)
+    }
+
    pub fn refresh_from_projection_view(
        &self,
        view: AuthStoreProjectionView,
@@ -1177,10 +1187,30 @@ impl InMemoryAuthStore {
            .lock()
            .map_err(|_| "认证仓储锁已中毒".to_string())?;
        state.apply_persistent_state(next_state);
+        self.revision.fetch_add(1, Ordering::Release);

        Ok(())
    }

+    pub fn refresh_from_projection_view_if_revision(
+        &self,
+        view: AuthStoreProjectionView,
+        expected_revision: u64,
+    ) -> Result<bool, String> {
+        let next_state = InMemoryAuthStoreState::from_projection_view(view)?;
+        let mut state = self
+            .inner
+            .lock()
+            .map_err(|_| "认证仓储锁已中毒".to_string())?;
+        if self.revision.load(Ordering::Acquire) != expected_revision {
+            return Ok(false);
+        }
+        state.apply_persistent_state(next_state);
+        self.revision.fetch_add(1, Ordering::Release);
+
+        Ok(true)
+    }
+
    pub fn export_projection_view(
        &self,
        updated_at_micros: i64,
@@ -1255,6 +1285,7 @@ impl InMemoryAuthStore {
            .collect::<Result<Vec<_>, String>>()?;

        Ok(AuthStoreProjectionView {
+            base_updated_at_micros: 0,
            updated_at_micros,
            users,
            identities,
@@ -1262,8 +1293,24 @@ impl InMemoryAuthStore {
        })
    }

+    pub fn export_projection_view_with_revision(
+        &self,
+        updated_at_micros: i64,
+    ) -> Result<(AuthStoreProjectionView, u64), String> {
+        for _ in 0..3 {
+            let before = self.revision.load(Ordering::Acquire);
+            let view = self.export_projection_view(updated_at_micros)?;
+            let after = self.revision.load(Ordering::Acquire);
+            if before == after {
+                return Ok((view, after));
+            }
+        }
+        Err("认证工作集在导出期间持续发生变化".to_string())
+    }
+
    fn persist_state(&self, state: &InMemoryAuthStoreState) -> Result<(), String> {
        let _ = state;
+        self.revision.fetch_add(1, Ordering::Release);
        Ok(())
    }

@@ -2783,6 +2830,7 @@ mod tests {

    fn empty_projection_store() -> InMemoryAuthStore {
        InMemoryAuthStore::from_projection_view(AuthStoreProjectionView {
+            base_updated_at_micros: 0,
            updated_at_micros: 0,
            users: vec![],
            identities: vec![],
@@ -3208,6 +3256,7 @@ mod tests {
    async fn phone_login_reuses_user_restored_from_projection() {
        let phone_service = build_phone_service(
            InMemoryAuthStore::from_projection_view(AuthStoreProjectionView {
+                base_updated_at_micros: 0,
                updated_at_micros: 1,
                users: vec![projection_user(
                    "user_existing_phone",
@@ -3359,6 +3408,30 @@ mod tests {
        ));
    }

+    #[test]
+    fn conditional_projection_refresh_rejects_stale_revision() {
+        let store = InMemoryAuthStore::default();
+        let projection = AuthStoreProjectionView {
+            base_updated_at_micros: 0,
+            updated_at_micros: 1,
+            users: vec![],
+            identities: vec![],
+            refresh_sessions: vec![],
+        };
+
+        assert_eq!(store.revision(), 0);
+        store
+            .refresh_from_projection_view(projection.clone())
+            .expect("initial projection refresh should succeed");
+        assert_eq!(store.revision(), 1);
+        assert!(
+            !store
+                .refresh_from_projection_view_if_revision(projection, 0)
+                .expect("stale projection refresh should be checked without error")
+        );
+        assert_eq!(store.revision(), 1);
+    }
+
    #[tokio::test]
    async fn empty_projection_restore_does_not_block_phone_login() {
        let phone_service = build_phone_service(empty_projection_store());
@@ -4248,6 +4321,7 @@ mod tests {
    #[tokio::test]
    async fn bind_wechat_phone_merges_when_existing_phone_restored_from_projection() {
        let store = InMemoryAuthStore::from_projection_view(AuthStoreProjectionView {
+            base_updated_at_micros: 0,
            updated_at_micros: 1,
            users: vec![projection_user(
                "user_existing_phone_bind",
@@ -52,6 +52,7 @@ pub(crate) fn map_auth_store_projection_view_input(
    view: module_auth::AuthStoreProjectionView,
 ) -> crate::module_bindings::AuthStoreProjectionView {
    crate::module_bindings::AuthStoreProjectionView {
+        base_updated_at_micros: view.base_updated_at_micros,
        updated_at_micros: view.updated_at_micros,
        users: view
            .users
@@ -111,6 +112,7 @@ fn map_auth_store_projection_view(
    view: crate::module_bindings::AuthStoreProjectionView,
 ) -> module_auth::AuthStoreProjectionView {
    module_auth::AuthStoreProjectionView {
+        base_updated_at_micros: view.base_updated_at_micros,
        updated_at_micros: view.updated_at_micros,
        users: view
            .users
@@ -35,6 +35,7 @@ pub(crate) fn map_auth_store_projection_view_input(
    view: module_auth::AuthStoreProjectionView,
 ) -> crate::module_bindings::AuthStoreProjectionView {
    crate::module_bindings::AuthStoreProjectionView {
+        base_updated_at_micros: view.base_updated_at_micros,
        updated_at_micros: view.updated_at_micros,
        users: view
            .users
@@ -94,6 +95,7 @@ fn map_auth_store_projection_view(
    view: crate::module_bindings::AuthStoreProjectionView,
 ) -> module_auth::AuthStoreProjectionView {
    module_auth::AuthStoreProjectionView {
+        base_updated_at_micros: view.base_updated_at_micros,
        updated_at_micros: view.updated_at_micros,
        users: view
            .users
@@ -15,6 +15,7 @@ pub struct AuthStoreProjectionView {
    pub users: Vec<AuthStoreProjectionUser>,
    pub identities: Vec<AuthStoreProjectionIdentity>,
    pub refresh_sessions: Vec<AuthStoreProjectionRefreshSession>,
+    pub base_updated_at_micros: i64,
 }

 impl __sdk::InModule for AuthStoreProjectionView {
@@ -15,6 +15,7 @@ pub struct AuthStoreProjectionView {
    pub users: Vec<AuthStoreProjectionUser>,
    pub identities: Vec<AuthStoreProjectionIdentity>,
    pub refresh_sessions: Vec<AuthStoreProjectionRefreshSession>,
+    pub base_updated_at_micros: i64,
 }

 #[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
@@ -97,7 +98,11 @@ pub fn validate_auth_session(
    ctx: &mut ProcedureContext,
    input: AuthSessionValidationInput,
 ) -> AuthSessionValidationProcedureResult {
-    match ctx.try_with_tx(|tx| validate_auth_session_tx(tx, input.clone())) {
+    let caller = ctx.sender();
+    match ctx.try_with_tx(|tx| {
+        require_auth_service_identity(tx, caller)?;
+        validate_auth_session_tx(tx, input.clone())
+    }) {
        Ok(active) => AuthSessionValidationProcedureResult {
            active,
            error_message: None,
@@ -146,7 +151,11 @@ pub fn sync_auth_store_projection(
    ctx: &mut ProcedureContext,
    input: AuthStoreProjectionView,
 ) -> AuthStoreProjectionSyncProcedureResult {
-    match ctx.try_with_tx(|tx| sync_auth_store_projection_tx(tx, input.clone())) {
+    let caller = ctx.sender();
+    match ctx.try_with_tx(|tx| {
+        require_auth_service_identity(tx, caller)?;
+        sync_auth_store_projection_tx(tx, input.clone())
+    }) {
        Ok(record) => AuthStoreProjectionSyncProcedureResult {
            ok: true,
            record: Some(record),
@@ -164,7 +173,11 @@ pub fn sync_auth_store_projection(
 pub fn export_auth_store_projection_from_tables(
    ctx: &mut ProcedureContext,
 ) -> AuthStoreProjectionProcedureResult {
-    match ctx.try_with_tx(|tx| export_auth_store_projection_from_tables_tx(tx)) {
+    let caller = ctx.sender();
+    match ctx.try_with_tx(|tx| {
+        require_auth_service_identity(tx, caller)?;
+        export_auth_store_projection_from_tables_tx(tx)
+    }) {
        Ok(record) => AuthStoreProjectionProcedureResult {
            ok: true,
            record: Some(record),
@@ -178,6 +191,14 @@ pub fn export_auth_store_projection_from_tables(
    }
 }

+fn require_auth_service_identity(
+    ctx: &ReducerContext,
+    caller: crate::Identity,
+) -> Result<(), String> {
+    crate::editor_project_storage::require_editor_generation_runtime_service_identity(ctx, caller)
+        .map_err(|_| "当前 identity 无权调用认证服务".to_string())
+}
+
 fn sync_auth_store_projection_tx(
    ctx: &ReducerContext,
    input: AuthStoreProjectionView,
@@ -188,6 +209,10 @@ fn sync_auth_store_projection_tx(
        .meta_id()
        .find(&AUTH_STORE_PROJECTION_META_ID.to_string())
        .map(|row| row.updated_at.to_micros_since_unix_epoch());
+    ensure_auth_projection_base_version(
+        input.base_updated_at_micros,
+        current_updated_at_micros.unwrap_or_default(),
+    )?;
    ensure_newer_auth_projection_version(current_updated_at_micros, input.updated_at_micros)?;

    let user_ids = input
@@ -327,20 +352,31 @@ 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.
+/// Full projections are emitted by API-local auth worksets. The base version
+/// is read immediately before the write and checked in this same transaction,
+/// so a stale API node cannot replace data written by another node meanwhile.
+/// The timestamp remains a diagnostic/monotonic watermark for accepted writes.
+fn ensure_auth_projection_base_version(
+    expected_updated_at_micros: i64,
+    current_updated_at_micros: i64,
+) -> Result<(), String> {
+    if expected_updated_at_micros != current_updated_at_micros {
+        return Err(format!(
+            "认证投影基线版本冲突：请求基线 {expected_updated_at_micros} 不等于当前版本 {current_updated_at_micros}"
+        ));
+    }
+
+    Ok(())
+}
+
 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 {
+        if incoming_updated_at_micros <= current_updated_at_micros {
            return Err(format!(
-                "认证投影版本冲突：请求版本 {incoming_updated_at_micros} 早于当前版本 {current_updated_at_micros}"
+                "认证投影版本冲突：请求版本 {incoming_updated_at_micros} 不晚于当前版本 {current_updated_at_micros}"
            ));
        }
    }
@@ -410,6 +446,7 @@ fn export_auth_store_projection_from_tables_tx(
        .collect();

    Ok(AuthStoreProjectionView {
+        base_updated_at_micros: updated_at_micros,
        updated_at_micros,
        users,
        identities,
@@ -563,7 +600,14 @@ mod tests {
    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), 10).is_err());
        assert!(ensure_newer_auth_projection_version(Some(10), 9).is_err());
    }
+
+    #[test]
+    fn auth_projection_base_version_must_match_current_version() {
+        assert!(ensure_auth_projection_base_version(0, 0).is_ok());
+        assert!(ensure_auth_projection_base_version(10, 10).is_ok());
+        assert!(ensure_auth_projection_base_version(9, 10).is_err());
+    }
 }