Surface dispatch-overdue wake-latency age on operator metrics and task_transport health

durable-workflow-ops · durable-workflow-ops · commit 6efc48780de5 · 2026-04-24T21:31:02.000Z
Adds tasks.oldest_dispatch_overdue_since and tasks.max_dispatch_overdue_age_ms
to OperatorMetrics::snapshot() so operators can answer "how long has the
oldest ready-but-unclaimed task been waiting for a working dispatch wake?"
from the metric alone, completing the coordination-health age set alongside
tasks.oldest_lease_expired_at, tasks.oldest_ready_due_at, runs.oldest_wait_started_at,
and backlog.oldest_compatibility_blocked_started_at.

The age is the effective COALESCE(last_dispatched_at, created_at) — the
timestamp the task has been waiting for a successful dispatch since (either
the last attempted dispatch that didn't stick, or the task's creation time
if it was never dispatched), computed across the dispatch_overdue subset
only. Refactors the existing filter into a private dispatchOverdueQuery()
helper so the count and the age use one authoritative query.

Forwards the same trio on HealthCheck::taskTransportCheck() data as
dispatch_overdue_tasks / oldest_dispatch_overdue_since /
max_dispatch_overdue_age_ms so wake-latency is legible from /healthz without
re-reading the metrics snapshot. The check's escalation predicate stays
unchanged (it still escalates only on tasks.unhealthy, which already counts
dispatch_overdue); the age data is observability so operators can tell
"dispatch wake is sporadically slow" apart from "dispatch wake has stalled
on this task for minutes".

Pins the new keys in docs/architecture/rollout-safety.md frozen metric
table and "Ready but unclaimed" stuck-detector entry, adds a
dispatch-overdue-age row assertion in RolloutSafetyDocumentationTest, and
extends the task_transport HealthCheckTest assertions to pin the new fields
in the health check data. Extends V2OperatorMetricsTest with two focused
tests proving the metric selects the earliest effective dispatch moment
across mixed dispatched/never-dispatched/failed/healthy task fixtures and
returns null/0 when no tasks are overdue, plus broad-snapshot and
task_transport forward assertions on the existing mixed-state fixture.
diff --git a/docs/architecture/rollout-safety.md b/docs/architecture/rollout-safety.md
@@ -411,6 +411,7 @@ change.
 | `tasks` | `dispatch_overdue`, `lease_expired` | lease and dispatch timing |
 | `tasks` | `oldest_lease_expired_at`, `max_lease_expired_age_ms` | earliest `lease_expires_at` among leased tasks whose lease has expired at snapshot time and the largest expired-lease age in milliseconds, mirroring the `backlog.oldest_compatibility_blocked_started_at` / `max_compatibility_blocked_age_ms` shape so operators can answer "how long has the worst leased task been expired without redelivery?" (the primary stuck-lease duplicate-risk age indicator) from the metric alone |
 | `tasks` | `oldest_ready_due_at`, `max_ready_due_age_ms` | earliest "ready since" timestamp among ready-due tasks (the effective `COALESCE(available_at, created_at)` — `available_at` when the task was delayed, otherwise the creation time that made it immediately actionable) and the largest ready-age in milliseconds, mirroring the `oldest_lease_expired_at` / `max_lease_expired_age_ms` shape so operators can read queue latency ("how long has the oldest actionable task been waiting to dispatch?") from the metric alone without walking `workflow_tasks` |
