Revert "Block default-queue starts on drained cohorts"

This reverts commit a1dc399.

Author: durable-workflow-ops

app/Http/Controllers/Api/BridgeAdapterController.php

@@ -113,40 +113,32 @@ private function startWorkflow(
         $workflowId = is_string($startTarget['workflow_id'] ?? null)
             ? $startTarget['workflow_id']
             : $this->workflowIdFor($adapter, $idempotencyKey);
-
-        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));
+        $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 {
@@ -168,7 +160,6 @@ private function startWorkflow(
                         'adapter' => $adapter,
                         'action' => 'start_workflow',
                         'idempotency_key' => $idempotencyKey,
-                        'task_queue' => $taskQueue,
                     ],
                 ),
             );

app/Http/Controllers/Api/WorkflowController.php

@@ -172,36 +172,31 @@ public function start(Request $request): JsonResponse
             ], 409);
         }
 
-        try {
-            $taskQueue = $this->workflowStartService->resolveTaskQueue(
-                $validated['workflow_type'],
-                $validated['task_queue'] ?? null,
-            );
-        } catch (LogicException $exception) {
-            throw ValidationException::withMessages([
-                'workflow_type' => [$exception->getMessage()],
-            ]);
-        }
-
-        $routingBlock = $this->taskQueueRoutingGate->workflowStartBlock((string) $namespace, $taskQueue);
+        $taskQueue = isset($validated['task_queue']) && is_string($validated['task_queue'])
+            ? trim($validated['task_queue'])
+            : null;
 
-        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);
+        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);
+            }
         }
 
         try {
@@ -214,7 +209,7 @@ public function start(Request $request): JsonResponse
                     commandName: 'start',
                     metadata: array_filter([
                         'workflow_type' => $validated['workflow_type'],
-                        'task_queue' => $taskQueue,
+                        'task_queue' => $validated['task_queue'] ?? null,
                         'duplicate_policy' => $validated['duplicate_policy'] ?? null,
                     ], static fn (mixed $value): bool => $value !== null),
                 ),

app/Support/ConfiguredWorkflowTypeValidator.php

@@ -7,30 +7,6 @@
 
 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
      */
@@ -49,7 +25,7 @@ public function validationMessage(string $workflowType): ?string
             return null;
         }
 
-        if ($this->resolveWorkflowClass($workflowType) !== null) {
+        if (class_exists($workflowType) && is_subclass_of($workflowType, Workflow::class)) {
             return null;
         }
 

app/Support/WorkflowStartService.php

@@ -7,8 +7,6 @@
 use Workflow\V2\CommandContext;
 use Workflow\V2\Contracts\WorkflowControlPlane;
 use Workflow\V2\Support\PayloadEnvelopeResolver;
-use Workflow\V2\Support\RoutingResolver;
-use Workflow\WorkflowMetadata;
 
 class WorkflowStartService
 {
@@ -40,37 +38,9 @@ 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);
 
-        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();
+        return $this->startRemoteWorkflow($workflowType, $workflowId, $validated, $namespace, $commandContext);
     }
 
     /**
@@ -102,7 +72,6 @@ public function resolveTaskQueue(string $workflowType, mixed $requestedTaskQueue
     private function startRemoteWorkflow(
         string $workflowType,
         ?string $workflowId,
-        string $taskQueue,
         array $validated,
         ?string $namespace = null,
         ?CommandContext $commandContext = null,
@@ -123,7 +92,9 @@ private function startRemoteWorkflow(
         $result = $this->controlPlane->start($workflowType, $workflowId, array_filter([
             'arguments' => $arguments,
             'payload_codec' => $payloadCodec,
-            'queue' => $taskQueue,
+            'queue' => isset($validated['task_queue']) && is_string($validated['task_queue'])
+                ? $validated['task_queue']
+                : null,
             'business_key' => isset($validated['business_key']) && is_string($validated['business_key'])
                 ? $validated['business_key']
                 : null,
@@ -185,19 +156,4 @@ 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,71 +304,6 @@ 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();