Fail closed incompatible workflow starts

Reject workflow start and signal-with-start attempts when only incompatible live workers exist.
Surface compatibility-blocked outcomes through workflow commands, schedules, and control-plane callers instead of starting runs that cannot execute.

Author: durable-workflow-ops

src/V2/CommandResult.php

@@ -130,4 +130,9 @@ public function reason(): ?string
     {
         return $this->command->commandReason();
     }
+
+    public function message(): ?string
+    {
+        return $this->command->commandMessage();
+    }
 }

src/V2/Enums/CommandOutcome.php

@@ -25,6 +25,7 @@ enum CommandOutcome: string
     case RejectedUnknownSignal = 'rejected_unknown_signal';
     case RejectedUnknownUpdate = 'rejected_unknown_update';
     case RejectedInvalidArguments = 'rejected_invalid_arguments';
+    case RejectedCompatibilityBlocked = 'rejected_compatibility_blocked';
     case RejectedPendingSignal = 'rejected_pending_signal';
     case RejectedWorkflowDefinitionUnavailable = 'rejected_workflow_definition_unavailable';
 }

src/V2/Models/WorkflowCommand.php

@@ -430,6 +430,19 @@ public function commandReason(): ?string
         return is_string($reason) && $reason !== '' ? $reason : null;
     }
 
+    public function commandMessage(): ?string
+    {
+        $payload = $this->payloadData();
+
+        if (! is_array($payload)) {
+            return null;
+        }
+
+        $message = $payload['message'] ?? null;
+
+        return is_string($message) && $message !== '' ? $message : null;
+    }
+
     /**
      * @return array<string, mixed>|null
      */

src/V2/Support/DefaultWorkflowControlPlane.php

@@ -30,6 +30,7 @@
 use Workflow\V2\Models\WorkflowTask;
 use Workflow\V2\UpdateResult;
 use Workflow\V2\WorkflowStub;
+use Workflow\V2\Support\WorkflowStartGate;
 
 final class DefaultWorkflowControlPlane implements WorkflowControlPlane
 {
@@ -156,6 +157,38 @@ public function start(string $workflowType, ?string $instanceId = null, array $o
                 $queue = $this->classPropertyDefault($resolvedClass, 'queue');
             }
 
+            $startBlockedReason = WorkflowStartGate::blockedReason(
+                WorkerCompatibility::current(),
+                $connection,
+                $queue,
+            );
+
+            if ($startBlockedReason !== null) {
+                $blockedMessage = WorkflowStartGate::blockedMessage(
+                    sprintf('Workflow instance [%s] cannot start.', $instance->id),
+                    WorkerCompatibility::current(),
+                    $connection,
+                    $queue,
+                ) ?? sprintf('Workflow instance [%s] cannot start.', $instance->id);
+
+                $command = WorkflowCommand::record($instance, null, $this->commandAttributes($commandContext, [
+                    'command_type' => CommandType::Start->value,
+                    'target_scope' => 'instance',
+                    'status' => CommandStatus::Rejected->value,
+                    'outcome' => CommandOutcome::RejectedCompatibilityBlocked->value,
+                    'payload_codec' => CodecRegistry::defaultCodec(),
+                    'payload' => Serializer::serializeWithCodec(CodecRegistry::defaultCodec(), array_filter([
+                        'reason' => $startBlockedReason,
+                        'message' => $blockedMessage,
+                        'arguments_blob' => is_string($arguments) ? $arguments : null,
+                    ], static fn (mixed $value): bool => $value !== null)),
+                    'rejection_reason' => $startBlockedReason,
+                    'rejected_at' => now(),
+                ]));
+
+                return;
+            }
+
             if ($instance->workflow_class !== $workflowClass) {
                 $instance->forceFill([
                     'workflow_class' => $workflowClass,
@@ -316,6 +349,7 @@ public function start(string $workflowType, ?string $instanceId = null, array $o
             'outcome' => $command?->outcome?->value ?? 'unknown',
             'task_id' => $task instanceof WorkflowTask ? $task->id : null,
             'reason' => $accepted ? null : ($command?->rejection_reason ?? 'start_failed'),
+            'message' => $accepted ? null : $command?->commandMessage(),
         ];
     }
 

src/V2/Support/ScheduleManager.php

@@ -14,6 +14,7 @@
 use Workflow\V2\Enums\RunStatus;
 use Workflow\V2\Enums\ScheduleOverlapPolicy;
 use Workflow\V2\Enums\ScheduleStatus;
+use Workflow\V2\Exceptions\WorkflowExecutionUnavailableException;
 use Workflow\V2\Models\WorkflowHistoryEvent;
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowSchedule;
@@ -392,11 +393,20 @@ public static function triggerDetailed(
                 self::closeExistingRun($schedule, $overlapPolicy);
             }
 
-            $startResult = self::startRun(
-                $schedule,
-                effectiveOverlapPolicy: $overlapPolicy->value,
-                context: $context,
-            );
+            try {
+                $startResult = self::startRun(
+                    $schedule,
+                    effectiveOverlapPolicy: $overlapPolicy->value,
+                    context: $context,
+                );
+            } catch (WorkflowExecutionUnavailableException $exception) {
+                self::recordSkip($schedule, $exception->blockedReason(), $context);
+                $schedule->forceFill([
+                    'next_fire_at' => $schedule->computeNextFireAtWithJitter(),
+                ])->save();
+
+                return new ScheduleTriggerResult('skipped', null, null, $exception->blockedReason());
+            }
 
             return new ScheduleTriggerResult('triggered', $startResult->instanceId, $startResult->runId, null);
         });
@@ -441,6 +451,15 @@ public static function tick(int $limit = 100): array
                         'instance_id' => $instanceId,
                     ];
                 }