+| `tasks` | `oldest_dispatch_overdue_since`, `max_dispatch_overdue_age_ms` | earliest `COALESCE(last_dispatched_at, created_at)` among dispatch-overdue tasks — the timestamp the worst-case ready-but-unclaimed task has been waiting for a successful dispatch wake since (either its last attempted dispatch that didn't stick or its creation time if it was never dispatched) — and the largest age in milliseconds, mirroring the `oldest_ready_due_at` / `max_ready_due_age_ms` shape so operators can read wake-latency ("how long has the oldest ready-but-unclaimed task been waiting for a working dispatch wake?") from the metric alone without walking `workflow_tasks` |
 | `tasks` | `unhealthy` | sum of transport failure and lease expiry counts (the primary duplicate-risk indicator) |
 | `backlog` | `runnable_tasks`, `delayed_tasks`, `leased_tasks` | authoritative backlog counts |
 | `backlog` | `unhealthy_tasks`, `repair_needed_runs`, `claim_failed_runs`, `compatibility_blocked_runs` | stuck/blocked roll-ups |
@@ -486,8 +487,16 @@ are authoritative and how they surface.
   authority for the redelivery decision.
 - **Ready but unclaimed.** A ready task that has sat past the
   repair window without being claimed is counted under
-  `tasks.dispatch_overdue` and considered by
-  `TaskRepairPolicy::readyTaskNeedsRedispatch()`.
+  `tasks.dispatch_overdue`, its worst-case waiting-for-dispatch
+  timestamp is surfaced through
+  `tasks.oldest_dispatch_overdue_since` and
+  `tasks.max_dispatch_overdue_age_ms`, and both keys are forwarded on
+  the `task_transport` health check (`dispatch_overdue_tasks`,
+  `oldest_dispatch_overdue_since`, `max_dispatch_overdue_age_ms`).
+  `TaskRepairPolicy::readyTaskNeedsRedispatch()` is the authority for
+  the redispatch decision; the age data is observability so operators
+  can tell the difference between "dispatch wake is sporadically
+  slow" and "dispatch wake has stalled on this task for minutes".
 - **Repair-needed runs.** Runs whose projected state shows
   `liveness_state = repair_needed` are counted under
   `runs.repair_needed` and surface through the
diff --git a/src/V2/Support/HealthCheck.php b/src/V2/Support/HealthCheck.php
@@ -235,6 +235,11 @@ private static function taskTransportCheck(array $tasks, array $backlog): array
                     ? $tasks['oldest_ready_due_at']
                     : null,
                 'max_ready_due_age_ms' => self::integer($tasks['max_ready_due_age_ms'] ?? 0),
+                'dispatch_overdue_tasks' => self::integer($tasks['dispatch_overdue'] ?? 0),
+                'oldest_dispatch_overdue_since' => is_string($tasks['oldest_dispatch_overdue_since'] ?? null)
+                    ? $tasks['oldest_dispatch_overdue_since']
+                    : null,
+                'max_dispatch_overdue_age_ms' => self::integer($tasks['max_dispatch_overdue_age_ms'] ?? 0),
                 'repair_needed_runs' => self::integer($backlog['repair_needed_runs'] ?? 0),
                 'claim_failed_runs' => self::integer($backlog['claim_failed_runs'] ?? 0),
                 'compatibility_blocked_runs' => self::integer($backlog['compatibility_blocked_runs'] ?? 0),
diff --git a/src/V2/Support/OperatorMetrics.php b/src/V2/Support/OperatorMetrics.php
@@ -121,6 +121,7 @@ private static function taskMetrics(CarbonInterface $now, ?string $namespace): a
     {
         $oldestLeaseExpiredAt = self::oldestLeaseExpiredAt($now, $namespace);
         $oldestReadyDueAt = self::oldestReadyDueAt($now, $namespace);
+        $oldestDispatchOverdueSince = self::oldestDispatchOverdueSince($now, $namespace);
 
         return [
             'open' => self::openTasks($namespace),
@@ -140,6 +141,10 @@ private static function taskMetrics(CarbonInterface $now, ?string $namespace): a
             'max_ready_due_age_ms' => $oldestReadyDueAt === null
                 ? 0
                 : (int) $oldestReadyDueAt->diffInMilliseconds($now),
+            'oldest_dispatch_overdue_since' => $oldestDispatchOverdueSince?->toJSON(),
+            'max_dispatch_overdue_age_ms' => $oldestDispatchOverdueSince === null
+                ? 0
+                : (int) $oldestDispatchOverdueSince->diffInMilliseconds($now),
             'unhealthy' => self::dispatchFailedTasks($namespace)
                 + self::claimFailedTasks($namespace)
                 + self::dispatchOverdueTasks($now, $namespace)
@@ -948,6 +953,36 @@ private static function claimFailedTasks(?string $namespace): int
     }
 
     private static function dispatchOverdueTasks(CarbonInterface $now, ?string $namespace): int
+    {
+        return self::dispatchOverdueQuery($now, $namespace)->count();
+    }
+
+    /**
+     * Earliest "waiting-for-dispatch" moment among dispatch-overdue tasks —
+     * the effective `COALESCE(last_dispatched_at, created_at)`, which is the
+     * timestamp the task has been waiting for a successful dispatch since
+     * (either the last attempted dispatch that didn't stick, or the task's
+     * creation time for tasks that were never dispatched at all). Rollout-safety
+     * surfaces this alongside `tasks.dispatch_overdue` so operators can read
+     * wake-latency ("how long has the oldest ready-but-unclaimed task been
+     * waiting for a working dispatch wake?") from the metric alone, mirroring
+     * the existing `oldest_ready_due_at` / `max_ready_due_age_ms` shape.
+     */
+    private static function oldestDispatchOverdueSince(CarbonInterface $now, ?string $namespace): ?CarbonInterface
+    {
+        /** @var WorkflowTask|null $task */
+        $task = self::dispatchOverdueQuery($now, $namespace)
+            ->orderByRaw('COALESCE(last_dispatched_at, created_at) asc')
+            ->first();
+
+        if (! $task instanceof WorkflowTask) {
+            return null;
+        }
+
+        return $task->last_dispatched_at ?? $task->created_at;
+    }
+
+    private static function dispatchOverdueQuery(CarbonInterface $now, ?string $namespace)
     {
         $cutoff = $now->copy()
             ->subSeconds(TaskRepairPolicy::redispatchAfterSeconds());
@@ -972,8 +1007,7 @@ private static function dispatchOverdueTasks(CarbonInterface $now, ?string $name
                     $neverDispatched->whereNull('last_dispatched_at')
                         ->where('created_at', '<=', $cutoff);
                 });
-            })
-            ->count();
+            });
     }
 
     private static function dispatchFailedQuery(?string $namespace)
