Surface rejected workflow starts in control-plane responses

Surface rejected workflow starts in control-plane responses

Author: durable-workflow-ops

app/Http/Controllers/Api/BridgeAdapterController.php

@@ -6,6 +6,7 @@
 use App\Support\ControlPlaneProtocol;
 use App\Support\NamespaceExternalPayloadStorage;
 use App\Support\NamespaceWorkflowScope;
+use App\Support\TaskQueueRoutingGate;
 use App\Support\WorkflowCommandContextFactory;
 use App\Support\WorkflowStartService;
 use Illuminate\Http\JsonResponse;
@@ -22,6 +23,7 @@ class BridgeAdapterController
     public function __construct(
         private readonly WorkflowStartService $workflowStartService,
         private readonly WorkflowControlPlane $workflowControlPlane,
+        private readonly TaskQueueRoutingGate $taskQueueRoutingGate,
         private readonly WorkflowCommandContextFactory $commandContexts,
         private readonly NamespaceExternalPayloadStorage $externalPayloadStorage,
     ) {}
@@ -111,6 +113,33 @@ 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 {
             $start = $this->workflowStartService->start(
@@ -150,8 +179,29 @@ private function startWorkflow(
 
         NamespaceWorkflowScope::bind($namespace, $start['workflow_id'], $start['workflow_type']);
 
+        $started = (bool) ($start['started'] ?? false);
         $duplicate = $start['outcome'] === 'returned_existing_active';
 
+        if (! $started && ! $duplicate) {
+            return $this->rejected(
+                $request,
+                $adapter,
+                'start_workflow',
+                $idempotencyKey,
+                $start['rejection_reason'] ?? $start['reason'] ?? 'unsupported_routing',
+                array_filter([
+                    'message' => $start['message'] ?? null,
+                    'target' => $this->redactedTarget($target + ['workflow_id' => $start['workflow_id']]),
+                    'correlation' => $correlation,
+                    'workflow_id' => $start['workflow_id'],
+                    'run_id' => $start['run_id'],
+                    'workflow_type' => $start['workflow_type'],
+                    'rejection_reason' => $start['rejection_reason'] ?? null,
+                    'control_plane_outcome' => $start['outcome'],
+                ], static fn (mixed $value): bool => $value !== null),
+            );
+        }
+
         return $this->outcome($request, [
             'adapter' => $adapter,
             'action' => 'start_workflow',

app/Http/Controllers/Api/WorkflowController.php

@@ -221,6 +221,7 @@ public function start(Request $request): JsonResponse
         }
 
         $workflowId = $start['workflow_id'];
+        $started = (bool) ($start['started'] ?? false);
 
         NamespaceWorkflowScope::bind(
             $namespace,
@@ -239,6 +240,11 @@ public function start(Request $request): JsonResponse
             'business_key' => $run?->business_key,
             'payload_codec' => $run?->payload_codec,
             'outcome' => $start['outcome'],
+            'command_status' => $started ? 'accepted' : 'rejected',
+            'command_source' => 'control_plane',
+            'reason' => $start['reason'],
+            'rejection_reason' => $start['rejection_reason'],
+            'message' => $start['message'],
         ], $this->startStatusCode($start['outcome']));
     }
 

app/Support/BridgeAdapterOutcomeContract.php

@@ -72,6 +72,12 @@ public static function manifest(): array
                     'idempotency_key',
                     'target',
                     'correlation',
+                    'workflow_id',
+                    'run_id',
+                    'workflow_type',
+                    'control_plane_outcome',
+                    'rejection_reason',
+                    'message',
                 ],
                 'redaction' => [
                     'never_echo' => ['authorization', 'signature', 'token', 'secret', 'raw_payload'],
@@ -109,6 +115,9 @@ public static function manifest(): array
                 'auth_failed',
                 'malformed_payload',
                 'duplicate_start',
+                'instance_already_started',
+                'compatibility_blocked',
+                'task_queue_draining',
                 'unsupported_routing',
                 'unsupported_action',
                 'payload_too_large',
@@ -195,6 +204,12 @@ public static function manifest(): array
                             'reason' => 'duplicate_start',
                             'control_plane_outcome' => 'returned_existing_active',
                         ],
+                        'incompatible_fleet' => [
+                            'http_status' => 422,
+                            'outcome' => 'rejected',
+                            'reason' => 'compatibility_blocked',
+                            'control_plane_outcome' => 'rejected_compatibility_blocked',
+                        ],
                         'unconfigured_workflow_type' => [
                             'http_status' => 422,
                             'outcome' => 'rejected',

app/Support/WorkflowStartService.php

@@ -19,11 +19,14 @@ public function __construct(
     /**
      * @param  array<string, mixed>  $validated
      * @return array{
+     *     started: bool,
      *     workflow_id: string,
      *     run_id: string|null,
      *     workflow_type: string,
      *     outcome: string|null,
      *     reason: string|null,
+     *     rejection_reason: string|null,
+     *     message: string|null,
      * }
      */
     public function start(
@@ -43,21 +46,27 @@ public function start(
     /**
      * @param  array<string, mixed>  $validated
      * @return array{
+     *     started: bool,
      *     workflow_id: string,
      *     run_id: string|null,
      *     workflow_type: string,
      *     outcome: string|null,
      *     reason: string|null,
+     *     rejection_reason: string|null,
+     *     message: string|null,
      * }
      */
     /**
      * @param  array<string, mixed>  $validated
      * @return array{
+     *     started: bool,
      *     workflow_id: string,
      *     run_id: string|null,
      *     workflow_type: string,
      *     outcome: string|null,
      *     reason: string|null,
+     *     rejection_reason: string|null,
+     *     message: string|null,
      * }
      */
     private function startRemoteWorkflow(
@@ -99,12 +108,23 @@ private function startRemoteWorkflow(
             'command_context' => $commandContext,
         ], static fn (mixed $value): bool => $value !== null));
 
+        $started = (bool) ($result['started'] ?? false);
+        $reason = isset($result['reason']) && is_string($result['reason'])
+            ? $result['reason']
+            : null;
+        $message = isset($result['message']) && is_string($result['message'])
+            ? $result['message']
+            : null;
+
         return [
+            'started' => $started,
             'workflow_id' => $result['workflow_instance_id'],
             'run_id' => $result['workflow_run_id'],
             'workflow_type' => $result['workflow_type'],
             'outcome' => $result['outcome'],
-            'reason' => $result['reason'],
+            'reason' => $reason,
+            'rejection_reason' => $started ? null : $reason,
+            'message' => $message,
         ];
     }
 

tests/Feature/BridgeAdapterControllerTest.php

@@ -2,12 +2,16 @@
 
 namespace Tests\Feature;
 
+use App\Models\WorkerBuildIdRollout;
+use App\Models\WorkerRegistration;
 use Illuminate\Foundation\Testing\RefreshDatabase;
 use Illuminate\Support\Facades\Queue;
 use Tests\Feature\Concerns\ServerTestHelpers;
 use Tests\Fixtures\InteractiveCommandWorkflow;
 use Tests\TestCase;
 use Workflow\V2\Models\WorkflowCommand;
+use Workflow\V2\Models\WorkflowRun;
+use Workflow\V2\Support\WorkerCompatibilityFleet;
 
 class BridgeAdapterControllerTest extends TestCase
 {
@@ -22,6 +26,14 @@ protected function setUp(): void
         $this->configureWorkflowTypes([
             'tests.interactive-command-workflow' => InteractiveCommandWorkflow::class,
         ]);
+        WorkerCompatibilityFleet::clear();
+    }
+
+    protected function tearDown(): void
+    {
+        WorkerCompatibilityFleet::clear();
+
+        parent::tearDown();
     }
 
     public function test_webhook_bridge_starts_workflow_and_dedupes_by_provider_event(): void
@@ -229,4 +241,109 @@ public function test_webhook_bridge_uses_named_rejections(): void
             ->assertJsonPath('reason', 'unsupported_action')
             ->assertJsonPath('action', 'not_supported');
     }
+
+    public function test_webhook_bridge_blocks_drained_task_queue_starts_without_an_active_worker_cohort(): void
+    {
+        Queue::fake();
+
+        WorkerRegistration::query()->create([
+            'worker_id' => 'draining-worker',
+            'namespace' => 'default',
+            'task_queue' => 'drain-queue',
+            'runtime' => 'php',
+            'sdk_version' => '1.0.0',
+            'build_id' => 'build-draining',
+            'supported_workflow_types' => ['tests.interactive-command-workflow'],
+            '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' => 'drain-queue',
+            'build_id' => 'build-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-drain-1',
+                'target' => [
+                    'workflow_id' => 'wf-bridge-drained-start',
+                    'workflow_type' => 'tests.interactive-command-workflow',
+                    'task_queue' => 'drain-queue',
+                ],
+            ]);
+
+        $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-start')
+            ->assertJsonPath('workflow_type', 'tests.interactive-command-workflow')
+            ->assertJsonPath('task_queue', 'drain-queue')
+            ->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-draining')
+            ->assertJsonPath('drain_intent', 'draining')
+            ->assertJsonPath(
+                'message',
+                'Task queue [drain-queue] 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();
+
+        config()->set('queue.default', 'redis');
+        config()->set('queue.connections.redis.driver', 'redis');
+        config()->set('workflows.v2.compatibility.current', 'build-a');
+        config()->set('workflows.v2.compatibility.supported', ['build-a']);
+        config()->set('workflows.v2.fleet.validation_mode', 'fail');
+
+        WorkerCompatibilityFleet::record(['build-b'], 'redis', 'default', 'worker-build-b');
+
+        $response = $this->withHeaders($this->apiHeaders())
+            ->postJson('/api/bridge-adapters/webhook/stripe', [
+                'action' => 'start_workflow',
+                'idempotency_key' => 'stripe-event-compat-1',
+                'target' => [
+                    'workflow_id' => 'wf-bridge-compatibility-blocked',
+                    'workflow_type' => 'tests.interactive-command-workflow',
+                ],
+            ]);
+
+        $response->assertStatus(422)
+            ->assertJsonPath('adapter', 'stripe')
+            ->assertJsonPath('action', 'start_workflow')
+            ->assertJsonPath('accepted', false)
+            ->assertJsonPath('outcome', 'rejected')
+            ->assertJsonPath('reason', 'compatibility_blocked')
+            ->assertJsonPath('rejection_reason', 'compatibility_blocked')
+            ->assertJsonPath('workflow_id', 'wf-bridge-compatibility-blocked')
+            ->assertJsonPath('run_id', null)
+            ->assertJsonPath('workflow_type', 'tests.interactive-command-workflow')
+            ->assertJsonPath('control_plane_outcome', 'rejected_compatibility_blocked')
+            ->assertJsonPath(
+                'message',
+                'Workflow instance [wf-bridge-compatibility-blocked] cannot start. Start blocked under fail validation mode. '
+                .'No active worker heartbeat advertises compatibility [build-a]. '
+                .'Active workers there advertise [build-b].',
+            );
+
+        $this->assertSame(0, WorkflowRun::query()->count());
+    }
 }