Block default-queue starts on drained cohorts

Resolve the effective default queue before the drain gate so implicit starts and webhook bridge starts fail closed while the default cohort is draining.

Update the compatibility-blocked assertions to match the queue-qualified rejection detail emitted after routing context resolution.

Author: rmcdaniel

app/Http/Controllers/Api/BridgeAdapterController.php

@@ -113,32 +113,40 @@ private function startWorkflow(
         $workflowId = is_string($startTarget['workflow_id'] ?? null)
             ? $startTarget['workflow_id']
             : $this->workflowIdFor($adapter, $idempotencyKey);
-        $taskQueue = is_string($startTarget['task_queue'] ?? null)
-            ? trim($startTarget['task_queue'])
-            : null;
-
-        if ($taskQueue !== null && $taskQueue !== '') {
-            $routingBlock = $this->taskQueueRoutingGate->workflowStartBlock((string) $namespace, $taskQueue);
-
-            if ($routingBlock !== null) {
-                return $this->rejected($request, $adapter, 'start_workflow', $idempotencyKey, 'task_queue_draining', array_filter([
-                    'message' => sprintf(
-                        'Task queue [%s] is draining and cannot accept new workflow starts until an active worker cohort is available.',
-                        $taskQueue,
-                    ),
-                    'target' => $this->redactedTarget($target + ['workflow_id' => $workflowId]),
-                    'correlation' => $correlation,
-                    'workflow_id' => $workflowId,
-                    'workflow_type' => $startTarget['workflow_type'],
-                    'task_queue' => $taskQueue,
-                    'routing_status' => $routingBlock['routing_status'],
-                    'active_worker_count' => $routingBlock['active_worker_count'],
-                    'draining_worker_count' => $routingBlock['draining_worker_count'],
-                    'stale_worker_count' => $routingBlock['stale_worker_count'],
-                    'draining_build_ids' => $routingBlock['draining_build_ids'],
-                    'drain_intent' => 'draining',
-                ], static fn (mixed $value): bool => $value !== null));
-            }
+
+        try {
+            $taskQueue = $this->workflowStartService->resolveTaskQueue(
+                $startTarget['workflow_type'],
+                $startTarget['task_queue'] ?? null,
+            );
+        } catch (LogicException $exception) {
+            return $this->rejected($request, $adapter, 'start_workflow', $idempotencyKey, 'unknown_target', [
+                'message' => $exception->getMessage(),
+                'target' => $this->redactedTarget($target),
+                'correlation' => $correlation,
+            ]);
+        }
+
+        $routingBlock = $this->taskQueueRoutingGate->workflowStartBlock((string) $namespace, $taskQueue);
+
+        if ($routingBlock !== null) {
+            return $this->rejected($request, $adapter, 'start_workflow', $idempotencyKey, 'task_queue_draining', array_filter([
+                'message' => sprintf(
+                    'Task queue [%s] is draining and cannot accept new workflow starts until an active worker cohort is available.',
+                    $taskQueue,
+                ),
+                'target' => $this->redactedTarget($target + ['workflow_id' => $workflowId]),
+                'correlation' => $correlation,
+                'workflow_id' => $workflowId,
+                'workflow_type' => $startTarget['workflow_type'],
+                'task_queue' => $taskQueue,
+                'routing_status' => $routingBlock['routing_status'],
+                'active_worker_count' => $routingBlock['active_worker_count'],
+                'draining_worker_count' => $routingBlock['draining_worker_count'],
+                'stale_worker_count' => $routingBlock['stale_worker_count'],
+                'draining_build_ids' => $routingBlock['draining_build_ids'],
+                'drain_intent' => 'draining',
+            ], static fn (mixed $value): bool => $value !== null));
         }
 
         try {
@@ -160,6 +168,7 @@ private function startWorkflow(
                         'adapter' => $adapter,
                         'action' => 'start_workflow',
                         'idempotency_key' => $idempotencyKey,
+                        'task_queue' => $taskQueue,
                     ],
                 ),
             );

app/Http/Controllers/Api/WorkflowController.php

@@ -172,31 +172,36 @@ public function start(Request $request): JsonResponse
             ], 409);
         }
 
-        $taskQueue = isset($validated['task_queue']) && is_string($validated['task_queue'])
-            ? trim($validated['task_queue'])
-            : null;
+        try {
+            $taskQueue = $this->workflowStartService->resolveTaskQueue(
+                $validated['workflow_type'],
+                $validated['task_queue'] ?? null,
+            );
+        } catch (LogicException $exception) {
+            throw ValidationException::withMessages([
+                'workflow_type' => [$exception->getMessage()],
+            ]);
+        }
 
