fix(store): avoid revalidating unchanged request observations (#387)

Validate observation shape on insert and changed JSON only, preserving atomic account deletion and historical snapshots. Add upgrade, validation-cost, bulk deletion, and rollback coverage. Co-authored-by: Codex <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-10-11 10:30:04 +08:00
parent 9bf86f7a7c
commit 85c64d35d5
5 changed files with 289 additions and 0 deletions
@@ -0,0 +1,125 @@
+//! 验证大批历史记录下账号删除的原子性、快照保留与审计提交
+
+use super::{
+    DeleteProviderAccounts, PgProviderAccountRepository, ProviderAccountAdminRepository,
+    ProviderAccountAdminScope, ProviderAccountRepository, TestDatabase, account, account_count,
+    audit, current_revision,
+};
+
+async fn seed_history(pool: &sqlx::PgPool, account_id: &str, count: i32) {
+    sqlx::query(
+        "insert into model_requests (
+           id, client_api_key_ref, provider_account_id, provider_account_ref, provider_kind,
+           operation, client_transport, started_at, deadline_at, outcome, completed_at,
+           input_tokens, output_tokens, total_tokens, request_observation_json
+         ) select $1 || '_' || n, 'historical_key', $1, $1, 'openai', 'responses', 'http_sse',
+           now(), now() + interval '1 hour', 'failed', now(), 10, 5, 15,
+           jsonb_build_object(
+             'request', jsonb_build_object('configRevision', 1, 'protocol', 'openai',
+               'endpoint', '/v1/responses', 'compact', false),
+             'account', jsonb_build_object('name', $1::text, 'email', 'test@example.invalid',
+               'authenticationKind', 'oauth'),
+             'routing', jsonb_build_object('scope', 'all', 'groupRefs', '[]'::jsonb,
+               'groupNamesSnapshot', '[]'::jsonb))
+           from generate_series(1, $2::integer) n",
+    )
+    .bind(account_id)
+    .bind(count)
+    .execute(pool)
+    .await
+    .expect("seed synthetic history");
+}
+
+#[tokio::test]
+async fn bulk_account_deletion_preserves_large_history_and_commits_one_revision_and_audit() {
+    let Some(db) = TestDatabase::create("account_delete_history").await else {
+        return;
+    };
+    let repository = PgProviderAccountRepository::new(db.pool.clone());
+    for (id, count) in [("delete_a", 21_421), ("delete_b", 128), ("keep", 3)] {
+        repository
+            .insert_provider_account(account(id, id))
+            .await
+            .unwrap();
+        seed_history(&db.pool, id, count).await;
+    }
+    // 保存全部历史字段，只允许本次删除解除选中账号的实时外键引用
+    sqlx::query("create table history_before as select * from model_requests")
+        .execute(&db.pool)
+        .await
+        .unwrap();
+    let before_revision = current_revision(&db.pool).await;
+    let revision = repository
+        .delete_provider_accounts_admin(DeleteProviderAccounts {
+            scope: ProviderAccountAdminScope {
+                provider_kind: "openai".to_owned(),
+            },
+            account_ids: vec!["delete_a".to_owned(), "delete_b".to_owned()],
+            audit: audit("delete_history", "delete", "provider_accounts"),
+        })
+        .await
+        .expect("delete accounts with substantial history");
+    assert_eq!(revision.get(), before_revision as u64 + 1);
+    for id in ["delete_a", "delete_b"] {
+        assert_eq!(account_count(&db.pool, id).await, 0);
+    }
+    assert_eq!(account_count(&db.pool, "keep").await, 1);
+    let changed_history: i64 = sqlx::query_scalar(
+        "select count(*) from history_before b full join model_requests r using (id)
+         where (to_jsonb(b) - 'provider_account_id') is distinct from
+               (to_jsonb(r) - 'provider_account_id')
+            or r.provider_account_id is distinct from
+               case when b.provider_account_id = 'keep' then 'keep' else null end",
+    )
+    .fetch_one(&db.pool)
+    .await
+    .unwrap();
+    assert_eq!(changed_history, 0);
+    let audit_revisions: Vec<i64> = sqlx::query_scalar(
+        "select config_revision from admin_audit_events where id = 'delete_history'",
+    )
+    .fetch_all(&db.pool)
+    .await
+    .unwrap();
+    assert_eq!(audit_revisions, vec![before_revision + 1]);
+    db.close().await;
+}
+
+#[tokio::test]
+async fn invalid_bulk_deletion_rolls_back_accounts_history_revision_and_audit() {
+    let Some(db) = TestDatabase::create("account_delete_rollback").await else {
+        return;
+    };
+    let repository = PgProviderAccountRepository::new(db.pool.clone());
+    repository
+        .insert_provider_account(account("keep", "keep"))
+        .await
+        .unwrap();
+    seed_history(&db.pool, "keep", 3).await;
+    let before_revision = current_revision(&db.pool).await;
+    repository
+        .delete_provider_accounts_admin(DeleteProviderAccounts {
+            scope: ProviderAccountAdminScope {
+                provider_kind: "openai".to_owned(),
+            },
+            account_ids: vec!["keep".to_owned(), "missing".to_owned()],
+            audit: audit("failed_delete", "delete", "provider_accounts"),
+        })
+        .await
+        .expect_err("one missing account must roll back the entire batch");
+    assert_eq!(account_count(&db.pool, "keep").await, 1);
+    assert_eq!(current_revision(&db.pool).await, before_revision);
+    let retained: i64 = sqlx::query_scalar(
+        "select count(*) from model_requests where provider_account_id = 'keep'",
+    )
+    .fetch_one(&db.pool)
+    .await
+    .unwrap();
+    assert_eq!(retained, 3);
+    let audits: i64 = sqlx::query_scalar("select count(*) from admin_audit_events")
+        .fetch_one(&db.pool)
+        .await
+        .unwrap();
+    assert_eq!(audits, 0);
+    db.close().await;
+}
@@ -8,6 +8,7 @@ use std::{

 mod admin_adapter;
 mod authorization;
+mod deletion;
 mod quota_forecast;
 mod timestamps;

@@ -408,3 +408,138 @@ async fn request_observation_reject_malformed_documents_and_keep_upstream_clock_
    assert_eq!(metrics, (2.450638, 0.0, None));
    db.close().await;
 }
