From 85c64d35d5c8de8635afa909442b52251019dfda Mon Sep 17 00:00:00 2001 From: hi_yueban Date: Sun, 11 Oct 2026 10:30:04 +0800 Subject: [PATCH] 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 --- .../postgres/provider_accounts/deletion.rs | 125 ++++++++++++++++ .../tests/postgres/provider_accounts/mod.rs | 1 + .../tests/postgres/schema_integrity.rs | 135 ++++++++++++++++++ backend/migrations/.frozen-sha256 | 1 + .../0028_request_observation_validation.sql | 27 ++++ 5 files changed, 289 insertions(+) create mode 100644 backend/crates/gateway-store/tests/postgres/provider_accounts/deletion.rs create mode 100644 backend/migrations/0028_request_observation_validation.sql diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts/deletion.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts/deletion.rs new file mode 100644 index 000000000..d30d62f6f --- /dev/null +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts/deletion.rs @@ -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 = 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; +} diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs index f88c00135..b5bdecfa5 100644 --- a/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs @@ -8,6 +8,7 @@ use std::{ mod admin_adapter; mod authorization; +mod deletion; mod quota_forecast; mod timestamps; diff --git a/backend/crates/gateway-store/tests/postgres/schema_integrity.rs b/backend/crates/gateway-store/tests/postgres/schema_integrity.rs index ca3239f22..e28760a73 100644 --- a/backend/crates/gateway-store/tests/postgres/schema_integrity.rs +++ b/backend/crates/gateway-store/tests/postgres/schema_integrity.rs @@ -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 = + 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; +} diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index ca1b14e6a..9592bf1cc 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -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 diff --git a/backend/migrations/0028_request_observation_validation.sql b/backend/migrations/0028_request_observation_validation.sql new file mode 100644 index 000000000..94ebb9542 --- /dev/null +++ b/backend/migrations/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;