diff --git a/tests/Feature/V2/V2OperatorMetricsTest.php b/tests/Feature/V2/V2OperatorMetricsTest.php
@@ -233,6 +233,13 @@ public function testSnapshotSummarizesDurableBacklogRepairCompatibilityAndWorker
             $snapshot['tasks']['oldest_ready_due_at'],
         );
         $this->assertSame(10 * 1000, $snapshot['tasks']['max_ready_due_age_ms']);
+        $this->assertSame(
+            Carbon::parse('2026-04-09 12:00:00')
+                ->subSeconds(10)
+                ->toJSON(),
+            $snapshot['tasks']['oldest_dispatch_overdue_since'],
+        );
+        $this->assertSame(10 * 1000, $snapshot['tasks']['max_dispatch_overdue_age_ms']);
         $this->assertSame(4, $snapshot['tasks']['unhealthy']);
         $this->assertSame(4, $snapshot['backlog']['runnable_tasks']);
         $this->assertSame(1, $snapshot['backlog']['delayed_tasks']);
@@ -351,6 +358,14 @@ public function testSnapshotSummarizesDurableBacklogRepairCompatibilityAndWorker
             $taskTransport['data']['oldest_ready_due_at'],
         );
         $this->assertSame(10 * 1000, $taskTransport['data']['max_ready_due_age_ms']);
