fix(data): 解耦候选记录的 API Key 历史身份

crates/aether-data/adapters/postgres/migrations/20260718000000_decouple_request_candidate_api_key_foreign_key.sql

@@ -0,0 +1,5 @@
+-- request_candidates 记录请求路由的历史事实。API Key 删除后，已有记录与迟到的
+-- 异步候选写入仍须保留原始身份快照，不能依赖当前认证目录中的可变行。
+
+ALTER TABLE ONLY public.request_candidates
+  DROP CONSTRAINT IF EXISTS request_candidates_api_key_id_fkey;

crates/aether-data/adapters/postgres/src/auth.rs

@@ -797,18 +797,6 @@ WHERE id = $1
   AND is_standalone = TRUE
 "#;
 
-const NULL_USAGE_API_KEY_FK_SQL: &str = r#"
-UPDATE usage
-SET api_key_id = NULL
-WHERE api_key_id = $1
-"#;
-
-const NULL_REQUEST_CANDIDATE_API_KEY_FK_SQL: &str = r#"
-UPDATE request_candidates
-SET api_key_id = NULL
-WHERE api_key_id = $1
-"#;
-
 const DISABLE_WALLET_BY_API_KEY_ID_SQL: &str = r#"
 UPDATE wallets
 SET status = 'disabled',
