Expose queue flow facts in visibility snapshots

Expose queue flow facts in visibility snapshots

Author: durable-workflow-ops

src/V2/Support/OperatorQueueVisibility.php

@@ -199,6 +199,7 @@ private static function stats(
             static fn (array $poller): bool => ($poller['is_stale'] ?? false) !== true,
         ));
         $stalePollers = count($pollers) - $activePollers;
+        $recentTaskFlow = self::recentTaskFlow($namespace, $taskQueue, $now);
 
         return [
             'approximate_backlog_count' => $readyCounts[TaskType::Workflow->value] + $readyCounts[TaskType::Activity->value],
@@ -229,6 +230,28 @@ private static function stats(
                 'stale_count' => $stalePollers,
                 'stale_after_seconds' => $staleAfterSeconds,
             ],
+            'tasks_added_last_minute' => $recentTaskFlow['tasks_added_last_minute'],
+            'tasks_dispatched_last_minute' => $recentTaskFlow['tasks_dispatched_last_minute'],
+        ];
+    }
+
+    /**
+     * @return array{tasks_added_last_minute: int, tasks_dispatched_last_minute: int}
+     */
+    private static function recentTaskFlow(string $namespace, string $taskQueue, CarbonInterface $now): array
+    {
+        $taskTable = self::taskTable();
+        $windowStart = self::databaseTimestamp($now->copy()->subMinute());
+        $query = self::baseTaskQuery($namespace, $taskQueue);
+
+        return [
+            'tasks_added_last_minute' => (clone $query)
+                ->where($taskTable . '.created_at', '>=', $windowStart)
+                ->count(),
+            'tasks_dispatched_last_minute' => (clone $query)
+                ->whereNotNull($taskTable . '.last_dispatched_at')
+                ->where($taskTable . '.last_dispatched_at', '>=', $windowStart)
+                ->count(),
         ];
     }
 

tests/Feature/V2/V2OperatorQueueVisibilityTest.php

@@ -59,12 +59,18 @@ public function testForQueueSummarizesBacklogPollersLeasesAndRepairCandidates():
                 'lease_expires_at' => now()
                     ->subMinute(),
                 'attempt_count' => 2,
+                'created_at' => now()
+                    ->subMinutes(2),
             ]
         );
         $activityTask = $this->createTask($run, '01JQUEUEVISIBLETASK000004', TaskType::Activity, TaskStatus::Leased, [
             'lease_owner' => 'worker-active',
             'lease_expires_at' => now()
                 ->addMinute(),
+            'created_at' => now()
+                ->subMinutes(2),
+            'last_dispatched_at' => now()
+                ->subSeconds(10),
         ]);
 
         ActivityAttempt::query()->create([
@@ -119,6 +125,8 @@ public function testForQueueSummarizesBacklogPollersLeasesAndRepairCandidates():
         $this->assertSame(1, $payload['stats']['pollers']['active_count']);
         $this->assertSame(1, $payload['stats']['pollers']['stale_count']);
         $this->assertSame(60, $payload['stats']['pollers']['stale_after_seconds']);
+        $this->assertSame(2, $payload['stats']['tasks_added_last_minute']);
+        $this->assertSame(1, $payload['stats']['tasks_dispatched_last_minute']);
         $this->assertSame('worker-active', $payload['pollers'][0]['worker_id']);
         $this->assertFalse($payload['pollers'][0]['is_stale']);
         $this->assertSame('worker-stale', $payload['pollers'][1]['worker_id']);
@@ -151,6 +159,10 @@ public function testForNamespaceListsTaskQueuesFromTasksAndInjectedPollerOnlyQue
 
         $this->createTask($defaultRun, '01JQUEUENSDEFAULTTASK001', TaskType::Workflow, TaskStatus::Ready, [
             'queue' => 'alpha',
+            'created_at' => now()
+                ->subSeconds(30),
+            'last_dispatched_at' => now()
+                ->subSeconds(15),
         ]);
         $this->createTask($otherRun, '01JQUEUENSOTHERTASK0001', TaskType::Workflow, TaskStatus::Ready, [
             'queue' => 'beta',
@@ -170,8 +182,12 @@ public function testForNamespaceListsTaskQueuesFromTasksAndInjectedPollerOnlyQue
         $this->assertSame(['alpha', 'poller-only'], array_column($snapshot['task_queues'], 'name'));
         $queues = collect($snapshot['task_queues'])->keyBy('name');
         $this->assertSame(1, $queues->get('alpha')['stats']['approximate_backlog_count']);
+        $this->assertSame(1, $queues->get('alpha')['stats']['tasks_added_last_minute']);
+        $this->assertSame(1, $queues->get('alpha')['stats']['tasks_dispatched_last_minute']);
         $this->assertSame(1, $queues->get('poller-only')['stats']['pollers']['active_count']);
         $this->assertSame(0, $queues->get('poller-only')['stats']['approximate_backlog_count']);
+        $this->assertSame(0, $queues->get('poller-only')['stats']['tasks_added_last_minute']);
+        $this->assertSame(0, $queues->get('poller-only')['stats']['tasks_dispatched_last_minute']);
         $this->assertFalse($queues->has('beta'));
     }