fix(order-stream): preserve Redshift telemetry field compatibility

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;

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:

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');