@@ -1493,16 +1481,6 @@ impl AuthApiKeyWriteRepository for SqlxAuthApiKeySnapshotReadRepository {
         api_key_id: &str,
     ) -> Result<bool, DataLayerError> {
         let mut tx = self.pool.begin().await.map_postgres_err()?;
-        sqlx::query(NULL_USAGE_API_KEY_FK_SQL)
-            .bind(api_key_id)
-            .execute(&mut *tx)
-            .await
-            .map_postgres_err()?;
-        sqlx::query(NULL_REQUEST_CANDIDATE_API_KEY_FK_SQL)
-            .bind(api_key_id)
-            .execute(&mut *tx)
-            .await
-            .map_postgres_err()?;
         sqlx::query(DISABLE_WALLET_BY_API_KEY_ID_SQL)
             .bind(api_key_id)
             .execute(&mut *tx)
@@ -1542,16 +1520,6 @@ impl AuthApiKeyWriteRepository for SqlxAuthApiKeySnapshotReadRepository {
 
     async fn delete_standalone_api_key(&self, api_key_id: &str) -> Result<bool, DataLayerError> {
         let mut tx = self.pool.begin().await.map_postgres_err()?;
-        sqlx::query(NULL_USAGE_API_KEY_FK_SQL)
-            .bind(api_key_id)
-            .execute(&mut *tx)
-            .await
-            .map_postgres_err()?;
-        sqlx::query(NULL_REQUEST_CANDIDATE_API_KEY_FK_SQL)
-            .bind(api_key_id)
-            .execute(&mut *tx)
-            .await
-            .map_postgres_err()?;
         sqlx::query(DISABLE_WALLET_BY_API_KEY_ID_SQL)
             .bind(api_key_id)
             .execute(&mut *tx)

crates/aether-data/runtime/schema/bootstrap/postgres/005_foreign_keys.sql

@@ -492,21 +492,6 @@ END $mig$;
 
 
 
---
--- Name: request_candidates request_candidates_api_key_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
---
-
-DO $mig$ BEGIN
-  ALTER TABLE ONLY public.request_candidates
-    ADD CONSTRAINT request_candidates_api_key_id_fkey FOREIGN KEY (api_key_id) REFERENCES public.api_keys(id) ON DELETE SET NULL;
-EXCEPTION
-  WHEN duplicate_object THEN NULL;
-  WHEN duplicate_table THEN NULL;
-  WHEN invalid_table_definition THEN NULL;
-END $mig$;
-
-
-
 --
 -- Name: request_candidates request_candidates_endpoint_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
 --

crates/aether-data/runtime/src/lifecycle/bootstrap/postgres.rs

@@ -7,7 +7,7 @@ use tracing::info;
 // Generated by build.rs from schema/bootstrap/postgres.
 pub(crate) static EMPTY_DATABASE_SNAPSHOT_SQL: &str =
     include_str!(concat!(env!("OUT_DIR"), "/empty_database_snapshot.sql"));
-pub(crate) const EMPTY_DATABASE_SNAPSHOT_CUTOFF_VERSION: i64 = 20260716000000;
+pub(crate) const EMPTY_DATABASE_SNAPSHOT_CUTOFF_VERSION: i64 = 20260718000000;
 
 const PUBLIC_BASE_TABLE_COUNT_SQL: &str = r#"
 SELECT COUNT(*)::BIGINT

crates/aether-data/runtime/src/lifecycle/migrate/tests.rs

@@ -6,9 +6,12 @@ use std::process::{Child, Command, Stdio};
 use std::time::{Duration, Instant};
 
 use sqlx::{
-    migrate::AppliedMigration, query, query_scalar, Connection, PgConnection, PgPool, SqlitePool,
+    migrate::{AppliedMigration, Migrate},
+    query, query_scalar, Connection, PgConnection, PgPool, SqlitePool,
 };
 
+use aether_data_contracts::repository::auth::AuthApiKeyWriteRepository;
+
 use super::{
     postgres::{all_up_migrations, pending_migrations_from_applied, POSTGRES_MIGRATOR},
     prepare_database_for_startup,
@@ -326,6 +329,7 @@ fn empty_database_snapshot_covers_current_cutoff_versions() {
             20260715130000,
             20260715130100,
             20260716000000,
+            20260718000000,
         ]
     );
 }
@@ -496,6 +500,25 @@ fn usage_identity_foreign_keys_are_decoupled_for_historical_ingestion() {
     }
 }
 
+#[test]
+fn request_candidate_api_key_identity_is_decoupled_for_historical_ingestion() {
+    let migration = POSTGRES_MIGRATOR
+        .iter()
+        .find(|migration| migration.version == 20260718000000)
+        .expect("request candidate API key identity decoupling migration should be embedded");
+
+    assert!(
+        migration
+            .sql
+            .contains("DROP CONSTRAINT IF EXISTS request_candidates_api_key_id_fkey"),
+        "migration should drop request_candidates_api_key_id_fkey"
+    );
+    assert!(
+        !EMPTY_DATABASE_SNAPSHOT_SQL.contains("ADD CONSTRAINT request_candidates_api_key_id_fkey"),
+        "fresh bootstrap snapshot should not recreate request_candidates_api_key_id_fkey"
+    );
+}
+
 #[test]
 fn empty_database_snapshot_sql_includes_payment_gateway_and_plans() {
     assert!(EMPTY_DATABASE_SNAPSHOT_SQL.contains("payment_provider character varying(64)"));
@@ -1579,6 +1602,7 @@ fn pending_migrations_from_applied_skips_versions_already_applied() {
             20260715130000,
             20260715130100,
             20260716000000,
+            20260718000000,
         ]
     );
 }
@@ -2121,6 +2145,282 @@ async fn prepare_database_for_startup_bootstraps_clean_database() {
     );
 }
 
