Surface rollout drain state on server health endpoints

Author: durable-workflow-ops

README.md

@@ -498,9 +498,9 @@ workflow-task command payload.
 - `GET /api/health` — Health check
 - `GET /api/ready` — Readiness check for migrations, default namespace, cache, auth config, and workflow v2 rollout-safety health
 - `GET /api/cluster/info` — Server capabilities, role topology, coordination-health summary, and version
-- `GET /api/system/health` — Full rollout-safety health snapshot for the requested namespace, including check status, categories, operator metrics, and structural limits
+- `GET /api/system/health` — Full rollout-safety health snapshot for the requested namespace, including check status, categories, routing-drain state, operator metrics, and structural limits
 - `GET /api/system/metrics` — Server metrics including bounded stuck workflow-task diagnostics
-- `GET /api/system/operator-metrics` — Full operator metrics snapshot (runs, tasks, backlog, repair, workers/fleet, backend, structural limits) for rollout-safety coordination health
+- `GET /api/system/operator-metrics` — Full operator metrics snapshot (runs, tasks, backlog, repair, workers/fleet, backend, structural limits) for namespace-scoped rollout-safety coordination health
 - `GET /api/system/repair` — Task repair diagnostics
 - `POST /api/system/repair/pass` — Run task repair sweep
 - `GET /api/system/activity-timeouts` — Expired activity execution diagnostics
@@ -696,11 +696,13 @@ control/execution topology.
 The same `GET /api/cluster/info` response now includes a versioned
 `coordination_health` manifest for rollout-safety coordination risk. It
 summarizes the current server-wide workflow v2 health status, warning and error
-check names, category counts, and the normalized check list that already powers
-the readiness gate. The manifest is intentionally `all_namespaces` scoped so it
-describes the server's fleet-wide coordination posture; use
-`GET /api/system/operator-metrics` when you need namespace-specific backlog and
-worker detail.
+check names, category counts, the normalized check list that already powers the
+readiness gate, and a `routing_drains` summary that lists namespaces and task
+queues with draining build-id cohorts. The manifest is intentionally
+`all_namespaces` scoped so it describes the server's fleet-wide coordination
+posture; use `GET /api/system/health` for the namespace-scoped
+`routing_drains` view and `GET /api/system/operator-metrics` when you need
+namespace-specific backlog and worker detail.
 
 The activity-grade external execution surface is published from
 `GET /api/cluster/info` at

app/Http/Controllers/Api/HealthController.php

@@ -10,6 +10,7 @@
 use App\Support\CoordinationHealthContract;
 use App\Support\ServerReadiness;
 use App\Support\ServerTopology;
+use App\Support\TaskQueueBuildIdRolloutSnapshot;
 use App\Support\WorkerProtocol;
 use Illuminate\Http\JsonResponse;
 use Illuminate\Http\Request;
@@ -26,6 +27,7 @@ class HealthController
 
     public function __construct(
         private readonly ServerReadiness $readiness,
+        private readonly TaskQueueBuildIdRolloutSnapshot $buildIdRollouts,
     ) {}
 
     public function check(): JsonResponse
@@ -120,7 +122,10 @@ public function clusterInfo(Request $request): JsonResponse
             ],
             'structural_limits' => StructuralLimits::snapshot(),
             'topology' => ServerTopology::info(),
-            'coordination_health' => CoordinationHealthContract::manifest($this->readiness->workflowStatus()),
+            'coordination_health' => CoordinationHealthContract::manifest(
+                $this->readiness->workflowStatus(),
+                $this->buildIdRollouts->routingDrains(),
+            ),
             'client_compatibility' => ClientCompatibility::info(),
             'auth_composition_contract' => AuthCompositionContract::manifest(),
             'control_plane' => ControlPlaneProtocol::info(),

app/Http/Controllers/Api/SystemController.php

@@ -5,6 +5,7 @@
 use App\Support\ControlPlaneProtocol;
 use App\Support\HistoryRetentionEnforcer;
 use App\Support\ProjectionDriftMetrics;
+use App\Support\TaskQueueBuildIdRolloutSnapshot;
 use App\Support\WorkflowTaskFailureMetrics;
 use Illuminate\Http\JsonResponse;
 use Illuminate\Http\Request;
