From 026e25879f36b49815141a28a0d2dc63544f925b Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 09:04:05 +0000 Subject: [PATCH 1/6] Reproduce default and opt-out workflow definition upgrade stalls --- .../WorkflowDefinitionEvolutionTest.php | 109 ++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 tests/Feature/WorkflowDefinitionEvolutionTest.php diff --git a/tests/Feature/WorkflowDefinitionEvolutionTest.php b/tests/Feature/WorkflowDefinitionEvolutionTest.php new file mode 100644 index 000000000..93ff05acc --- /dev/null +++ b/tests/Feature/WorkflowDefinitionEvolutionTest.php @@ -0,0 +1,109 @@ +createNamespace('default'); + config(['server.polling.timeout' => 0, 'workflows.v2.compatibility.pin_to_recorded_fingerprint' => true]); + } + + public function test_default_replacement_can_claim_when_the_recorded_definition_is_gone(): void + { + [$run, $task, $started] = $this->original(); + $original = [$started->getRawOriginal('payload'), $run->run_deadline_at, $run->execution_deadline_at]; + $this->deleteJson('/api/workers/original-worker', [], $this->apiHeaders())->assertOk(); + $this->worker('replacement-worker', 'definition-v2'); + + $this->poll('replacement-worker')->assertOk() + ->assertJsonPath('task.run_id', $run->id) + ->assertJsonPath('task.task_id', $task->id); + + self::assertSame($original, [ + $started->fresh()->getRawOriginal('payload'), $run->fresh()->run_deadline_at, $run->fresh()->execution_deadline_at, + ]); + self::assertSame(1, $task->fresh()->attempt_count); + self::assertSame(0, $task->fresh()->repair_count); + } + + public function test_explicit_opt_out_does_not_wait_for_a_matching_peer(): void + { + [$run, $task] = $this->original(); + $this->worker('replacement-worker', 'definition-v2'); + config(['workflows.v2.compatibility.pin_to_recorded_fingerprint' => false]); + + $this->poll('replacement-worker')->assertOk() + ->assertJsonPath('task.run_id', $run->id) + ->assertJsonPath('task.task_id', $task->id); + } + + public function test_default_keeps_the_matching_registered_peer_preference(): void + { + [, $task] = $this->original(); + $this->worker('replacement-worker', 'definition-v2'); + + $this->poll('replacement-worker')->assertOk()->assertJsonPath('task', null); + self::assertSame(TaskStatus::Ready, $task->fresh()->status); + self::assertNull($task->fresh()->lease_owner); + self::assertSame(0, $task->fresh()->attempt_count); + + $this->poll('original-worker')->assertOk()->assertJsonPath('task.task_id', $task->id); + } + + private function original(): array + { + $this->worker('original-worker', 'definition-v1'); + $start = $this->postJson('/api/workflows', [ + 'workflow_id' => 'definition-original', 'workflow_type' => self::TYPE, 'task_queue' => self::QUEUE, + 'run_timeout_seconds' => 60, 'execution_timeout_seconds' => 120, + ], $this->apiHeaders())->assertCreated(); + $run = WorkflowRun::query()->findOrFail($start->json('run_id')); + $task = WorkflowTask::query()->where('workflow_run_id', $run->id)->firstOrFail(); + $started = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id) + ->where('event_type', 'WorkflowStarted')->firstOrFail(); + self::assertSame('worker', $started->payload['workflow_definition_fingerprint_source']); + self::assertSame('definition-v1', $started->payload['workflow_definition_fingerprint']); + + return [$run, $task, $started]; + } + + private function worker(string $id, string $fingerprint): void + { + $this->postJson('/api/worker/register', [ + 'worker_id' => $id, 'task_queue' => self::QUEUE, 'runtime' => 'python', 'sdk_version' => '2.6.1', + 'capability_manifest' => $this->portableWorkerAffinityRefusalManifest(), + 'supported_workflow_types' => [self::TYPE], + 'workflow_definition_fingerprints' => [self::TYPE => $fingerprint], + 'workflow_command_contracts' => [self::TYPE => ['signals' => [], 'queries' => [], 'updates' => [], 'update_validators' => []]], + ], $this->workerHeaders())->assertCreated(); + } + + private function poll(string $id): \Illuminate\Testing\TestResponse + { + return $this->postJson('/api/worker/workflow-tasks/poll', [ + 'worker_id' => $id, 'task_queue' => self::QUEUE, 'timeout_seconds' => 0, + ], $this->workerHeaders()); + } +} From e03bd5fc39166723a8a80430d6fb47e5b1629376 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 09:21:23 +0000 Subject: [PATCH 2/6] Allow strict replay by replacement workflow definitions and explain routing --- .../Controllers/Api/TaskQueueController.php | 38 ++-- app/Http/Controllers/Api/WorkerController.php | 10 +- .../Controllers/Api/WorkflowController.php | 1 + app/Support/ServerWorkflowControlPlane.php | 26 ++- app/Support/WorkflowDefinitionRouting.php | 179 ++++++++++++++++++ app/Support/WorkflowReplayEligibility.php | 40 ++++ app/Support/WorkflowRunDiagnostics.php | 13 ++ app/Support/WorkflowTaskPoller.php | 68 ++----- config/dw-contract.php | 2 +- docs/server-reference.md | 20 +- .../WorkflowDefinitionEvolutionTest.php | 178 ++++++++++++++++- 11 files changed, 501 insertions(+), 74 deletions(-) create mode 100644 app/Support/WorkflowDefinitionRouting.php create mode 100644 app/Support/WorkflowReplayEligibility.php diff --git a/app/Http/Controllers/Api/TaskQueueController.php b/app/Http/Controllers/Api/TaskQueueController.php index f27274fda..907d18fe7 100644 --- a/app/Http/Controllers/Api/TaskQueueController.php +++ b/app/Http/Controllers/Api/TaskQueueController.php @@ -10,6 +10,7 @@ use App\Support\TaskQueueAdmission; use App\Support\TaskQueueBuildIdRolloutSnapshot; use App\Support\TaskQueuePriorityFairnessSurface; +use App\Support\WorkflowDefinitionRouting; use App\Support\WorkflowQueryTaskBroker; use Illuminate\Http\JsonResponse; use Illuminate\Http\Request; @@ -28,6 +29,7 @@ public function __construct( private readonly TaskQueueBuildIdRolloutSnapshot $buildIdRollouts, private readonly TaskQueuePriorityFairnessSurface $priorityFairness, private readonly DeploymentLifecycleService $deployments, + private readonly WorkflowDefinitionRouting $definitionRouting, ) {} public function index(Request $request): JsonResponse @@ -67,22 +69,26 @@ public function show(Request $request, string $taskQueue): JsonResponse $namespace = (string) $request->attributes->get('namespace'); - return ControlPlaneProtocol::json( - $this->withAdmission( + $payload = $this->withAdmission( + $namespace, + $this->withRecentTaskFlow( $namespace, - $this->withRecentTaskFlow( + $taskQueue, + StandaloneWorkerVisibility::queueDetail( $namespace, $taskQueue, - StandaloneWorkerVisibility::queueDetail( - $namespace, - $taskQueue, - WorkerRegistration::class, - now(), - $this->workerStaleAfterSeconds(), - )->toArray(), - ), + WorkerRegistration::class, + now(), + $this->workerStaleAfterSeconds(), + )->toArray(), ), ); + $payload['definition_routing'] = $this->definitionRouting->forQueue($namespace, $taskQueue); + if ($payload['definition_routing']['blocked_sample_count'] > 0) { + $payload['repair']['needs_attention'] = true; + } + + return ControlPlaneProtocol::json($payload); } /** @@ -263,7 +269,7 @@ private function setBuildIdDrainIntent(Request $request, string $taskQueue, stri } /** - * @param array $deployment + * @param array $deployment * @return array */ private function buildIdLifecyclePayload( @@ -351,7 +357,7 @@ private function normalizeDeploymentLifecycleTimestamps(array $deployment): arra } /** - * @param list> $blockages + * @param list> $blockages */ private function orderBlockages(array &$blockages): void { @@ -630,6 +636,7 @@ private function activityAttemptStateCounts(string $namespace, string $taskQueue if ($attemptStatus === ActivityAttemptStatus::Running->value) { $counts['running_count'] += $count; $counts['open_count'] += $count; + continue; } @@ -639,6 +646,7 @@ private function activityAttemptStateCounts(string $namespace, string $taskQueue ) { $counts['retrying_count'] += $count; $counts['open_count'] += $count; + continue; } @@ -646,6 +654,7 @@ private function activityAttemptStateCounts(string $namespace, string $taskQueue $counts['timed_out_count'] += $count; $counts['expired_count'] += $count; $counts['terminal_count'] += $count; + continue; } @@ -656,18 +665,21 @@ private function activityAttemptStateCounts(string $namespace, string $taskQueue ) { $counts['timed_out_count'] += $count; $counts['terminal_count'] += $count; + continue; } if ($attemptStatus === ActivityAttemptStatus::Failed->value) { $counts['failed_count'] += $count; $counts['terminal_count'] += $count; + continue; } if ($attemptStatus === ActivityAttemptStatus::Completed->value) { $counts['completed_count'] += $count; $counts['terminal_count'] += $count; + continue; } diff --git a/app/Http/Controllers/Api/WorkerController.php b/app/Http/Controllers/Api/WorkerController.php index 9d9e3c6ee..a90ca6ccf 100644 --- a/app/Http/Controllers/Api/WorkerController.php +++ b/app/Http/Controllers/Api/WorkerController.php @@ -46,6 +46,7 @@ use App\Support\WorkflowHistoryPageToken; use App\Support\WorkflowMetadataCapabilityPolicy; use App\Support\WorkflowQueryTaskBroker; +use App\Support\WorkflowReplayEligibility; use App\Support\WorkflowStreamCommandProcessor; use App\Support\WorkflowStreamService; use App\Support\WorkflowTaskLeaseRecovery; @@ -407,7 +408,10 @@ public function register(Request $request): JsonResponse $this->validateWorkerTaskCapacity($maxWorkflowTasks, $maxActivityTasks); $maxWorkerSessions = $validated['max_concurrent_worker_sessions'] ?? 10; $taskSlots = is_array($validated['task_slots'] ?? null) ? $validated['task_slots'] : []; - $processMetrics = $this->normalizeProcessMetrics($validated['process_metrics'] ?? null); + $processMetrics = [ + ...($this->normalizeProcessMetrics($validated['process_metrics'] ?? null) ?? []), + WorkflowReplayEligibility::PROTOCOL_METRIC => WorkerProtocol::requestVersion($request), + ]; $releaseLeasesForRegistration = $this->shouldReleaseLeasesForWorkerRegistration($existing, $processMetrics); try { @@ -935,6 +939,10 @@ public function heartbeat(Request $request): JsonResponse if (array_key_exists('process_metrics', $validated)) { $update['process_metrics'] = $this->normalizeProcessMetrics($validated['process_metrics']); } + $update['process_metrics'] = [ + ...($update['process_metrics'] ?? $worker->process_metrics ?? []), + WorkflowReplayEligibility::PROTOCOL_METRIC => WorkerProtocol::requestVersion($request), + ]; if (array_key_exists('heartbeat_interval_seconds', $validated) && $validated['heartbeat_interval_seconds'] !== null) { diff --git a/app/Http/Controllers/Api/WorkflowController.php b/app/Http/Controllers/Api/WorkflowController.php index 7af65aa5b..d2c323820 100644 --- a/app/Http/Controllers/Api/WorkflowController.php +++ b/app/Http/Controllers/Api/WorkflowController.php @@ -1250,6 +1250,7 @@ private function formatRun(WorkflowRun $run, string $namespace, array $descripti 'compatibility_supported_in_fleet' => $this->compatibilitySupportedInFleet($namespace, $run), 'compatibility_fleet_reason' => $this->compatibilityFleetReason($namespace, $run), 'payload_codec' => $run->payload_codec, + 'definition_routing' => $description['definition_routing'] ?? null, 'execution_timeout_seconds' => $description['execution_timeout_seconds'] ?? null, 'run_timeout_seconds' => $runDescription['run_timeout_seconds'] ?? null, 'execution_deadline_at' => $runDescription['execution_deadline_at'] ?? null, diff --git a/app/Support/ServerWorkflowControlPlane.php b/app/Support/ServerWorkflowControlPlane.php index 162735706..3094a9476 100644 --- a/app/Support/ServerWorkflowControlPlane.php +++ b/app/Support/ServerWorkflowControlPlane.php @@ -26,6 +26,7 @@ public function __construct( private readonly ControlPlaneFailureDiagnostics $failureDiagnostics, private readonly NamespaceDurableStateQuota $durableStateQuota, private readonly WorkflowStartVersionPin $versionPin, + private readonly WorkflowDefinitionRouting $definitionRouting, ) {} public function start(string $workflowType, ?string $instanceId = null, array $options = []): array @@ -148,10 +149,12 @@ public function terminate(string $instanceId, array $options = []): array public function repair(string $instanceId, array $options = []): array { - return $this->mutateCommand( + $result = $this->mutateCommand( $instanceId, 'repair', HistoryEventType::RepairRequested, $options, fn (): array => $this->inner->repair($instanceId, $options), ); + + return $this->withDefinitionRouting($instanceId, $options, $result); } public function redrive(string $instanceId, string $failedRunId, array $options = []): array @@ -309,7 +312,26 @@ private function recoveredCommandResult(Throwable $exception, WorkflowCommand $c public function describe(string $instanceId, array $options = []): array { - return $this->inner->describe($instanceId, $options); + return $this->withDefinitionRouting($instanceId, $options, $this->inner->describe($instanceId, $options)); + } + + private function withDefinitionRouting(string $instanceId, array $options, array $result): array + { + $namespace = $this->namespace($options); + if ($namespace === null || ($result['found'] ?? true) === false + || in_array($result['reason'] ?? null, ['instance_not_found', 'run_not_found'], true)) { + return $result; + } + + $runId = data_get($result, 'run.run_id') ?? $result['workflow_run_id'] ?? $options['run_id'] ?? null; + $run = WorkflowRun::query()->where('namespace', $namespace)->where('workflow_instance_id', $instanceId) + ->when(is_string($runId), static fn ($query) => $query->whereKey($runId)) + ->orderByDesc('run_number')->first(); + if ($run instanceof WorkflowRun) { + $result['definition_routing'] = $this->definitionRouting->inspect($run); + } + + return $result; } /** diff --git a/app/Support/WorkflowDefinitionRouting.php b/app/Support/WorkflowDefinitionRouting.php new file mode 100644 index 000000000..cb1f264fa --- /dev/null +++ b/app/Support/WorkflowDefinitionRouting.php @@ -0,0 +1,179 @@ + */ + public function forWorker( + WorkflowRun $run, + WorkerRegistration $worker, + ?string $taskQueue = null, + ?string $requiredCompatibility = null, + ): array { + $decision = $this->context($run, $taskQueue, $requiredCompatibility); + $recorded = $decision['recorded_fingerprint']; + $advertised = self::stringValue($worker->workflow_definition_fingerprints[$run->workflow_type] ?? null); + $decision['advertised_fingerprint'] = $advertised; + + if (! $decision['worker_pinned']) { + return [...$decision, 'allowed' => true, 'status' => 'not_worker_pinned']; + } + if (! $decision['pinning_enabled']) { + return [...$decision, 'allowed' => true, 'status' => 'pinning_disabled']; + } + if ($recorded === null || $advertised === null) { + return [...$decision, 'allowed' => false, 'status' => $recorded === null + ? 'recorded_definition_missing' : 'worker_definition_missing']; + } + if (hash_equals($recorded, $advertised)) { + return [...$decision, 'allowed' => true, 'status' => 'matching_definition']; + } + + foreach ($this->eligibleWorkers($run, $decision['task_queue'], $decision['required_build_id']) as $peer) { + if (self::stringValue($peer->workflow_definition_fingerprints[$run->workflow_type] ?? null) === $recorded) { + return [...$decision, 'allowed' => false, 'status' => 'matching_definition_preferred', + 'preferred_worker_id' => $peer->worker_id]; + } + } + + // Match the embedded resolver: prefer the recorded source if it is + // available, otherwise let the current definition validate replay. + // This never changes the run's original fingerprint or build pin. + return [...$decision, 'allowed' => true, 'status' => 'replay_required']; + } + + /** @return array */ + public function inspect(WorkflowRun $run, ?string $taskQueue = null, ?string $requiredCompatibility = null): array + { + $task = WorkflowTask::query()->where('workflow_run_id', $run->id) + ->where('task_type', TaskType::Workflow->value)->where('status', TaskStatus::Ready->value) + ->where(static fn ($query) => $query->whereNull('available_at')->orWhere('available_at', '<=', now()->format('Y-m-d H:i:s.u'))) + ->when($taskQueue !== null, static fn ($query) => $query->where('queue', $taskQueue)) + ->orderBy('id')->first(); + if ($task instanceof WorkflowTask) { + $taskQueue ??= $task->queue; + $requiredCompatibility ??= TaskCompatibility::resolve($task, $run); + } + $context = $this->context($run, $taskQueue, $requiredCompatibility); + $matching = $changed = $missing = 0; + $examples = []; + + foreach ($this->eligibleWorkers($run, (string) $context['task_queue'], $context['required_build_id']) as $worker) { + $fingerprint = self::stringValue($worker->workflow_definition_fingerprints[$run->workflow_type] ?? null); + if ($fingerprint === null) { + $missing++; + } elseif ($fingerprint === $context['recorded_fingerprint']) { + $matching++; + } else { + $changed++; + } + if (count($examples) < 10) { + $examples[] = ['worker_id' => $worker->worker_id, 'fingerprint' => $fingerprint]; + } + } + + $eligible = $matching + $changed + $missing; + $status = match (true) { + ! $context['worker_pinned'] => 'not_worker_pinned', + ! $context['pinning_enabled'] => 'pinning_disabled', + $context['recorded_fingerprint'] === null => 'recorded_definition_missing', + $matching > 0 => 'matching_definition_available', + $changed > 0 => 'replay_required', + $missing > 0 => 'worker_definition_missing', + default => 'no_eligible_worker', + }; + $runnable = ! $run->status->isTerminal() && $task instanceof WorkflowTask; + $blocked = $runnable && $context['worker_pinned'] && $context['pinning_enabled'] + && in_array($status, ['recorded_definition_missing', 'worker_definition_missing', 'no_eligible_worker'], true); + + return [...$context, 'status' => $status, 'blocked' => $blocked, 'runnable_task' => $runnable, + 'eligible_worker_count' => $eligible, 'matching_worker_count' => $matching, + 'changed_worker_count' => $changed, 'missing_fingerprint_count' => $missing, + 'workers' => $examples, 'workers_truncated' => $eligible > count($examples), + 'resolution' => $blocked + ? 'Register a fresh worker for the original namespace, task queue, workflow type, build and replay capabilities with its definition fingerprint. Do not rewrite the run history.' + : ($status === 'replay_required' + ? 'Current workers may replay the original history. Use versioned branches for command changes; incompatible replay still refuses.' + : 'Recorded build, namespace, type and replay-capability guards remain in force.'), + ]; + } + + /** Bounded examples of due ready runs, never a repair or a payload read. */ + public function forQueue(string $namespace, string $taskQueue): array + { + $runs = WorkflowRun::query()->where('namespace', $namespace)->where('queue', $taskQueue) + ->whereIn('id', WorkflowTask::query()->select('workflow_run_id') + ->where('task_type', TaskType::Workflow->value)->where('status', TaskStatus::Ready->value) + ->where('queue', $taskQueue) + ->where(static fn ($query) => $query->whereNull('available_at')->orWhere('available_at', '<=', now()->format('Y-m-d H:i:s.u')))) + ->orderBy('id')->limit(11)->get(); + $examples = []; + foreach ($runs->take(10) as $run) { + if (! $run->status->isTerminal()) { + $examples[] = ['workflow_id' => $run->workflow_instance_id, 'run_id' => $run->id, + ...$this->inspect($run, $taskQueue)]; + } + } + + return ['sample_limit' => 10, 'has_more_ready_runs' => $runs->count() > 10, + 'blocked_sample_count' => count(array_filter($examples, static fn (array $row): bool => $row['blocked'])), + 'runs' => $examples]; + } + + /** @return array */ + private function context(WorkflowRun $run, ?string $taskQueue, ?string $requiredCompatibility): array + { + $started = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id) + ->where('event_type', 'WorkflowStarted')->orderBy('sequence')->first(); + $recorded = self::stringValue($started?->payload['workflow_definition_fingerprint'] ?? null); + + return [ + 'pinning_enabled' => (bool) config('workflows.v2.compatibility.pin_to_recorded_fingerprint', true), + 'worker_pinned' => ($started?->payload['workflow_definition_fingerprint_source'] ?? null) === 'worker', + 'recorded_fingerprint' => $recorded, + 'workflow_type' => $run->workflow_type, + 'task_queue' => $taskQueue ?? $run->queue, + 'required_build_id' => $requiredCompatibility ?? self::stringValue($run->compatibility), + ]; + } + + /** @return Generator */ + private function eligibleWorkers( + WorkflowRun $run, + string $taskQueue, + ?string $requiredCompatibility, + ): Generator { + $workers = WorkerRegistration::query()->where('namespace', $run->namespace) + ->where('task_queue', $taskQueue)->where('status', 'active'); + if ($requiredCompatibility !== null) { + $workers->where('build_id', $requiredCompatibility); + } + + foreach ($workers->cursor(['namespace', 'worker_id', 'task_queue', 'build_id', 'status', 'last_heartbeat_at', + 'workflow_definition_fingerprints', 'supported_workflow_types', 'max_concurrent_workflow_tasks', + 'capabilities', 'process_metrics']) as $worker) { + if (! in_array($run->workflow_type, $worker->supported_workflow_types ?? [], true) + || ($worker->max_concurrent_workflow_tasks !== null && (int) $worker->max_concurrent_workflow_tasks < 1)) { + continue; + } + if (WorkflowReplayEligibility::allows($worker, $run, WorkflowReplayEligibility::recordedProtocol($worker))) { + yield $worker; + } + } + } + + private static function stringValue(mixed $value): ?string + { + return is_string($value) && trim($value) !== '' ? trim($value) : null; + } +} diff --git a/app/Support/WorkflowReplayEligibility.php b/app/Support/WorkflowReplayEligibility.php new file mode 100644 index 000000000..172dc4b0f --- /dev/null +++ b/app/Support/WorkflowReplayEligibility.php @@ -0,0 +1,40 @@ +namespace !== $run->namespace || ! WorkerPollFence::isFresh($worker)) { + return false; + } + + $capabilities = is_array($worker->capabilities) ? $worker->capabilities : []; + + if (is_string($run->cancellation_request_command_id) + && ! CooperativeCancellationPolicy::supports($capabilities, $protocolVersion)) { + return false; + } + if (is_string($run->cancellation_request_command_id) + && NexusCooperativeCancellationPolicy::requiresCapabilityForReplay($run) + && ! NexusCooperativeCancellationPolicy::supports($capabilities, $protocolVersion)) { + return false; + } + + return PreparedLocalActivityPolicy::canReplayRun($run->id, $capabilities, $protocolVersion) + && WorkflowMetadataCapabilityPolicy::canReplayRun($run->id, $capabilities, $protocolVersion); + } + + public static function recordedProtocol(WorkerRegistration $worker): ?string + { + $version = $worker->process_metrics[self::PROTOCOL_METRIC] ?? null; + + return is_string($version) && $version !== '' ? $version : null; + } +} diff --git a/app/Support/WorkflowRunDiagnostics.php b/app/Support/WorkflowRunDiagnostics.php index 6f6490561..810d3f1be 100644 --- a/app/Support/WorkflowRunDiagnostics.php +++ b/app/Support/WorkflowRunDiagnostics.php @@ -41,6 +41,8 @@ class WorkflowRunDiagnostics private const LAST_EVENT_PAYLOAD_KEY_LIMIT = 20; + public function __construct(private readonly WorkflowDefinitionRouting $definitionRouting) {} + /** * @return array */ @@ -84,6 +86,7 @@ public function forRun(string $namespace, WorkflowRun $run, bool $includeLastEve 'recent_failures_truncated' => $failureRows->count() > self::FAILURE_LIMIT, 'latest_workflow_task_failure' => $latestWorkflowTaskFailure, 'compatibility' => $this->compatibility($namespace, $run, $summary, $taskQueue), + 'definition_routing' => $this->definitionRouting->inspect($run), 'cancellation_cascade_supported' => class_exists(CancellationCascadeView::class), 'cancellation_cascade' => class_exists(CancellationCascadeView::class) && $run->namespace === $namespace ? CancellationCascadeView::forRun($run) @@ -95,6 +98,16 @@ public function forRun(string $namespace, WorkflowRun $run, bool $includeLastEve ...$this->pendingWorkflowRoutingFindings($namespace, $run, $taskRows->all(), $taskQueue), ]; + if ($payload['definition_routing']['blocked']) { + $payload['findings'][] = [ + 'severity' => 'warning', 'code' => 'workflow_definition_routing_blocked', + 'message' => 'A ready workflow task has no eligible definition route.', + 'routing_status' => $payload['definition_routing']['status'], + 'expected_resolution' => $payload['definition_routing']['resolution'], + ]; + $payload['diagnostic_status'] = 'needs_attention'; + } + return $payload; } diff --git a/app/Support/WorkflowTaskPoller.php b/app/Support/WorkflowTaskPoller.php index 6c21da0b8..4cc917342 100644 --- a/app/Support/WorkflowTaskPoller.php +++ b/app/Support/WorkflowTaskPoller.php @@ -12,7 +12,6 @@ use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; use Workflow\V2\Jobs\RunTimerTask; -use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowSignal; use Workflow\V2\Models\WorkflowTask; @@ -42,6 +41,7 @@ public function __construct( private readonly PollRequestLeaseBinding $pollLeaseBindings, private readonly WorkflowUpdateValidationTaskBroker $updateValidationTasks, private readonly WorkflowTaskKindSelector $taskKindSelector, + private readonly WorkflowDefinitionRouting $definitionRouting, ) {} /** @@ -1041,9 +1041,11 @@ private function claimReadyTask( } if (! $this->matchesRecordedWorkflowDefinition( + $namespace, + $leaseOwner, $runId, - $readyTask['workflow_type'] ?? null, - $workflowDefinitionFingerprints, + $taskQueue, + $effectiveCompatibility, )) { continue; } @@ -2325,63 +2327,25 @@ private function workerCanReplayRun( ->where('worker_id', $workerId) ->first(); - if (! $worker instanceof WorkerRegistration || ! WorkerPollFence::isFresh($worker)) { - return false; - } - - $capabilities = is_array($worker->capabilities) - ? array_values(array_filter( - $worker->capabilities, - static fn (mixed $capability): bool => is_string($capability) && trim($capability) !== '', - )) - : []; - $run = WorkflowRun::query()->where('namespace', $namespace)->find($runId); - if ($run instanceof WorkflowRun && is_string($run->cancellation_request_command_id) - && ! CooperativeCancellationPolicy::supports($capabilities, $protocolVersion)) { - return false; - } - if ($run instanceof WorkflowRun && is_string($run->cancellation_request_command_id) - && NexusCooperativeCancellationPolicy::requiresCapabilityForReplay($run) - && ! NexusCooperativeCancellationPolicy::supports($capabilities, $protocolVersion)) { - return false; - } - return PreparedLocalActivityPolicy::canReplayRun($runId, $capabilities, $protocolVersion) - && WorkflowMetadataCapabilityPolicy::canReplayRun( - $runId, - $capabilities, - $protocolVersion, - ); + return $worker instanceof WorkerRegistration && $run instanceof WorkflowRun + && WorkflowReplayEligibility::allows($worker, $run, $protocolVersion); } - /** @param array $workflowDefinitionFingerprints */ private function matchesRecordedWorkflowDefinition( + string $namespace, + string $workerId, string $runId, - mixed $workflowType, - array $workflowDefinitionFingerprints, + string $taskQueue, + ?string $requiredCompatibility, ): bool { - $started = WorkflowHistoryEvent::query() - ->where('workflow_run_id', $runId) - ->where('event_type', 'WorkflowStarted') - ->orderBy('sequence') - ->first(); - - if (($started?->payload['workflow_definition_fingerprint_source'] ?? null) !== 'worker') { - return true; - } - - $recorded = $this->nonEmptyString($started?->payload['workflow_definition_fingerprint'] ?? null); - - if ($recorded === null) { - return false; - } - - $advertised = is_string($workflowType) - ? $this->nonEmptyString($workflowDefinitionFingerprints[$workflowType] ?? null) - : null; + $run = WorkflowRun::query()->where('namespace', $namespace)->find($runId); + $worker = WorkerRegistration::query()->where('namespace', $namespace) + ->where('worker_id', $workerId)->first(); - return $advertised !== null && hash_equals($recorded, $advertised); + return $run instanceof WorkflowRun && $worker instanceof WorkerRegistration + && $this->definitionRouting->forWorker($run, $worker, $taskQueue, $requiredCompatibility)['allowed']; } /** diff --git a/config/dw-contract.php b/config/dw-contract.php index f9f23786c..2e8abf2f3 100644 --- a/config/dw-contract.php +++ b/config/dw-contract.php @@ -889,7 +889,7 @@ 'legacy' => 'WORKFLOW_V2_COMPATIBILITY_HEARTBEAT_TTL', ], 'DW_V2_PIN_TO_RECORDED_FINGERPRINT' => [ - 'description' => 'When true (default), in-flight runs resolve their workflow class from the fingerprint recorded at WorkflowStarted.', + 'description' => 'Prefer an eligible worker matching the recorded definition. If none remains, current code must validate strict replay. False removes only this source preference, not build pins or replay checks.', 'default' => 'true', 'since' => '2.0.0', 'legacy' => 'WORKFLOW_V2_PIN_TO_RECORDED_FINGERPRINT', diff --git a/docs/server-reference.md b/docs/server-reference.md index 8d836aa79..d793ee967 100644 --- a/docs/server-reference.md +++ b/docs/server-reference.md @@ -418,6 +418,24 @@ Workers that omit fingerprints during re-registration cannot clear previously stored fingerprints for workflow types they still advertise; the server keeps the stored value until a new worker id is used. +Existing runs retain their original definition fingerprint, build pin, history +and deadlines. By default, the Server prefers a fresh eligible worker with the +recorded definition. When that definition is no longer available in the run's +namespace, queue, workflow type and build, a replacement definition can claim +the original task and must pass the SDK's strict replay checks. Use versioned +branches such as `get_version` for changes to durable commands. A different +source fingerprint alone is not proof of incompatible replay. + +`DW_V2_PIN_TO_RECORDED_FINGERPRINT=false` removes the matching-source preference. +It does not remove build pins, namespace or queue isolation, workflow-type and +capability checks, immutable registration identity, lease fences or strict replay. +Run detail, run debug, repair responses and queue detail expose +`definition_routing`. `replay_required` is an allowed fallback, while +`blocked=true` means a due ready workflow task has no eligible definition route. +Repair can fix transport state; it cannot manufacture a compatible worker or +rewrite the original definition. Queue observations contain at most ten run +examples and explicitly report whether more ready runs exist. + ### 5. Poll for Workflow Tasks The server holds the connection open (long-poll) until a task is ready or @@ -1715,7 +1733,7 @@ inside the package's `config/workflows.php` via | `DW_V2_SUPPORTED_COMPATIBILITIES` | (unset) | Comma-separated marker list the worker accepts, or `*` for any. | | `DW_V2_COMPATIBILITY_NAMESPACE` | (unset) | Compatibility namespace for independent fleets sharing one database. | | `DW_V2_COMPATIBILITY_HEARTBEAT_TTL` | `30` | Seconds a worker-compatibility heartbeat remains valid. | -| `DW_V2_PIN_TO_RECORDED_FINGERPRINT` | `true` | Resolve in-flight runs from the fingerprint recorded at WorkflowStarted. | +| `DW_V2_PIN_TO_RECORDED_FINGERPRINT` | `true` | Prefer an eligible recorded-definition worker; otherwise require strict replay by current code. False removes this preference, not build pins or replay checks. | | `DW_V2_CONTINUE_AS_NEW_EVENT_THRESHOLD` | `10000` | History event count at which the package signals continue-as-new. | | `DW_V2_CONTINUE_AS_NEW_SIZE_BYTES_THRESHOLD` | `5242880` | Serialized-history byte count at which the package signals continue-as-new. | | `DW_V2_HISTORY_EVENT_WARNING_THRESHOLD` | `8000` | Event count at which history pressure becomes `approaching`. | diff --git a/tests/Feature/WorkflowDefinitionEvolutionTest.php b/tests/Feature/WorkflowDefinitionEvolutionTest.php index 93ff05acc..3bbd0ac77 100644 --- a/tests/Feature/WorkflowDefinitionEvolutionTest.php +++ b/tests/Feature/WorkflowDefinitionEvolutionTest.php @@ -2,11 +2,16 @@ namespace Tests\Feature; +use App\Models\WorkerRegistration; +use App\Support\WorkflowReplayEligibility; use Illuminate\Foundation\Testing\RefreshDatabase; use Illuminate\Support\Facades\Queue; +use Illuminate\Testing\TestResponse; +use PHPUnit\Framework\Attributes\DataProvider; use Tests\Feature\Concerns\ServerTestHelpers; use Tests\TestCase; use Workflow\V2\Enums\TaskStatus; +use Workflow\V2\Models\WorkflowCommand; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowTask; @@ -33,7 +38,7 @@ protected function setUp(): void public function test_default_replacement_can_claim_when_the_recorded_definition_is_gone(): void { [$run, $task, $started] = $this->original(); - $original = [$started->getRawOriginal('payload'), $run->run_deadline_at, $run->execution_deadline_at]; + $original = [$started->getRawOriginal('payload'), $run->getRawOriginal('run_deadline_at'), $run->getRawOriginal('execution_deadline_at')]; $this->deleteJson('/api/workers/original-worker', [], $this->apiHeaders())->assertOk(); $this->worker('replacement-worker', 'definition-v2'); @@ -42,7 +47,7 @@ public function test_default_replacement_can_claim_when_the_recorded_definition_ ->assertJsonPath('task.task_id', $task->id); self::assertSame($original, [ - $started->fresh()->getRawOriginal('payload'), $run->fresh()->run_deadline_at, $run->fresh()->execution_deadline_at, + $started->fresh()->getRawOriginal('payload'), $run->fresh()->getRawOriginal('run_deadline_at'), $run->fresh()->getRawOriginal('execution_deadline_at'), ]); self::assertSame(1, $task->fresh()->attempt_count); self::assertSame(0, $task->fresh()->repair_count); @@ -72,6 +77,169 @@ public function test_default_keeps_the_matching_registered_peer_preference(): vo $this->poll('original-worker')->assertOk()->assertJsonPath('task.task_id', $task->id); } + #[DataProvider('ineligibleRecordedWorkers')] + public function test_an_ineligible_recorded_peer_does_not_strand_a_replacement(array $attributes): void + { + [, $task] = $this->original(); + WorkerRegistration::query()->where('worker_id', 'original-worker')->firstOrFail() + ->forceFill($attributes)->save(); + $this->worker('replacement-worker', 'definition-v2'); + + $this->poll('replacement-worker')->assertOk()->assertJsonPath('task.task_id', $task->id); + self::assertSame(1, $task->fresh()->attempt_count); + self::assertSame(0, $task->fresh()->repair_count); + } + + public static function ineligibleRecordedWorkers(): array + { + return [ + 'stale heartbeat' => [['last_heartbeat_at' => '2000-01-01 00:00:00']], + 'missing heartbeat' => [['last_heartbeat_at' => null]], + 'draining' => [['status' => 'draining']], + 'other namespace' => [['namespace' => 'other']], + 'other queue' => [['task_queue' => 'other']], + 'other type' => [['supported_workflow_types' => ['other']]], + 'no workflow capacity' => [['max_concurrent_workflow_tasks' => 0]], + 'missing definition' => [['workflow_definition_fingerprints' => []]], + ]; + } + + public function test_busy_matching_peer_retains_preference_without_inventing_slot_authority(): void + { + [, $task] = $this->original(); + WorkerRegistration::query()->where('worker_id', 'original-worker')->update(['available_workflow_slots' => 0]); + $this->worker('replacement-worker', 'definition-v2'); + + $this->poll('replacement-worker')->assertOk()->assertJsonPath('task', null); + self::assertSame(TaskStatus::Ready, $task->fresh()->status); + self::assertSame(0, $task->fresh()->attempt_count); + } + + public function test_fallback_does_not_remove_the_original_build_pin(): void + { + [$run, $task] = $this->original(); + $run->forceFill(['compatibility' => 'build-v1'])->save(); + $task->forceFill(['compatibility' => 'build-v1'])->save(); + $this->deleteJson('/api/workers/original-worker', [], $this->apiHeaders())->assertOk(); + $this->worker('replacement-worker', 'definition-v2', ['build_id' => 'build-v2']); + + $this->poll('replacement-worker', ['build_id' => 'build-v2'])->assertOk()->assertJsonPath('task', null); + self::assertSame(TaskStatus::Ready, $task->fresh()->status); + self::assertSame(0, $task->fresh()->attempt_count); + self::assertSame('build-v1', $run->fresh()->compatibility); + } + + public function test_a_matching_peer_on_another_build_cannot_suppress_a_pinned_replacement(): void + { + [$run, $task] = $this->original(); + $run->forceFill(['compatibility' => 'build-v1'])->save(); + $task->forceFill(['compatibility' => 'build-v1'])->save(); + WorkerRegistration::query()->where('worker_id', 'original-worker')->update(['build_id' => 'build-v2']); + $this->worker('replacement-worker', 'definition-v2', ['build_id' => 'build-v1']); + + $this->poll('replacement-worker', ['build_id' => 'build-v1'])->assertOk()->assertJsonPath('task.task_id', $task->id); + self::assertSame('build-v1', $run->fresh()->compatibility); + } + + public function test_matching_peer_must_support_the_recorded_replay_protocol(): void + { + [$run, $task] = $this->original(); + $worker = WorkerRegistration::query()->where('worker_id', 'original-worker')->firstOrFail(); + $worker->forceFill(['process_metrics' => [WorkflowReplayEligibility::PROTOCOL_METRIC => '1.16']])->save(); + WorkflowHistoryEvent::query()->create([ + 'workflow_run_id' => $run->id, + 'sequence' => WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->max('sequence') + 1, + 'event_type' => 'TimerScheduled', + 'payload' => ['condition_wait_occurrence_id' => 'original-condition-occurrence'], + ]); + $this->worker('replacement-worker', 'definition-v2'); + + $this->poll('replacement-worker')->assertOk()->assertJsonPath('task.task_id', $task->id); + } + + public function test_client_process_metrics_cannot_spoof_the_saved_protocol(): void + { + $this->worker('original-worker', 'definition-v1', [ + 'process_metrics' => [WorkflowReplayEligibility::PROTOCOL_METRIC => '999.0', 'memory_bytes' => 123], + ]); + $worker = WorkerRegistration::query()->where('worker_id', 'original-worker')->firstOrFail(); + self::assertSame($this->workerHeaders()['X-Durable-Workflow-Protocol-Version'], WorkflowReplayEligibility::recordedProtocol($worker)); + self::assertSame(123, $worker->process_metrics['memory_bytes']); + + $this->postJson('/api/worker/heartbeat', [ + 'worker_id' => 'original-worker', 'process_metrics' => [WorkflowReplayEligibility::PROTOCOL_METRIC => '999.0', 'memory_bytes' => 456], + ], $this->workerHeaders())->assertOk(); + self::assertSame($this->workerHeaders()['X-Durable-Workflow-Protocol-Version'], WorkflowReplayEligibility::recordedProtocol($worker->fresh())); + self::assertSame(456, $worker->fresh()->process_metrics['memory_bytes']); + } + + public function test_run_debug_queue_and_repair_distinguish_fallback_from_a_real_block(): void + { + [$run, $task] = $this->original(); + $this->deleteJson('/api/workers/original-worker', [], $this->apiHeaders())->assertOk(); + $this->worker('replacement-worker', 'definition-v2'); + $this->getJson('/api/workflows/definition-original', $this->apiHeaders())->assertOk() + ->assertJsonPath('definition_routing.status', 'replay_required') + ->assertJsonPath('definition_routing.blocked', false) + ->assertJsonPath('definition_routing.recorded_fingerprint', 'definition-v1'); + $this->getJson('/api/workflows/definition-original/runs/'.$run->id, $this->apiHeaders())->assertOk() + ->assertJsonPath('definition_routing.status', 'replay_required'); + $this->getJson('/api/workflows/definition-original/debug', $this->apiHeaders())->assertOk() + ->assertJsonPath('definition_routing.blocked', false) + ->assertJsonMissing(['code' => 'workflow_definition_routing_blocked']); + + WorkerRegistration::query()->where('worker_id', 'replacement-worker')->firstOrFail() + ->forceFill(['workflow_definition_fingerprints' => []])->save(); + $this->getJson('/api/workflows/definition-original', $this->apiHeaders())->assertOk() + ->assertJsonPath('definition_routing.status', 'worker_definition_missing') + ->assertJsonPath('definition_routing.blocked', true); + $this->getJson('/api/workflows/definition-original/debug', $this->apiHeaders())->assertOk() + ->assertJsonPath('diagnostic_status', 'needs_attention') + ->assertJsonFragment(['code' => 'workflow_definition_routing_blocked']); + $this->getJson('/api/task-queues/'.self::QUEUE, $this->apiHeaders())->assertOk() + ->assertJsonPath('definition_routing.blocked_sample_count', 1) + ->assertJsonPath('definition_routing.runs.0.run_id', $run->id) + ->assertJsonPath('repair.needs_attention', true); + $repair = $this->postJson('/api/workflows/definition-original/repair', [ + 'request_id' => 'definition-routing-repair', + ], $this->apiHeaders())->assertOk() + ->assertJsonPath('definition_routing.status', 'worker_definition_missing') + ->assertJsonPath('definition_routing.blocked', true); + $command = WorkflowCommand::query()->findOrFail($repair->json('command_id')); + self::assertSame($command->commandReason(), $repair->json('reason')); + self::assertSame($command->outcome->value, $repair->json('outcome')); + self::assertSame(0, $task->fresh()->attempt_count); + self::assertSame(0, $task->fresh()->repair_count); + + $this->createNamespace('other'); + $this->getJson('/api/task-queues/'.self::QUEUE, $this->apiHeaders('other'))->assertOk() + ->assertJsonPath('definition_routing.runs', []); + } + + #[DataProvider('notRunnableTasks')] + public function test_definition_drift_is_not_a_ready_task_block_when_delivery_is_not_due(string $case): void + { + [$run, $task] = $this->original(); + $this->deleteJson('/api/workers/original-worker', [], $this->apiHeaders())->assertOk(); + if ($case === 'terminal') { + $run->forceFill(['status' => 'completed', 'closed_at' => now()])->save(); + } elseif ($case === 'leased') { + $task->forceFill(['status' => TaskStatus::Leased, 'lease_owner' => 'original-worker'])->save(); + } else { + $task->forceFill(['available_at' => now()->addMinutes(1)])->save(); + } + $this->getJson('/api/workflows/definition-original', $this->apiHeaders())->assertOk() + ->assertJsonPath('definition_routing.runnable_task', false) + ->assertJsonPath('definition_routing.blocked', false); + $this->getJson('/api/workflows/definition-original/debug', $this->apiHeaders())->assertOk() + ->assertJsonMissing(['code' => 'workflow_definition_routing_blocked']); + } + + public static function notRunnableTasks(): array + { + return [['terminal'], ['leased'], ['future']]; + } + private function original(): array { $this->worker('original-worker', 'definition-v1'); @@ -89,7 +257,7 @@ private function original(): array return [$run, $task, $started]; } - private function worker(string $id, string $fingerprint): void + private function worker(string $id, string $fingerprint, array $overrides = []): void { $this->postJson('/api/worker/register', [ 'worker_id' => $id, 'task_queue' => self::QUEUE, 'runtime' => 'python', 'sdk_version' => '2.6.1', @@ -97,13 +265,15 @@ private function worker(string $id, string $fingerprint): void 'supported_workflow_types' => [self::TYPE], 'workflow_definition_fingerprints' => [self::TYPE => $fingerprint], 'workflow_command_contracts' => [self::TYPE => ['signals' => [], 'queries' => [], 'updates' => [], 'update_validators' => []]], + ...$overrides, ], $this->workerHeaders())->assertCreated(); } - private function poll(string $id): \Illuminate\Testing\TestResponse + private function poll(string $id, array $overrides = []): TestResponse { return $this->postJson('/api/worker/workflow-tasks/poll', [ 'worker_id' => $id, 'task_queue' => self::QUEUE, 'timeout_seconds' => 0, + ...$overrides, ], $this->workerHeaders()); } } From 57de9ec39b4107f32d2a7bcb6b070a4158cf3dc1 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 09:28:33 +0000 Subject: [PATCH 3/6] Bind routing diagnostics to the canonical selected or current run --- app/Support/ServerWorkflowControlPlane.php | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/app/Support/ServerWorkflowControlPlane.php b/app/Support/ServerWorkflowControlPlane.php index 3094a9476..caa2da4fe 100644 --- a/app/Support/ServerWorkflowControlPlane.php +++ b/app/Support/ServerWorkflowControlPlane.php @@ -323,10 +323,11 @@ private function withDefinitionRouting(string $instanceId, array $options, array return $result; } - $runId = data_get($result, 'run.run_id') ?? $result['workflow_run_id'] ?? $options['run_id'] ?? null; - $run = WorkflowRun::query()->where('namespace', $namespace)->where('workflow_instance_id', $instanceId) - ->when(is_string($runId), static fn ($query) => $query->whereKey($runId)) - ->orderByDesc('run_number')->first(); + $runId = data_get($result, 'run.run_id') ?? $result['resolved_run_id'] ?? $result['run_id'] + ?? $result['workflow_run_id'] ?? $options['run_id'] ?? null; + $run = is_string($runId) + ? NamespaceWorkflowScope::run($namespace, $instanceId, $runId) + : NamespaceWorkflowScope::currentRun($namespace, $instanceId); if ($run instanceof WorkflowRun) { $result['definition_routing'] = $this->definitionRouting->inspect($run); } From 75d6f3801c8a00bcef016045afe4ecc8d6e942b3 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 10:34:49 +0000 Subject: [PATCH 4/6] Clarify replay routing decisions and preserve explicit metrics clearing --- app/Http/Controllers/Api/WorkerController.php | 2 +- app/Support/WorkflowDefinitionRouting.php | 33 +++++++++---------- app/Support/WorkflowReplayEligibility.php | 9 ++--- app/Support/WorkflowTaskPoller.php | 3 -- .../WorkflowDefinitionEvolutionTest.php | 12 +++++++ 5 files changed, 32 insertions(+), 27 deletions(-) diff --git a/app/Http/Controllers/Api/WorkerController.php b/app/Http/Controllers/Api/WorkerController.php index a90ca6ccf..4da9a14eb 100644 --- a/app/Http/Controllers/Api/WorkerController.php +++ b/app/Http/Controllers/Api/WorkerController.php @@ -940,7 +940,7 @@ public function heartbeat(Request $request): JsonResponse $update['process_metrics'] = $this->normalizeProcessMetrics($validated['process_metrics']); } $update['process_metrics'] = [ - ...($update['process_metrics'] ?? $worker->process_metrics ?? []), + ...(array_key_exists('process_metrics', $update) ? ($update['process_metrics'] ?? []) : ($worker->process_metrics ?? [])), WorkflowReplayEligibility::PROTOCOL_METRIC => WorkerProtocol::requestVersion($request), ]; diff --git a/app/Support/WorkflowDefinitionRouting.php b/app/Support/WorkflowDefinitionRouting.php index cb1f264fa..2719ee015 100644 --- a/app/Support/WorkflowDefinitionRouting.php +++ b/app/Support/WorkflowDefinitionRouting.php @@ -25,18 +25,16 @@ public function forWorker( $advertised = self::stringValue($worker->workflow_definition_fingerprints[$run->workflow_type] ?? null); $decision['advertised_fingerprint'] = $advertised; - if (! $decision['worker_pinned']) { - return [...$decision, 'allowed' => true, 'status' => 'not_worker_pinned']; - } - if (! $decision['pinning_enabled']) { - return [...$decision, 'allowed' => true, 'status' => 'pinning_disabled']; - } - if ($recorded === null || $advertised === null) { - return [...$decision, 'allowed' => false, 'status' => $recorded === null - ? 'recorded_definition_missing' : 'worker_definition_missing']; - } - if (hash_equals($recorded, $advertised)) { - return [...$decision, 'allowed' => true, 'status' => 'matching_definition']; + $immediate = match (true) { + ! $decision['worker_pinned'] => ['allowed' => true, 'status' => 'not_worker_pinned'], + ! $decision['pinning_enabled'] => ['allowed' => true, 'status' => 'pinning_disabled'], + $recorded === null => ['allowed' => false, 'status' => 'recorded_definition_missing'], + $advertised === null => ['allowed' => false, 'status' => 'worker_definition_missing'], + hash_equals($recorded, $advertised) => ['allowed' => true, 'status' => 'matching_definition'], + default => null, + }; + if ($immediate !== null) { + return [...$decision, ...$immediate]; } foreach ($this->eligibleWorkers($run, $decision['task_queue'], $decision['required_build_id']) as $peer) { @@ -95,16 +93,17 @@ public function inspect(WorkflowRun $run, ?string $taskQueue = null, ?string $re $runnable = ! $run->status->isTerminal() && $task instanceof WorkflowTask; $blocked = $runnable && $context['worker_pinned'] && $context['pinning_enabled'] && in_array($status, ['recorded_definition_missing', 'worker_definition_missing', 'no_eligible_worker'], true); + $resolution = match (true) { + $blocked => 'Register a fresh worker for the original namespace, task queue, workflow type, build and replay capabilities with its definition fingerprint. Do not rewrite the run history.', + $status === 'replay_required' => 'Current workers may replay the original history. Use versioned branches for command changes; incompatible replay still refuses.', + default => 'Recorded build, namespace, type and replay-capability guards remain in force.', + }; return [...$context, 'status' => $status, 'blocked' => $blocked, 'runnable_task' => $runnable, 'eligible_worker_count' => $eligible, 'matching_worker_count' => $matching, 'changed_worker_count' => $changed, 'missing_fingerprint_count' => $missing, 'workers' => $examples, 'workers_truncated' => $eligible > count($examples), - 'resolution' => $blocked - ? 'Register a fresh worker for the original namespace, task queue, workflow type, build and replay capabilities with its definition fingerprint. Do not rewrite the run history.' - : ($status === 'replay_required' - ? 'Current workers may replay the original history. Use versioned branches for command changes; incompatible replay still refuses.' - : 'Recorded build, namespace, type and replay-capability guards remain in force.'), + 'resolution' => $resolution, ]; } diff --git a/app/Support/WorkflowReplayEligibility.php b/app/Support/WorkflowReplayEligibility.php index 172dc4b0f..da4744b2d 100644 --- a/app/Support/WorkflowReplayEligibility.php +++ b/app/Support/WorkflowReplayEligibility.php @@ -18,12 +18,9 @@ public static function allows(WorkerRegistration $worker, WorkflowRun $run, ?str $capabilities = is_array($worker->capabilities) ? $worker->capabilities : []; if (is_string($run->cancellation_request_command_id) - && ! CooperativeCancellationPolicy::supports($capabilities, $protocolVersion)) { - return false; - } - if (is_string($run->cancellation_request_command_id) - && NexusCooperativeCancellationPolicy::requiresCapabilityForReplay($run) - && ! NexusCooperativeCancellationPolicy::supports($capabilities, $protocolVersion)) { + && (! CooperativeCancellationPolicy::supports($capabilities, $protocolVersion) + || (NexusCooperativeCancellationPolicy::requiresCapabilityForReplay($run) + && ! NexusCooperativeCancellationPolicy::supports($capabilities, $protocolVersion)))) { return false; } diff --git a/app/Support/WorkflowTaskPoller.php b/app/Support/WorkflowTaskPoller.php index 4cc917342..f21fb6877 100644 --- a/app/Support/WorkflowTaskPoller.php +++ b/app/Support/WorkflowTaskPoller.php @@ -749,7 +749,6 @@ function () use ( historyPageSize: $historyPageSize, acceptHistoryEncoding: $acceptHistoryEncoding, supportedWorkflowTypes: $supportedWorkflowTypes, - workflowDefinitionFingerprints: $workflowDefinitionFingerprints, workerPollFence: $workerPollFence, ), ), @@ -798,7 +797,6 @@ function () use ( historyPageSize: $historyPageSize, acceptHistoryEncoding: $acceptHistoryEncoding, supportedWorkflowTypes: $supportedWorkflowTypes, - workflowDefinitionFingerprints: $workflowDefinitionFingerprints, workerPollFence: $workerPollFence, ), ), @@ -968,7 +966,6 @@ private function claimReadyTask( ?int $historyPageSize = null, ?string $acceptHistoryEncoding = null, array $supportedWorkflowTypes = [], - array $workflowDefinitionFingerprints = [], array $workerPollFence = [], ): ?array { $readyTasks = $this->pollReplayEligibleReadyTasks( diff --git a/tests/Feature/WorkflowDefinitionEvolutionTest.php b/tests/Feature/WorkflowDefinitionEvolutionTest.php index 3bbd0ac77..f8398e18b 100644 --- a/tests/Feature/WorkflowDefinitionEvolutionTest.php +++ b/tests/Feature/WorkflowDefinitionEvolutionTest.php @@ -173,6 +173,18 @@ public function test_client_process_metrics_cannot_spoof_the_saved_protocol(): v self::assertSame(456, $worker->fresh()->process_metrics['memory_bytes']); } + public function test_explicit_null_clears_client_metrics_without_clearing_server_protocol(): void + { + $this->worker('original-worker', 'definition-v1', ['process_metrics' => ['memory_bytes' => 123]]); + $this->postJson('/api/worker/heartbeat', [ + 'worker_id' => 'original-worker', 'process_metrics' => null, + ], $this->workerHeaders())->assertOk(); + $worker = WorkerRegistration::query()->where('worker_id', 'original-worker')->firstOrFail(); + self::assertSame([ + WorkflowReplayEligibility::PROTOCOL_METRIC => $this->workerHeaders()['X-Durable-Workflow-Protocol-Version'], + ], $worker->process_metrics); + } + public function test_run_debug_queue_and_repair_distinguish_fallback_from_a_real_block(): void { [$run, $task] = $this->original(); From ebfcbff19970cd8324df0e6792bef6ed929925de Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 15:03:02 +0000 Subject: [PATCH 5/6] Fence entry-schema failures and require safe replacement workers --- app/Http/Controllers/Api/WorkerController.php | 1 + app/Support/WorkflowDefinitionRouting.php | 18 ++++- docs/server-reference.md | 7 +- .../WorkflowDefinitionEvolutionTest.php | 69 +++++++++++++++++++ 4 files changed, 91 insertions(+), 4 deletions(-) diff --git a/app/Http/Controllers/Api/WorkerController.php b/app/Http/Controllers/Api/WorkerController.php index 4da9a14eb..b192dc0f6 100644 --- a/app/Http/Controllers/Api/WorkerController.php +++ b/app/Http/Controllers/Api/WorkerController.php @@ -161,6 +161,7 @@ class WorkerController 'version_marker_kind_mismatch', 'version_marker_sequence_invalid', 'version_range_invalid', + 'workflow_entry_arguments_mismatch', 'workflow_nondeterministic', ]; diff --git a/app/Support/WorkflowDefinitionRouting.php b/app/Support/WorkflowDefinitionRouting.php index 2719ee015..3109f8bab 100644 --- a/app/Support/WorkflowDefinitionRouting.php +++ b/app/Support/WorkflowDefinitionRouting.php @@ -13,6 +13,8 @@ final class WorkflowDefinitionRouting { + public const ENTRY_VALIDATION_CAPABILITY = 'workflow_entry_arguments_validation'; + /** @return array */ public function forWorker( WorkflowRun $run, @@ -44,6 +46,10 @@ public function forWorker( } } + if (! in_array(self::ENTRY_VALIDATION_CAPABILITY, $worker->capabilities ?? [], true)) { + return [...$decision, 'allowed' => false, 'status' => 'entry_validation_unavailable']; + } + // Match the embedded resolver: prefer the recorded source if it is // available, otherwise let the current definition validate replay. // This never changes the run's original fingerprint or build pin. @@ -63,7 +69,7 @@ public function inspect(WorkflowRun $run, ?string $taskQueue = null, ?string $re $requiredCompatibility ??= TaskCompatibility::resolve($task, $run); } $context = $this->context($run, $taskQueue, $requiredCompatibility); - $matching = $changed = $missing = 0; + $matching = $changed = $missing = $unsafeReplacement = 0; $examples = []; foreach ($this->eligibleWorkers($run, (string) $context['task_queue'], $context['required_build_id']) as $worker) { @@ -74,6 +80,9 @@ public function inspect(WorkflowRun $run, ?string $taskQueue = null, ?string $re $matching++; } else { $changed++; + if (! in_array(self::ENTRY_VALIDATION_CAPABILITY, $worker->capabilities ?? [], true)) { + $unsafeReplacement++; + } } if (count($examples) < 10) { $examples[] = ['worker_id' => $worker->worker_id, 'fingerprint' => $fingerprint]; @@ -86,14 +95,16 @@ public function inspect(WorkflowRun $run, ?string $taskQueue = null, ?string $re ! $context['pinning_enabled'] => 'pinning_disabled', $context['recorded_fingerprint'] === null => 'recorded_definition_missing', $matching > 0 => 'matching_definition_available', - $changed > 0 => 'replay_required', + $changed > $unsafeReplacement => 'replay_required', + $unsafeReplacement > 0 => 'entry_validation_unavailable', $missing > 0 => 'worker_definition_missing', default => 'no_eligible_worker', }; $runnable = ! $run->status->isTerminal() && $task instanceof WorkflowTask; $blocked = $runnable && $context['worker_pinned'] && $context['pinning_enabled'] - && in_array($status, ['recorded_definition_missing', 'worker_definition_missing', 'no_eligible_worker'], true); + && in_array($status, ['recorded_definition_missing', 'worker_definition_missing', 'no_eligible_worker', 'entry_validation_unavailable'], true); $resolution = match (true) { + $blocked && $status === 'entry_validation_unavailable' => 'Upgrade the replacement worker SDK to an implementation advertising workflow_entry_arguments_validation, or restore the recorded definition. The current replacement cannot safely refuse incompatible entry input.', $blocked => 'Register a fresh worker for the original namespace, task queue, workflow type, build and replay capabilities with its definition fingerprint. Do not rewrite the run history.', $status === 'replay_required' => 'Current workers may replay the original history. Use versioned branches for command changes; incompatible replay still refuses.', default => 'Recorded build, namespace, type and replay-capability guards remain in force.', @@ -102,6 +113,7 @@ public function inspect(WorkflowRun $run, ?string $taskQueue = null, ?string $re return [...$context, 'status' => $status, 'blocked' => $blocked, 'runnable_task' => $runnable, 'eligible_worker_count' => $eligible, 'matching_worker_count' => $matching, 'changed_worker_count' => $changed, 'missing_fingerprint_count' => $missing, + 'entry_validation_unavailable_count' => $unsafeReplacement, 'workers' => $examples, 'workers_truncated' => $eligible > count($examples), 'resolution' => $resolution, ]; diff --git a/docs/server-reference.md b/docs/server-reference.md index 00af2598a..2c3fe5e44 100644 --- a/docs/server-reference.md +++ b/docs/server-reference.md @@ -422,7 +422,12 @@ Existing runs retain their original definition fingerprint, build pin, history and deadlines. By default, the Server prefers a fresh eligible worker with the recorded definition. When that definition is no longer available in the run's namespace, queue, workflow type and build, a replacement definition can claim -the original task and must pass the SDK's strict replay checks. Use versioned +the original task only when it advertises `workflow_entry_arguments_validation`, +and must pass the SDK's strict replay checks. This worker capability declares +pre-entry refusal without failing the business run; older worker implementations +do not gain it from their runtime or protocol version. `entry_validation_unavailable` +in `definition_routing` identifies a replacement that needs an SDK upgrade. +Use versioned branches such as `get_version` for changes to durable commands. A different source fingerprint alone is not proof of incompatible replay. diff --git a/tests/Feature/WorkflowDefinitionEvolutionTest.php b/tests/Feature/WorkflowDefinitionEvolutionTest.php index f8398e18b..52e973657 100644 --- a/tests/Feature/WorkflowDefinitionEvolutionTest.php +++ b/tests/Feature/WorkflowDefinitionEvolutionTest.php @@ -4,6 +4,7 @@ use App\Models\WorkerRegistration; use App\Support\WorkflowReplayEligibility; +use App\Support\WorkflowDefinitionRouting; use Illuminate\Foundation\Testing\RefreshDatabase; use Illuminate\Support\Facades\Queue; use Illuminate\Testing\TestResponse; @@ -64,6 +65,24 @@ public function test_explicit_opt_out_does_not_wait_for_a_matching_peer(): void ->assertJsonPath('task.task_id', $task->id); } + public function test_legacy_replacement_cannot_gain_unsafe_default_fallback(): void + { + [$run, $task] = $this->original(); + $this->deleteJson('/api/workers/original-worker', [], $this->apiHeaders())->assertOk(); + $this->worker('legacy-replacement', 'definition-v2', ['capabilities' => []]); + $this->poll('legacy-replacement')->assertOk()->assertJsonPath('task', null); + self::assertSame(0, $task->fresh()->attempt_count); + $this->getJson('/api/workflows/definition-original', $this->apiHeaders())->assertOk() + ->assertJsonPath('definition_routing.status', 'entry_validation_unavailable') + ->assertJsonPath('definition_routing.blocked', true) + ->assertJsonPath('definition_routing.entry_validation_unavailable_count', 1); + $this->worker('safe-replacement', 'definition-v2'); + $this->getJson('/api/workflows/definition-original', $this->apiHeaders())->assertOk() + ->assertJsonPath('definition_routing.status', 'replay_required') + ->assertJsonPath('definition_routing.blocked', false); + $this->poll('safe-replacement')->assertOk()->assertJsonPath('task.run_id', $run->id); + } + public function test_default_keeps_the_matching_registered_peer_preference(): void { [, $task] = $this->original(); @@ -252,6 +271,55 @@ public static function notRunnableTasks(): array return [['terminal'], ['leased'], ['future']]; } + #[DataProvider('entryFailureTypes')] + public function test_entry_binding_failure_preserves_the_original_until_supported_repair(string $failureType): void + { + [$run, $task, $started] = $this->original(); + $original = [$started->getRawOriginal('payload'), $run->getRawOriginal('run_deadline_at'), $run->getRawOriginal('execution_deadline_at')]; + $claim = $this->poll('original-worker')->assertOk(); + $failure = [ + 'lease_owner' => $claim->json('task.lease_owner'), + 'workflow_task_attempt' => $claim->json('task.workflow_task_attempt'), + 'failure' => [ + 'type' => $failureType, + 'reason' => 'workflow_entry_arguments_mismatch', + 'message' => 'The recorded input cannot bind to the deployed entry.', + ], + ]; + + $this->postJson('/api/worker/workflow-tasks/'.$task->id.'/fail', [ + ...$failure, 'lease_owner' => 'another-worker', + ], $this->workerHeaders())->assertConflict(); + self::assertSame(TaskStatus::Leased, $task->fresh()->status); + + $this->postJson('/api/worker/workflow-tasks/'.$task->id.'/fail', $failure, $this->workerHeaders()) + ->assertOk()->assertJsonPath('recorded', true)->assertJsonPath('next_task_id', null); + self::assertSame(TaskStatus::Failed, $task->fresh()->status); + self::assertTrue($task->fresh()->payload['replay_blocked'] ?? false); + self::assertSame('workflow_entry_arguments_mismatch', $task->fresh()->payload['failure_reason']); + self::assertSame(1, WorkflowTask::query()->where('workflow_run_id', $run->id)->count()); + self::assertNotContains($run->fresh()->status->value, ['failed', 'completed']); + $this->poll('original-worker')->assertOk()->assertJsonPath('task', null); + + $this->postJson('/api/workflows/definition-original/repair', [ + 'request_id' => 'entry-compatible-repair', + ], $this->apiHeaders())->assertOk(); + $this->poll('original-worker')->assertOk()->assertJsonPath('task.task_id', $task->id); + self::assertSame(1, $task->fresh()->repair_count); + self::assertSame($original, [ + $started->fresh()->getRawOriginal('payload'), $run->fresh()->getRawOriginal('run_deadline_at'), $run->fresh()->getRawOriginal('execution_deadline_at'), + ]); + } + + public static function entryFailureTypes(): array + { + return [ + 'PHP' => ['DurableWorkflow\\Exception\\WorkflowEntryArgumentsMismatch'], + 'Python' => ['WorkflowEntryArgumentsMismatchError'], + 'Rust' => ['WorkflowEntryArgumentsMismatch'], + ]; + } + private function original(): array { $this->worker('original-worker', 'definition-v1'); @@ -273,6 +341,7 @@ private function worker(string $id, string $fingerprint, array $overrides = []): { $this->postJson('/api/worker/register', [ 'worker_id' => $id, 'task_queue' => self::QUEUE, 'runtime' => 'python', 'sdk_version' => '2.6.1', + 'capabilities' => [WorkflowDefinitionRouting::ENTRY_VALIDATION_CAPABILITY], 'capability_manifest' => $this->portableWorkerAffinityRefusalManifest(), 'supported_workflow_types' => [self::TYPE], 'workflow_definition_fingerprints' => [self::TYPE => $fingerprint], From 85a81e67e2db0469a86831efe0171e8ce69dbdf4 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 15:24:26 +0000 Subject: [PATCH 6/6] Keep replacement safety decision in one final return --- app/Support/WorkflowDefinitionRouting.php | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/app/Support/WorkflowDefinitionRouting.php b/app/Support/WorkflowDefinitionRouting.php index 3109f8bab..0579f3ba0 100644 --- a/app/Support/WorkflowDefinitionRouting.php +++ b/app/Support/WorkflowDefinitionRouting.php @@ -46,14 +46,13 @@ public function forWorker( } } - if (! in_array(self::ENTRY_VALIDATION_CAPABILITY, $worker->capabilities ?? [], true)) { - return [...$decision, 'allowed' => false, 'status' => 'entry_validation_unavailable']; - } - // Match the embedded resolver: prefer the recorded source if it is // available, otherwise let the current definition validate replay. // This never changes the run's original fingerprint or build pin. - return [...$decision, 'allowed' => true, 'status' => 'replay_required']; + $entryValidation = in_array(self::ENTRY_VALIDATION_CAPABILITY, $worker->capabilities ?? [], true); + + return [...$decision, 'allowed' => $entryValidation, + 'status' => $entryValidation ? 'replay_required' : 'entry_validation_unavailable']; } /** @return array */