+            } catch (WorkflowExecutionUnavailableException $exception) {
+                $schedule->refresh();
+                self::recordSkip($schedule, $exception->blockedReason());
+                $results[] = [
+                    'schedule_id' => $schedule->schedule_id,
+                    'instance_id' => null,
+                    'outcome' => 'skipped',
+                    'reason' => $exception->blockedReason(),
+                ];
             } catch (\Throwable $e) {
                 $schedule->refresh();
                 $schedule->recordFailure($e->getMessage());
@@ -617,13 +636,19 @@ private static function triggerForBackfill(
                 self::closeExistingRun($schedule, $effectivePolicy);
             }
 
-            return self::startRun(
-                $schedule,
-                occurrenceTime: $occurrenceTime,
-                outcome: 'backfilled',
-                effectiveOverlapPolicy: $effectivePolicy->value,
-                context: $context,
-            )->instanceId;
+            try {
+                return self::startRun(
+                    $schedule,
+                    occurrenceTime: $occurrenceTime,
+                    outcome: 'backfilled',
+                    effectiveOverlapPolicy: $effectivePolicy->value,
+                    context: $context,
+                )->instanceId;
+            } catch (WorkflowExecutionUnavailableException $exception) {
+                self::recordSkip($schedule, $exception->blockedReason(), $context);
+
+                return null;
+            }
         });
     }
 
@@ -642,6 +667,8 @@ private static function startRun(
 
         try {
             $result = $starter->start($schedule, $occurrenceTime, $outcome, $effectiveOverlapPolicy);
+        } catch (WorkflowExecutionUnavailableException $exception) {
+            throw $exception;
         } catch (\Throwable $e) {
             $schedule->recordFailure($e->getMessage());
             $schedule->save();

src/V2/Support/WorkflowStartGate.php

@@ -0,0 +1,50 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Support;
+
+final class WorkflowStartGate
+{
+    public const BLOCKED_COMPATIBILITY = 'compatibility_blocked';
+
+    public static function blockedReason(
+        ?string $required,
+        ?string $connection = null,
+        ?string $queue = null,
+    ): ?string {
+        if (config('workflows.v2.fleet.validation_mode') !== 'fail') {
+            return null;
+        }
+
+        if ($required === null) {
+            return null;
+        }
+
+        if (WorkerCompatibilityFleet::activeWorkerCount($connection, $queue) === 0) {
+            return null;
+        }
+
+        if (WorkerCompatibilityFleet::supports($required, $connection, $queue)) {
+            return null;
+        }
+
+        return self::BLOCKED_COMPATIBILITY;
+    }
+
+    public static function blockedMessage(
+        string $prefix,
+        ?string $required,
+        ?string $connection = null,
+        ?string $queue = null,
+    ): ?string {
+        if (self::blockedReason($required, $connection, $queue) === null) {
+            return null;
+        }
+
+        $reason = WorkerCompatibilityFleet::mismatchReason($required, $connection, $queue)
+            ?? 'No compatible worker is live for the requested queue.';
+
+        return sprintf('%s Start blocked under fail validation mode. %s', $prefix, $reason);
+    }
+}