@@ -17,6 +18,10 @@
 
 class SystemController
 {
+    public function __construct(
+        private readonly TaskQueueBuildIdRolloutSnapshot $buildIdRollouts,
+    ) {}
+
     public function repairPass(Request $request): JsonResponse
     {
         if ($response = ControlPlaneProtocol::rejectUnsupported($request)) {
@@ -104,6 +109,7 @@ public function health(Request $request): JsonResponse
 
         $namespace = (string) $request->attributes->get('namespace');
         $snapshot = HealthCheck::snapshot(null, $namespace);
+        $snapshot['routing_drains'] = $this->buildIdRollouts->routingDrains($namespace);
 
         return ControlPlaneProtocol::json([
             'namespace' => $namespace,

app/Http/Controllers/Api/TaskQueueController.php

@@ -6,6 +6,7 @@
 use App\Models\WorkerRegistration;
 use App\Support\ControlPlaneProtocol;
 use App\Support\TaskQueueAdmission;
+use App\Support\TaskQueueBuildIdRolloutSnapshot;
 use App\Support\WorkflowQueryTaskBroker;
 use Illuminate\Http\JsonResponse;
 use Illuminate\Http\Request;
@@ -18,6 +19,7 @@ class TaskQueueController
     public function __construct(
         private readonly WorkflowQueryTaskBroker $queryTasks,
         private readonly TaskQueueAdmission $admission,
+        private readonly TaskQueueBuildIdRolloutSnapshot $buildIdRollouts,
     ) {}
 
     public function index(Request $request): JsonResponse
@@ -90,156 +92,12 @@ public function buildIds(Request $request, string $taskQueue): JsonResponse
             return $response;
         }
 
-        $namespace = (string) $request->attributes->get('namespace');
-        $staleAfter = $this->workerStaleAfterSeconds();
-        $now = now();
-
-        $workers = WorkerRegistration::query()
-            ->where('namespace', $namespace)
-            ->where('task_queue', $taskQueue)
-            ->orderByDesc('last_heartbeat_at')
-            ->orderBy('worker_id')
-            ->get();
-
-        $groups = [];
-
-        foreach ($workers as $worker) {
-            $buildId = is_string($worker->build_id) && trim($worker->build_id) !== ''
-                ? trim($worker->build_id)
-                : null;
-            $key = $buildId ?? '__unversioned__';
-
-            $heartbeat = $worker->last_heartbeat_at;
-            $isStale = $heartbeat
-                && $heartbeat->lt($now->copy()->subSeconds($staleAfter));
-            $declaredStatus = is_string($worker->status) ? $worker->status : 'active';
-            $effectiveStatus = $isStale ? 'stale' : $declaredStatus;
-
-            $groups[$key] ??= [
-                'build_id' => $buildId,
-                'active_worker_count' => 0,
-                'stale_worker_count' => 0,
-                'draining_worker_count' => 0,
-                'total_worker_count' => 0,
-                'runtimes' => [],
-                'sdk_versions' => [],
-                'last_heartbeat_at' => null,
-                'first_seen_at' => null,
-            ];
-
-            $groups[$key]['total_worker_count']++;
-            if ($effectiveStatus === 'stale') {
-                $groups[$key]['stale_worker_count']++;
-            } elseif ($effectiveStatus === 'draining') {
-                $groups[$key]['draining_worker_count']++;
-            } else {
-                $groups[$key]['active_worker_count']++;
-            }
-
-            if (is_string($worker->runtime) && trim($worker->runtime) !== '') {
-                $groups[$key]['runtimes'][trim($worker->runtime)] = true;
-            }
-            if (is_string($worker->sdk_version) && trim($worker->sdk_version) !== '') {
-                $groups[$key]['sdk_versions'][trim($worker->sdk_version)] = true;
-            }
-
-            if ($heartbeat !== null) {
-                $existing = $groups[$key]['last_heartbeat_at'];
-                if ($existing === null || $heartbeat->gt($existing)) {
-                    $groups[$key]['last_heartbeat_at'] = $heartbeat;
-                }
-            }
-
-            $createdAt = $worker->created_at;
-            if ($createdAt !== null) {
-                $existing = $groups[$key]['first_seen_at'];
-                if ($existing === null || $createdAt->lt($existing)) {
-                    $groups[$key]['first_seen_at'] = $createdAt;
-                }
-            }
-        }
-
-        $rolloutMap = $this->rolloutsForTaskQueue($namespace, $taskQueue);
-
-        $buildIds = [];
-        foreach ($groups as $group) {
-            $runtimes = array_keys($group['runtimes']);
-            sort($runtimes);
-            $sdkVersions = array_keys($group['sdk_versions']);
-            sort($sdkVersions);
-
-            $rolloutKey = WorkerBuildIdRollout::buildIdKey($group['build_id']);
-            $rollout = $rolloutMap[$rolloutKey] ?? null;
-            $drainIntent = $rollout?->drain_intent ?? WorkerBuildIdRollout::DRAIN_INTENT_ACTIVE;
-
-            $buildIds[] = [
-                'build_id' => $group['build_id'],
-                'rollout_status' => $this->buildIdRolloutStatus(
-                    $group['active_worker_count'],
-                    $group['draining_worker_count'],
-                    $group['stale_worker_count'],
-                    $drainIntent,
-                ),
-                'drain_intent' => $drainIntent,
-                'drained_at' => $rollout?->drained_at?->toJSON(),
-                'active_worker_count' => $group['active_worker_count'],
-                'draining_worker_count' => $group['draining_worker_count'],
-                'stale_worker_count' => $group['stale_worker_count'],
-                'total_worker_count' => $group['total_worker_count'],
-                'runtimes' => $runtimes,
-                'sdk_versions' => $sdkVersions,
-                'last_heartbeat_at' => $group['last_heartbeat_at']?->toJSON(),
-                'first_seen_at' => $group['first_seen_at']?->toJSON(),
-            ];
-        }
-
-        // Surface rollout intent even for cohorts whose worker rows have
-        // been removed: operators still need to see "this build_id is
-        // marked draining" until they explicitly resume it.
-        foreach ($rolloutMap as $key => $rollout) {
-            if (isset($groups[$key === '' ? '__unversioned__' : $key])) {
-                continue;
-            }
-
-            $buildIds[] = [
-                'build_id' => $rollout->publicBuildId(),
-                'rollout_status' => $this->buildIdRolloutStatus(
-                    0,
-                    0,
-                    0,
-                    $rollout->drain_intent,
-                ),
-                'drain_intent' => $rollout->drain_intent,
-                'drained_at' => $rollout->drained_at?->toJSON(),
-                'active_worker_count' => 0,
-                'draining_worker_count' => 0,
-                'stale_worker_count' => 0,
-                'total_worker_count' => 0,
-                'runtimes' => [],
-                'sdk_versions' => [],
-                'last_heartbeat_at' => null,
-                'first_seen_at' => null,
-            ];
-        }
-
-        usort($buildIds, function (array $a, array $b): int {
-            $rankA = $this->buildIdRolloutRank($a);
-            $rankB = $this->buildIdRolloutRank($b);
-            if ($rankA !== $rankB) {
-                return $rankA <=> $rankB;
-            }
-            $heartA = $a['last_heartbeat_at'] ?? '';
-            $heartB = $b['last_heartbeat_at'] ?? '';
-
-            return strcmp($heartB, $heartA);
-        });
-
-        return ControlPlaneProtocol::json([
-            'namespace' => $namespace,
-            'task_queue' => $taskQueue,
-            'stale_after_seconds' => $staleAfter,
-            'build_ids' => $buildIds,
-        ]);
+        return ControlPlaneProtocol::json(
+            $this->buildIdRollouts->forTaskQueue(
+                (string) $request->attributes->get('namespace'),
+                $taskQueue,
+            ),
+        );
     }
 
     /**
@@ -332,77 +190,6 @@ private function setBuildIdDrainIntent(Request $request, string $taskQueue, stri
         ]);
     }
 
-    /**
-     * @return array<string, WorkerBuildIdRollout>
-     */
-    private function rolloutsForTaskQueue(string $namespace, string $taskQueue): array
-    {
-        $rollouts = WorkerBuildIdRollout::query()
-            ->where('namespace', $namespace)
-            ->where('task_queue', $taskQueue)
-            ->get();
-
-        $map = [];
-        foreach ($rollouts as $rollout) {
-            $map[(string) $rollout->build_id] = $rollout;
-        }
-
-        return $map;
-    }
-
-    private function buildIdRolloutStatus(
-        int $active,
-        int $draining,
-        int $stale,
-        string $drainIntent = WorkerBuildIdRollout::DRAIN_INTENT_ACTIVE,
-    ): string {
-        $intentDraining = $drainIntent === WorkerBuildIdRollout::DRAIN_INTENT_DRAINING;
-
-        if ($active > 0) {
-            return $intentDraining || $draining > 0 ? 'active_with_draining' : 'active';
-        }
-
-        if ($draining > 0) {
-            return 'draining';
-        }
-
-        if ($intentDraining) {
-            // Operator intent is to drain, but no live workers remain to
-            // acknowledge it. Keep the cohort visible as draining so the
-            // rollout state is clear even after stale workers are purged.
-            return 'draining';
-        }
-
-        return $stale > 0 ? 'stale_only' : 'no_workers';
-    }
-
-    /**
-     * Sort key: rollout-active builds first, then draining, then stale.
-     * Within each rollout-status bucket the unversioned cohort sorts last
-     * so the named builds an operator is rolling out are visible above
-     * the legacy default — but a stale named build still sorts below an
-     * active unversioned cohort.
-     *
-     * @param  array<string, mixed>  $entry
-     */
-    private function buildIdRolloutRank(array $entry): int
-    {
-        $statusRank = match ($entry['rollout_status'] ?? '') {
-            'active' => 0,
-            'active_with_draining' => 1,
-            'draining' => 2,
-            'stale_only' => 3,
-            default => 4,
-        };
-
-        $rank = $statusRank * 2;
-        if (($entry['build_id'] ?? null) === null) {
-            $rank += 1;
-        }
-
-        return $rank;
-    }
-
     /**
      * @param  array<string, mixed>  $payload
      * @return array<string, mixed>