+        $this->assertSame(1, $taskTransport['data']['dispatch_overdue_tasks']);
+        $this->assertSame(
+            Carbon::parse('2026-04-09 12:00:00')
+                ->subSeconds(10)
+                ->toJSON(),
+            $taskTransport['data']['oldest_dispatch_overdue_since'],
+        );
+        $this->assertSame(10 * 1000, $taskTransport['data']['max_dispatch_overdue_age_ms']);
     }
 
     public function testSnapshotCountsStaleRunSummaryProjectionRows(): void
@@ -1020,6 +1035,117 @@ public function testSnapshotSurfacesRunWaitAgeForEveryRunningWaitNotJustCompatib
         $this->assertSame(7 * 60 * 1000, $resumePaths['data']['max_wait_age_ms']);
     }
 
+    public function testSnapshotSurfacesDispatchOverdueAgeFromEarliestEffectiveDispatchMoment(): void
+    {
+        Carbon::setTestNow('2026-04-09 12:00:00');
+        $this->beforeApplicationDestroyed(static function (): void {
+            Carbon::setTestNow();
+        });
+
+        $now = Carbon::now();
+
+        $run = $this->createRunWithSummary(
+            instanceId: 'dispatch-overdue-age-instance',
+            runId: '01JDISPATCHOVDRUN0000000001',
+            status: 'running',
+            statusBucket: 'running',
+            livenessState: 'running',
+        );
+
+        // Worst-case: ready task whose last successful dispatch was 90s ago.
+        $this->createTask($run, '01JDISPATCHOVDTASK000000001', TaskStatus::Ready->value, [
+            'available_at' => $now->copy()
+                ->subSeconds(120),
+            'last_dispatched_at' => $now->copy()
+                ->subSeconds(90),
+            'created_at' => $now->copy()
+                ->subSeconds(150),
+        ]);
+
+        // Never-dispatched ready task created well past the cutoff — counted
+        // as overdue but its effective age (created_at = 30s ago) is newer
+        // than the 90s task above, so it must not win the "oldest since".
+        $this->createTask($run, '01JDISPATCHOVDTASK000000002', TaskStatus::Ready->value, [
+            'available_at' => $now->copy()
+                ->subSeconds(30),
+            'created_at' => $now->copy()
+                ->subSeconds(30),
+        ]);
+
+        // Ready task dispatched well within the cutoff — healthy, NOT overdue.
+        $this->createTask($run, '01JDISPATCHOVDTASK000000003', TaskStatus::Ready->value, [
+            'available_at' => $now->copy()
+                ->subSecond(),
+            'last_dispatched_at' => $now->copy()
+                ->subSecond(),
+            'created_at' => $now->copy()
+                ->subSecond(),
+        ]);
+
+        // Dispatch-failed ready task — applyDispatchHealthy excludes it from
+        // the overdue set even though its created_at is well past cutoff.
+        $this->createTask($run, '01JDISPATCHOVDTASK000000004', TaskStatus::Ready->value, [
+            'available_at' => $now->copy()
+                ->subSeconds(120),
+            'last_dispatch_attempt_at' => $now->copy()
+                ->subSeconds(30),
+            'last_dispatch_error' => 'Queue transport unavailable.',
+            'created_at' => $now->copy()
+                ->subSeconds(120),
+        ]);
+
+        $snapshot = OperatorMetrics::snapshot($now);
+
+        $expectedOldestDispatchOverdueSince = $now->copy()
+            ->subSeconds(90)
+            ->toJSON();
+
+        $this->assertSame(2, $snapshot['tasks']['dispatch_overdue']);
+        $this->assertSame($expectedOldestDispatchOverdueSince, $snapshot['tasks']['oldest_dispatch_overdue_since']);
+        $this->assertSame(90 * 1000, $snapshot['tasks']['max_dispatch_overdue_age_ms']);
+
+        $healthSnapshot = HealthCheck::snapshot($now);
+        $taskTransport = collect($healthSnapshot['checks'])->firstWhere('name', 'task_transport');
+        $this->assertNotNull($taskTransport);
+        $this->assertSame(2, $taskTransport['data']['dispatch_overdue_tasks']);
+        $this->assertSame($expectedOldestDispatchOverdueSince, $taskTransport['data']['oldest_dispatch_overdue_since']);
+        $this->assertSame(90 * 1000, $taskTransport['data']['max_dispatch_overdue_age_ms']);
+    }
+
+    public function testSnapshotReportsDispatchOverdueAgeAsZeroWhenNoTasksAreOverdue(): void
+    {
+        Carbon::setTestNow('2026-04-09 12:00:00');
+        $this->beforeApplicationDestroyed(static function (): void {
+            Carbon::setTestNow();
+        });
+
+        $now = Carbon::now();
+
+        $run = $this->createRunWithSummary(
+            instanceId: 'dispatch-overdue-none-instance',
+            runId: '01JDISPATCHNONERUN0000000001',
+            status: 'running',
+            statusBucket: 'running',
+            livenessState: 'running',
+        );
+
+        // Fresh ready task — dispatched within the cutoff, not overdue.
+        $this->createTask($run, '01JDISPATCHNONETASK000000001', TaskStatus::Ready->value, [
+            'available_at' => $now->copy()
+                ->subSecond(),
+            'last_dispatched_at' => $now->copy()
+                ->subSecond(),
+            'created_at' => $now->copy()
+                ->subSecond(),
+        ]);
+
+        $snapshot = OperatorMetrics::snapshot($now);
+
+        $this->assertSame(0, $snapshot['tasks']['dispatch_overdue']);
+        $this->assertNull($snapshot['tasks']['oldest_dispatch_overdue_since']);
+        $this->assertSame(0, $snapshot['tasks']['max_dispatch_overdue_age_ms']);
+    }
+
     public function testSnapshotReportsRunWaitAgeAsZeroWhenNoRunsAreWaiting(): void
     {
         Carbon::setTestNow('2026-04-09 12:00:00');
diff --git a/tests/Unit/V2/HealthCheckTest.php b/tests/Unit/V2/HealthCheckTest.php
@@ -552,6 +552,9 @@ public function testSnapshotWarnsWhenOpenRunHasNoDurableResumePath(): void
         $this->assertSame(200, HealthCheck::httpStatus($snapshot));
         $this->assertSame('ok', $taskTransport['status']);
         $this->assertSame(0, $taskTransport['data']['unhealthy_tasks']);
+        $this->assertSame(0, $taskTransport['data']['dispatch_overdue_tasks']);
+        $this->assertNull($taskTransport['data']['oldest_dispatch_overdue_since']);
+        $this->assertSame(0, $taskTransport['data']['max_dispatch_overdue_age_ms']);
         $this->assertSame(1, $taskTransport['data']['repair_needed_runs']);
         $this->assertSame('warning', $resumePaths['status']);
         $this->assertSame(1, $resumePaths['data']['repair_needed_runs']);
diff --git a/tests/Unit/V2/RolloutSafetyDocumentationTest.php b/tests/Unit/V2/RolloutSafetyDocumentationTest.php
@@ -361,6 +361,17 @@ public function testContractDocumentFreezesRunWaitAgeRow(): void
         );
     }
 
+    public function testContractDocumentFreezesDispatchOverdueAgeRow(): void
+    {
+        $contents = $this->documentContents();
+
+        $this->assertMatchesRegularExpression(
+            '/\|\s*`tasks`\s*\|[^|]*`oldest_dispatch_overdue_since`[^|]*`max_dispatch_overdue_age_ms`/',
+            $contents,
+            'Rollout safety contract must pin the tasks dispatch-overdue age row so operators can read wake-latency ("how long has the oldest ready-but-unclaimed task been waiting for a working dispatch wake?") from OperatorMetrics::snapshot() without walking workflow_tasks.',
+        );
+    }
+
     public function testContractDocumentFreezesHealthCheckNames(): void
     {
         $contents = $this->documentContents();