+#[tokio::test]
+async fn postgres_request_candidates_preserve_deleted_api_key_identity() {
+    let Some(server) = ManagedPostgresServer::try_start()
+        .await
+        .expect("postgres request candidate lifecycle test should start or skip")
+    else {
+        return;
+    };
+
+    let pool = PgPool::connect(server.database_url())
+        .await
+        .expect("pool should connect");
+    let pending = prepare_database_for_startup(&pool)
+        .await
+        .expect("clean database bootstrap should succeed");
+    assert!(
+        pending.is_empty(),
+        "clean database bootstrap should not leave pending migrations: {pending:?}"
+    );
+
+    query(
+        r#"
+INSERT INTO public.users (id, username, email_verified)
+VALUES ('request-candidate-user', 'request-candidate-user', TRUE)
+"#,
+    )
+    .execute(&pool)
+    .await
+    .expect("request candidate user fixture should be inserted");
+    query(
+        r#"
+INSERT INTO public.api_keys (id, user_id, key_hash)
+VALUES ('deleted-api-key', 'request-candidate-user', 'request-candidate-key-hash')
+"#,
+    )
+    .execute(&pool)
+    .await
+    .expect("request candidate API key fixture should be inserted");
+    query(
+        r#"
+INSERT INTO public.request_candidates (
+  id,
+  request_id,
+  api_key_id,
+  candidate_index,
+  status
+) VALUES (
+  'request-candidate-before-delete',
+  'request-before-delete',
+  'deleted-api-key',
+  0,
+  'pending'
+)
+"#,
+    )
+    .execute(&pool)
+    .await
+    .expect("request candidate fixture should be inserted");
+    query(
+        r#"
+INSERT INTO public.usage (
+  id,
+  request_id,
+  api_key_id,
+  provider_name,
+  model
+) VALUES (
+  'usage-before-api-key-delete',
+  'usage-request-before-api-key-delete',
+  'deleted-api-key',
+  'historical-provider',
+  'historical-model'
+)
+"#,
+    )
+    .execute(&pool)
+    .await
+    .expect("usage fixture should be inserted");
+
+    let repository = aether_data_postgres::SqlxAuthApiKeySnapshotReadRepository::new(pool.clone());
+    assert!(repository
+        .delete_user_api_key("request-candidate-user", "deleted-api-key")
+        .await
+        .expect("API key deletion should succeed"));
+
+    let api_key_id: Option<String> = query_scalar(
+        "SELECT api_key_id FROM public.request_candidates WHERE id = 'request-candidate-before-delete'",
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("request candidate API key identity should be readable");
+    assert_eq!(api_key_id.as_deref(), Some("deleted-api-key"));
+
+    let usage_api_key_id: Option<String> = query_scalar(
+        "SELECT api_key_id FROM public.usage WHERE id = 'usage-before-api-key-delete'",
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("usage API key identity should be readable");
+    assert_eq!(usage_api_key_id.as_deref(), Some("deleted-api-key"));
+
+    query(
+        r#"
+INSERT INTO public.request_candidates (
+  id,
+  request_id,
+  api_key_id,
+  candidate_index,
+  status
+) VALUES (
+  'request-candidate-after-delete',
+  'request-after-delete',
+  'deleted-api-key',
+  0,
+  'pending'
+)
+"#,
+    )
+    .execute(&pool)
+    .await
+    .expect("late request candidate should preserve a deleted API key identity");
+
+    let late_api_key_id: Option<String> = query_scalar(
+        "SELECT api_key_id FROM public.request_candidates WHERE id = 'request-candidate-after-delete'",
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("late request candidate API key identity should be readable");
+    assert_eq!(late_api_key_id.as_deref(), Some("deleted-api-key"));
+
+    query(
+        r#"
+INSERT INTO public.api_keys (id, user_id, key_hash, is_standalone)
+VALUES (
+  'deleted-standalone-api-key',
+  'request-candidate-user',
+  'request-candidate-standalone-hash',
+  TRUE
+)
+"#,
+    )
+    .execute(&pool)
+    .await
+    .expect("standalone API key fixture should be inserted");
+    query(
+        r#"
+INSERT INTO public.request_candidates (
+  id,
+  request_id,
+  api_key_id,
+  candidate_index,
+  status
+) VALUES (
+  'candidate-standalone-before-delete',
+  'standalone-before-delete',
+  'deleted-standalone-api-key',
+  0,
+  'pending'
+)
+"#,
+    )
+    .execute(&pool)
+    .await
+    .expect("standalone request candidate fixture should be inserted");
+
+    assert!(repository
+        .delete_standalone_api_key("deleted-standalone-api-key")
+        .await
+        .expect("standalone API key deletion should succeed"));
+
+    let standalone_api_key_id: Option<String> = query_scalar(
+        "SELECT api_key_id FROM public.request_candidates WHERE id = 'candidate-standalone-before-delete'",
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("standalone request candidate API key identity should be readable");
+    assert_eq!(
+        standalone_api_key_id.as_deref(),
+        Some("deleted-standalone-api-key")
+    );
+}
+
+#[tokio::test]
+async fn postgres_request_candidate_migration_decouples_legacy_api_key_foreign_key() {
+    const PREVIOUS_SNAPSHOT_CUTOFF_VERSION: i64 = 20260716000000;
+
+    let Some(server) = ManagedPostgresServer::try_start()
+        .await
+        .expect("postgres request candidate migration test should start or skip")
+    else {
+        return;
+    };
+
+    let mut conn = PgConnection::connect(server.database_url())
+        .await
+        .expect("postgres migration connection should open");
+    conn.ensure_migrations_table()
+        .await
+        .expect("migration table should be created");
+    for migration in POSTGRES_MIGRATOR
+        .iter()
+        .filter(|migration| migration.version <= PREVIOUS_SNAPSHOT_CUTOFF_VERSION)
+    {
+        conn.apply(migration)
+            .await
+            .expect("legacy postgres migration should apply");
+    }
+    drop(conn);
+
+    let pool = PgPool::connect(server.database_url())
+        .await
+        .expect("pool should connect");
+    let legacy_constraint_exists: bool = query_scalar(
+        r#"
+SELECT EXISTS (
+  SELECT 1
+  FROM pg_catalog.pg_constraint
+  WHERE conname = 'request_candidates_api_key_id_fkey'
+    AND conrelid = 'public.request_candidates'::regclass
+)
+"#,
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("legacy request candidate constraint should be readable");
+    assert!(legacy_constraint_exists);
+
+    super::run_migrations(&pool)
+        .await
+        .expect("request candidate API key decoupling migration should apply");
+
+    let upgraded_constraint_exists: bool = query_scalar(
+        r#"
+SELECT EXISTS (
+  SELECT 1
+  FROM pg_catalog.pg_constraint
+  WHERE conname = 'request_candidates_api_key_id_fkey'
+    AND conrelid = 'public.request_candidates'::regclass
+)
+"#,
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("upgraded request candidate constraint should be readable");
+    assert!(!upgraded_constraint_exists);
+
+    query(
+        r#"
+INSERT INTO public.request_candidates (
+  id,
+  request_id,
+  api_key_id,
+  candidate_index,
+  status
+) VALUES (
+  'legacy-late-request-candidate',
+  'legacy-late-request',
+  'legacy-deleted-api-key',
+  0,
+  'pending'
+)
+"#,
+    )
+    .execute(&pool)
+    .await
+    .expect("upgraded schema should accept a late historical API key identity");
+
+    let api_key_id: Option<String> = query_scalar(
+        "SELECT api_key_id FROM public.request_candidates WHERE id = 'legacy-late-request-candidate'",
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("upgraded request candidate identity should be readable");
+    assert_eq!(api_key_id.as_deref(), Some("legacy-deleted-api-key"));
+}
+
 #[tokio::test]
 async fn postgres_usage_billing_facts_total_tokens_counts_cached_input_once() {
     const LEGACY_VIEW_MIGRATION_VERSION: i64 = 20260505130000;

docker-compose.yml

@@ -32,6 +32,12 @@ services:
       interval: 5s
       timeout: 5s
       retries: 5
+    logging:
+      driver: local
+      options:
+        max-size: "100m"
+        max-file: "3"
+        compress: "true"
     restart: unless-stopped
 
   redis: