Expose recent queue flow on task queue APIs

Expose recent queue flow on task queue APIs

Author: durable-workflow-ops

README.md

@@ -821,7 +821,7 @@ heartbeat, complete, and fail responses include `run_closed_reason` and
 
 ### Task Queues
 - `GET /api/task-queues` — List task queues
-- `GET /api/task-queues/{name}` — Task queue details and pollers
+- `GET /api/task-queues/{name}` — Task queue details, pollers, and recent add/dispatch flow
 
 Task queue responses include an `admission` object so operators can separate
 worker-local capacity from server-side queue and query-task admission limits. Workflow

app/Http/Controllers/Api/TaskQueueController.php

@@ -9,6 +9,8 @@
 use App\Support\WorkflowQueryTaskBroker;
 use Illuminate\Http\JsonResponse;
 use Illuminate\Http\Request;
+use Workflow\V2\Enums\TaskType;
+use Workflow\V2\Models\WorkflowTask;
 use Workflow\V2\Support\StandaloneWorkerVisibility;
 
 class TaskQueueController
@@ -35,7 +37,7 @@ public function index(Request $request): JsonResponse
         $payload = [
             'namespace' => $snapshot->namespace,
             'task_queues' => array_map(function ($detail) use ($namespace): array {
-                $summary = $detail->toSummaryArray();
+                $summary = $this->withRecentTaskFlow($namespace, $detail->name, $detail->toSummaryArray());
                 $summary['pollers'] = $detail->pollers();
                 $summary = $this->withAdmission($namespace, $summary);
                 unset($summary['pollers']);
@@ -56,13 +58,20 @@ public function show(Request $request, string $taskQueue): JsonResponse
         $namespace = (string) $request->attributes->get('namespace');
 
         return ControlPlaneProtocol::json(
-            $this->withAdmission($namespace, StandaloneWorkerVisibility::queueDetail(
+            $this->withAdmission(
                 $namespace,
-                $taskQueue,
-                WorkerRegistration::class,
-                now(),
-                $this->workerStaleAfterSeconds(),
-            )->toArray()),
+                $this->withRecentTaskFlow(
+                    $namespace,
+                    $taskQueue,
+                    StandaloneWorkerVisibility::queueDetail(
+                        $namespace,
+                        $taskQueue,
+                        WorkerRegistration::class,
+                        now(),
+                        $this->workerStaleAfterSeconds(),
+                    )->toArray(),
+                ),
+            ),
         );
     }
 
@@ -541,6 +550,42 @@ private function taskAdmissionStatus(int $activeWorkerCount, int $configuredSlot
         return $leasedCount >= $configuredSlots ? 'saturated' : 'accepting';
     }
 
+    /**
+     * @param  array<string, mixed>  $payload
+     * @return array<string, mixed>
+     */
+    private function withRecentTaskFlow(string $namespace, string $taskQueue, array $payload): array
+    {
+        $stats = is_array($payload['stats'] ?? null) ? $payload['stats'] : [];
+        $stats = array_merge($stats, $this->recentTaskFlow($namespace, $taskQueue));
+        $payload['stats'] = $stats;
+
+        return $payload;
+    }
+
+    /**
+     * @return array{tasks_added_last_minute: int, tasks_dispatched_last_minute: int}
+     */
+    private function recentTaskFlow(string $namespace, string $taskQueue): array
+    {
+        $windowStart = now()->subMinute();
+
+        $query = WorkflowTask::query()
+            ->where('namespace', $namespace)
+            ->where('queue', $taskQueue)
+            ->whereIn('task_type', [TaskType::Workflow->value, TaskType::Activity->value]);
+
+        return [
+            'tasks_added_last_minute' => (clone $query)
+                ->where('created_at', '>=', $windowStart)
+                ->count(),
+            'tasks_dispatched_last_minute' => (clone $query)
+                ->whereNotNull('last_dispatched_at')
+                ->where('last_dispatched_at', '>=', $windowStart)
+                ->count(),
+        ];
+    }
+
     private function workerStaleAfterSeconds(): int
     {
         $configured = config('server.workers.stale_after_seconds');

tests/Feature/TaskQueueVisibilityTest.php

@@ -110,6 +110,8 @@ public function test_it_reports_workflow_backlog_stale_pollers_and_expired_workf
             ->assertHeader('X-Durable-Workflow-Control-Plane-Version', '2')
             ->assertJsonPath('name', 'external-workflows')
             ->assertJsonPath('stats.approximate_backlog_count', 1)
+            ->assertJsonPath('stats.tasks_added_last_minute', 2)
+            ->assertJsonPath('stats.tasks_dispatched_last_minute', 2)
             ->assertJsonPath('stats.workflow_tasks.ready_count', 1)
             ->assertJsonPath('stats.workflow_tasks.leased_count', 1)
             ->assertJsonPath('stats.workflow_tasks.expired_lease_count', 1)
@@ -153,6 +155,8 @@ public function test_it_reports_workflow_backlog_stale_pollers_and_expired_workf
 
         $list->assertOk()
             ->assertJsonPath('task_queues.0.name', 'external-workflows')
+            ->assertJsonPath('task_queues.0.stats.tasks_added_last_minute', 2)
+            ->assertJsonPath('task_queues.0.stats.tasks_dispatched_last_minute', 2)
             ->assertJsonPath('task_queues.0.admission.workflow_tasks.configured_slot_count', 10)
             ->assertJsonPath('task_queues.0.admission.query_tasks.status', 'full')
             ->assertJsonMissingPath('task_queues.0.pollers');
@@ -214,6 +218,8 @@ public function test_it_reports_activity_backlog_and_current_activity_attempt_le
         $describe->assertOk()
             ->assertHeader('X-Durable-Workflow-Control-Plane-Version', '2')
             ->assertJsonPath('stats.approximate_backlog_count', 1)
+            ->assertJsonPath('stats.tasks_added_last_minute', 2)
+            ->assertJsonPath('stats.tasks_dispatched_last_minute', 2)
             ->assertJsonPath('stats.workflow_tasks.ready_count', 0)
             ->assertJsonPath('stats.activity_tasks.ready_count', 1)
             ->assertJsonPath('stats.activity_tasks.leased_count', 1)