+
+#[tokio::test]
+async fn observation_validation_upgrade_preserves_history_and_rejects_invalid_writes() {
+    let Some(db) = TestDatabase::create_through("observation_validation_upgrade", 27).await else {
+        return;
+    };
+    seed_request(&db.pool).await;
+    let before: serde_json::Value =
+        sqlx::query_scalar("select to_jsonb(mr) from model_requests mr")
+            .fetch_one(&db.pool)
+            .await
+            .unwrap();
+
+    super::TEST_MIGRATOR.run(&db.pool).await.unwrap();
+    let after: serde_json::Value = sqlx::query_scalar("select to_jsonb(mr) from model_requests mr")
+        .fetch_one(&db.pool)
+        .await
+        .unwrap();
+    assert_eq!(after, before);
+
+    for statement in [
+        "insert into model_requests (
+           id, client_api_key_ref, operation, client_transport, started_at, deadline_at,
+           outcome, completed_at, request_observation_json
+         ) select 'invalid_insert', client_api_key_ref, operation, client_transport,
+           started_at, deadline_at, outcome, completed_at,
+           request_observation_json || '{\"unknown\":true}'::jsonb from model_requests",
+        "update model_requests set request_observation_json =
+           request_observation_json || '{\"unknown\":true}'::jsonb",
+    ] {
+        let error = sqlx::query(sqlx::AssertSqlSafe(statement))
+            .execute(&db.pool)
+            .await
+            .expect_err("invalid observations must still fail immediately");
+        assert_check_rejected(&error);
+        assert_eq!(
+            error.as_database_error().unwrap().constraint(),
+            Some("model_requests_observation_shape_ck")
+        );
+    }
+    let rows: Vec<serde_json::Value> =
+        sqlx::query_scalar("select to_jsonb(mr) from model_requests mr")
+            .fetch_all(&db.pool)
+            .await
+            .unwrap();
+    assert_eq!(rows, vec![before]);
+    db.close().await;
+}
+
+#[tokio::test]
+async fn unchanged_observations_do_not_repeat_full_document_validation() {
+    let Some(db) = TestDatabase::create("observation_validation_cost").await else {
+        return;
+    };
+    // 统计原校验器调用次数，避免性能回归断言依赖 CI 机器的运行速度
+    sqlx::query("create sequence observation_validation_calls")
+        .execute(&db.pool)
+        .await
+        .unwrap();
+    let definition: String = sqlx::query_scalar(
+        "select pg_get_functiondef('model_request_observation_valid(jsonb)'::regprocedure)",
+    )
+    .fetch_one(&db.pool)
+    .await
+    .unwrap();
+    let instrumented = definition.replacen(
+        "begin\n",
+        "begin\n  perform nextval('observation_validation_calls');\n",
+        1,
+    );
+    assert_ne!(instrumented, definition);
+    sqlx::raw_sql(sqlx::AssertSqlSafe(instrumented))
+        .execute(&db.pool)
+        .await
+        .unwrap();
+    seed_request(&db.pool).await;
+    sqlx::query("update model_requests set deadline_at = deadline_at + interval '1 second'")
+        .execute(&db.pool)
+        .await
+        .unwrap();
+    sqlx::query("update model_requests set request_observation_json = request_observation_json")
+        .execute(&db.pool)
+        .await
+        .unwrap();
+    let calls: i64 = sqlx::query_scalar("select last_value from observation_validation_calls")
+        .fetch_one(&db.pool)
+        .await
+        .unwrap();
+    assert_eq!(calls, 1, "only the initial insert needs full validation");
+
+    sqlx::query(
+        "update model_requests set request_observation_json = jsonb_set(
+           request_observation_json, '{request,userAgent}', '\"changed\"'::jsonb)",
+    )
+    .execute(&db.pool)
+    .await
+    .unwrap();
+    let calls: i64 = sqlx::query_scalar("select last_value from observation_validation_calls")
+        .fetch_one(&db.pool)
+        .await
+        .unwrap();
+    assert_eq!(
+        calls, 2,
+        "changed observations must run the original validator"
+    );
+    db.close().await;
+}
+
+#[tokio::test]
+async fn observation_validation_checks_the_final_row_after_before_triggers() {
+    let Some(db) = TestDatabase::create("observation_validation_final_row").await else {
+        return;
+    };
+    seed_request(&db.pool).await;
+    sqlx::raw_sql(
+        "create function change_observation_before_update() returns trigger language plpgsql as $$
+         begin
+           new.request_observation_json := new.request_observation_json || '{\"unknown\":true}'::jsonb;
+           return new;
+         end;
+         $$;
+         create trigger change_observation_before_update before update on model_requests
+           for each row execute function change_observation_before_update();",
+    )
+    .execute(&db.pool)
+    .await
+    .unwrap();
+    let error =
+        sqlx::query("update model_requests set deadline_at = deadline_at + interval '1 second'")
+            .execute(&db.pool)
+            .await
+            .expect_err("indirect JSON changes must not bypass validation");
+    assert_check_rejected(&error);
+    db.close().await;
+}
@@ -25,3 +25,4 @@ fa99dde0fe0b3d7da6693d76b2735629dde0dc255ea8c53f0354857a4e80669b  0024_preferred
 299d347add09b1e6f086732c0db8269efb640db4ecccb948e17c067b572afc43  0025_request_observation.sql
 d4b599066d7382a1c9cb6b3c003200046579e4fce3531f232cbd665a9d92c542  0026_codex_privacy_policy.sql
 8c9bbecd8315a26e928213a82a2ec651c35976e0643b82caef1d1bcfeee2916f  0027_request_observation_projection.sql
+b22b914d64291af7a6227083721fe608cc33fe0cedb9539b86bc6dbf7c045849  0028_request_observation_validation.sql
@@ -0,0 +1,27 @@
+-- 观测 JSON 只在插入或内容变化时校验，外键置空不重复遍历历史快照
+create function check_model_request_observation() returns trigger
+language plpgsql as $$
+begin
+  if model_request_observation_valid(new.request_observation_json) is not true then
+    raise exception using
+      errcode = '23514',
+      message = 'request observation violates shape constraint',
+      constraint = 'model_requests_observation_shape_ck',
+      table = tg_table_name,
+      schema = tg_table_schema;
+  end if;
+  return null;
+end;
+$$;
+
+-- AFTER 检查最终写入值，也覆盖其他 BEFORE 触发器造成的 JSON 变化
+create trigger model_requests_observation_insert
+  after insert on model_requests
+  for each row execute function check_model_request_observation();
+create trigger model_requests_observation_update
+  after update on model_requests
+  for each row
+  when (old.request_observation_json is distinct from new.request_observation_json)
+  execute function check_model_request_observation();
+
+alter table model_requests drop constraint model_requests_observation_shape_ck;