-        if ($taskQueue !== null && $taskQueue !== '') {
-            $routingBlock = $this->taskQueueRoutingGate->workflowStartBlock((string) $namespace, $taskQueue);
-
-            if ($routingBlock !== null) {
-                return ControlPlaneProtocol::jsonForRequest($request, array_filter([
-                    'workflow_id' => $workflowId,
-                    'workflow_type' => $validated['workflow_type'],
-                    'task_queue' => $taskQueue,
-                    'message' => sprintf(
-                        'Task queue [%s] is draining and cannot accept new workflow starts until an active worker cohort is available.',
-                        $taskQueue,
-                    ),
-                    'reason' => 'task_queue_draining',
-                    'routing_status' => $routingBlock['routing_status'],
-                    'active_worker_count' => $routingBlock['active_worker_count'],
-                    'draining_worker_count' => $routingBlock['draining_worker_count'],
-                    'stale_worker_count' => $routingBlock['stale_worker_count'],
-                    'draining_build_ids' => $routingBlock['draining_build_ids'],
-                    'drain_intent' => 'draining',
-                ], static fn (mixed $value): bool => $value !== null), 409);
-            }
+        $routingBlock = $this->taskQueueRoutingGate->workflowStartBlock((string) $namespace, $taskQueue);
+
+        if ($routingBlock !== null) {
+            return ControlPlaneProtocol::jsonForRequest($request, array_filter([
+                'workflow_id' => $workflowId,
+                'workflow_type' => $validated['workflow_type'],
+                'task_queue' => $taskQueue,
+                'message' => sprintf(
+                    'Task queue [%s] is draining and cannot accept new workflow starts until an active worker cohort is available.',
+                    $taskQueue,
+                ),
+                'reason' => 'task_queue_draining',
+                'routing_status' => $routingBlock['routing_status'],
+                'active_worker_count' => $routingBlock['active_worker_count'],
+                'draining_worker_count' => $routingBlock['draining_worker_count'],
+                'stale_worker_count' => $routingBlock['stale_worker_count'],
+                'draining_build_ids' => $routingBlock['draining_build_ids'],
+                'drain_intent' => 'draining',
+            ], static fn (mixed $value): bool => $value !== null), 409);
         }
 
         try {
@@ -209,7 +214,7 @@ public function start(Request $request): JsonResponse
                     commandName: 'start',
                     metadata: array_filter([
                         'workflow_type' => $validated['workflow_type'],
-                        'task_queue' => $validated['task_queue'] ?? null,
+                        'task_queue' => $taskQueue,
                         'duplicate_policy' => $validated['duplicate_policy'] ?? null,
                     ], static fn (mixed $value): bool => $value !== null),
                 ),

app/Support/ConfiguredWorkflowTypeValidator.php

@@ -7,6 +7,30 @@
 
 final class ConfiguredWorkflowTypeValidator
 {
+    /**
+     * @return class-string<Workflow>|null
+     */
+    public function resolveWorkflowClass(string $workflowType): ?string
+    {
+        if (class_exists($workflowType) && is_subclass_of($workflowType, Workflow::class)) {
+            return $workflowType;
+        }
+
+        $configured = config('workflows.v2.types.workflows', []);
+
+        if (! is_array($configured) || ! array_key_exists($workflowType, $configured)) {
+            return null;
+        }
+
+        $workflowClass = $configured[$workflowType];
+
+        if (! is_string($workflowClass) || ! class_exists($workflowClass) || ! is_subclass_of($workflowClass, Workflow::class)) {
+            return null;
+        }
+
+        return $workflowClass;
+    }
+
     /**
      * @throws LogicException
      */
@@ -25,7 +49,7 @@ public function validationMessage(string $workflowType): ?string
             return null;
         }
 
-        if (class_exists($workflowType) && is_subclass_of($workflowType, Workflow::class)) {
+        if ($this->resolveWorkflowClass($workflowType) !== null) {
             return null;
         }
 

app/Support/WorkflowStartService.php

@@ -7,6 +7,8 @@
 use Workflow\V2\CommandContext;
 use Workflow\V2\Contracts\WorkflowControlPlane;
 use Workflow\V2\Support\PayloadEnvelopeResolver;
+use Workflow\V2\Support\RoutingResolver;
+use Workflow\WorkflowMetadata;
 
 class WorkflowStartService
 {
@@ -38,9 +40,37 @@ public function start(
         $workflowId = isset($validated['workflow_id']) && is_string($validated['workflow_id'])
             ? $validated['workflow_id']
             : null;
+        $taskQueue = $this->resolveTaskQueue($workflowType, $validated['task_queue'] ?? null);
+
+        return $this->startRemoteWorkflow(
+            $workflowType,
+            $workflowId,
+            $taskQueue,
+            $validated,
+            $namespace,
+            $commandContext,
+        );
+    }
+
+    public function resolveTaskQueue(string $workflowType, mixed $requestedTaskQueue = null): string
+    {
         $this->workflowTypes->assertLoadable($workflowType);
 
-        return $this->startRemoteWorkflow($workflowType, $workflowId, $validated, $namespace, $commandContext);
+        if (is_string($requestedTaskQueue) && trim($requestedTaskQueue) !== '') {
+            return trim($requestedTaskQueue);
+        }
+
+        $workflowClass = $this->workflowTypes->resolveWorkflowClass($workflowType);
+
+        if ($workflowClass !== null) {
+            $resolvedQueue = RoutingResolver::workflowQueue($workflowClass, new WorkflowMetadata([]));
+
+            if (is_string($resolvedQueue) && trim($resolvedQueue) !== '') {
+                return trim($resolvedQueue);
+            }
+        }
+
+        return $this->defaultTaskQueue();
     }
 
     /**
@@ -72,6 +102,7 @@ public function start(
     private function startRemoteWorkflow(
         string $workflowType,
         ?string $workflowId,
+        string $taskQueue,
         array $validated,
         ?string $namespace = null,
         ?CommandContext $commandContext = null,
@@ -92,9 +123,7 @@ private function startRemoteWorkflow(
         $result = $this->controlPlane->start($workflowType, $workflowId, array_filter([
             'arguments' => $arguments,
             'payload_codec' => $payloadCodec,
-            'queue' => isset($validated['task_queue']) && is_string($validated['task_queue'])
-                ? $validated['task_queue']
-                : null,
+            'queue' => $taskQueue,
             'business_key' => isset($validated['business_key']) && is_string($validated['business_key'])
                 ? $validated['business_key']
                 : null,
@@ -156,4 +185,19 @@ private function intValue(array $validated, string $key): ?int
 
         return is_int($value) ? $value : null;
     }
+
+    private function defaultTaskQueue(): string
+    {
+        $connection = config('queue.default');
+
+        if (! is_string($connection) || trim($connection) === '') {
+            return 'default';
+        }
+
+        $queue = config('queue.connections.'.trim($connection).'.queue', 'default');
+
+        return is_string($queue) && trim($queue) !== ''
+            ? trim($queue)
+            : 'default';
+    }
 }

tests/Feature/BridgeAdapterControllerTest.php

@@ -304,6 +304,71 @@ public function test_webhook_bridge_blocks_drained_task_queue_starts_without_an_
         $this->assertFalse(WorkflowRun::query()->exists());
     }
 
+    public function test_webhook_bridge_rejects_start_when_the_implicit_default_queue_is_draining(): void
+    {
+        Queue::fake();
+
+        config()->set('queue.default', 'redis');
+        config()->set('queue.connections.redis.driver', 'redis');
+        config()->set('queue.connections.redis.queue', 'default');
+
+        WorkerRegistration::query()->create([
+            'worker_id' => 'worker-default-draining',
+            'namespace' => 'default',
+            'task_queue' => 'default',
+            'runtime' => 'php',
+            'sdk_version' => '1.0.0',
+            'build_id' => 'build-default-draining',
+            'supported_workflow_types' => [],
+            'workflow_definition_fingerprints' => [],
+            'supported_activity_types' => [],
+            'max_concurrent_workflow_tasks' => 100,
+            'max_concurrent_activity_tasks' => 100,
+            'last_heartbeat_at' => now(),
+            'status' => 'draining',
+        ]);
+
+        WorkerBuildIdRollout::query()->create([
+            'namespace' => 'default',
+            'task_queue' => 'default',
+            'build_id' => 'build-default-draining',
+            'drain_intent' => 'draining',
+            'drained_at' => now(),
+        ]);
+
+        $response = $this->withHeaders($this->apiHeaders())
+            ->postJson('/api/bridge-adapters/webhook/stripe', [
+                'action' => 'start_workflow',
+                'idempotency_key' => 'stripe-event-default-drain-1',
+                'target' => [
+                    'workflow_id' => 'wf-bridge-drained-default-start',
+                    'workflow_type' => 'tests.interactive-command-workflow',
+                ],
+            ]);
+
+        $response->assertStatus(422)
+            ->assertJsonPath('adapter', 'stripe')
+            ->assertJsonPath('action', 'start_workflow')
+            ->assertJsonPath('accepted', false)
+            ->assertJsonPath('outcome', 'rejected')
+            ->assertJsonPath('reason', 'task_queue_draining')
+            ->assertJsonPath('workflow_id', 'wf-bridge-drained-default-start')
+            ->assertJsonPath('workflow_type', 'tests.interactive-command-workflow')
+            ->assertJsonPath('task_queue', 'default')
+            ->assertJsonPath('routing_status', 'draining')
+            ->assertJsonPath('active_worker_count', 0)
+            ->assertJsonPath('draining_worker_count', 1)
+            ->assertJsonPath('stale_worker_count', 0)
+            ->assertJsonPath('draining_build_ids.0', 'build-default-draining')
+            ->assertJsonPath('drain_intent', 'draining')
+            ->assertJsonPath(
+                'message',
+                'Task queue [default] is draining and cannot accept new workflow starts until an active worker cohort is available.',
+            );
+
+        $this->assertFalse(WorkflowRun::query()->exists());
+    }
+
     public function test_webhook_bridge_surfaces_fail_closed_start_rejection_detail(): void
     {
         Queue::fake();
@@ -340,7 +405,7 @@ public function test_webhook_bridge_surfaces_fail_closed_start_rejection_detail(
             ->assertJsonPath(
                 'message',
                 'Workflow instance [wf-bridge-compatibility-blocked] cannot start. Start blocked under fail validation mode. '
-                .'No active worker heartbeat advertises compatibility [build-a]. '
+                .'No active worker heartbeat for queue [default] advertises compatibility [build-a]. '
                 .'Active workers there advertise [build-b].',
             );