Implement workflow-level queue_time tracking and update Redis helper for job dependencies

Author: can-gaa-hou

aws/lambda/cross_repo_ci_relay/callback/callback_handler.py

@@ -121,8 +121,15 @@ def _update_state_and_compute_metrics(
 
     Writes IN_PROGRESS or COMPLETED state (with the current timestamp), then
     reads back the stored record to compute:
-    - ``queue_time``:     dispatch → in_progress  (set on "in_progress" callbacks)
-    - ``execution_time``: in_progress → completed  (set on "completed" callbacks)
+    - ``queue_time``:     dispatch → first job's in_progress in the run (set on
+                          "in_progress" callbacks). Workflow-level, not
+                          per-job: a run's jobs commonly wait on one another
+                          (e.g. tests waiting on a build job), but they all
+                          share one dispatch timestamp, so every job in the run
+                          is given the same queue_time, anchored to whichever
+                          job started first. See ``record_workflow_started``.
+    - ``execution_time``: in_progress → completed  (set on "completed"
+                          callbacks). Stays per-job: each job's own duration.
 
     Both metrics default to None when the required prior state is unavailable
     (e.g. Redis cache miss or rerun without matching prior record).
@@ -175,9 +182,21 @@ def _update_state_and_compute_metrics(
         return ci_metrics
 
     if state == CallbackState.IN_PROGRESS:
+        # Anchor queue_time to the first job in the run to start, not to this
+        # job's own in_progress time -- otherwise a job that waits on an
+        # earlier one (e.g. tests waiting on a build) would have that wait
+        # folded into its queue_time.
+        workflow_start_ts = redis_helper.record_workflow_started(
+            config,
+            delivery_id,
+            verified_repo,
+            run_id,
+            run_attempt,
+            updated_workflow_record.timestamp,
+        )
         ci_metrics["queue_time"] = _safe_delta(
             dispatch_record.timestamp,
-            updated_workflow_record.timestamp,
+            workflow_start_ts,
             "queue_time",
         )
     else:

aws/lambda/cross_repo_ci_relay/tests/test_callback_handler.py

@@ -68,6 +68,11 @@ def setUp(self):
         self.mock_redis = self.patcher_redis.start()
         self.mock_redis.create_client.return_value = MagicMock()
 
+        # Default: workflow-level queue_time passthrough, i.e. this job is
+        # treated as the first (and only) job in its run. Tests exercising
+        # multiple jobs in one run override this explicitly.
+        self.mock_redis.record_workflow_started.side_effect = lambda *a, **kw: a[5]
+
         # Setup default: dispatch exists, workflow state is None (in_progress not yet reported)
         def default_get_state(
             cfg,
@@ -188,6 +193,56 @@ def test_execution_time_calculated_from_state_records(self):
         _, trusted_arg, _ = self.mock_hud.call_args[0]
         self.assertEqual(trusted_arg["ci_metrics"]["execution_time"], 30.0)
 
+    # --- queue_time is workflow-level (anchored to the first job to start) ---
+
+    def test_queue_time_uses_first_jobs_start_for_a_later_job(self):
+        """A later job (e.g. "test", waiting on a "build" job) gets queue_time
+        anchored to the first job's start, not its own (later) in_progress
+        time -- otherwise the wait on "build" would inflate its queue_time."""
+        dispatch_record = CallbackStateRecord(CallbackState.DISPATCHED, 1000.0, {})
+        # This job reports in_progress at 1200 (long after dispatch), but the
+        # "build" job already claimed the workflow-start marker at 1030.
+        self.mock_redis.get_callback_state.side_effect = [
+            dispatch_record,  # dispatch lookup
+            None,  # workflow state: not yet set
+            CallbackStateRecord(CallbackState.IN_PROGRESS, 1200.0, {}),  # re-read
+        ]
+        self.mock_redis.record_workflow_started.side_effect = None
+        self.mock_redis.record_workflow_started.return_value = 1030.0
+
+        handle(
+            _cfg(),
+            _body(status="in_progress", job_name="test"),
+            verified_repo="org/repo",
+        )
+
+        _, trusted_arg, _ = self.mock_hud.call_args[0]
+        # dispatch (1000) -> first job's start (1030), NOT this job's own 1200.
+        self.assertEqual(trusted_arg["ci_metrics"]["queue_time"], 30.0)
+
+    def test_execution_time_unaffected_by_workflow_level_queue_time(self):
+        """execution_time stays purely job-level (in_progress -> completed for
+        this same job) and completed callbacks never touch the workflow-start
+        marker -- that's only relevant to in_progress/queue_time."""
+        dispatch_record = CallbackStateRecord(CallbackState.DISPATCHED, 1000.0, {})
+        workflow_record = CallbackStateRecord(CallbackState.IN_PROGRESS, 1200.0, {})
+        completed_record = CallbackStateRecord(CallbackState.COMPLETED, 1260.0, {})
+        self.mock_redis.get_callback_state.side_effect = [
+            dispatch_record,
+            workflow_record,
+            completed_record,
+        ]
+
+        handle(
+            _cfg(),
+            _body(status="completed", job_name="test"),
+            verified_repo="org/repo",
+        )
+
+        _, trusted_arg, _ = self.mock_hud.call_args[0]
+        self.assertEqual(trusted_arg["ci_metrics"]["execution_time"], 60.0)
+        self.mock_redis.record_workflow_started.assert_not_called()
+
     # --- HUD 4xx propagates (5xx is swallowed inside forward_to_hud) ---
 
     def test_hud_4xx_propagates(self):
@@ -293,6 +348,7 @@ def setUp(self):
 
         self.patcher_redis = patch("callback.callback_handler.redis_helper")
         self.mock_redis = self.patcher_redis.start()
+        self.mock_redis.record_workflow_started.side_effect = lambda *a, **kw: a[5]
 
         def _get_state(
             cfg,
@@ -521,6 +577,7 @@ def test_nightly_callback_skips_redis(self):
         self.mock_redis.set_callback_state.assert_not_called()
         self.mock_redis.add_in_progress_tracker.assert_not_called()
         self.mock_redis.remove_in_progress_tracker.assert_not_called()
+        self.mock_redis.record_workflow_started.assert_not_called()
 
     def test_nightly_callback_has_no_timing_metrics(self):
         handle(_cfg(), self._nightly_body(), verified_repo="org/repo")

aws/lambda/cross_repo_ci_relay/tests/test_redis_helper.py

@@ -12,6 +12,7 @@
     create_client,
     get_cached_yaml,
     get_callback_state,
+    record_workflow_started,
     set_cached_yaml,
     set_callback_state,
 )
@@ -267,6 +268,62 @@ def get_side_effect(
             )
 
 
+class TestWorkflowStart(unittest.TestCase):
+    """queue_time is workflow-level: the first job in a run to report
+    in_progress claims the shared start marker; later jobs in the same run
+    read that same value back instead of recording their own (later) time."""
+
+    def setUp(self):
+        redis_helper._cached_client = None
+        redis_helper._cached_client_url = None
+
+    def test_first_job_claims_and_returns_its_own_timestamp(self):
+        """SET NX succeeds when no other job has claimed the marker yet."""
+        client = MagicMock()
+        client.set.return_value = True
+
+        result = record_workflow_started(
+            _cfg(), "del-123", "org/repo", 99999, 1, 1030.0, client=client
+        )
+
+        self.assertEqual(result, 1030.0)
+        client.set.assert_called_once_with(
+            "crcr:workflow_start:del-123:org/repo:99999:1",
+            1030.0,
+            nx=True,
+            ex=3600,
+        )
+        client.get.assert_not_called()
+
+    def test_later_job_reads_back_first_jobs_timestamp(self):
+        """SET NX fails once another job already claimed the marker; the
+        later job's own (later) timestamp is discarded for the stored one."""
+        client = MagicMock()
+        client.set.return_value = None
+        client.get.return_value = "1030.0"
+
+        result = record_workflow_started(
+            _cfg(), "del-123", "org/repo", 99999, 1, 1200.0, client=client
+        )
+
+        self.assertEqual(result, 1030.0)
+        client.get.assert_called_once_with(
+            "crcr:workflow_start:del-123:org/repo:99999:1"
+        )
+
+    def test_redis_error_on_set_falls_back_to_own_timestamp(self):
+        """A Redis outage degrades to per-job queue_time rather than blocking
+        the callback: the caller's own timestamp is returned unchanged."""
+        client = MagicMock()
+        client.set.side_effect = redis_lib.exceptions.RedisError("boom")
+
+        result = record_workflow_started(
+            _cfg(), "del-123", "org/repo", 99999, 1, 1200.0, client=client
+        )
+
+        self.assertEqual(result, 1200.0)
+
+
 class TestRateLimit(unittest.TestCase):
     def setUp(self):
         redis_helper._cached_client = None

aws/lambda/cross_repo_ci_relay/utils/redis_helper.py

@@ -22,6 +22,7 @@
 _IN_PROGRESS_ZSET = "crcr:in_progress"
 _DISPATCH_JOB_PREFIX = "crcr:dispatch_job:"
 _CHECK_RUN_WANTED_PREFIX = "crcr:check_run_wanted:"
+_WORKFLOW_START_PREFIX = "crcr:workflow_start:"
 _cached_client: redis_lib.Redis | None = None
 _cached_client_url: str | None = None
 
@@ -466,6 +467,62 @@ def set_callback_state(
         raise
 
 
+def _workflow_start_key(
+    delivery_id: str, downstream_repo: str, run_id: int, run_attempt: int
+) -> str:
+    """Redis key for the workflow-level "first job started" marker.
+
+    Keyed by delivery_id + repo + run_id + run_attempt, with no job_name:
+    every job within one run shares this single timestamp.
+    """
+    return (
+        f"{_WORKFLOW_START_PREFIX}{delivery_id}:{downstream_repo}:"
+        f"{run_id}:{run_attempt}"
+    )
+
+
+def record_workflow_started(
+    config: RelayConfig,
+    delivery_id: str,
+    downstream_repo: str,
+    run_id: int,
+    run_attempt: int,
+    timestamp: float,
+    client: redis_lib.Redis | None = None,
+) -> float:
+    """Claim or read the workflow-level "first job started" timestamp.
+
+    A run's jobs commonly form a dependency chain (e.g. a build job that later
+    test jobs wait on), but every job shares the same dispatch timestamp. If
+    queue_time were computed per job against that shared dispatch timestamp,
+    later jobs would have the earlier jobs' build/wait time folded into their
+    queue_time. Instead, queue_time is measured once per workflow run, from
+    dispatch to the first job's in_progress: this uses SET NX so the first
+    caller's timestamp "wins" and is stored; every later job in the same run
+    reads that same stored value back instead of recording its own (later)
+    in_progress time.
+
+    Falls back to returning ``timestamp`` unchanged on Redis errors or a
+    malformed stored value, so a transient outage degrades to (rare) per-job
+    queue_time rather than blocking the callback.
+    """
+    key = _workflow_start_key(delivery_id, downstream_repo, run_id, run_attempt)
+    try:
+        if client is None:
+            client = create_client(config)
+        won = client.set(key, timestamp, nx=True, ex=config.crcr_status_ttl)
+        if won:
+            return timestamp
+        existing = client.get(key)
+        return float(existing) if existing is not None else timestamp
+    except RedisError:
+        logger.exception("record_workflow_started: redis error key=%s", key)
+        return timestamp
+    except (TypeError, ValueError):
+        logger.exception("record_workflow_started: malformed stored value key=%s", key)
+        return timestamp
+
+
 def _in_progress_member(
     delivery_id: str,
     downstream_repo: str,

aws/lambda/cross_repo_ci_relay/webhook/event_handler.py

@@ -45,7 +45,8 @@ def _dispatch_one(
     # Set dispatch state with timestamp to prove valid webhook occurred.
     # Keyed by delivery_id + repo + run_id + run_attempt.
     # Uses DISPATCH_RUN_ID/DISPATCH_RUN_ATTEMPT sentinels for repo-level dispatch.
-    # Timestamp is used for queue_time calculation (dispatch → in_progress).
+    # Timestamp is used for queue_time calculation (dispatch → first job's
+    # in_progress in the run; see redis_helper.record_workflow_started).
     redis_helper.set_callback_state(
         config,
         client_payload["delivery_id"],