From 321e6943bbf2cc7a0e3c28e652ea90c0619edb16 Mon Sep 17 00:00:00 2001 From: zloglevel Date: Wed, 19 Aug 2026 10:02:21 +0800 Subject: [PATCH] fix(order-stream): preserve Redshift telemetry field compatibility Signed-off-by: zloglevel --- crates/order-stream/tests/telemetry.rs | 31 ++++++- infra/order-stream/README.md | 4 + ..._preserve_telemetry_completion_aliases.sql | 80 +++++++++++++++++++ 3 files changed, 113 insertions(+), 2 deletions(-) create mode 100644 infra/order-stream/redshift-migrations/005_preserve_telemetry_completion_aliases.sql diff --git a/crates/order-stream/tests/telemetry.rs b/crates/order-stream/tests/telemetry.rs index 881e9bae0c..6401b6bed7 100644 --- a/crates/order-stream/tests/telemetry.rs +++ b/crates/order-stream/tests/telemetry.rs @@ -231,7 +231,8 @@ async fn telemetry_end_to_end() { let mut eval_found = false; for attempt in 1..=max_attempts { let row = sqlx::query( - "SELECT broker_address, outcome FROM telemetry.request_evaluations \ + "SELECT broker_address, outcome, preflight_duration_ms \ + FROM telemetry.request_evaluations \ WHERE request_id = $1 \ LIMIT 1", ) @@ -243,8 +244,10 @@ async fn telemetry_end_to_end() { Ok(Some(r)) => { let db_broker: String = r.get("broker_address"); let db_outcome: String = r.get("outcome"); + let db_preflight_duration: i64 = r.get("preflight_duration_ms"); assert_eq!(db_broker, broker_addr_str, "eval broker_address mismatch"); assert_eq!(db_outcome, "Locked", "eval outcome mismatch"); + assert_eq!(db_preflight_duration, 800, "eval preflight_duration_ms mismatch"); eprintln!(" Evaluation found on attempt {attempt}"); eval_found = true; break; @@ -266,7 +269,10 @@ async fn telemetry_end_to_end() { let mut comp_found = false; for attempt in 1..=max_attempts { let row = sqlx::query( - "SELECT broker_address, outcome, total_duration_secs FROM telemetry.request_completions \ + "SELECT broker_address, outcome, total_duration_secs, \ + committed_to_application_proof_duration_secs, proving_duration_secs, \ + actual_total_proving_time_secs, actual_proving_time_secs \ + FROM telemetry.request_completions \ WHERE request_id = $1 \ LIMIT 1", ) @@ -279,9 +285,30 @@ async fn telemetry_end_to_end() { let db_broker: String = r.get("broker_address"); let db_outcome: String = r.get("outcome"); let db_total_dur: i64 = r.get("total_duration_secs"); + let db_application_proof_dur: i64 = + r.get("committed_to_application_proof_duration_secs"); + let db_legacy_proving_dur: i64 = r.get("proving_duration_secs"); + let db_actual_total_proving_dur: i64 = r.get("actual_total_proving_time_secs"); + let db_legacy_actual_proving_dur: i64 = r.get("actual_proving_time_secs"); assert_eq!(db_broker, broker_addr_str, "comp broker_address mismatch"); assert_eq!(db_outcome, "Fulfilled", "comp outcome mismatch"); assert_eq!(db_total_dur, 55, "comp total_duration_secs mismatch"); + assert_eq!( + db_application_proof_dur, 30, + "comp application proof duration mismatch" + ); + assert_eq!( + db_legacy_proving_dur, db_application_proof_dur, + "legacy proving alias mismatch" + ); + assert_eq!( + db_actual_total_proving_dur, 30, + "comp actual proving duration mismatch" + ); + assert_eq!( + db_legacy_actual_proving_dur, db_actual_total_proving_dur, + "legacy actual proving alias mismatch" + ); eprintln!(" Completion found on attempt {attempt}"); comp_found = true; break; diff --git a/infra/order-stream/README.md b/infra/order-stream/README.md index 0793fd9b02..7f150de44f 100644 --- a/infra/order-stream/README.md +++ b/infra/order-stream/README.md @@ -59,6 +59,10 @@ In deployment pipelines, apply script is run automatically after each deployment To add a new migration, create the next numbered file (e.g. `004_add_new_field.sql`) and re-run `apply.sh`. Since Redshift does not allow `CREATE OR REPLACE VIEW` when the column list changes, migrations that add or remove columns must `DROP VIEW` and recreate it (and re-grant permissions to the `readonly` user). +The completion view exposes both the current timing fields and the historical +`proving_duration_secs` / `actual_proving_time_secs` aliases. The broker no +longer emits `preflight_cache_hit`; use `preflight_duration_ms` instead. + ## Querying Redshift Connect using any Postgres client with the read-only credentials: diff --git a/infra/order-stream/redshift-migrations/005_preserve_telemetry_completion_aliases.sql b/infra/order-stream/redshift-migrations/005_preserve_telemetry_completion_aliases.sql new file mode 100644 index 0000000000..23ff38f2c6 --- /dev/null +++ b/infra/order-stream/redshift-migrations/005_preserve_telemetry_completion_aliases.sql @@ -0,0 +1,80 @@ +-- 005_preserve_telemetry_completion_aliases.sql +-- Keep the renamed completion timing fields available under their historical +-- view column names so existing dashboards and queries continue to work. The +-- fallback to the old JSON keys also preserves rows ingested before migration 3. + +DROP VIEW IF EXISTS telemetry.request_completions; +CREATE VIEW telemetry.request_completions AS +SELECT + broker_address, order_id, request_id, request_digest, proof_type, outcome, + error_code, error_reason, + lock_duration_secs, + committed_to_application_proof_duration_secs, + proving_duration_secs, + aggregation_duration_secs, submission_duration_secs, total_duration_secs, + estimated_proving_time_secs, + actual_total_proving_time_secs, + actual_proving_time_secs, + concurrent_proving_jobs_start, concurrent_proving_jobs_end, + total_cycles, fulfillment_type, + stark_proving_secs, proof_compression_secs, + set_builder_proving_secs, assessor_proving_secs, + assessor_compression_proof_secs, + received_at_timestamp, completed_at, received_at +FROM ( + SELECT + data.broker_address::VARCHAR(42) AS broker_address, + data.order_id::VARCHAR(256) AS order_id, + data.request_id::VARCHAR(78) AS request_id, + data.request_digest::VARCHAR(78) AS request_digest, + data.proof_type::VARCHAR(32) AS proof_type, + data.outcome::VARCHAR(32) AS outcome, + data.error_code::VARCHAR(32) AS error_code, + data.error_reason::VARCHAR(512) AS error_reason, + data.lock_duration_secs::BIGINT AS lock_duration_secs, + COALESCE( + data.committed_to_application_proof_duration_secs::BIGINT, + data.proving_duration_secs::BIGINT + ) AS committed_to_application_proof_duration_secs, + COALESCE( + data.committed_to_application_proof_duration_secs::BIGINT, + data.proving_duration_secs::BIGINT + ) AS proving_duration_secs, + data.aggregation_duration_secs::BIGINT AS aggregation_duration_secs, + data.submission_duration_secs::BIGINT AS submission_duration_secs, + data.total_duration_secs::BIGINT AS total_duration_secs, + data.estimated_proving_time_secs::BIGINT AS estimated_proving_time_secs, + COALESCE( + data.actual_total_proving_time_secs::BIGINT, + data.actual_proving_time_secs::BIGINT + ) AS actual_total_proving_time_secs, + COALESCE( + data.actual_total_proving_time_secs::BIGINT, + data.actual_proving_time_secs::BIGINT + ) AS actual_proving_time_secs, + data.concurrent_proving_jobs_start::INTEGER AS concurrent_proving_jobs_start, + data.concurrent_proving_jobs_end::INTEGER AS concurrent_proving_jobs_end, + data.total_cycles::BIGINT AS total_cycles, + data.fulfillment_type::VARCHAR(32) AS fulfillment_type, + data.stark_proving_secs::DOUBLE PRECISION AS stark_proving_secs, + data.proof_compression_secs::DOUBLE PRECISION AS proof_compression_secs, + data.set_builder_proving_secs::DOUBLE PRECISION AS set_builder_proving_secs, + data.assessor_proving_secs::DOUBLE PRECISION AS assessor_proving_secs, + data.assessor_compression_proof_secs::DOUBLE PRECISION AS assessor_compression_proof_secs, + data.received_at_timestamp::BIGINT AS received_at_timestamp, + data.completed_at::TIMESTAMPTZ AS completed_at, + received_at, + ROW_NUMBER() OVER ( + PARTITION BY data.broker_address, + data.order_id + ORDER BY received_at DESC + ) AS rn + FROM telemetry.completions_raw +) +WHERE rn = 1; + +-- Re-grant permissions lost by DROP VIEW +GRANT SELECT ON telemetry.request_completions TO readonly; + +INSERT INTO telemetry._migrations (version, name) +VALUES (5, '005_preserve_telemetry_completion_aliases');