diff --git a/backend/crates/gateway-store/tests/postgres/schema_integrity.rs b/backend/crates/gateway-store/tests/postgres/schema_integrity.rs index 147933bc8..c3a032312 100644 --- a/backend/crates/gateway-store/tests/postgres/schema_integrity.rs +++ b/backend/crates/gateway-store/tests/postgres/schema_integrity.rs @@ -78,29 +78,6 @@ async fn seed_request(pool: &sqlx::PgPool) { .expect("seed request without optional facts"); } -async fn seed_request_at_27(pool: &sqlx::PgPool) { - sqlx::query( - "insert into model_requests ( - id, client_api_key_ref, operation, client_transport, started_at, deadline_at, outcome, completed_at, request_observation_json - ) values ( - 'req_integrity', 'deleted_key', 'responses', 'http_sse', now(), now() + interval '1 hour', 'failed', now(), - jsonb_strip_nulls(jsonb_build_object( - 'request', jsonb_build_object( - 'configRevision', 1, - 'protocol', 'openai', - 'endpoint', '/v1/responses', - 'compact', false), - 'routing', jsonb_build_object( - 'scope', 'all', - 'groupRefs', '{}'::text[], - 'groupNamesSnapshot', '[]'::jsonb))) - )", - ) - .execute(pool) - .await - .expect("seed request without optional facts"); -} - fn assert_check_rejected(error: &sqlx::Error) { assert_eq!( error @@ -430,54 +407,6 @@ async fn request_observation_reject_malformed_documents_and_keep_upstream_clock_ 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_at_27(&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_to(28, &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 { @@ -596,10 +525,10 @@ async fn observation_validation_checks_the_final_row_after_before_triggers() { #[tokio::test] async fn conditional_business_validation_preserves_each_constraint_and_cross_column_updates() { - let Some(db) = TestDatabase::create_through("observation_dependencies", 28).await else { + let Some(db) = TestDatabase::create("observation_dependencies").await else { return; }; - seed_request_at_27(&db.pool).await; + seed_request(&db.pool).await; let cases = [ ( "model_requests_upstream_duration_ck", @@ -645,25 +574,19 @@ async fn conditional_business_validation_preserves_each_constraint_and_cross_col "cost_source = 'calculated', cost_amount = 1", ), ]; - for version in [28, 29] { - super::TEST_MIGRATOR - .run_to(version, &db.pool) - .await - .unwrap(); - for (constraint, assignment) in cases { - let error = sqlx::query(sqlx::AssertSqlSafe(format!( - "update model_requests set {assignment} where id = 'req_integrity'" - ))) - .execute(&db.pool) - .await - .expect_err("invalid final row must fail"); - assert_check_rejected(&error); - assert_eq!( - error.as_database_error().unwrap().constraint(), - Some(constraint), - "migration {version}: {assignment}" - ); - } + for (constraint, assignment) in cases { + let error = sqlx::query(sqlx::AssertSqlSafe(format!( + "update model_requests set {assignment} where id = 'req_integrity'" + ))) + .execute(&db.pool) + .await + .expect_err("invalid final row must fail"); + assert_check_rejected(&error); + assert_eq!( + error.as_database_error().unwrap().constraint(), + Some(constraint), + "{assignment}" + ); } sqlx::query("update model_requests set continuation_requested = true, request_observation_json = request_observation_json || jsonb_build_object('continuation', jsonb_build_object('previousResponseIdHash', repeat('a',64)))") .execute(&db.pool).await.unwrap(); @@ -677,31 +600,3 @@ async fn conditional_business_validation_preserves_each_constraint_and_cross_col ); db.close().await; } - -#[tokio::test] -async fn lease_migration_moves_running_expiry_and_preserves_terminal_expiry() { - let Some(db) = TestDatabase::create_through("lease_upgrade", 30).await else { - return; - }; - seed_request_at_27(&db.pool).await; - sqlx::query("insert into model_requests (id, client_api_key_ref, operation, client_transport, started_at, deadline_at, request_observation_json) - select 'running_upgrade', client_api_key_ref, operation, client_transport, started_at, deadline_at, request_observation_json from model_requests where id = 'req_integrity'") - .execute(&db.pool).await.unwrap(); - let expiry: chrono::DateTime = - sqlx::query_scalar("select deadline_at from model_requests where id = 'running_upgrade'") - .fetch_one(&db.pool) - .await - .unwrap(); - super::TEST_MIGRATOR.run_to(31, &db.pool).await.unwrap(); - let running: (Option>, chrono::DateTime) = sqlx::query_as("select request.recovery_lease_expires_at, lease.lease_expires_at from model_requests request join model_request_leases lease on lease.request_id = request.id where request.id = 'running_upgrade'") - .fetch_one(&db.pool).await.unwrap(); - assert_eq!(running, (None, expiry)); - let terminal: chrono::DateTime = sqlx::query_scalar( - "select recovery_lease_expires_at from model_requests where id = 'req_integrity'", - ) - .fetch_one(&db.pool) - .await - .unwrap(); - assert_eq!(terminal, expiry); - db.close().await; -} diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index be2e31e98..5280d9c9c 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -25,11 +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 -a5e08257f7baf6c26181fa7e367cecd43de1ef9adfef737f51c626014705fbf5 0029_request_observation_dependencies.sql -f5831bbdfdfdf7beaf083ace8797bd42ad45ce7000cef2bd22a60db3d629f28d 0030_historical_request_identity.sql -8009684fac1a797d07173a5b2fcb3cb9bc466cae65b9bfeb854750fe3c1eca13 0031_request_recovery_leases.sql -e514ec3b88d8099464d09388ffc0e505009891d3905bc96250cf4d92249be784 0032_request_token_names.sql -c67e3215c5ce77e7dc7f05017bc7bfd12b295d212a87abb8fb41a95494277aeb 0033_client_key_charge_event_index.sql -99c3aca3e3ed4b52508d04b402b7f6cc6287388c789e4fab337034902f60cc71 0034_request_analytics.sql -d66b0476ad3e7e4d6a4bfa0eadc8f0e6e6c64e5417165aa785ef91510dbc85fe 0035_request_analytics_delete_invalidation.sql +31d6668707186d7c19cdd507e20162a08c7fbb8ba41ff4a4439d0a5e08369cad 0028_request_analytics.sql diff --git a/backend/migrations/0028_request_analytics.sql b/backend/migrations/0028_request_analytics.sql new file mode 100644 index 000000000..2c37dd715 --- /dev/null +++ b/backend/migrations/0028_request_analytics.sql @@ -0,0 +1,800 @@ +-- 请求历史身份、观测校验、恢复租约与可重建统计 +-- 迁移在停止旧写入者的窗口执行,旧二进制不可直接回退 + +create or replace function model_request_observation_valid(document jsonb) returns boolean +language plpgsql immutable parallel safe as $$ +declare + spec record; + item jsonb; + number numeric; +begin + if document is null or jsonb_typeof(document) <> 'object' + or octet_length(document::text) > 1048576 then + return false; + end if; + for spec in select * from (values + ('{}'::text[], array['account','continuation','error','recovery','request','response','routing','scheduling','timings','transport']::text[]), + ('{request}'::text[], array['clientApiKeyNameSnapshot','clientIp','compact','configRevision','endpoint','protocol','reasoningEffort','reasoningPreset','subagentKind','userAgent']::text[]), + ('{account}'::text[], array['authenticationKind','email','name','planTypeSnapshot','notesSnapshot']::text[]), + ('{routing}'::text[], array['groupNamesSnapshot','groupRefs','scope']::text[]), + ('{error}'::text[], array['message','providerErrorCode','retryAfterMs']::text[]), + ('{timings}'::text[], array['local','upstream']::text[]), + ('{timings,local}'::text[], array['connectMs','firstEventMs','firstReasoningMs','firstTextMs','firstTokenMs','headersMs','latencyMs','transportDecisionWaitMs']::text[]), + ('{timings,upstream}'::text[], array['processingMs','responseMs','apiOverheadMs','engineMs','engineIapiTtftMs','engineServiceTtftMs','engineIapiTbtMs','engineServiceTbtMs']::text[]), + ('{scheduling}'::text[], array['accountSelectionWaitMs','admissionDecisionMs','capacityTotalSlots','capacityUsedSlots']::text[]), + ('{transport}'::text[], array['connection','httpVersion','websocketPool']::text[]), + ('{transport,connection}'::text[], array['ageMs','exitReason','id','idleMs']::text[]), + ('{response}'::text[], array['responseModel']::text[]), + ('{continuation}'::text[], array['previousResponseIdHash','unavailableReason']::text[]), + ('{recovery}'::text[], array['retryDelayMs','totalLatencyMs']::text[])) as groups(path, allowed_keys) loop + item := document #> spec.path; + if item is not null and (jsonb_typeof(item) <> 'object' or item - spec.allowed_keys <> '{}'::jsonb) then + return false; + end if; + end loop; + for spec in select * from (values + ('{request,configRevision}'::text[], 'number'), + ('{request,protocol}'::text[], 'string'), + ('{request,endpoint}'::text[], 'string'), + ('{request,clientApiKeyNameSnapshot}'::text[], 'string'), + ('{account,planTypeSnapshot}'::text[], 'string'), + ('{account,notesSnapshot}'::text[], 'string'), + ('{account,name}'::text[], 'string'), + ('{account,email}'::text[], 'string'), + ('{account,authenticationKind}'::text[], 'string'), + ('{routing,scope}'::text[], 'string'), + ('{routing,groupRefs}'::text[], 'array'), + ('{routing,groupNamesSnapshot}'::text[], 'array'), + ('{error,providerErrorCode}'::text[], 'string'), + ('{error,message}'::text[], 'string'), + ('{error,retryAfterMs}'::text[], 'number'), + ('{request,clientIp}'::text[], 'string'), + ('{request,userAgent}'::text[], 'string'), + ('{request,reasoningEffort}'::text[], 'string'), + ('{request,reasoningPreset}'::text[], 'string'), + ('{request,subagentKind}'::text[], 'string'), + ('{request,compact}'::text[], 'boolean'), + ('{timings,local,transportDecisionWaitMs}'::text[], 'number'), + ('{timings,local,connectMs}'::text[], 'number'), + ('{timings,local,headersMs}'::text[], 'number'), + ('{timings,local,firstEventMs}'::text[], 'number'), + ('{timings,local,firstReasoningMs}'::text[], 'number'), + ('{timings,local,firstTextMs}'::text[], 'number'), + ('{timings,local,firstTokenMs}'::text[], 'number'), + ('{timings,local,latencyMs}'::text[], 'number'), + ('{timings,upstream,processingMs}'::text[], 'number'), + ('{timings,upstream,apiOverheadMs}'::text[], 'milliseconds'), + ('{timings,upstream,engineMs}'::text[], 'milliseconds'), + ('{timings,upstream,engineIapiTtftMs}'::text[], 'milliseconds'), + ('{timings,upstream,engineServiceTtftMs}'::text[], 'milliseconds'), + ('{timings,upstream,engineIapiTbtMs}'::text[], 'milliseconds'), + ('{timings,upstream,engineServiceTbtMs}'::text[], 'milliseconds'), + ('{scheduling,admissionDecisionMs}'::text[], 'number'), + ('{scheduling,accountSelectionWaitMs}'::text[], 'number'), + ('{scheduling,capacityUsedSlots}'::text[], 'number'), + ('{scheduling,capacityTotalSlots}'::text[], 'number'), + ('{transport,httpVersion}'::text[], 'string'), + ('{transport,websocketPool}'::text[], 'string'), + ('{transport,connection,id}'::text[], 'string'), + ('{transport,connection,exitReason}'::text[], 'string'), + ('{transport,connection,ageMs}'::text[], 'number'), + ('{transport,connection,idleMs}'::text[], 'number'), + ('{response,responseModel}'::text[], 'string'), + ('{continuation,previousResponseIdHash}'::text[], 'string'), + ('{continuation,unavailableReason}'::text[], 'string'), + ('{recovery,retryDelayMs}'::text[], 'number'), + ('{recovery,totalLatencyMs}'::text[], 'number'), + ('{timings,upstream,responseMs}'::text[], 'number')) as fields(path, kind) loop + item := document #> spec.path; + if item is null or item = 'null'::jsonb then continue; end if; + if spec.kind = 'milliseconds' then + if jsonb_typeof(item) <> 'number' then return false; end if; + number := (item #>> '{}')::numeric; + if number < 0 or number > 9223372036854775807 then return false; end if; + continue; + end if; + if jsonb_typeof(item) <> spec.kind then return false; end if; + if spec.kind = 'number' then + number := (item #>> '{}')::numeric; + if item #>> '{}' !~ '^[0-9]+$' or number > 9223372036854775807 then return false; end if; + elsif spec.kind = 'array' then + if exists (select 1 from jsonb_array_elements(item) member where jsonb_typeof(member) <> 'string') then return false; end if; + end if; + end loop; + if (document #> '{request,configRevision}' is not null + and document #> '{request,configRevision}' <> 'null'::jsonb + and document #> '{request,compact}' in ('true'::jsonb, 'false'::jsonb) + and coalesce(length(document #>> '{request,protocol}'), 0) > 0 + and coalesce(length(document #>> '{request,endpoint}'), 0) > 0 + and jsonb_typeof(document #> '{routing,scope}') = 'string' + and jsonb_typeof(document #> '{routing,groupRefs}') = 'array' + and jsonb_typeof(document #> '{routing,groupNamesSnapshot}') = 'array') is not true then + return false; + end if; + if document #>> '{request,clientIp}' is not null then + perform (document #>> '{request,clientIp}')::inet; + end if; + return true; +exception when invalid_text_representation or numeric_value_out_of_range then + return false; +end; +$$; + +-- 按最终行的真实依赖执行校验,纯状态写入不展开无关 JSON +create function check_model_request_observation() returns trigger +language plpgsql as $$ +declare + json_changed boolean := tg_op = 'INSERT' or old.request_observation_json is distinct from new.request_observation_json; +begin + if json_changed then + if model_request_observation_valid(new.request_observation_json) is not true then + raise exception using errcode = '23514', constraint = 'model_requests_observation_shape_ck', + message = 'request observation violates shape constraint', table = tg_table_name, schema = tg_table_schema; + end if; + end if; + if json_changed then + declare + timings_upstream_response_ms bigint; + begin + timings_upstream_response_ms := (new.request_observation_json #>> '{timings,upstream,responseMs}')::bigint; + if (timings_upstream_response_ms > 0) is false then + raise exception using errcode = '23514', constraint = 'model_requests_upstream_duration_ck', + message = 'request observation violates model_requests_upstream_duration_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end; + end if; + if json_changed then + if ((new.request_observation_json #>> '{request,configRevision}')::bigint > 0) is false then + raise exception using errcode = '23514', constraint = 'model_requests_revision_attempt_ck', message = 'request observation violates model_requests_revision_attempt_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end if; + if json_changed then + if ((new.request_observation_json #>> '{error,retryAfterMs}')::bigint is null or (new.request_observation_json #>> '{error,retryAfterMs}')::bigint >= 0) is false then + raise exception using errcode = '23514', constraint = 'model_requests_status_ck', message = 'request observation violates model_requests_status_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end if; + if json_changed then + declare + transport_websocket_pool text; + begin + transport_websocket_pool := (new.request_observation_json #>> '{transport,websocketPool}'); + if (transport_websocket_pool is null or transport_websocket_pool in ('new', 'reuse')) is false then + raise exception using errcode = '23514', constraint = 'model_requests_websocket_pool_ck', + message = 'request observation violates model_requests_websocket_pool_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end; + end if; + if json_changed then + declare + routing_group_names_snapshot jsonb; + routing_scope text; + routing_group_refs jsonb; + begin + routing_group_names_snapshot := (new.request_observation_json #> '{routing,groupNamesSnapshot}'); + routing_scope := (new.request_observation_json #>> '{routing,scope}'); + routing_group_refs := (new.request_observation_json #> '{routing,groupRefs}'); + if (jsonb_typeof(routing_group_names_snapshot) = 'array' + and ( + ( + routing_scope in ('legacy_provider', 'all') + and jsonb_array_length(routing_group_refs) = 0 + and jsonb_array_length(routing_group_names_snapshot) = 0 + ) + or ( + routing_scope = 'groups' + and jsonb_array_length(routing_group_refs) > 0 + and not jsonb_path_exists(new.request_observation_json, '$.routing.groupRefs[*] ? (@ == null)') + and jsonb_array_length(routing_group_names_snapshot) + = jsonb_array_length(routing_group_refs) + ) + )) is false then + raise exception using errcode = '23514', constraint = 'model_requests_routing_group_names_ck', + message = 'request observation violates model_requests_routing_group_names_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end; + end if; + if json_changed then + declare + routing_scope text; + begin + routing_scope := (new.request_observation_json #>> '{routing,scope}'); + if (routing_scope in ('legacy_provider', 'all', 'groups')) is false then + raise exception using errcode = '23514', constraint = 'model_requests_routing_scope_ck', + message = 'request observation violates model_requests_routing_scope_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end; + end if; + if json_changed then + declare + timings_local_transport_decision_wait_ms bigint; + timings_local_connect_ms bigint; + timings_local_headers_ms bigint; + timings_local_first_event_ms bigint; + timings_local_first_reasoning_ms bigint; + timings_local_first_text_ms bigint; + timings_local_first_token_ms bigint; + timings_upstream_processing_ms bigint; + timings_local_latency_ms bigint; + begin + timings_local_transport_decision_wait_ms := (new.request_observation_json #>> '{timings,local,transportDecisionWaitMs}')::bigint; + timings_local_connect_ms := (new.request_observation_json #>> '{timings,local,connectMs}')::bigint; + timings_local_headers_ms := (new.request_observation_json #>> '{timings,local,headersMs}')::bigint; + timings_local_first_event_ms := (new.request_observation_json #>> '{timings,local,firstEventMs}')::bigint; + timings_local_first_reasoning_ms := (new.request_observation_json #>> '{timings,local,firstReasoningMs}')::bigint; + timings_local_first_text_ms := (new.request_observation_json #>> '{timings,local,firstTextMs}')::bigint; + timings_local_first_token_ms := (new.request_observation_json #>> '{timings,local,firstTokenMs}')::bigint; + timings_upstream_processing_ms := (new.request_observation_json #>> '{timings,upstream,processingMs}')::bigint; + timings_local_latency_ms := (new.request_observation_json #>> '{timings,local,latencyMs}')::bigint; + if ((timings_local_transport_decision_wait_ms is null or timings_local_transport_decision_wait_ms >= 0) + and (timings_local_connect_ms is null or timings_local_connect_ms >= 0) + and (timings_local_headers_ms is null or timings_local_headers_ms >= 0) + and (timings_local_first_event_ms is null or timings_local_first_event_ms >= 0) + and (timings_local_first_reasoning_ms is null or timings_local_first_reasoning_ms >= 0) + and (timings_local_first_text_ms is null or timings_local_first_text_ms >= 0) + and (timings_local_first_token_ms is null or timings_local_first_token_ms >= 0) + and (timings_upstream_processing_ms is null or timings_upstream_processing_ms >= 0) + and (timings_local_latency_ms is null or timings_local_latency_ms >= 0) + and ( + timings_local_latency_ms is null + or ( + (timings_local_transport_decision_wait_ms is null or timings_local_transport_decision_wait_ms <= timings_local_latency_ms) + and (timings_local_connect_ms is null or timings_local_connect_ms <= timings_local_latency_ms) + and (timings_local_headers_ms is null or timings_local_headers_ms <= timings_local_latency_ms) + and (timings_local_first_event_ms is null or timings_local_first_event_ms <= timings_local_latency_ms) + and (timings_local_first_reasoning_ms is null or timings_local_first_reasoning_ms <= timings_local_latency_ms) + and (timings_local_first_text_ms is null or timings_local_first_text_ms <= timings_local_latency_ms) + and (timings_local_first_token_ms is null or timings_local_first_token_ms <= timings_local_latency_ms) + ) + )) is false then + raise exception using errcode = '23514', constraint = 'model_requests_latency_ck', + message = 'request observation violates model_requests_latency_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end; + end if; + if json_changed then + declare + scheduling_admission_decision_ms bigint; + scheduling_account_selection_wait_ms bigint; + scheduling_capacity_used_slots bigint; + scheduling_capacity_total_slots bigint; + begin + scheduling_admission_decision_ms := (new.request_observation_json #>> '{scheduling,admissionDecisionMs}')::bigint; + scheduling_account_selection_wait_ms := (new.request_observation_json #>> '{scheduling,accountSelectionWaitMs}')::bigint; + scheduling_capacity_used_slots := (new.request_observation_json #>> '{scheduling,capacityUsedSlots}')::bigint; + scheduling_capacity_total_slots := (new.request_observation_json #>> '{scheduling,capacityTotalSlots}')::bigint; + if ((scheduling_admission_decision_ms is null or scheduling_admission_decision_ms >= 0) + and (scheduling_account_selection_wait_ms is null or scheduling_account_selection_wait_ms >= 0) + and ( + (scheduling_capacity_used_slots is null and scheduling_capacity_total_slots is null) + or ( + scheduling_capacity_used_slots >= 0 + and scheduling_capacity_total_slots > 0 + and scheduling_capacity_used_slots <= scheduling_capacity_total_slots + ) + )) is false then + raise exception using errcode = '23514', constraint = 'model_requests_runtime_pressure_ck', + message = 'request observation violates model_requests_runtime_pressure_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end; + end if; + if json_changed or old.continuation_requested is distinct from new.continuation_requested then + if (((new.request_observation_json #>> '{continuation,previousResponseIdHash}') is null or (new.request_observation_json #>> '{continuation,previousResponseIdHash}') ~ '^[0-9a-f]{64}$') and (new.continuation_requested or (new.request_observation_json #>> '{continuation,previousResponseIdHash}') is null)) is false then + raise exception using errcode = '23514', constraint = 'model_requests_continuation_hashes_ck', message = 'request observation violates model_requests_continuation_hashes_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end if; + if json_changed then + declare + continuation_unavailable_reason text; + begin + continuation_unavailable_reason := (new.request_observation_json #>> '{continuation,unavailableReason}'); + if (continuation_unavailable_reason is null + or ( + octet_length(continuation_unavailable_reason) between 1 and 64 + and continuation_unavailable_reason ~ '^[a-z][a-z0-9_]*$' + )) is false then + raise exception using errcode = '23514', constraint = 'model_requests_continuation_reason_ck', + message = 'request observation violates model_requests_continuation_reason_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end; + end if; + if json_changed then + declare + transport_connection_id text; + transport_connection_exit_reason text; + transport_connection_age_ms bigint; + transport_connection_idle_ms bigint; + begin + transport_connection_id := (new.request_observation_json #>> '{transport,connection,id}'); + transport_connection_exit_reason := (new.request_observation_json #>> '{transport,connection,exitReason}'); + transport_connection_age_ms := (new.request_observation_json #>> '{transport,connection,ageMs}')::bigint; + transport_connection_idle_ms := (new.request_observation_json #>> '{transport,connection,idleMs}')::bigint; + if (( + transport_connection_id is null + and transport_connection_exit_reason is null + and transport_connection_age_ms is null + and transport_connection_idle_ms is null + ) + or ( + octet_length(transport_connection_id) between 1 and 128 + and transport_connection_id !~ '[[:cntrl:]]' + and octet_length(transport_connection_exit_reason) between 1 and 64 + and transport_connection_exit_reason ~ '^[a-z][a-z0-9_]*$' + and transport_connection_age_ms >= 0 + and transport_connection_idle_ms between 0 and transport_connection_age_ms + )) is false then + raise exception using errcode = '23514', constraint = 'model_requests_upstream_connection_ck', + message = 'request observation violates model_requests_upstream_connection_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end; + end if; + if json_changed or old.completed_at is distinct from new.completed_at or old.recovered_at is distinct from new.recovered_at or old.recovery_attempt_count is distinct from new.recovery_attempt_count or old.recovery_request_id is distinct from new.recovery_request_id then + declare + recovery_retry_delay_ms bigint; + recovery_total_latency_ms bigint; + begin + recovery_retry_delay_ms := (new.request_observation_json #>> '{recovery,retryDelayMs}')::bigint; + recovery_total_latency_ms := (new.request_observation_json #>> '{recovery,totalLatencyMs}')::bigint; + if (new.recovery_attempt_count >= 0 + and ( + ( + new.recovered_at is null + and new.recovery_request_id is null + and recovery_retry_delay_ms is null + and recovery_total_latency_ms is null + ) + or ( + new.recovered_at is not null + and new.recovered_at >= new.completed_at + and new.recovery_request_id is not null + and octet_length(new.recovery_request_id) between 1 and 128 + and new.recovery_attempt_count > 0 + and recovery_retry_delay_ms >= 0 + and recovery_total_latency_ms >= recovery_retry_delay_ms + ) + )) is false then + raise exception using errcode = '23514', constraint = 'model_requests_recovery_ck', + message = 'request observation violates model_requests_recovery_ck', table = tg_table_name, schema = tg_table_schema; + end if; + end; + end if; + if json_changed then + if (num_nonnulls(new.request_observation_json #>> '{scheduling,capacityUsedSlots}', new.request_observation_json #>> '{scheduling,capacityTotalSlots}') in (0, 2) + and num_nonnulls(new.request_observation_json #>> '{transport,connection,id}', new.request_observation_json #>> '{transport,connection,exitReason}', + new.request_observation_json #>> '{transport,connection,ageMs}', new.request_observation_json #>> '{transport,connection,idleMs}') in (0, 4)) is false then + raise exception using errcode = '23514', constraint = 'model_requests_fact_completeness_ck', + message = 'request observation violates model_requests_fact_completeness_ck', table = tg_table_name, schema = tg_table_schema; + + end if; + end if; + if json_changed or old.recovered_at is distinct from new.recovered_at or old.completed_at is distinct from new.completed_at then + if (new.recovered_at is null or (new.completed_at is not null + and new.request_observation_json #>> '{recovery,retryDelayMs}' is not null + and new.request_observation_json #>> '{recovery,totalLatencyMs}' is not null)) is false then + raise exception using errcode = '23514', constraint = 'model_requests_fact_completeness_ck', + message = 'request observation violates model_requests_fact_completeness_ck', table = tg_table_name, schema = tg_table_schema; + + end if; + end if; + return null; +end; +$$; + +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 + or old.completed_at is distinct from new.completed_at + or old.continuation_requested is distinct from new.continuation_requested + or old.recovered_at is distinct from new.recovered_at + or old.recovery_attempt_count is distinct from new.recovery_attempt_count + or old.recovery_request_id is distinct from new.recovery_request_id +) execute function check_model_request_observation(); + +alter table model_requests + drop constraint model_requests_observation_shape_ck, + drop constraint model_requests_upstream_duration_ck, + drop constraint model_requests_revision_attempt_ck, + drop constraint model_requests_status_ck, + drop constraint model_requests_websocket_pool_ck, + drop constraint model_requests_routing_scope_ck, + drop constraint model_requests_routing_group_names_ck, + drop constraint model_requests_latency_ck, + drop constraint model_requests_runtime_pressure_ck, + drop constraint model_requests_continuation_hashes_ck, + drop constraint model_requests_continuation_reason_ck, + drop constraint model_requests_upstream_connection_ck, + drop constraint model_requests_recovery_ck, + drop constraint model_requests_fact_completeness_ck; + +-- 不读取 JSON 的廉价约束保留为 CHECK,也保持与其他标量 CHECK 的错误优先级 +alter table model_requests + add constraint model_requests_revision_attempt_ck check (attempt_count >= 0), + add constraint model_requests_status_ck check ( + (client_status_code is null or client_status_code between 100 and 599) + and (upstream_status_code is null or upstream_status_code between 100 and 599)), + add constraint model_requests_continuation_hashes_ck check ( + continuation_affinity_hash is null or continuation_affinity_hash ~ '^[0-9a-f]{64}$'), + add constraint model_requests_recovery_ck check (recovery_attempt_count >= 0), + add constraint model_requests_fact_completeness_ck check ( + cost_source = 'unavailable' or cost_currency is not null); + +-- 历史引用与快照不依赖实体是否仍存在 +drop view model_request_observations; +alter table model_requests + drop constraint model_requests_client_fk, + drop constraint model_requests_account_fk, + drop constraint model_requests_client_ref_ck, + drop constraint model_requests_account_ref_ck, + drop column client_api_key_id, + drop column provider_account_id, + add constraint model_requests_client_ref_ck check (octet_length(client_api_key_ref) between 1 and 128), + add constraint model_requests_account_ref_ck check (provider_account_ref is null or octet_length(provider_account_ref) between 1 and 128); +alter table ops_events + drop constraint ops_events_account_fk, + drop constraint ops_events_account_ref_ck, + drop constraint ops_events_account_snapshot_present_ck, + drop column provider_account_id, + add column provider_account_plan_type_snapshot text, + add constraint ops_events_account_ref_ck check (provider_account_ref is null or octet_length(provider_account_ref) between 1 and 128); + +-- 租约只保存运行态,历史表不承担高频续期更新 +create table model_request_leases ( + request_id text primary key references model_requests(id) on delete cascade, + lease_expires_at timestamptz not null +); +create index model_request_leases_expiry_idx on model_request_leases (lease_expires_at, request_id); +insert into model_request_leases (request_id, lease_expires_at) +select id, deadline_at from model_requests where outcome = 'running'; +drop index model_requests_running_deadline_idx; +alter table model_requests rename column deadline_at to recovery_lease_expires_at; +alter table model_requests alter column recovery_lease_expires_at drop not null; +update model_requests set recovery_lease_expires_at = null where outcome = 'running'; + +-- 主请求锁先于租约锁,终结和恢复共用同一顺序;续期只取得租约锁 +create function finish_model_request_lease() returns trigger +language plpgsql as $$ +declare + expires_at timestamptz; +begin + if old.outcome = 'running' and new.outcome <> 'running' then + select lease_expires_at into expires_at from model_request_leases + where request_id = new.id for update; + new.recovery_lease_expires_at := coalesce(expires_at, old.recovery_lease_expires_at, old.started_at); + delete from model_request_leases where request_id = new.id; + end if; + return new; +end; +$$; +create trigger model_requests_finish_lease before update on model_requests + for each row when (old.outcome = 'running' and new.outcome <> 'running') + execute function finish_model_request_lease(); + +-- 只更名,不改变 Token 的包含关系或费用语义 +alter table model_requests rename column cached_tokens to cached_input_tokens; +alter table model_requests rename column cache_write_tokens to cache_write_input_tokens; +alter table model_requests rename column reasoning_tokens to reasoning_output_tokens; + +create view model_request_observations as +select mr.*, + coalesce(lease.lease_expires_at, mr.recovery_lease_expires_at) as effective_recovery_lease_expires_at, + (mr.request_observation_json #>> '{request,configRevision}')::bigint as config_revision, + (mr.request_observation_json #>> '{request,protocol}') as protocol, + (mr.request_observation_json #>> '{request,endpoint}') as endpoint, + (mr.request_observation_json #>> '{account,name}') as provider_account_name_snapshot, + (mr.request_observation_json #>> '{account,email}') as provider_account_email_snapshot, + (mr.request_observation_json #>> '{account,authenticationKind}') as provider_account_authentication_kind_snapshot, + (mr.request_observation_json #>> '{routing,scope}') as routing_scope, + array(select jsonb_array_elements_text(mr.request_observation_json #> '{routing,groupRefs}')) as routing_group_refs, + (mr.request_observation_json #> '{routing,groupNamesSnapshot}') as routing_group_names_snapshot, + (mr.request_observation_json #>> '{error,providerErrorCode}') as provider_error_code, + (mr.request_observation_json #>> '{error,message}') as error_message, + (mr.request_observation_json #>> '{error,retryAfterMs}')::bigint as retry_after_ms, + (mr.request_observation_json #>> '{request,clientIp}')::inet as client_ip, + (mr.request_observation_json #>> '{request,userAgent}') as user_agent, + (mr.request_observation_json #>> '{request,reasoningEffort}') as reasoning_effort, + (mr.request_observation_json #>> '{request,reasoningPreset}') as reasoning_preset, + (mr.request_observation_json #>> '{request,subagentKind}') as subagent_kind, + (mr.request_observation_json #>> '{request,compact}')::boolean as compact, + measured.transport_decision_wait_ms as transport_decision_wait_ms, + measured.connect_ms as connect_ms, + measured.headers_ms as headers_ms, + measured.first_event_ms as first_event_ms, + measured.first_reasoning_ms as first_reasoning_ms, + measured.first_text_ms as first_text_ms, + measured.first_token_ms as first_token_ms, + measured.latency_ms as latency_ms, + measured.provider_processing_ms as provider_processing_ms, + measured.admission_decision_ms as admission_decision_ms, + measured.account_selection_wait_ms as account_selection_wait_ms, + measured.capacity_used_slots as capacity_used_slots, + measured.capacity_total_slots as capacity_total_slots, + (mr.request_observation_json #>> '{transport,httpVersion}') as http_version, + (mr.request_observation_json #>> '{transport,websocketPool}') as websocket_pool, + (mr.request_observation_json #>> '{transport,connection,id}') as upstream_connection_id, + (mr.request_observation_json #>> '{transport,connection,exitReason}') as upstream_connection_exit_reason, + (mr.request_observation_json #>> '{transport,connection,ageMs}')::bigint as upstream_connection_age_ms, + (mr.request_observation_json #>> '{transport,connection,idleMs}')::bigint as upstream_connection_idle_ms, + (mr.request_observation_json #>> '{response,responseModel}') as upstream_response_model, + (mr.request_observation_json #>> '{continuation,previousResponseIdHash}') as continuation_previous_response_id_hash, + (mr.request_observation_json #>> '{continuation,unavailableReason}') as continuation_unavailable_reason, + (mr.request_observation_json #>> '{recovery,retryDelayMs}')::bigint as recovery_retry_delay_ms, + (mr.request_observation_json #>> '{recovery,totalLatencyMs}')::bigint as recovery_total_latency_ms, + (mr.request_observation_json #>> '{timings,upstream,apiOverheadMs}')::double precision as upstream_api_overhead_ms, + (mr.request_observation_json #>> '{timings,upstream,engineMs}')::double precision as upstream_engine_ms, + (mr.request_observation_json #>> '{timings,upstream,engineIapiTtftMs}')::double precision as upstream_engine_iapi_ttft_ms, + (mr.request_observation_json #>> '{timings,upstream,engineServiceTtftMs}')::double precision as upstream_engine_service_ttft_ms, + (mr.request_observation_json #>> '{timings,upstream,engineIapiTbtMs}')::double precision as upstream_engine_iapi_tbt_ms, + (mr.request_observation_json #>> '{timings,upstream,engineServiceTbtMs}')::double precision as upstream_engine_service_tbt_ms, + measured.upstream_response_ms as upstream_response_ms, + mr.request_observation_json #>> '{request,clientApiKeyNameSnapshot}' as client_api_key_name_snapshot, + mr.request_observation_json #>> '{account,planTypeSnapshot}' as provider_account_plan_type_snapshot, + mr.request_observation_json #>> '{account,notesSnapshot}' as provider_account_notes_snapshot +from model_requests mr +left join model_request_leases lease on lease.request_id = mr.id +cross join lateral ( + select (mr.request_observation_json #>> '{timings,local,transportDecisionWaitMs}')::bigint as transport_decision_wait_ms, + (mr.request_observation_json #>> '{timings,local,connectMs}')::bigint as connect_ms, + (mr.request_observation_json #>> '{timings,local,headersMs}')::bigint as headers_ms, + (mr.request_observation_json #>> '{timings,local,firstEventMs}')::bigint as first_event_ms, + (mr.request_observation_json #>> '{timings,local,firstReasoningMs}')::bigint as first_reasoning_ms, + (mr.request_observation_json #>> '{timings,local,firstTextMs}')::bigint as first_text_ms, + (mr.request_observation_json #>> '{timings,local,firstTokenMs}')::bigint as first_token_ms, + (mr.request_observation_json #>> '{timings,local,latencyMs}')::bigint as latency_ms, + (mr.request_observation_json #>> '{timings,upstream,processingMs}')::bigint as provider_processing_ms, + (mr.request_observation_json #>> '{scheduling,admissionDecisionMs}')::bigint as admission_decision_ms, + (mr.request_observation_json #>> '{scheduling,accountSelectionWaitMs}')::bigint as account_selection_wait_ms, + (mr.request_observation_json #>> '{scheduling,capacityUsedSlots}')::bigint as capacity_used_slots, + (mr.request_observation_json #>> '{scheduling,capacityTotalSlots}')::bigint as capacity_total_slots, + (mr.request_observation_json #>> '{timings,upstream,responseMs}')::bigint as upstream_response_ms +) measured; + +-- 删除 Key 时按稳定身份定位费用事件,避免扫描其他 Key 的完整账本 +create index client_key_charge_events_client_api_key_id + on client_key_charge_events (client_api_key_id); + +-- 请求观测的可重建窄事实、持久化工作队列与分层统计 +create sequence request_analytics_change_version; +create table request_analytics_facts as select + mr.id, mr.started_at, mr.completed_at, mr.client_api_key_ref, mr.provider_account_ref, mr.routing_group_refs, mr.operation, mr.provider_kind, mr.requested_model_id, mr.upstream_model_id, mr.outcome, mr.client_status_code, mr.upstream_status_code, mr.client_transport, mr.upstream_transport, mr.attempt_count, mr.client_response_id, mr.upstream_response_id, mr.upstream_request_id, mr.recovered_at, mr.downstream_committed_at, mr.request_kind, mr.image_generation_requested, mr.image_generation_succeeded, mr.error_kind, mr.input_tokens, mr.output_tokens, mr.cached_input_tokens, mr.cache_write_input_tokens, mr.reasoning_output_tokens, mr.image_input_tokens, mr.image_output_tokens, mr.total_tokens, mr.cost_source, mr.cost_currency, mr.cost_amount, mr.first_token_ms, mr.latency_ms, mr.admission_decision_ms, mr.account_selection_wait_ms, mr.upstream_response_ms, mr.capacity_used_slots, mr.capacity_total_slots +from model_request_observations mr with no data; +alter table request_analytics_facts add primary key (id), + add column standard_cost_amount numeric(20,10), + add column no_cache_cost_amount numeric(20,10), + add column source_version bigint not null, + add column schema_version integer not null default 1, + add column ops_failure_count bigint not null default 0, + add column ops_rate_limited_count bigint not null default 0, + add column ops_auth_failure_count bigint not null default 0, + add column ops_provider_5xx_count bigint not null default 0; +create index request_analytics_facts_started on request_analytics_facts (started_at, id); +create index request_analytics_facts_key on request_analytics_facts (client_api_key_ref, started_at, id); +create index request_analytics_facts_account on request_analytics_facts (provider_account_ref, started_at, id); +-- 分组集合未预聚合,窄事实回退仍需按集合定位,避免低命中筛选扫描全部保留历史 +create index request_analytics_facts_groups on request_analytics_facts using gin (routing_group_refs); +-- 用量列表的精确总数与账号最近使用只需要交付事实,部分索引避免为判定资格读取宽行 +create index model_requests_ops_completed on model_requests (completed_at,id) +where error_kind is not null and error_kind <> 'cancelled'; +create index model_requests_delivered_started on model_requests (started_at,id) +where outcome='succeeded' and downstream_committed_at is not null + and (provider_kind is distinct from 'openai' or request_kind is distinct from 'prewarm') + and ((client_transport='websocket' and client_status_code is null) or client_status_code between 200 and 399) + and (requested_model_id is not null or upstream_model_id is not null or image_generation_requested + or input_tokens is not null or output_tokens is not null or cached_input_tokens is not null + or cache_write_input_tokens is not null or reasoning_output_tokens is not null + or image_input_tokens is not null or image_output_tokens is not null + or total_tokens is not null or cost_amount is not null); +create index request_analytics_facts_delivered_account on request_analytics_facts (provider_account_ref,started_at desc,id desc) +where outcome='succeeded' and downstream_committed_at is not null + and (provider_kind is distinct from 'openai' or request_kind is distinct from 'prewarm') + and ((client_transport='websocket' and client_status_code is null) or client_status_code between 200 and 399) + and (requested_model_id is not null or upstream_model_id is not null or image_generation_requested + or input_tokens is not null or output_tokens is not null or cached_input_tokens is not null + or cache_write_input_tokens is not null or reasoning_output_tokens is not null + or image_input_tokens is not null or image_output_tokens is not null + or total_tokens is not null or cost_amount is not null); +-- 不透明 ID 没有私有长度上限,等值 HASH 索引避免长值超过 B-tree 索引项限制 +create index request_analytics_facts_client_response on request_analytics_facts using hash (client_response_id) where client_response_id is not null; +create index request_analytics_facts_upstream_response on request_analytics_facts using hash (upstream_response_id) where upstream_response_id is not null; +create index request_analytics_facts_upstream_request on request_analytics_facts using hash (upstream_request_id) where upstream_request_id is not null; +create table request_analytics_pending_requests ( + request_id text primary key, + change_version bigint not null default nextval('request_analytics_change_version'), + oldest_pending_at timestamptz not null default clock_timestamp(), + previous_bucket timestamptz, + deletion_pending boolean not null default false +); +create index request_analytics_pending_deletions on request_analytics_pending_requests (previous_bucket) where deletion_pending; +create index request_analytics_pending_requests_age on request_analytics_pending_requests (oldest_pending_at, request_id); +create table request_analytics_pending_buckets ( + bucket_start timestamptz primary key, + change_version bigint not null default nextval('request_analytics_change_version'), + oldest_pending_at timestamptz not null, + build_version bigint, + build_phase smallint not null default 0 check (build_phase between 0 and 4), + build_started_at timestamptz, + build_id text, + build_collection text, + build_dimensions text[], + build_outcome text +); +create index request_analytics_pending_buckets_age on request_analytics_pending_buckets(oldest_pending_at,bucket_start); +create table request_analytics_15m ( + bucket_start timestamptz not null, + collection text not null, + dimensions text[] not null, + outcome text not null, + schema_version integer not null, + histogram_version integer not null, + counts jsonb not null, + request_count bigint generated always as ((counts->>'request_count')::bigint) stored, + costs jsonb not null, + distributions jsonb not null, + primary key (bucket_start, collection, dimensions, outcome) +); +create table request_analytics_hourly (like request_analytics_15m including all); +create index request_analytics_15m_collection on request_analytics_15m (collection,bucket_start); +create index request_analytics_hourly_collection on request_analytics_hourly (collection,bucket_start); +create index request_analytics_15m_dimension on request_analytics_15m (collection,(dimensions[1]),bucket_start); +create index request_analytics_hourly_dimension on request_analytics_hourly (collection,(dimensions[1]),bucket_start); +-- 排名只读取确定派生的计数,覆盖索引避免为高基数账号和密钥读取宽分布行 +create index request_analytics_15m_rank on request_analytics_15m (collection,bucket_start) include (dimensions,outcome,schema_version,histogram_version,request_count) where collection in ('account','key'); +create index request_analytics_hourly_rank on request_analytics_hourly (collection,bucket_start) include (dimensions,outcome,schema_version,histogram_version,request_count) where collection in ('account','key'); +-- 高频身份排名只保存日级精确计数,不复制分布与诊断详情 +create table request_analytics_daily_counts ( + bucket_start timestamptz not null, + collection text not null check (collection in ('account','key')), + dimension text not null, + outcome text not null, + schema_version integer not null, + histogram_version integer not null, + request_count bigint not null check (request_count >= 0), + primary key (collection,bucket_start,dimension,outcome,schema_version,histogram_version) +); + +-- 计数由已持久化小时行的 OLD/NEW 差额派生,重放同值不会重复累计 +-- 未发布小时所在的日期不参与日级查询,分段重建与版本清理仍可逐批提交 +create function sync_request_analytics_daily_counts() returns trigger language plpgsql as $$ +declare + old_day timestamptz; + new_day timestamptz; + delta bigint; +begin + if TG_OP <> 'INSERT' then + old_day := date_bin(interval '1 day',OLD.bucket_start,timestamptz '2000-01-01 UTC'); + end if; + if TG_OP <> 'DELETE' then + new_day := date_bin(interval '1 day',NEW.bucket_start,timestamptz '2000-01-01 UTC'); + end if; + if TG_OP = 'UPDATE' and + (old_day,OLD.collection,OLD.dimensions[1],OLD.outcome,OLD.schema_version,OLD.histogram_version) + is not distinct from + (new_day,NEW.collection,NEW.dimensions[1],NEW.outcome,NEW.schema_version,NEW.histogram_version) then + if NEW.collection in ('account','key') then + delta := NEW.request_count - OLD.request_count; + if delta <> 0 then + update request_analytics_daily_counts set request_count=request_count+delta + where collection=NEW.collection and bucket_start=new_day and dimension=NEW.dimensions[1] + and outcome=NEW.outcome and schema_version=NEW.schema_version and histogram_version=NEW.histogram_version; + end if; + end if; + return null; + end if; + if TG_OP <> 'INSERT' and OLD.collection in ('account','key') then + update request_analytics_daily_counts set request_count=request_count-OLD.request_count + where collection=OLD.collection and bucket_start=old_day and dimension=OLD.dimensions[1] + and outcome=OLD.outcome and schema_version=OLD.schema_version and histogram_version=OLD.histogram_version; + delete from request_analytics_daily_counts + where collection=OLD.collection and bucket_start=old_day and dimension=OLD.dimensions[1] + and outcome=OLD.outcome and schema_version=OLD.schema_version and histogram_version=OLD.histogram_version + and request_count=0; + end if; + if TG_OP <> 'DELETE' and NEW.collection in ('account','key') then + insert into request_analytics_daily_counts values(new_day,NEW.collection,NEW.dimensions[1],NEW.outcome,NEW.schema_version,NEW.histogram_version,NEW.request_count) + on conflict(collection,bucket_start,dimension,outcome,schema_version,histogram_version) + do update set request_count=request_analytics_daily_counts.request_count+excluded.request_count; + end if; + return null; +end; +$$; +create trigger request_analytics_hourly_daily_counts + after insert or update or delete on request_analytics_hourly + for each row execute function sync_request_analytics_daily_counts(); + +create table request_analytics_build_state ( + singleton boolean primary key default true check (singleton), + schema_version integer not null default 1, + histogram_version integer not null default 1, + coverage_start timestamptz, + coverage_end timestamptz, + backfill_cursor_started_at timestamptz, + backfill_cursor_id text, + backfill_complete boolean not null default false, + reset_phase smallint not null default 0 check (reset_phase between 0 and 4), + reset_cursor jsonb, + generation bigint not null default 0, + updated_at timestamptz not null default clock_timestamp(), + check (reset_phase <> 0 or reset_cursor is null) +); +insert into request_analytics_build_state (singleton) values (true); + +create function enqueue_request_analytics(request_ref text, request_start timestamptz, source_deleted boolean default false) +returns void language sql as $$ + insert into request_analytics_pending_requests(request_id, previous_bucket, deletion_pending) + select request_ref, date_bin(interval '15 minutes', coalesce(request_start,(select started_at from request_analytics_facts where id=request_ref)), timestamptz '2000-01-01 UTC'), source_deleted + where request_ref is not null + on conflict (request_id) do update + set change_version = nextval('request_analytics_change_version'), + previous_bucket = coalesce(request_analytics_pending_requests.previous_bucket, excluded.previous_bucket), + deletion_pending = request_analytics_pending_requests.deletion_pending or excluded.deletion_pending +$$; +create function observe_request_analytics_change() returns trigger language plpgsql as $$ +begin + if TG_OP = 'DELETE' then + perform enqueue_request_analytics(OLD.id, OLD.started_at, true); + return OLD; + end if; + perform enqueue_request_analytics(NEW.id, NEW.started_at); + return NEW; +end +$$; +create trigger request_analytics_insert after insert on model_requests +for each row execute function observe_request_analytics_change(); +create trigger request_analytics_delete after delete on model_requests +for each row execute function observe_request_analytics_change(); +create trigger request_analytics_update after update on model_requests +for each row when (ROW(OLD.id, OLD.started_at, OLD.completed_at, OLD.client_api_key_ref, OLD.provider_account_ref, OLD.operation, OLD.provider_kind, OLD.requested_model_id, OLD.upstream_model_id, OLD.outcome, OLD.client_status_code, OLD.upstream_status_code, OLD.client_transport, OLD.upstream_transport, OLD.attempt_count, OLD.client_response_id, OLD.upstream_response_id, OLD.upstream_request_id, OLD.recovered_at, OLD.downstream_committed_at, OLD.request_kind, OLD.image_generation_requested, OLD.image_generation_succeeded, OLD.error_kind, OLD.input_tokens, OLD.output_tokens, OLD.cached_input_tokens, OLD.cache_write_input_tokens, OLD.reasoning_output_tokens, OLD.image_input_tokens, OLD.image_output_tokens, OLD.total_tokens, OLD.cost_source, OLD.cost_currency, OLD.cost_amount) is distinct from ROW(NEW.id, NEW.started_at, NEW.completed_at, NEW.client_api_key_ref, NEW.provider_account_ref, NEW.operation, NEW.provider_kind, NEW.requested_model_id, NEW.upstream_model_id, NEW.outcome, NEW.client_status_code, NEW.upstream_status_code, NEW.client_transport, NEW.upstream_transport, NEW.attempt_count, NEW.client_response_id, NEW.upstream_response_id, NEW.upstream_request_id, NEW.recovered_at, NEW.downstream_committed_at, NEW.request_kind, NEW.image_generation_requested, NEW.image_generation_succeeded, NEW.error_kind, NEW.input_tokens, NEW.output_tokens, NEW.cached_input_tokens, NEW.cache_write_input_tokens, NEW.reasoning_output_tokens, NEW.image_input_tokens, NEW.image_output_tokens, NEW.total_tokens, NEW.cost_source, NEW.cost_currency, NEW.cost_amount) + or OLD.billing_snapshot_json is distinct from NEW.billing_snapshot_json + or OLD.service_tier is distinct from NEW.service_tier + or OLD.request_observation_json->'timings' is distinct from NEW.request_observation_json->'timings' + or OLD.request_observation_json->'scheduling' is distinct from NEW.request_observation_json->'scheduling' + or OLD.request_observation_json->'routing' is distinct from NEW.request_observation_json->'routing') +execute function observe_request_analytics_change(); +create function observe_ops_analytics_change() returns trigger language plpgsql as $$ +begin + if TG_OP <> 'INSERT' then + perform enqueue_request_analytics(OLD.model_request_id, null, TG_OP = 'DELETE'); + end if; + if TG_OP <> 'DELETE' then + perform enqueue_request_analytics(NEW.model_request_id, null); + return NEW; + end if; + return OLD; +end +$$; +create trigger request_analytics_ops_insert after insert on ops_events +for each row execute function observe_ops_analytics_change(); +create trigger request_analytics_ops_delete after delete on ops_events +for each row execute function observe_ops_analytics_change(); +create trigger request_analytics_ops_update after update on ops_events +for each row when (ROW(OLD.model_request_id, OLD.failure_kind, OLD.status_code, OLD.attempt_index) + is distinct from ROW(NEW.model_request_id, NEW.failure_kind, NEW.status_code, NEW.attempt_index)) +execute function observe_ops_analytics_change(); + +-- 删除在源事务内使缓存失效,语句级触发避免批量清理争用全局行 +-- 空清理不更新统计代次,避免与启动回填的快照事务发生无效写冲突 +create function invalidate_deleted_request_analytics() returns trigger language plpgsql as $$ +begin + if exists (select 1 from deleted_requests) then + update request_analytics_build_state set generation = generation + 1 where singleton; + end if; + return null; +end +$$; + +create trigger request_analytics_delete_invalidation after delete on model_requests +referencing old table as deleted_requests +for each statement execute function invalidate_deleted_request_analytics(); +create function invalidate_deleted_ops_analytics() returns trigger language plpgsql as $$ +begin + -- 父请求删除已经负责失效;独立删除仍有关联源请求的 Ops 才需要另行失效 + if exists (select 1 from deleted_ops join model_requests mr on mr.id = deleted_ops.model_request_id) then + update request_analytics_build_state set generation = generation + 1 where singleton; + end if; + return null; +end +$$; +create trigger request_analytics_ops_delete_invalidation after delete on ops_events +referencing old table as deleted_ops +for each statement execute function invalidate_deleted_ops_analytics(); diff --git a/backend/migrations/0028_request_observation_validation.sql b/backend/migrations/0028_request_observation_validation.sql deleted file mode 100644 index 94ebb9542..000000000 --- a/backend/migrations/0028_request_observation_validation.sql +++ /dev/null @@ -1,27 +0,0 @@ --- 观测 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; diff --git a/backend/migrations/0029_request_observation_dependencies.sql b/backend/migrations/0029_request_observation_dependencies.sql deleted file mode 100644 index 015aa7db3..000000000 --- a/backend/migrations/0029_request_observation_dependencies.sql +++ /dev/null @@ -1,414 +0,0 @@ --- 结构校验只约束路由字段类型,枚举范围由同名业务约束负责 -create or replace function model_request_observation_valid(document jsonb) returns boolean -language plpgsql immutable parallel safe as $$ -declare - spec record; - item jsonb; - number numeric; -begin - if document is null or jsonb_typeof(document) <> 'object' - or octet_length(document::text) > 1048576 then - return false; - end if; - for spec in select * from (values - ('{}'::text[], array['account','continuation','error','recovery','request','response','routing','scheduling','timings','transport']::text[]), - ('{request}'::text[], array['clientIp','compact','configRevision','endpoint','protocol','reasoningEffort','reasoningPreset','subagentKind','userAgent']::text[]), - ('{account}'::text[], array['authenticationKind','email','name']::text[]), - ('{routing}'::text[], array['groupNamesSnapshot','groupRefs','scope']::text[]), - ('{error}'::text[], array['message','providerErrorCode','retryAfterMs']::text[]), - ('{timings}'::text[], array['local','upstream']::text[]), - ('{timings,local}'::text[], array['connectMs','firstEventMs','firstReasoningMs','firstTextMs','firstTokenMs','headersMs','latencyMs','transportDecisionWaitMs']::text[]), - ('{timings,upstream}'::text[], array['processingMs','responseMs','apiOverheadMs','engineMs','engineIapiTtftMs','engineServiceTtftMs','engineIapiTbtMs','engineServiceTbtMs']::text[]), - ('{scheduling}'::text[], array['accountSelectionWaitMs','admissionDecisionMs','capacityTotalSlots','capacityUsedSlots']::text[]), - ('{transport}'::text[], array['connection','httpVersion','websocketPool']::text[]), - ('{transport,connection}'::text[], array['ageMs','exitReason','id','idleMs']::text[]), - ('{response}'::text[], array['responseModel']::text[]), - ('{continuation}'::text[], array['previousResponseIdHash','unavailableReason']::text[]), - ('{recovery}'::text[], array['retryDelayMs','totalLatencyMs']::text[])) as groups(path, allowed_keys) loop - item := document #> spec.path; - if item is not null and (jsonb_typeof(item) <> 'object' or item - spec.allowed_keys <> '{}'::jsonb) then - return false; - end if; - end loop; - for spec in select * from (values - ('{request,configRevision}'::text[], 'number'), - ('{request,protocol}'::text[], 'string'), - ('{request,endpoint}'::text[], 'string'), - ('{account,name}'::text[], 'string'), - ('{account,email}'::text[], 'string'), - ('{account,authenticationKind}'::text[], 'string'), - ('{routing,scope}'::text[], 'string'), - ('{routing,groupRefs}'::text[], 'array'), - ('{routing,groupNamesSnapshot}'::text[], 'array'), - ('{error,providerErrorCode}'::text[], 'string'), - ('{error,message}'::text[], 'string'), - ('{error,retryAfterMs}'::text[], 'number'), - ('{request,clientIp}'::text[], 'string'), - ('{request,userAgent}'::text[], 'string'), - ('{request,reasoningEffort}'::text[], 'string'), - ('{request,reasoningPreset}'::text[], 'string'), - ('{request,subagentKind}'::text[], 'string'), - ('{request,compact}'::text[], 'boolean'), - ('{timings,local,transportDecisionWaitMs}'::text[], 'number'), - ('{timings,local,connectMs}'::text[], 'number'), - ('{timings,local,headersMs}'::text[], 'number'), - ('{timings,local,firstEventMs}'::text[], 'number'), - ('{timings,local,firstReasoningMs}'::text[], 'number'), - ('{timings,local,firstTextMs}'::text[], 'number'), - ('{timings,local,firstTokenMs}'::text[], 'number'), - ('{timings,local,latencyMs}'::text[], 'number'), - ('{timings,upstream,processingMs}'::text[], 'number'), - ('{timings,upstream,apiOverheadMs}'::text[], 'milliseconds'), - ('{timings,upstream,engineMs}'::text[], 'milliseconds'), - ('{timings,upstream,engineIapiTtftMs}'::text[], 'milliseconds'), - ('{timings,upstream,engineServiceTtftMs}'::text[], 'milliseconds'), - ('{timings,upstream,engineIapiTbtMs}'::text[], 'milliseconds'), - ('{timings,upstream,engineServiceTbtMs}'::text[], 'milliseconds'), - ('{scheduling,admissionDecisionMs}'::text[], 'number'), - ('{scheduling,accountSelectionWaitMs}'::text[], 'number'), - ('{scheduling,capacityUsedSlots}'::text[], 'number'), - ('{scheduling,capacityTotalSlots}'::text[], 'number'), - ('{transport,httpVersion}'::text[], 'string'), - ('{transport,websocketPool}'::text[], 'string'), - ('{transport,connection,id}'::text[], 'string'), - ('{transport,connection,exitReason}'::text[], 'string'), - ('{transport,connection,ageMs}'::text[], 'number'), - ('{transport,connection,idleMs}'::text[], 'number'), - ('{response,responseModel}'::text[], 'string'), - ('{continuation,previousResponseIdHash}'::text[], 'string'), - ('{continuation,unavailableReason}'::text[], 'string'), - ('{recovery,retryDelayMs}'::text[], 'number'), - ('{recovery,totalLatencyMs}'::text[], 'number'), - ('{timings,upstream,responseMs}'::text[], 'number')) as fields(path, kind) loop - item := document #> spec.path; - if item is null or item = 'null'::jsonb then continue; end if; - if spec.kind = 'milliseconds' then - if jsonb_typeof(item) <> 'number' then return false; end if; - number := (item #>> '{}')::numeric; - if number < 0 or number > 9223372036854775807 then return false; end if; - continue; - end if; - if jsonb_typeof(item) <> spec.kind then return false; end if; - if spec.kind = 'number' then - number := (item #>> '{}')::numeric; - if item #>> '{}' !~ '^[0-9]+$' or number > 9223372036854775807 then return false; end if; - elsif spec.kind = 'array' then - if exists (select 1 from jsonb_array_elements(item) member where jsonb_typeof(member) <> 'string') then return false; end if; - end if; - end loop; - if (document #> '{request,configRevision}' is not null - and document #> '{request,configRevision}' <> 'null'::jsonb - and document #> '{request,compact}' in ('true'::jsonb, 'false'::jsonb) - and coalesce(length(document #>> '{request,protocol}'), 0) > 0 - and coalesce(length(document #>> '{request,endpoint}'), 0) > 0 - and jsonb_typeof(document #> '{routing,scope}') = 'string' - and jsonb_typeof(document #> '{routing,groupRefs}') = 'array' - and jsonb_typeof(document #> '{routing,groupNamesSnapshot}') = 'array') is not true then - return false; - end if; - if document #>> '{request,clientIp}' is not null then - perform (document #>> '{request,clientIp}')::inet; - end if; - return true; -exception when invalid_text_representation or numeric_value_out_of_range then - return false; -end; -$$; - --- 按最终行的真实依赖执行校验,纯状态写入不展开无关 JSON -create or replace function check_model_request_observation() returns trigger -language plpgsql as $$ -declare - json_changed boolean := tg_op = 'INSERT' or old.request_observation_json is distinct from new.request_observation_json; -begin - if json_changed then - if model_request_observation_valid(new.request_observation_json) is not true then - raise exception using errcode = '23514', constraint = 'model_requests_observation_shape_ck', - message = 'request observation violates shape constraint', table = tg_table_name, schema = tg_table_schema; - end if; - end if; - if json_changed then - declare - timings_upstream_response_ms bigint; - begin - timings_upstream_response_ms := (new.request_observation_json #>> '{timings,upstream,responseMs}')::bigint; - if (timings_upstream_response_ms > 0) is false then - raise exception using errcode = '23514', constraint = 'model_requests_upstream_duration_ck', - message = 'request observation violates model_requests_upstream_duration_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end; - end if; - if json_changed then - if ((new.request_observation_json #>> '{request,configRevision}')::bigint > 0) is false then - raise exception using errcode = '23514', constraint = 'model_requests_revision_attempt_ck', message = 'request observation violates model_requests_revision_attempt_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end if; - if json_changed then - if ((new.request_observation_json #>> '{error,retryAfterMs}')::bigint is null or (new.request_observation_json #>> '{error,retryAfterMs}')::bigint >= 0) is false then - raise exception using errcode = '23514', constraint = 'model_requests_status_ck', message = 'request observation violates model_requests_status_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end if; - if json_changed then - declare - transport_websocket_pool text; - begin - transport_websocket_pool := (new.request_observation_json #>> '{transport,websocketPool}'); - if (transport_websocket_pool is null or transport_websocket_pool in ('new', 'reuse')) is false then - raise exception using errcode = '23514', constraint = 'model_requests_websocket_pool_ck', - message = 'request observation violates model_requests_websocket_pool_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end; - end if; - if json_changed then - declare - routing_group_names_snapshot jsonb; - routing_scope text; - routing_group_refs jsonb; - begin - routing_group_names_snapshot := (new.request_observation_json #> '{routing,groupNamesSnapshot}'); - routing_scope := (new.request_observation_json #>> '{routing,scope}'); - routing_group_refs := (new.request_observation_json #> '{routing,groupRefs}'); - if (jsonb_typeof(routing_group_names_snapshot) = 'array' - and ( - ( - routing_scope in ('legacy_provider', 'all') - and jsonb_array_length(routing_group_refs) = 0 - and jsonb_array_length(routing_group_names_snapshot) = 0 - ) - or ( - routing_scope = 'groups' - and jsonb_array_length(routing_group_refs) > 0 - and not jsonb_path_exists(new.request_observation_json, '$.routing.groupRefs[*] ? (@ == null)') - and jsonb_array_length(routing_group_names_snapshot) - = jsonb_array_length(routing_group_refs) - ) - )) is false then - raise exception using errcode = '23514', constraint = 'model_requests_routing_group_names_ck', - message = 'request observation violates model_requests_routing_group_names_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end; - end if; - if json_changed then - declare - routing_scope text; - begin - routing_scope := (new.request_observation_json #>> '{routing,scope}'); - if (routing_scope in ('legacy_provider', 'all', 'groups')) is false then - raise exception using errcode = '23514', constraint = 'model_requests_routing_scope_ck', - message = 'request observation violates model_requests_routing_scope_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end; - end if; - if json_changed then - declare - timings_local_transport_decision_wait_ms bigint; - timings_local_connect_ms bigint; - timings_local_headers_ms bigint; - timings_local_first_event_ms bigint; - timings_local_first_reasoning_ms bigint; - timings_local_first_text_ms bigint; - timings_local_first_token_ms bigint; - timings_upstream_processing_ms bigint; - timings_local_latency_ms bigint; - begin - timings_local_transport_decision_wait_ms := (new.request_observation_json #>> '{timings,local,transportDecisionWaitMs}')::bigint; - timings_local_connect_ms := (new.request_observation_json #>> '{timings,local,connectMs}')::bigint; - timings_local_headers_ms := (new.request_observation_json #>> '{timings,local,headersMs}')::bigint; - timings_local_first_event_ms := (new.request_observation_json #>> '{timings,local,firstEventMs}')::bigint; - timings_local_first_reasoning_ms := (new.request_observation_json #>> '{timings,local,firstReasoningMs}')::bigint; - timings_local_first_text_ms := (new.request_observation_json #>> '{timings,local,firstTextMs}')::bigint; - timings_local_first_token_ms := (new.request_observation_json #>> '{timings,local,firstTokenMs}')::bigint; - timings_upstream_processing_ms := (new.request_observation_json #>> '{timings,upstream,processingMs}')::bigint; - timings_local_latency_ms := (new.request_observation_json #>> '{timings,local,latencyMs}')::bigint; - if ((timings_local_transport_decision_wait_ms is null or timings_local_transport_decision_wait_ms >= 0) - and (timings_local_connect_ms is null or timings_local_connect_ms >= 0) - and (timings_local_headers_ms is null or timings_local_headers_ms >= 0) - and (timings_local_first_event_ms is null or timings_local_first_event_ms >= 0) - and (timings_local_first_reasoning_ms is null or timings_local_first_reasoning_ms >= 0) - and (timings_local_first_text_ms is null or timings_local_first_text_ms >= 0) - and (timings_local_first_token_ms is null or timings_local_first_token_ms >= 0) - and (timings_upstream_processing_ms is null or timings_upstream_processing_ms >= 0) - and (timings_local_latency_ms is null or timings_local_latency_ms >= 0) - and ( - timings_local_latency_ms is null - or ( - (timings_local_transport_decision_wait_ms is null or timings_local_transport_decision_wait_ms <= timings_local_latency_ms) - and (timings_local_connect_ms is null or timings_local_connect_ms <= timings_local_latency_ms) - and (timings_local_headers_ms is null or timings_local_headers_ms <= timings_local_latency_ms) - and (timings_local_first_event_ms is null or timings_local_first_event_ms <= timings_local_latency_ms) - and (timings_local_first_reasoning_ms is null or timings_local_first_reasoning_ms <= timings_local_latency_ms) - and (timings_local_first_text_ms is null or timings_local_first_text_ms <= timings_local_latency_ms) - and (timings_local_first_token_ms is null or timings_local_first_token_ms <= timings_local_latency_ms) - ) - )) is false then - raise exception using errcode = '23514', constraint = 'model_requests_latency_ck', - message = 'request observation violates model_requests_latency_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end; - end if; - if json_changed then - declare - scheduling_admission_decision_ms bigint; - scheduling_account_selection_wait_ms bigint; - scheduling_capacity_used_slots bigint; - scheduling_capacity_total_slots bigint; - begin - scheduling_admission_decision_ms := (new.request_observation_json #>> '{scheduling,admissionDecisionMs}')::bigint; - scheduling_account_selection_wait_ms := (new.request_observation_json #>> '{scheduling,accountSelectionWaitMs}')::bigint; - scheduling_capacity_used_slots := (new.request_observation_json #>> '{scheduling,capacityUsedSlots}')::bigint; - scheduling_capacity_total_slots := (new.request_observation_json #>> '{scheduling,capacityTotalSlots}')::bigint; - if ((scheduling_admission_decision_ms is null or scheduling_admission_decision_ms >= 0) - and (scheduling_account_selection_wait_ms is null or scheduling_account_selection_wait_ms >= 0) - and ( - (scheduling_capacity_used_slots is null and scheduling_capacity_total_slots is null) - or ( - scheduling_capacity_used_slots >= 0 - and scheduling_capacity_total_slots > 0 - and scheduling_capacity_used_slots <= scheduling_capacity_total_slots - ) - )) is false then - raise exception using errcode = '23514', constraint = 'model_requests_runtime_pressure_ck', - message = 'request observation violates model_requests_runtime_pressure_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end; - end if; - if json_changed or old.continuation_requested is distinct from new.continuation_requested then - if (((new.request_observation_json #>> '{continuation,previousResponseIdHash}') is null or (new.request_observation_json #>> '{continuation,previousResponseIdHash}') ~ '^[0-9a-f]{64}$') and (new.continuation_requested or (new.request_observation_json #>> '{continuation,previousResponseIdHash}') is null)) is false then - raise exception using errcode = '23514', constraint = 'model_requests_continuation_hashes_ck', message = 'request observation violates model_requests_continuation_hashes_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end if; - if json_changed then - declare - continuation_unavailable_reason text; - begin - continuation_unavailable_reason := (new.request_observation_json #>> '{continuation,unavailableReason}'); - if (continuation_unavailable_reason is null - or ( - octet_length(continuation_unavailable_reason) between 1 and 64 - and continuation_unavailable_reason ~ '^[a-z][a-z0-9_]*$' - )) is false then - raise exception using errcode = '23514', constraint = 'model_requests_continuation_reason_ck', - message = 'request observation violates model_requests_continuation_reason_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end; - end if; - if json_changed then - declare - transport_connection_id text; - transport_connection_exit_reason text; - transport_connection_age_ms bigint; - transport_connection_idle_ms bigint; - begin - transport_connection_id := (new.request_observation_json #>> '{transport,connection,id}'); - transport_connection_exit_reason := (new.request_observation_json #>> '{transport,connection,exitReason}'); - transport_connection_age_ms := (new.request_observation_json #>> '{transport,connection,ageMs}')::bigint; - transport_connection_idle_ms := (new.request_observation_json #>> '{transport,connection,idleMs}')::bigint; - if (( - transport_connection_id is null - and transport_connection_exit_reason is null - and transport_connection_age_ms is null - and transport_connection_idle_ms is null - ) - or ( - octet_length(transport_connection_id) between 1 and 128 - and transport_connection_id !~ '[[:cntrl:]]' - and octet_length(transport_connection_exit_reason) between 1 and 64 - and transport_connection_exit_reason ~ '^[a-z][a-z0-9_]*$' - and transport_connection_age_ms >= 0 - and transport_connection_idle_ms between 0 and transport_connection_age_ms - )) is false then - raise exception using errcode = '23514', constraint = 'model_requests_upstream_connection_ck', - message = 'request observation violates model_requests_upstream_connection_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end; - end if; - if json_changed or old.completed_at is distinct from new.completed_at or old.recovered_at is distinct from new.recovered_at or old.recovery_attempt_count is distinct from new.recovery_attempt_count or old.recovery_request_id is distinct from new.recovery_request_id then - declare - recovery_retry_delay_ms bigint; - recovery_total_latency_ms bigint; - begin - recovery_retry_delay_ms := (new.request_observation_json #>> '{recovery,retryDelayMs}')::bigint; - recovery_total_latency_ms := (new.request_observation_json #>> '{recovery,totalLatencyMs}')::bigint; - if (new.recovery_attempt_count >= 0 - and ( - ( - new.recovered_at is null - and new.recovery_request_id is null - and recovery_retry_delay_ms is null - and recovery_total_latency_ms is null - ) - or ( - new.recovered_at is not null - and new.recovered_at >= new.completed_at - and new.recovery_request_id is not null - and octet_length(new.recovery_request_id) between 1 and 128 - and new.recovery_attempt_count > 0 - and recovery_retry_delay_ms >= 0 - and recovery_total_latency_ms >= recovery_retry_delay_ms - ) - )) is false then - raise exception using errcode = '23514', constraint = 'model_requests_recovery_ck', - message = 'request observation violates model_requests_recovery_ck', table = tg_table_name, schema = tg_table_schema; - end if; - end; - end if; - if json_changed then - if (num_nonnulls(new.request_observation_json #>> '{scheduling,capacityUsedSlots}', new.request_observation_json #>> '{scheduling,capacityTotalSlots}') in (0, 2) - and num_nonnulls(new.request_observation_json #>> '{transport,connection,id}', new.request_observation_json #>> '{transport,connection,exitReason}', - new.request_observation_json #>> '{transport,connection,ageMs}', new.request_observation_json #>> '{transport,connection,idleMs}') in (0, 4)) is false then - raise exception using errcode = '23514', constraint = 'model_requests_fact_completeness_ck', - message = 'request observation violates model_requests_fact_completeness_ck', table = tg_table_name, schema = tg_table_schema; - - end if; - end if; - if json_changed or old.recovered_at is distinct from new.recovered_at or old.completed_at is distinct from new.completed_at then - if (new.recovered_at is null or (new.completed_at is not null - and new.request_observation_json #>> '{recovery,retryDelayMs}' is not null - and new.request_observation_json #>> '{recovery,totalLatencyMs}' is not null)) is false then - raise exception using errcode = '23514', constraint = 'model_requests_fact_completeness_ck', - message = 'request observation violates model_requests_fact_completeness_ck', table = tg_table_name, schema = tg_table_schema; - - end if; - end if; - return null; -end; -$$; - -drop trigger model_requests_observation_update on model_requests; -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 - or old.completed_at is distinct from new.completed_at - or old.continuation_requested is distinct from new.continuation_requested - or old.recovered_at is distinct from new.recovered_at - or old.recovery_attempt_count is distinct from new.recovery_attempt_count - or old.recovery_request_id is distinct from new.recovery_request_id -) execute function check_model_request_observation(); - -alter table model_requests - drop constraint model_requests_upstream_duration_ck, - drop constraint model_requests_revision_attempt_ck, - drop constraint model_requests_status_ck, - drop constraint model_requests_websocket_pool_ck, - drop constraint model_requests_routing_scope_ck, - drop constraint model_requests_routing_group_names_ck, - drop constraint model_requests_latency_ck, - drop constraint model_requests_runtime_pressure_ck, - drop constraint model_requests_continuation_hashes_ck, - drop constraint model_requests_continuation_reason_ck, - drop constraint model_requests_upstream_connection_ck, - drop constraint model_requests_recovery_ck, - drop constraint model_requests_fact_completeness_ck; - --- 不读取 JSON 的廉价约束保留为 CHECK,也保持与其他标量 CHECK 的错误优先级 -alter table model_requests - add constraint model_requests_revision_attempt_ck check (attempt_count >= 0), - add constraint model_requests_status_ck check ( - (client_status_code is null or client_status_code between 100 and 599) - and (upstream_status_code is null or upstream_status_code between 100 and 599)), - add constraint model_requests_continuation_hashes_ck check ( - continuation_affinity_hash is null or continuation_affinity_hash ~ '^[0-9a-f]{64}$'), - add constraint model_requests_recovery_ck check (recovery_attempt_count >= 0), - add constraint model_requests_fact_completeness_ck check ( - (provider_account_id is null or provider_account_ref is not null) - and (cost_source = 'unavailable' or cost_currency is not null)); diff --git a/backend/migrations/0030_historical_request_identity.sql b/backend/migrations/0030_historical_request_identity.sql deleted file mode 100644 index 731d39502..000000000 --- a/backend/migrations/0030_historical_request_identity.sql +++ /dev/null @@ -1,211 +0,0 @@ --- 历史引用与快照不依赖实体是否仍存在,切换前必须停止旧写入者 -create or replace function model_request_observation_valid(document jsonb) returns boolean -language plpgsql immutable parallel safe as $$ -declare - spec record; - item jsonb; - number numeric; -begin - if document is null or jsonb_typeof(document) <> 'object' - or octet_length(document::text) > 1048576 then - return false; - end if; - for spec in select * from (values - ('{}'::text[], array['account','continuation','error','recovery','request','response','routing','scheduling','timings','transport']::text[]), - ('{request}'::text[], array['clientApiKeyNameSnapshot','clientIp','compact','configRevision','endpoint','protocol','reasoningEffort','reasoningPreset','subagentKind','userAgent']::text[]), - ('{account}'::text[], array['authenticationKind','email','name','planTypeSnapshot','notesSnapshot']::text[]), - ('{routing}'::text[], array['groupNamesSnapshot','groupRefs','scope']::text[]), - ('{error}'::text[], array['message','providerErrorCode','retryAfterMs']::text[]), - ('{timings}'::text[], array['local','upstream']::text[]), - ('{timings,local}'::text[], array['connectMs','firstEventMs','firstReasoningMs','firstTextMs','firstTokenMs','headersMs','latencyMs','transportDecisionWaitMs']::text[]), - ('{timings,upstream}'::text[], array['processingMs','responseMs','apiOverheadMs','engineMs','engineIapiTtftMs','engineServiceTtftMs','engineIapiTbtMs','engineServiceTbtMs']::text[]), - ('{scheduling}'::text[], array['accountSelectionWaitMs','admissionDecisionMs','capacityTotalSlots','capacityUsedSlots']::text[]), - ('{transport}'::text[], array['connection','httpVersion','websocketPool']::text[]), - ('{transport,connection}'::text[], array['ageMs','exitReason','id','idleMs']::text[]), - ('{response}'::text[], array['responseModel']::text[]), - ('{continuation}'::text[], array['previousResponseIdHash','unavailableReason']::text[]), - ('{recovery}'::text[], array['retryDelayMs','totalLatencyMs']::text[])) as groups(path, allowed_keys) loop - item := document #> spec.path; - if item is not null and (jsonb_typeof(item) <> 'object' or item - spec.allowed_keys <> '{}'::jsonb) then - return false; - end if; - end loop; - for spec in select * from (values - ('{request,configRevision}'::text[], 'number'), - ('{request,protocol}'::text[], 'string'), - ('{request,endpoint}'::text[], 'string'), - ('{request,clientApiKeyNameSnapshot}'::text[], 'string'), - ('{account,planTypeSnapshot}'::text[], 'string'), - ('{account,notesSnapshot}'::text[], 'string'), - ('{account,name}'::text[], 'string'), - ('{account,email}'::text[], 'string'), - ('{account,authenticationKind}'::text[], 'string'), - ('{routing,scope}'::text[], 'string'), - ('{routing,groupRefs}'::text[], 'array'), - ('{routing,groupNamesSnapshot}'::text[], 'array'), - ('{error,providerErrorCode}'::text[], 'string'), - ('{error,message}'::text[], 'string'), - ('{error,retryAfterMs}'::text[], 'number'), - ('{request,clientIp}'::text[], 'string'), - ('{request,userAgent}'::text[], 'string'), - ('{request,reasoningEffort}'::text[], 'string'), - ('{request,reasoningPreset}'::text[], 'string'), - ('{request,subagentKind}'::text[], 'string'), - ('{request,compact}'::text[], 'boolean'), - ('{timings,local,transportDecisionWaitMs}'::text[], 'number'), - ('{timings,local,connectMs}'::text[], 'number'), - ('{timings,local,headersMs}'::text[], 'number'), - ('{timings,local,firstEventMs}'::text[], 'number'), - ('{timings,local,firstReasoningMs}'::text[], 'number'), - ('{timings,local,firstTextMs}'::text[], 'number'), - ('{timings,local,firstTokenMs}'::text[], 'number'), - ('{timings,local,latencyMs}'::text[], 'number'), - ('{timings,upstream,processingMs}'::text[], 'number'), - ('{timings,upstream,apiOverheadMs}'::text[], 'milliseconds'), - ('{timings,upstream,engineMs}'::text[], 'milliseconds'), - ('{timings,upstream,engineIapiTtftMs}'::text[], 'milliseconds'), - ('{timings,upstream,engineServiceTtftMs}'::text[], 'milliseconds'), - ('{timings,upstream,engineIapiTbtMs}'::text[], 'milliseconds'), - ('{timings,upstream,engineServiceTbtMs}'::text[], 'milliseconds'), - ('{scheduling,admissionDecisionMs}'::text[], 'number'), - ('{scheduling,accountSelectionWaitMs}'::text[], 'number'), - ('{scheduling,capacityUsedSlots}'::text[], 'number'), - ('{scheduling,capacityTotalSlots}'::text[], 'number'), - ('{transport,httpVersion}'::text[], 'string'), - ('{transport,websocketPool}'::text[], 'string'), - ('{transport,connection,id}'::text[], 'string'), - ('{transport,connection,exitReason}'::text[], 'string'), - ('{transport,connection,ageMs}'::text[], 'number'), - ('{transport,connection,idleMs}'::text[], 'number'), - ('{response,responseModel}'::text[], 'string'), - ('{continuation,previousResponseIdHash}'::text[], 'string'), - ('{continuation,unavailableReason}'::text[], 'string'), - ('{recovery,retryDelayMs}'::text[], 'number'), - ('{recovery,totalLatencyMs}'::text[], 'number'), - ('{timings,upstream,responseMs}'::text[], 'number')) as fields(path, kind) loop - item := document #> spec.path; - if item is null or item = 'null'::jsonb then continue; end if; - if spec.kind = 'milliseconds' then - if jsonb_typeof(item) <> 'number' then return false; end if; - number := (item #>> '{}')::numeric; - if number < 0 or number > 9223372036854775807 then return false; end if; - continue; - end if; - if jsonb_typeof(item) <> spec.kind then return false; end if; - if spec.kind = 'number' then - number := (item #>> '{}')::numeric; - if item #>> '{}' !~ '^[0-9]+$' or number > 9223372036854775807 then return false; end if; - elsif spec.kind = 'array' then - if exists (select 1 from jsonb_array_elements(item) member where jsonb_typeof(member) <> 'string') then return false; end if; - end if; - end loop; - if (document #> '{request,configRevision}' is not null - and document #> '{request,configRevision}' <> 'null'::jsonb - and document #> '{request,compact}' in ('true'::jsonb, 'false'::jsonb) - and coalesce(length(document #>> '{request,protocol}'), 0) > 0 - and coalesce(length(document #>> '{request,endpoint}'), 0) > 0 - and jsonb_typeof(document #> '{routing,scope}') = 'string' - and jsonb_typeof(document #> '{routing,groupRefs}') = 'array' - and jsonb_typeof(document #> '{routing,groupNamesSnapshot}') = 'array') is not true then - return false; - end if; - if document #>> '{request,clientIp}' is not null then - perform (document #>> '{request,clientIp}')::inet; - end if; - return true; -exception when invalid_text_representation or numeric_value_out_of_range then - return false; -end; -$$; - -drop view model_request_observations; -alter table model_requests - drop constraint model_requests_client_fk, - drop constraint model_requests_account_fk, - drop constraint model_requests_client_ref_ck, - drop constraint model_requests_account_ref_ck, - drop constraint model_requests_fact_completeness_ck, - drop column client_api_key_id, - drop column provider_account_id, - add constraint model_requests_fact_completeness_ck check (cost_source = 'unavailable' or cost_currency is not null), - add constraint model_requests_client_ref_ck check (octet_length(client_api_key_ref) between 1 and 128), - add constraint model_requests_account_ref_ck check (provider_account_ref is null or octet_length(provider_account_ref) between 1 and 128); -alter table ops_events - drop constraint ops_events_account_fk, - drop constraint ops_events_account_ref_ck, - drop constraint ops_events_account_snapshot_present_ck, - drop column provider_account_id, - add column provider_account_plan_type_snapshot text, - add constraint ops_events_account_ref_ck check (provider_account_ref is null or octet_length(provider_account_ref) between 1 and 128); - -create or replace view model_request_observations as -select mr.*, - (mr.request_observation_json #>> '{request,configRevision}')::bigint as config_revision, - (mr.request_observation_json #>> '{request,protocol}') as protocol, - (mr.request_observation_json #>> '{request,endpoint}') as endpoint, - (mr.request_observation_json #>> '{account,name}') as provider_account_name_snapshot, - (mr.request_observation_json #>> '{account,email}') as provider_account_email_snapshot, - (mr.request_observation_json #>> '{account,authenticationKind}') as provider_account_authentication_kind_snapshot, - (mr.request_observation_json #>> '{routing,scope}') as routing_scope, - array(select jsonb_array_elements_text(mr.request_observation_json #> '{routing,groupRefs}')) as routing_group_refs, - (mr.request_observation_json #> '{routing,groupNamesSnapshot}') as routing_group_names_snapshot, - (mr.request_observation_json #>> '{error,providerErrorCode}') as provider_error_code, - (mr.request_observation_json #>> '{error,message}') as error_message, - (mr.request_observation_json #>> '{error,retryAfterMs}')::bigint as retry_after_ms, - (mr.request_observation_json #>> '{request,clientIp}')::inet as client_ip, - (mr.request_observation_json #>> '{request,userAgent}') as user_agent, - (mr.request_observation_json #>> '{request,reasoningEffort}') as reasoning_effort, - (mr.request_observation_json #>> '{request,reasoningPreset}') as reasoning_preset, - (mr.request_observation_json #>> '{request,subagentKind}') as subagent_kind, - (mr.request_observation_json #>> '{request,compact}')::boolean as compact, - measured.transport_decision_wait_ms as transport_decision_wait_ms, - measured.connect_ms as connect_ms, - measured.headers_ms as headers_ms, - measured.first_event_ms as first_event_ms, - measured.first_reasoning_ms as first_reasoning_ms, - measured.first_text_ms as first_text_ms, - measured.first_token_ms as first_token_ms, - measured.latency_ms as latency_ms, - measured.provider_processing_ms as provider_processing_ms, - measured.admission_decision_ms as admission_decision_ms, - measured.account_selection_wait_ms as account_selection_wait_ms, - measured.capacity_used_slots as capacity_used_slots, - measured.capacity_total_slots as capacity_total_slots, - (mr.request_observation_json #>> '{transport,httpVersion}') as http_version, - (mr.request_observation_json #>> '{transport,websocketPool}') as websocket_pool, - (mr.request_observation_json #>> '{transport,connection,id}') as upstream_connection_id, - (mr.request_observation_json #>> '{transport,connection,exitReason}') as upstream_connection_exit_reason, - (mr.request_observation_json #>> '{transport,connection,ageMs}')::bigint as upstream_connection_age_ms, - (mr.request_observation_json #>> '{transport,connection,idleMs}')::bigint as upstream_connection_idle_ms, - (mr.request_observation_json #>> '{response,responseModel}') as upstream_response_model, - (mr.request_observation_json #>> '{continuation,previousResponseIdHash}') as continuation_previous_response_id_hash, - (mr.request_observation_json #>> '{continuation,unavailableReason}') as continuation_unavailable_reason, - (mr.request_observation_json #>> '{recovery,retryDelayMs}')::bigint as recovery_retry_delay_ms, - (mr.request_observation_json #>> '{recovery,totalLatencyMs}')::bigint as recovery_total_latency_ms, - (mr.request_observation_json #>> '{timings,upstream,apiOverheadMs}')::double precision as upstream_api_overhead_ms, - (mr.request_observation_json #>> '{timings,upstream,engineMs}')::double precision as upstream_engine_ms, - (mr.request_observation_json #>> '{timings,upstream,engineIapiTtftMs}')::double precision as upstream_engine_iapi_ttft_ms, - (mr.request_observation_json #>> '{timings,upstream,engineServiceTtftMs}')::double precision as upstream_engine_service_ttft_ms, - (mr.request_observation_json #>> '{timings,upstream,engineIapiTbtMs}')::double precision as upstream_engine_iapi_tbt_ms, - (mr.request_observation_json #>> '{timings,upstream,engineServiceTbtMs}')::double precision as upstream_engine_service_tbt_ms, - measured.upstream_response_ms as upstream_response_ms, - mr.request_observation_json #>> '{request,clientApiKeyNameSnapshot}' as client_api_key_name_snapshot, - mr.request_observation_json #>> '{account,planTypeSnapshot}' as provider_account_plan_type_snapshot, - mr.request_observation_json #>> '{account,notesSnapshot}' as provider_account_notes_snapshot -from model_requests mr -cross join lateral ( - select (mr.request_observation_json #>> '{timings,local,transportDecisionWaitMs}')::bigint as transport_decision_wait_ms, - (mr.request_observation_json #>> '{timings,local,connectMs}')::bigint as connect_ms, - (mr.request_observation_json #>> '{timings,local,headersMs}')::bigint as headers_ms, - (mr.request_observation_json #>> '{timings,local,firstEventMs}')::bigint as first_event_ms, - (mr.request_observation_json #>> '{timings,local,firstReasoningMs}')::bigint as first_reasoning_ms, - (mr.request_observation_json #>> '{timings,local,firstTextMs}')::bigint as first_text_ms, - (mr.request_observation_json #>> '{timings,local,firstTokenMs}')::bigint as first_token_ms, - (mr.request_observation_json #>> '{timings,local,latencyMs}')::bigint as latency_ms, - (mr.request_observation_json #>> '{timings,upstream,processingMs}')::bigint as provider_processing_ms, - (mr.request_observation_json #>> '{scheduling,admissionDecisionMs}')::bigint as admission_decision_ms, - (mr.request_observation_json #>> '{scheduling,accountSelectionWaitMs}')::bigint as account_selection_wait_ms, - (mr.request_observation_json #>> '{scheduling,capacityUsedSlots}')::bigint as capacity_used_slots, - (mr.request_observation_json #>> '{scheduling,capacityTotalSlots}')::bigint as capacity_total_slots, - (mr.request_observation_json #>> '{timings,upstream,responseMs}')::bigint as upstream_response_ms -) measured; diff --git a/backend/migrations/0031_request_recovery_leases.sql b/backend/migrations/0031_request_recovery_leases.sql deleted file mode 100644 index a3e6d0f15..000000000 --- a/backend/migrations/0031_request_recovery_leases.sql +++ /dev/null @@ -1,107 +0,0 @@ --- 租约只保存运行态,历史表不承担高频续期更新 --- 迁移在停止旧写入者的窗口执行,旧二进制不可直接回退 -create table model_request_leases ( - request_id text primary key references model_requests(id) on delete cascade, - lease_expires_at timestamptz not null -); -create index model_request_leases_expiry_idx on model_request_leases (lease_expires_at, request_id); -insert into model_request_leases (request_id, lease_expires_at) -select id, deadline_at from model_requests where outcome = 'running'; -drop view model_request_observations; -drop index model_requests_running_deadline_idx; -alter table model_requests rename column deadline_at to recovery_lease_expires_at; -alter table model_requests alter column recovery_lease_expires_at drop not null; -update model_requests set recovery_lease_expires_at = null where outcome = 'running'; - --- 主请求锁先于租约锁,终结和恢复共用同一顺序;续期只取得租约锁 -create function finish_model_request_lease() returns trigger -language plpgsql as $$ -declare - expires_at timestamptz; -begin - if old.outcome = 'running' and new.outcome <> 'running' then - select lease_expires_at into expires_at from model_request_leases - where request_id = new.id for update; - new.recovery_lease_expires_at := coalesce(expires_at, old.recovery_lease_expires_at, old.started_at); - delete from model_request_leases where request_id = new.id; - end if; - return new; -end; -$$; -create trigger model_requests_finish_lease before update on model_requests - for each row when (old.outcome = 'running' and new.outcome <> 'running') - execute function finish_model_request_lease(); - -create or replace view model_request_observations as -select mr.*, - coalesce(lease.lease_expires_at, mr.recovery_lease_expires_at) as effective_recovery_lease_expires_at, - (mr.request_observation_json #>> '{request,configRevision}')::bigint as config_revision, - (mr.request_observation_json #>> '{request,protocol}') as protocol, - (mr.request_observation_json #>> '{request,endpoint}') as endpoint, - (mr.request_observation_json #>> '{account,name}') as provider_account_name_snapshot, - (mr.request_observation_json #>> '{account,email}') as provider_account_email_snapshot, - (mr.request_observation_json #>> '{account,authenticationKind}') as provider_account_authentication_kind_snapshot, - (mr.request_observation_json #>> '{routing,scope}') as routing_scope, - array(select jsonb_array_elements_text(mr.request_observation_json #> '{routing,groupRefs}')) as routing_group_refs, - (mr.request_observation_json #> '{routing,groupNamesSnapshot}') as routing_group_names_snapshot, - (mr.request_observation_json #>> '{error,providerErrorCode}') as provider_error_code, - (mr.request_observation_json #>> '{error,message}') as error_message, - (mr.request_observation_json #>> '{error,retryAfterMs}')::bigint as retry_after_ms, - (mr.request_observation_json #>> '{request,clientIp}')::inet as client_ip, - (mr.request_observation_json #>> '{request,userAgent}') as user_agent, - (mr.request_observation_json #>> '{request,reasoningEffort}') as reasoning_effort, - (mr.request_observation_json #>> '{request,reasoningPreset}') as reasoning_preset, - (mr.request_observation_json #>> '{request,subagentKind}') as subagent_kind, - (mr.request_observation_json #>> '{request,compact}')::boolean as compact, - measured.transport_decision_wait_ms as transport_decision_wait_ms, - measured.connect_ms as connect_ms, - measured.headers_ms as headers_ms, - measured.first_event_ms as first_event_ms, - measured.first_reasoning_ms as first_reasoning_ms, - measured.first_text_ms as first_text_ms, - measured.first_token_ms as first_token_ms, - measured.latency_ms as latency_ms, - measured.provider_processing_ms as provider_processing_ms, - measured.admission_decision_ms as admission_decision_ms, - measured.account_selection_wait_ms as account_selection_wait_ms, - measured.capacity_used_slots as capacity_used_slots, - measured.capacity_total_slots as capacity_total_slots, - (mr.request_observation_json #>> '{transport,httpVersion}') as http_version, - (mr.request_observation_json #>> '{transport,websocketPool}') as websocket_pool, - (mr.request_observation_json #>> '{transport,connection,id}') as upstream_connection_id, - (mr.request_observation_json #>> '{transport,connection,exitReason}') as upstream_connection_exit_reason, - (mr.request_observation_json #>> '{transport,connection,ageMs}')::bigint as upstream_connection_age_ms, - (mr.request_observation_json #>> '{transport,connection,idleMs}')::bigint as upstream_connection_idle_ms, - (mr.request_observation_json #>> '{response,responseModel}') as upstream_response_model, - (mr.request_observation_json #>> '{continuation,previousResponseIdHash}') as continuation_previous_response_id_hash, - (mr.request_observation_json #>> '{continuation,unavailableReason}') as continuation_unavailable_reason, - (mr.request_observation_json #>> '{recovery,retryDelayMs}')::bigint as recovery_retry_delay_ms, - (mr.request_observation_json #>> '{recovery,totalLatencyMs}')::bigint as recovery_total_latency_ms, - (mr.request_observation_json #>> '{timings,upstream,apiOverheadMs}')::double precision as upstream_api_overhead_ms, - (mr.request_observation_json #>> '{timings,upstream,engineMs}')::double precision as upstream_engine_ms, - (mr.request_observation_json #>> '{timings,upstream,engineIapiTtftMs}')::double precision as upstream_engine_iapi_ttft_ms, - (mr.request_observation_json #>> '{timings,upstream,engineServiceTtftMs}')::double precision as upstream_engine_service_ttft_ms, - (mr.request_observation_json #>> '{timings,upstream,engineIapiTbtMs}')::double precision as upstream_engine_iapi_tbt_ms, - (mr.request_observation_json #>> '{timings,upstream,engineServiceTbtMs}')::double precision as upstream_engine_service_tbt_ms, - measured.upstream_response_ms as upstream_response_ms, - mr.request_observation_json #>> '{request,clientApiKeyNameSnapshot}' as client_api_key_name_snapshot, - mr.request_observation_json #>> '{account,planTypeSnapshot}' as provider_account_plan_type_snapshot, - mr.request_observation_json #>> '{account,notesSnapshot}' as provider_account_notes_snapshot -from model_requests mr -left join model_request_leases lease on lease.request_id = mr.id -cross join lateral ( - select (mr.request_observation_json #>> '{timings,local,transportDecisionWaitMs}')::bigint as transport_decision_wait_ms, - (mr.request_observation_json #>> '{timings,local,connectMs}')::bigint as connect_ms, - (mr.request_observation_json #>> '{timings,local,headersMs}')::bigint as headers_ms, - (mr.request_observation_json #>> '{timings,local,firstEventMs}')::bigint as first_event_ms, - (mr.request_observation_json #>> '{timings,local,firstReasoningMs}')::bigint as first_reasoning_ms, - (mr.request_observation_json #>> '{timings,local,firstTextMs}')::bigint as first_text_ms, - (mr.request_observation_json #>> '{timings,local,firstTokenMs}')::bigint as first_token_ms, - (mr.request_observation_json #>> '{timings,local,latencyMs}')::bigint as latency_ms, - (mr.request_observation_json #>> '{timings,upstream,processingMs}')::bigint as provider_processing_ms, - (mr.request_observation_json #>> '{scheduling,admissionDecisionMs}')::bigint as admission_decision_ms, - (mr.request_observation_json #>> '{scheduling,accountSelectionWaitMs}')::bigint as account_selection_wait_ms, - (mr.request_observation_json #>> '{scheduling,capacityUsedSlots}')::bigint as capacity_used_slots, - (mr.request_observation_json #>> '{scheduling,capacityTotalSlots}')::bigint as capacity_total_slots, - (mr.request_observation_json #>> '{timings,upstream,responseMs}')::bigint as upstream_response_ms -) measured; diff --git a/backend/migrations/0032_request_token_names.sql b/backend/migrations/0032_request_token_names.sql deleted file mode 100644 index 5b6fd9ce1..000000000 --- a/backend/migrations/0032_request_token_names.sql +++ /dev/null @@ -1,7 +0,0 @@ --- 只更名,不改变 Token 的包含关系或费用语义 -alter table model_requests rename column cached_tokens to cached_input_tokens; -alter table model_requests rename column cache_write_tokens to cache_write_input_tokens; -alter table model_requests rename column reasoning_tokens to reasoning_output_tokens; -alter view model_request_observations rename column cached_tokens to cached_input_tokens; -alter view model_request_observations rename column cache_write_tokens to cache_write_input_tokens; -alter view model_request_observations rename column reasoning_tokens to reasoning_output_tokens; diff --git a/backend/migrations/0033_client_key_charge_event_index.sql b/backend/migrations/0033_client_key_charge_event_index.sql deleted file mode 100644 index dbc8b758b..000000000 --- a/backend/migrations/0033_client_key_charge_event_index.sql +++ /dev/null @@ -1,3 +0,0 @@ --- 删除 Key 时按稳定身份定位费用事件,避免扫描其他 Key 的完整账本 -create index client_key_charge_events_client_api_key_id - on client_key_charge_events (client_api_key_id); diff --git a/backend/migrations/0034_request_analytics.sql b/backend/migrations/0034_request_analytics.sql deleted file mode 100644 index 9013a1741..000000000 --- a/backend/migrations/0034_request_analytics.sql +++ /dev/null @@ -1,239 +0,0 @@ --- 请求观测的可重建窄事实、持久化工作队列与分层统计 -create sequence request_analytics_change_version; -create table request_analytics_facts as select - mr.id, mr.started_at, mr.completed_at, mr.client_api_key_ref, mr.provider_account_ref, mr.routing_group_refs, mr.operation, mr.provider_kind, mr.requested_model_id, mr.upstream_model_id, mr.outcome, mr.client_status_code, mr.upstream_status_code, mr.client_transport, mr.upstream_transport, mr.attempt_count, mr.client_response_id, mr.upstream_response_id, mr.upstream_request_id, mr.recovered_at, mr.downstream_committed_at, mr.request_kind, mr.image_generation_requested, mr.image_generation_succeeded, mr.error_kind, mr.input_tokens, mr.output_tokens, mr.cached_input_tokens, mr.cache_write_input_tokens, mr.reasoning_output_tokens, mr.image_input_tokens, mr.image_output_tokens, mr.total_tokens, mr.cost_source, mr.cost_currency, mr.cost_amount, mr.first_token_ms, mr.latency_ms, mr.admission_decision_ms, mr.account_selection_wait_ms, mr.upstream_response_ms, mr.capacity_used_slots, mr.capacity_total_slots -from model_request_observations mr with no data; -alter table request_analytics_facts add primary key (id), - add column standard_cost_amount numeric(20,10), - add column no_cache_cost_amount numeric(20,10), - add column source_version bigint not null, - add column schema_version integer not null default 1, - add column ops_failure_count bigint not null default 0, - add column ops_rate_limited_count bigint not null default 0, - add column ops_auth_failure_count bigint not null default 0, - add column ops_provider_5xx_count bigint not null default 0; -create index request_analytics_facts_started on request_analytics_facts (started_at, id); -create index request_analytics_facts_key on request_analytics_facts (client_api_key_ref, started_at, id); -create index request_analytics_facts_account on request_analytics_facts (provider_account_ref, started_at, id); --- 分组集合未预聚合,窄事实回退仍需按集合定位,避免低命中筛选扫描全部保留历史 -create index request_analytics_facts_groups on request_analytics_facts using gin (routing_group_refs); --- 用量列表的精确总数与账号最近使用只需要交付事实,部分索引避免为判定资格读取宽行 -create index model_requests_ops_completed on model_requests (completed_at,id) -where error_kind is not null and error_kind <> 'cancelled'; -create index model_requests_delivered_started on model_requests (started_at,id) -where outcome='succeeded' and downstream_committed_at is not null - and (provider_kind is distinct from 'openai' or request_kind is distinct from 'prewarm') - and ((client_transport='websocket' and client_status_code is null) or client_status_code between 200 and 399) - and (requested_model_id is not null or upstream_model_id is not null or image_generation_requested - or input_tokens is not null or output_tokens is not null or cached_input_tokens is not null - or cache_write_input_tokens is not null or reasoning_output_tokens is not null - or image_input_tokens is not null or image_output_tokens is not null - or total_tokens is not null or cost_amount is not null); -create index request_analytics_facts_delivered_account on request_analytics_facts (provider_account_ref,started_at desc,id desc) -where outcome='succeeded' and downstream_committed_at is not null - and (provider_kind is distinct from 'openai' or request_kind is distinct from 'prewarm') - and ((client_transport='websocket' and client_status_code is null) or client_status_code between 200 and 399) - and (requested_model_id is not null or upstream_model_id is not null or image_generation_requested - or input_tokens is not null or output_tokens is not null or cached_input_tokens is not null - or cache_write_input_tokens is not null or reasoning_output_tokens is not null - or image_input_tokens is not null or image_output_tokens is not null - or total_tokens is not null or cost_amount is not null); --- 不透明 ID 没有私有长度上限,等值 HASH 索引避免长值超过 B-tree 索引项限制 -create index request_analytics_facts_client_response on request_analytics_facts using hash (client_response_id) where client_response_id is not null; -create index request_analytics_facts_upstream_response on request_analytics_facts using hash (upstream_response_id) where upstream_response_id is not null; -create index request_analytics_facts_upstream_request on request_analytics_facts using hash (upstream_request_id) where upstream_request_id is not null; -create table request_analytics_pending_requests ( - request_id text primary key, - change_version bigint not null default nextval('request_analytics_change_version'), - oldest_pending_at timestamptz not null default clock_timestamp(), - previous_bucket timestamptz, - deletion_pending boolean not null default false -); -create index request_analytics_pending_deletions on request_analytics_pending_requests (previous_bucket) where deletion_pending; -create index request_analytics_pending_requests_age on request_analytics_pending_requests (oldest_pending_at, request_id); -create table request_analytics_pending_buckets ( - bucket_start timestamptz primary key, - change_version bigint not null default nextval('request_analytics_change_version'), - oldest_pending_at timestamptz not null, - build_version bigint, - build_phase smallint not null default 0 check (build_phase between 0 and 4), - build_started_at timestamptz, - build_id text, - build_collection text, - build_dimensions text[], - build_outcome text -); -create index request_analytics_pending_buckets_age on request_analytics_pending_buckets(oldest_pending_at,bucket_start); -create table request_analytics_15m ( - bucket_start timestamptz not null, - collection text not null, - dimensions text[] not null, - outcome text not null, - schema_version integer not null, - histogram_version integer not null, - counts jsonb not null, - request_count bigint generated always as ((counts->>'request_count')::bigint) stored, - costs jsonb not null, - distributions jsonb not null, - primary key (bucket_start, collection, dimensions, outcome) -); -create table request_analytics_hourly (like request_analytics_15m including all); -create index request_analytics_15m_collection on request_analytics_15m (collection,bucket_start); -create index request_analytics_hourly_collection on request_analytics_hourly (collection,bucket_start); -create index request_analytics_15m_dimension on request_analytics_15m (collection,(dimensions[1]),bucket_start); -create index request_analytics_hourly_dimension on request_analytics_hourly (collection,(dimensions[1]),bucket_start); --- 排名只读取确定派生的计数,覆盖索引避免为高基数账号和密钥读取宽分布行 -create index request_analytics_15m_rank on request_analytics_15m (collection,bucket_start) include (dimensions,outcome,schema_version,histogram_version,request_count) where collection in ('account','key'); -create index request_analytics_hourly_rank on request_analytics_hourly (collection,bucket_start) include (dimensions,outcome,schema_version,histogram_version,request_count) where collection in ('account','key'); --- 高频身份排名只保存日级精确计数,不复制分布与诊断详情 -create table request_analytics_daily_counts ( - bucket_start timestamptz not null, - collection text not null check (collection in ('account','key')), - dimension text not null, - outcome text not null, - schema_version integer not null, - histogram_version integer not null, - request_count bigint not null check (request_count >= 0), - primary key (collection,bucket_start,dimension,outcome,schema_version,histogram_version) -); - --- 计数由已持久化小时行的 OLD/NEW 差额派生,重放同值不会重复累计 --- 未发布小时所在的日期不参与日级查询,分段重建与版本清理仍可逐批提交 -create function sync_request_analytics_daily_counts() returns trigger language plpgsql as $$ -declare - old_day timestamptz; - new_day timestamptz; - delta bigint; -begin - if TG_OP <> 'INSERT' then - old_day := date_bin(interval '1 day',OLD.bucket_start,timestamptz '2000-01-01 UTC'); - end if; - if TG_OP <> 'DELETE' then - new_day := date_bin(interval '1 day',NEW.bucket_start,timestamptz '2000-01-01 UTC'); - end if; - if TG_OP = 'UPDATE' and - (old_day,OLD.collection,OLD.dimensions[1],OLD.outcome,OLD.schema_version,OLD.histogram_version) - is not distinct from - (new_day,NEW.collection,NEW.dimensions[1],NEW.outcome,NEW.schema_version,NEW.histogram_version) then - if NEW.collection in ('account','key') then - delta := NEW.request_count - OLD.request_count; - if delta <> 0 then - update request_analytics_daily_counts set request_count=request_count+delta - where collection=NEW.collection and bucket_start=new_day and dimension=NEW.dimensions[1] - and outcome=NEW.outcome and schema_version=NEW.schema_version and histogram_version=NEW.histogram_version; - end if; - end if; - return null; - end if; - if TG_OP <> 'INSERT' and OLD.collection in ('account','key') then - update request_analytics_daily_counts set request_count=request_count-OLD.request_count - where collection=OLD.collection and bucket_start=old_day and dimension=OLD.dimensions[1] - and outcome=OLD.outcome and schema_version=OLD.schema_version and histogram_version=OLD.histogram_version; - delete from request_analytics_daily_counts - where collection=OLD.collection and bucket_start=old_day and dimension=OLD.dimensions[1] - and outcome=OLD.outcome and schema_version=OLD.schema_version and histogram_version=OLD.histogram_version - and request_count=0; - end if; - if TG_OP <> 'DELETE' and NEW.collection in ('account','key') then - insert into request_analytics_daily_counts values(new_day,NEW.collection,NEW.dimensions[1],NEW.outcome,NEW.schema_version,NEW.histogram_version,NEW.request_count) - on conflict(collection,bucket_start,dimension,outcome,schema_version,histogram_version) - do update set request_count=request_analytics_daily_counts.request_count+excluded.request_count; - end if; - return null; -end; -$$; -create trigger request_analytics_hourly_daily_counts - after insert or update or delete on request_analytics_hourly - for each row execute function sync_request_analytics_daily_counts(); - -create table request_analytics_build_state ( - singleton boolean primary key default true check (singleton), - schema_version integer not null default 1, - histogram_version integer not null default 1, - coverage_start timestamptz, - coverage_end timestamptz, - backfill_cursor_started_at timestamptz, - backfill_cursor_id text, - backfill_complete boolean not null default false, - reset_phase smallint not null default 0 check (reset_phase between 0 and 4), - reset_cursor jsonb, - generation bigint not null default 0, - updated_at timestamptz not null default clock_timestamp(), - check (reset_phase <> 0 or reset_cursor is null) -); -insert into request_analytics_build_state (singleton) values (true); - -create function enqueue_request_analytics(request_ref text, request_start timestamptz, source_deleted boolean default false) -returns void language sql as $$ - insert into request_analytics_pending_requests(request_id, previous_bucket, deletion_pending) - select request_ref, date_bin(interval '15 minutes', coalesce(request_start,(select started_at from request_analytics_facts where id=request_ref)), timestamptz '2000-01-01 UTC'), source_deleted - where request_ref is not null - on conflict (request_id) do update - set change_version = nextval('request_analytics_change_version'), - previous_bucket = coalesce(request_analytics_pending_requests.previous_bucket, excluded.previous_bucket), - deletion_pending = request_analytics_pending_requests.deletion_pending or excluded.deletion_pending -$$; -create function observe_request_analytics_change() returns trigger language plpgsql as $$ -begin - if TG_OP = 'DELETE' then - perform enqueue_request_analytics(OLD.id, OLD.started_at, true); - return OLD; - end if; - perform enqueue_request_analytics(NEW.id, NEW.started_at); - return NEW; -end -$$; -create trigger request_analytics_insert after insert on model_requests -for each row execute function observe_request_analytics_change(); -create trigger request_analytics_delete after delete on model_requests -for each row execute function observe_request_analytics_change(); -create trigger request_analytics_update after update on model_requests -for each row when (ROW(OLD.id, OLD.started_at, OLD.completed_at, OLD.client_api_key_ref, OLD.provider_account_ref, OLD.operation, OLD.provider_kind, OLD.requested_model_id, OLD.upstream_model_id, OLD.outcome, OLD.client_status_code, OLD.upstream_status_code, OLD.client_transport, OLD.upstream_transport, OLD.attempt_count, OLD.client_response_id, OLD.upstream_response_id, OLD.upstream_request_id, OLD.recovered_at, OLD.downstream_committed_at, OLD.request_kind, OLD.image_generation_requested, OLD.image_generation_succeeded, OLD.error_kind, OLD.input_tokens, OLD.output_tokens, OLD.cached_input_tokens, OLD.cache_write_input_tokens, OLD.reasoning_output_tokens, OLD.image_input_tokens, OLD.image_output_tokens, OLD.total_tokens, OLD.cost_source, OLD.cost_currency, OLD.cost_amount) is distinct from ROW(NEW.id, NEW.started_at, NEW.completed_at, NEW.client_api_key_ref, NEW.provider_account_ref, NEW.operation, NEW.provider_kind, NEW.requested_model_id, NEW.upstream_model_id, NEW.outcome, NEW.client_status_code, NEW.upstream_status_code, NEW.client_transport, NEW.upstream_transport, NEW.attempt_count, NEW.client_response_id, NEW.upstream_response_id, NEW.upstream_request_id, NEW.recovered_at, NEW.downstream_committed_at, NEW.request_kind, NEW.image_generation_requested, NEW.image_generation_succeeded, NEW.error_kind, NEW.input_tokens, NEW.output_tokens, NEW.cached_input_tokens, NEW.cache_write_input_tokens, NEW.reasoning_output_tokens, NEW.image_input_tokens, NEW.image_output_tokens, NEW.total_tokens, NEW.cost_source, NEW.cost_currency, NEW.cost_amount) - or OLD.billing_snapshot_json is distinct from NEW.billing_snapshot_json - or OLD.service_tier is distinct from NEW.service_tier - or OLD.request_observation_json->'timings' is distinct from NEW.request_observation_json->'timings' - or OLD.request_observation_json->'scheduling' is distinct from NEW.request_observation_json->'scheduling' - or OLD.request_observation_json->'routing' is distinct from NEW.request_observation_json->'routing') -execute function observe_request_analytics_change(); -create function observe_ops_analytics_change() returns trigger language plpgsql as $$ -begin - if TG_OP <> 'INSERT' then - perform enqueue_request_analytics(OLD.model_request_id, null, TG_OP = 'DELETE'); - end if; - if TG_OP <> 'DELETE' then - perform enqueue_request_analytics(NEW.model_request_id, null); - return NEW; - end if; - return OLD; -end -$$; -create trigger request_analytics_ops_insert after insert on ops_events -for each row execute function observe_ops_analytics_change(); -create trigger request_analytics_ops_delete after delete on ops_events -for each row execute function observe_ops_analytics_change(); -create trigger request_analytics_ops_update after update on ops_events -for each row when (ROW(OLD.model_request_id, OLD.failure_kind, OLD.status_code, OLD.attempt_index) - is distinct from ROW(NEW.model_request_id, NEW.failure_kind, NEW.status_code, NEW.attempt_index)) -execute function observe_ops_analytics_change(); - --- 删除在源事务内使缓存失效,语句级触发避免批量清理争用全局行 -create function invalidate_deleted_request_analytics() returns trigger language plpgsql as $$ -begin - update request_analytics_build_state set generation = generation + 1 where singleton; - return null; -end -$$; -create trigger request_analytics_delete_invalidation after delete on model_requests -for each statement execute function invalidate_deleted_request_analytics(); -create function invalidate_deleted_ops_analytics() returns trigger language plpgsql as $$ -begin - -- 父请求删除已经负责失效;独立删除仍有关联源请求的 Ops 才需要另行失效 - if exists (select 1 from deleted_ops join model_requests mr on mr.id = deleted_ops.model_request_id) then - update request_analytics_build_state set generation = generation + 1 where singleton; - end if; - return null; -end -$$; -create trigger request_analytics_ops_delete_invalidation after delete on ops_events -referencing old table as deleted_ops -for each statement execute function invalidate_deleted_ops_analytics(); diff --git a/backend/migrations/0035_request_analytics_delete_invalidation.sql b/backend/migrations/0035_request_analytics_delete_invalidation.sql deleted file mode 100644 index 3ce5fbaf2..000000000 --- a/backend/migrations/0035_request_analytics_delete_invalidation.sql +++ /dev/null @@ -1,14 +0,0 @@ --- 空清理不更新统计代次,避免与启动回填的快照事务发生无效写冲突 -create or replace function invalidate_deleted_request_analytics() returns trigger language plpgsql as $$ -begin - if exists (select 1 from deleted_requests) then - update request_analytics_build_state set generation = generation + 1 where singleton; - end if; - return null; -end -$$; - -drop trigger request_analytics_delete_invalidation on model_requests; -create trigger request_analytics_delete_invalidation after delete on model_requests -referencing old table as deleted_requests -for each statement execute function invalidate_deleted_request_analytics();