Expose recent backlog task flow on operator metrics

Expose recent backlog task flow on operator metrics

Author: durable-workflow-ops

docs/architecture/rollout-safety.md

@@ -419,6 +419,7 @@ change.
 | `activities` | `retrying`, `oldest_retrying_started_at`, `max_retrying_age_ms` | activity executions currently in the retry window (Pending status with `attempt_count > 0`), the earliest `started_at` among them, and the largest retrying age in milliseconds, mirroring the `tasks.oldest_lease_expired_at` / `max_lease_expired_age_ms` shape on the task path so operators can answer "how long has the worst-case activity been chewing retries?" — the primary retry-rate age indicator on the activity path — from the metric alone without walking `activity_executions` |
 | `activities` | `timeout_overdue`, `oldest_timeout_overdue_at`, `max_timeout_overdue_age_ms` | open activity executions whose `schedule_deadline_at` (Pending), `close_deadline_at` (Running), `schedule_to_close_deadline_at`, or `heartbeat_deadline_at` (Running) is `<= $now` and is therefore waiting for `ActivityTimeoutEnforcer` to enforce the timeout, the earliest such expired deadline timestamp across the four enforcement-relevant columns, and the largest overdue age in milliseconds. The `timeout_overdue` predicate mirrors `ActivityTimeoutEnforcer::expiredExecutionIds()` exactly, so the count is the operator-visible view of the same enforcement backlog — the activity-path counterpart of `tasks.lease_expired`. Both surface stuck work that the corresponding sweep has not yet reclaimed; sustained non-zero readings indicate the activity-timeout sweep is lagging or stalled and that worker liveness via heartbeat or start-to-close has stopped on at least one execution. The age data lets operators read "how long has the worst-case activity been past a timeout deadline without enforcement?" — the primary stuck-activity duplicate-risk age indicator on the activity path — from the metric alone without walking `activity_executions` |
 | `backlog` | `runnable_tasks`, `delayed_tasks`, `leased_tasks` | authoritative backlog counts |
+| `backlog` | `tasks_added_last_minute`, `tasks_dispatched_last_minute` | trailing-60-second queue-flow facts: distinct durable task rows created in the last minute and distinct durable task rows whose latest successful `last_dispatched_at` falls in the last minute. These are intentionally task-row facts, not a transport-attempt counter stream; repeated redispatches of the same durable task collapse to one count because `workflow_tasks` retains only the latest successful dispatch timestamp |
 | `backlog` | `unhealthy_tasks`, `repair_needed_runs`, `claim_failed_runs`, `compatibility_blocked_runs` | stuck/blocked roll-ups |
 | `backlog` | `oldest_compatibility_blocked_started_at`, `max_compatibility_blocked_age_ms` | earliest wait-start timestamp among compatibility-blocked runs and the largest blocked age in milliseconds, mirroring the `repair.oldest_missing_run_started_at` / `max_missing_run_age_ms` shape so operators can answer "how stale is the worst mixed-build block?" from the metric alone |
 | `repair` | `missing_task_candidates`, `selected_missing_task_candidates`, `oldest_missing_run_started_at`, `max_missing_run_age_ms` | stuck-run detectors per `TaskRepairCandidates` |

src/V2/Support/OperatorMetrics.php

@@ -188,6 +188,8 @@ private static function backlogMetrics(CarbonInterface $now, ?string $namespace)
             'runnable_tasks' => self::readyDueTasks($now, $namespace),
             'delayed_tasks' => self::delayedTasks($now, $namespace),
             'leased_tasks' => self::leasedTasks($namespace),
+            'tasks_added_last_minute' => self::tasksAddedLastMinute($now, $namespace),
+            'tasks_dispatched_last_minute' => self::tasksDispatchedLastMinute($now, $namespace),
             'retrying_activities' => self::retryingActivities($namespace),
             'unhealthy_tasks' => self::dispatchFailedTasks($namespace)
                 + self::claimFailedTasks($namespace)
@@ -1141,6 +1143,38 @@ private static function leasedTasks(?string $namespace): int
             ->count();
     }
 
+    /**
+     * Distinct durable task rows created inside the trailing 60-second window.
+     * This is the best stable "tasks added" fact the durable model exposes
+     * without inventing a second transport-only counter stream.
+     */
+    private static function tasksAddedLastMinute(CarbonInterface $now, ?string $namespace): int
+    {
+        $windowStart = $now->copy()
+            ->subMinute();
+
+        return self::scopedRunModelQuery(self::taskModel(), $namespace)
+            ->where('created_at', '>=', $windowStart)
+            ->count();
+    }
+
+    /**
+     * Distinct durable task rows whose latest successful dispatch landed
+     * inside the trailing 60-second window. Repeated redispatches of the same
+     * durable task collapse to one row because `workflow_tasks` retains only
+     * the latest successful `last_dispatched_at`.
+     */
+    private static function tasksDispatchedLastMinute(CarbonInterface $now, ?string $namespace): int
+    {
+        $windowStart = $now->copy()
+            ->subMinute();
+
+        return self::scopedRunModelQuery(self::taskModel(), $namespace)
+            ->whereNotNull('last_dispatched_at')
+            ->where('last_dispatched_at', '>=', $windowStart)
+            ->count();
+    }
+
     private static function leaseExpiredTasks(CarbonInterface $now, ?string $namespace): int
     {
         return self::scopedRunModelQuery(self::taskModel(), $namespace)

tests/Feature/V2/V2OperatorMetricsTest.php

@@ -260,6 +260,8 @@ public function testSnapshotSummarizesDurableBacklogRepairCompatibilityAndWorker
         $this->assertSame(4, $snapshot['backlog']['runnable_tasks']);
         $this->assertSame(1, $snapshot['backlog']['delayed_tasks']);
         $this->assertSame(2, $snapshot['backlog']['leased_tasks']);
+        $this->assertSame(6, $snapshot['backlog']['tasks_added_last_minute']);
+        $this->assertSame(2, $snapshot['backlog']['tasks_dispatched_last_minute']);
         $this->assertSame(4, $snapshot['backlog']['unhealthy_tasks']);
         $this->assertSame(1, $snapshot['backlog']['repair_needed_runs']);
         $this->assertSame(1, $snapshot['backlog']['claim_failed_runs']);

tests/Unit/V2/RolloutSafetyDocumentationTest.php

@@ -145,6 +145,8 @@ final class RolloutSafetyDocumentationTest extends TestCase
         'runnable_tasks',
         'delayed_tasks',
         'leased_tasks',
+        'tasks_added_last_minute',
+        'tasks_dispatched_last_minute',
         'unhealthy_tasks',
         'repair_needed_runs',
         'claim_failed_runs',
@@ -356,6 +358,17 @@ public function testContractDocumentFreezesCompatibilityBlockedAgeRow(): void
         );
     }
 
+    public function testContractDocumentFreezesBacklogRateRow(): void
+    {
+        $contents = $this->documentContents();
+
+        $this->assertMatchesRegularExpression(
+            '/\|\s*`backlog`\s*\|[^|]*`tasks_added_last_minute`[^|]*`tasks_dispatched_last_minute`/',
+            $contents,
+            'Rollout safety contract must pin the backlog add/dispatch throughput row so operators can read recent queue flow from OperatorMetrics::snapshot() without inferring it from unrelated counters.',
+        );
+    }
+
     public function testContractDocumentFreezesLeaseExpiredAgeRow(): void
     {
         $contents = $this->documentContents();