From 4798a1465f179e681dafff603731fa899783ff07 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 05:59:31 +0000 Subject: [PATCH 1/9] feat: guard prepared retry admission with retained authority --- .../Api/PreparedLocalActivityController.php | 132 +++++++-- app/Support/PreparedLocalRetryAdmission.php | 259 ++++++++++++++++++ app/Support/RuntimePayloadCompletionLease.php | 18 +- docs/contracts/prepared-local-retry-v2.md | 37 +++ .../worker-protocol-api.openapi.yaml | 51 +++- .../PreparedLocalRetryAdmissionTest.php | 139 ++++++++++ 6 files changed, 613 insertions(+), 23 deletions(-) create mode 100644 app/Support/PreparedLocalRetryAdmission.php create mode 100644 tests/Feature/PreparedLocalRetryAdmissionTest.php diff --git a/app/Http/Controllers/Api/PreparedLocalActivityController.php b/app/Http/Controllers/Api/PreparedLocalActivityController.php index e6ef5eae0..e58c2d1b5 100644 --- a/app/Http/Controllers/Api/PreparedLocalActivityController.php +++ b/app/Http/Controllers/Api/PreparedLocalActivityController.php @@ -12,6 +12,7 @@ use App\Support\PayloadCodecContract; use App\Support\PreparedLocalActivityAdmissionRefused; use App\Support\PreparedLocalActivityPolicy; +use App\Support\PreparedLocalRetryAdmission; use App\Support\ServiceModeTimerDispatcher; use App\Support\WorkerPollFence; use App\Support\WorkerProtocol; @@ -24,6 +25,7 @@ use Workflow\V2\Contracts\CancellationScopeAdmission; use Workflow\V2\Contracts\PreparedCancellationScopeTaskBridge; use Workflow\V2\Contracts\PreparedLocalActivityTaskBridge; +use Workflow\V2\Contracts\PreparedLocalActivityRetryTaskBridge; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Exceptions\ExternalPayloadIntegrityException; use Workflow\V2\Exceptions\StructuralLimitExceededException; @@ -66,6 +68,13 @@ private function admit(Request $request, string $taskId, bool $recover): JsonRes $request->validate([ 'descriptor.cancellation_scope_id' => ['sometimes', 'required', 'string', 'max:255'], ]); + $retry = PreparedLocalRetryAdmission::descriptorIntent($validated['descriptor']); + if ($retry) { + if (! PreparedLocalRetryAdmission::backendSupported()) { + return self::refused($validated['lease_owner'], ['installed_runtime_prepared_local_retry'], PreparedLocalRetryAdmission::CAPABILITY); + } + PreparedLocalRetryAdmission::validatePolicy($validated['descriptor']); + } $namespace = (string) $request->attributes->get('namespace'); $task = NamespaceWorkflowScope::task($namespace, $taskId); if ($task === null) { @@ -79,12 +88,15 @@ private function admit(Request $request, string $taskId, bool $recover): JsonRes if ($claim === null) { return self::refused($validated['lease_owner'], ['worker_claim_capability']); } + if ($retry && ! PreparedLocalRetryAdmission::claimSupports($claim)) { + return self::refused($validated['lease_owner'], ['worker_claim_retry_capability'], PreparedLocalRetryAdmission::CAPABILITY); + } try { if ($response = $this->cancellationScopeRefusal($request, $task, $validated['descriptor'], (int) $validated['sequence'])) { return $response; } $validated['descriptor'] = $this->resolvePayload($validated['descriptor'], 'arguments', 'descriptor', $namespace); - $reply = $this->mutations->run(fn (): array|JsonResponse => DB::transaction(function () use ($request, $namespace, $taskId, $validated, $claim, $recover): array|JsonResponse { + $reply = $this->mutations->run(fn (): array|JsonResponse => DB::transaction(function () use ($request, $namespace, $taskId, $validated, $claim, $recover, $retry): array|JsonResponse { $quotaSnapshot = $this->quota->snapshotForMutation($namespace, self::quotaResources()); // Registration precedes Native's attempt/execution/run/task // locks. Do not lock the hosting task before entering Native. @@ -97,15 +109,44 @@ private function admit(Request $request, string $taskId, bool $recover): JsonRes } /** @var PreparedLocalActivityTaskBridge $bridge */ $bridge = app(WorkflowTaskBridge::class); - $reply = $recover - ? $bridge->recoverLocalActivity($taskId, $validated['lease_owner'], (int) $validated['workflow_task_attempt'], (int) $validated['sequence'], $validated['descriptor'], WorkerProtocol::requestVersion($request)) - : $bridge->prepareLocalActivity($taskId, $validated['lease_owner'], (int) $validated['workflow_task_attempt'], (int) $validated['sequence'], $validated['worker_attempt_id'], $validated['descriptor'], WorkerProtocol::requestVersion($request)); + $recoveryAuthority = null; + if ($retry) { + $currentTask = NamespaceWorkflowScope::task($namespace, $taskId); + if (! PreparedLocalRetryAdmission::backendSupported() || ! $bridge instanceof PreparedLocalActivityRetryTaskBridge + || ! $currentTask instanceof WorkflowTask || PreparedLocalActivityPolicy::currentClaim($currentTask) !== $claim + || ! PreparedLocalRetryAdmission::claimSupports($claim)) { + return self::refused($validated['lease_owner'], ['worker_claim_retry_capability'], PreparedLocalRetryAdmission::CAPABILITY); + } + if ($recover) { + $recoveryAuthority = PreparedLocalRetryAdmission::recoveryAuthority($currentTask, (int) $validated['sequence'], $validated['lease_owner'], (int) $validated['workflow_task_attempt']); + if ($recoveryAuthority === null) { + return self::refused($validated['lease_owner'], ['original_prepared_local_retry_claim'], PreparedLocalRetryAdmission::CAPABILITY); + } + } + $reply = $recover + ? $bridge->recoverLocalActivityWithRetryPolicy($taskId, $validated['lease_owner'], (int) $validated['workflow_task_attempt'], (int) $validated['sequence'], $validated['descriptor'], WorkerProtocol::requestVersion($request)) + : $bridge->prepareLocalActivityWithRetryPolicy($taskId, $validated['lease_owner'], (int) $validated['workflow_task_attempt'], (int) $validated['sequence'], $validated['worker_attempt_id'], $validated['descriptor'], WorkerProtocol::requestVersion($request)); + } else { + $reply = $recover + ? $bridge->recoverLocalActivity($taskId, $validated['lease_owner'], (int) $validated['workflow_task_attempt'], (int) $validated['sequence'], $validated['descriptor'], WorkerProtocol::requestVersion($request)) + : $bridge->prepareLocalActivity($taskId, $validated['lease_owner'], (int) $validated['workflow_task_attempt'], (int) $validated['sequence'], $validated['worker_attempt_id'], $validated['descriptor'], WorkerProtocol::requestVersion($request)); + } + if ($retry && $recover && ($reply['recovered'] ?? false)) { + $originalTask = $recoveryAuthority['task']->refresh(); + $originalAttempt = $recoveryAuthority['attempt']->refresh(); + $originalClaim = $recoveryAuthority['claim']; + if (($reply['activity_attempt_id'] ?? null) !== $originalAttempt->id + || PreparedLocalRetryAdmission::originalClaim($originalTask, $originalAttempt, $originalClaim['lease_owner'], $originalClaim['attempt']) !== $originalClaim) { + throw new PreparedLocalActivityAdmissionRefused; + } + } if (! $recover && ($reply['prepared'] ?? false)) { $task = NamespaceWorkflowScope::taskQuery($namespace)->lockForUpdate()->find($taskId); $attemptId = $reply['activity_attempt_id'] ?? null; $attempt = is_string($attemptId) ? ActivityAttempt::query()->find($attemptId) : null; if (! $task instanceof WorkflowTask || ! $attempt instanceof ActivityAttempt - || ! PreparedLocalActivityPolicy::remember($task, $attempt, $claim)) { + || ! ($retry ? PreparedLocalRetryAdmission::remember($task, $attempt, $claim) + : PreparedLocalActivityPolicy::remember($task, $attempt, $claim))) { throw new PreparedLocalActivityAdmissionRefused; } } @@ -196,15 +237,35 @@ private function attemptOperation(Request $request, string $taskId, string $atte 'request_id' => [$operation === 'acknowledge' ? 'required' : 'nullable', 'string', 'max:255'], 'progress' => ['nullable', 'array'], ]); + $retryIntent = $operation === 'outcome' && PreparedLocalRetryAdmission::reportIntent($validated['report']); + if ($retryIntent) { + if (! PreparedLocalRetryAdmission::backendSupported()) { + return self::refused($validated['lease_owner'], ['installed_runtime_prepared_local_retry'], PreparedLocalRetryAdmission::CAPABILITY); + } + if ($response = $this->retryReportRefusal($attemptId, $validated['report'])) { + return $response; + } + } $namespace = (string) $request->attributes->get('namespace'); $task = NamespaceWorkflowScope::task($namespace, $taskId); $attempt = ActivityAttempt::query()->where('workflow_task_id', $taskId)->find($attemptId); if ($task === null || $attempt === null || $attempt->workflow_run_id !== $task->workflow_run_id) { return WorkerProtocol::json(['reason' => 'activity_attempt_not_found'], 404); } - $claim = PreparedLocalActivityPolicy::originalClaim($task, $attempt, $validated['lease_owner'], (int) $validated['workflow_task_attempt']); + $retry = PreparedLocalRetryAdmission::retainedIntent($task, $attempt); + if ($retryIntent && ! $retry) { + return self::refused($validated['lease_owner'], ['original_prepared_local_retry_claim'], PreparedLocalRetryAdmission::CAPABILITY); + } + $claim = $retry + ? PreparedLocalRetryAdmission::originalClaim($task, $attempt, $validated['lease_owner'], (int) $validated['workflow_task_attempt']) + : PreparedLocalActivityPolicy::originalClaim($task, $attempt, $validated['lease_owner'], (int) $validated['workflow_task_attempt']); if ($claim === null) { - return self::refused($validated['lease_owner'], ['original_prepared_local_claim']); + return self::refused($validated['lease_owner'], [$retry ? 'original_prepared_local_retry_claim' : 'original_prepared_local_claim'], + $retry ? PreparedLocalRetryAdmission::CAPABILITY : PreparedLocalActivityPolicy::CAPABILITY); + } + if ($retry && $operation === 'outcome' + && ($response = $this->retryReportRefusal($attemptId, $validated['report']))) { + return $response; } try { $bridge = app(WorkflowTaskBridge::class); @@ -218,10 +279,9 @@ private function attemptOperation(Request $request, string $taskId, string $atte // This stop-only observation grants no renewal, heartbeat // or publication authority. Keep those mutations behind // the registration and quota locks below. - $control = $this->mutations->run(fn (): array => $bridge->controlLocalActivity( - $attemptId, $validated['lease_owner'], (int) $validated['workflow_task_attempt'], - false, WorkerProtocol::requestVersion($request) - )); + $control = $this->mutations->run(fn (): array => $retry + ? $bridge->controlLocalActivityWithRetryPolicy($attemptId, $validated['lease_owner'], (int) $validated['workflow_task_attempt'], false, WorkerProtocol::requestVersion($request)) + : $bridge->controlLocalActivity($attemptId, $validated['lease_owner'], (int) $validated['workflow_task_attempt'], false, WorkerProtocol::requestVersion($request))); if ($operation !== 'outcome' && isset($control['cancellation_scope'])) { return $this->reply($request, $control); } @@ -230,13 +290,25 @@ private function attemptOperation(Request $request, string $taskId, string $atte if ($operation === 'outcome' && ($validated['report']['outcome'] ?? null) === 'completed') { $validated['report'] = $this->resolvePayload($validated['report'], 'result', 'report', $namespace); } - $reply = $this->mutations->run(fn (): array => DB::transaction(function () use ($request, $validated, $claim, $attemptId, $operation): array { + $reply = $this->mutations->run(fn (): array|JsonResponse => DB::transaction(function () use ($request, $validated, $claim, $taskId, $attemptId, $operation, $retry): array|JsonResponse { $quotaSnapshot = $this->quota->snapshotForMutation($claim['namespace'], self::quotaResources()); /** @var PreparedLocalActivityTaskBridge $bridge */ $bridge = app(WorkflowTaskBridge::class); $version = WorkerProtocol::requestVersion($request); $epoch = (int) $validated['workflow_task_attempt']; $owner = $validated['lease_owner']; + if ($retry) { + $task = NamespaceWorkflowScope::task($claim['namespace'], $taskId); + $attempt = ActivityAttempt::query()->where('workflow_task_id', $taskId)->find($attemptId); + if (! $bridge instanceof PreparedLocalActivityRetryTaskBridge || ! $task instanceof WorkflowTask || ! $attempt instanceof ActivityAttempt + || PreparedLocalRetryAdmission::originalClaim($task, $attempt, $owner, $epoch) !== $claim) { + return self::refused($owner, ['original_prepared_local_retry_claim'], PreparedLocalRetryAdmission::CAPABILITY); + } + if ($operation === 'outcome' + && ($response = $this->retryReportRefusal($attemptId, $validated['report']))) { + return $response; + } + } if ($operation === 'acknowledge') { // A dead, drained or replaced registration may still have // a surviving supervisor reporting its actual joined stop. @@ -258,19 +330,29 @@ private function attemptOperation(Request $request, string $taskId, string $atte // A stale registration can observe and fence its old // cancelled callback. It cannot publish shielded // cleanup merely because cancellation was accepted. - $control = $bridge->controlLocalActivity($attemptId, $owner, $epoch, false, $version); + $control = $retry + ? $bridge->controlLocalActivityWithRetryPolicy($attemptId, $owner, $epoch, false, $version) + : $bridge->controlLocalActivity($attemptId, $owner, $epoch, false, $version); if (! in_array($control['reason'] ?? null, ['cancellation_requested', 'cancellation_deadline_expired'], true)) { return ['recorded' => false, 'reason' => 'stale_worker_registration']; } } - $reply = $bridge->recordLocalActivityOutcome($attemptId, $owner, $epoch, $validated['report'], $version); + $reply = $retry + ? $bridge->recordLocalActivityRetryOutcome($attemptId, $owner, $epoch, $validated['report'], $version) + : $bridge->recordLocalActivityOutcome($attemptId, $owner, $epoch, $validated['report'], $version); $this->quota->assertNoIncreasePastLimit($quotaSnapshot); return $reply; } - $reply = $operation === 'heartbeat' && $current - ? $bridge->heartbeatLocalActivity($attemptId, $owner, $epoch, $validated['progress'] ?? [], $version) - : $bridge->controlLocalActivity($attemptId, $owner, $epoch, $current && ($validated['renew_lease'] ?? false), $version); + if ($retry) { + $reply = $operation === 'heartbeat' && $current + ? $bridge->heartbeatLocalActivityWithRetryPolicy($attemptId, $owner, $epoch, $validated['progress'] ?? [], $version) + : $bridge->controlLocalActivityWithRetryPolicy($attemptId, $owner, $epoch, $current && ($validated['renew_lease'] ?? false), $version); + } else { + $reply = $operation === 'heartbeat' && $current + ? $bridge->heartbeatLocalActivity($attemptId, $owner, $epoch, $validated['progress'] ?? [], $version) + : $bridge->controlLocalActivity($attemptId, $owner, $epoch, $current && ($validated['renew_lease'] ?? false), $version); + } if (! $current && ($reply['active'] ?? false)) { $reply = [...$reply, 'active' => false, 'renewed' => false, 'heartbeat_recorded' => false, 'stop_required' => true, 'reason' => 'stale_worker_registration']; @@ -293,6 +375,22 @@ private function attemptOperation(Request $request, string $taskId, string $atte return $this->reply($request, $reply); } + private function retryReportRefusal(string $attemptId, array $report): ?JsonResponse + { + $reason = PreparedLocalRetryAdmission::reportRefusal($report); + if ($reason === null) { + return null; + } + if ($reason === 'invalid_activity_retry_delay') { + return WorkerProtocol::json(['reason' => $reason, 'message' => $reason], 422); + } + + return WorkerProtocol::json(['recorded' => false, 'duplicate' => false, + 'reason' => 'invalid_local_activity_outcome', 'activity_attempt_id' => $attemptId, + 'claim_released' => false, 'created_task_ids' => [], + 'history_refresh_page_token' => WorkflowHistoryPageToken::encode(0)], 409); + } + /** @return array> */ private static function claimRules(): array { diff --git a/app/Support/PreparedLocalRetryAdmission.php b/app/Support/PreparedLocalRetryAdmission.php new file mode 100644 index 000000000..499f8a37a --- /dev/null +++ b/app/Support/PreparedLocalRetryAdmission.php @@ -0,0 +1,259 @@ +payload[self::CLAIMS_KEY][$attempt->id]['schema'] ?? null) === self::SCHEMA; + } + + public static function claimSupports(array $claim): bool + { + return is_array($claim['capabilities'] ?? null) + && is_string($claim['protocol_version'] ?? null) + && in_array(self::CAPABILITY, $claim['capabilities'], true) + && in_array(PreparedLocalActivityPolicy::CAPABILITY, $claim['capabilities'], true) + && CooperativeCancellationPolicy::supports($claim['capabilities'], $claim['protocol_version'] ?? null); + } + + /** Pure policy checks precede any argument resolution or preparation. */ + public static function validatePolicy(array $descriptor): void + { + if (! class_exists(PreparedLocalRetryPolicy::class)) { + throw new LogicException('The installed prepared retry policy role is unavailable.'); + } + try { + if (! is_array($descriptor['retry_policy'] ?? null)) { + throw new InvalidArgumentException; + } + PreparedLocalRetryPolicy::normalize($descriptor['retry_policy'], array_intersect_key( + $descriptor, array_flip(['start_to_close_timeout', 'schedule_to_close_timeout', 'heartbeat_timeout']), + )); + } catch (InvalidArgumentException) { + throw ValidationException::withMessages([ + 'descriptor.retry_policy' => ['Expected a valid version-2 prepared retry policy.'], + ]); + } + } + + /** No lookup, payload resolution, callback or clock sample occurs here. */ + public static function reportRefusal(array $report): ?string + { + if (! class_exists(PreparedLocalActivityReport::class)) { + throw new LogicException('The installed prepared retry report role is unavailable.'); + } + + return PreparedLocalActivityReport::refusal($report); + } + + /** Persist only after the actual optional bridge prepared under the outer transaction. */ + public static function remember(WorkflowTask $task, ActivityAttempt $attempt, array $claim): bool + { + if (DB::transactionLevel() < 1 || ! self::backendSupported() + || PreparedLocalActivityPolicy::currentClaim($task) !== $claim || ! self::claimSupports($claim)) { + return false; + } + $graph = self::graph($task, $attempt, $claim['lease_owner'], $claim['attempt']); + if ($graph === null) { + return false; + } + $payload = is_array($task->payload) ? $task->payload : []; + $receipts = $payload[self::CLAIMS_KEY] ?? []; + if (! is_array($receipts)) { + return false; + } + if (array_key_exists($attempt->id, $receipts)) { + // A v1 receipt, or a changed v2 graph, is never rewritten. + return self::originalClaim($task, $attempt, $claim['lease_owner'], $claim['attempt']) === $claim; + } + $receipts[$attempt->id] = ['schema' => self::SCHEMA, 'claim' => $claim, + 'activity_execution_id' => $attempt->activity_execution_id, + 'activity_attempt_id' => $attempt->id, 'worker_attempt_id' => $attempt->worker_attempt_id, + 'local_activity_contract_version' => 2, ...$graph]; + $payload[self::CLAIMS_KEY] = $receipts; + $task->forceFill(['payload' => $payload])->save(); + + return true; + } + + /** Original authority survives replacement of the hosting task's current claim. */ + public static function originalClaim(WorkflowTask $task, ActivityAttempt $attempt, string $owner, int $epoch): ?array + { + $receipt = $task->payload[self::CLAIMS_KEY][$attempt->id] ?? null; + $claim = is_array($receipt) ? ($receipt['claim'] ?? null) : null; + if (! self::backendSupported() || ! is_array($receipt) || ! is_array($claim) + || ! self::exactKeys($receipt, self::RECEIPT_KEYS) + || ($receipt['schema'] ?? null) !== self::SCHEMA + || ($receipt['local_activity_contract_version'] ?? null) !== 2 + || ($receipt['activity_execution_id'] ?? null) !== $attempt->activity_execution_id + || ($receipt['activity_attempt_id'] ?? null) !== $attempt->id + || ($receipt['worker_attempt_id'] ?? null) !== $attempt->worker_attempt_id + || ($claim['task_id'] ?? null) !== $task->id || ($claim['run_id'] ?? null) !== $task->workflow_run_id + || ($claim['namespace'] ?? null) !== $task->namespace || ($claim['worker_id'] ?? null) !== $owner + || ($claim['lease_owner'] ?? null) !== $owner || ($claim['attempt'] ?? null) !== $epoch + || ! self::claimSupports($claim)) { + return null; + } + $graph = self::graph($task, $attempt, $owner, $epoch); + if ($graph === null || ($receipt['preparation_history_event_id'] ?? null) !== $graph['preparation_history_event_id'] + || ($receipt['descriptor_fingerprint'] ?? null) !== $graph['descriptor_fingerprint']) { + return null; + } + try { + if (! is_array($receipt['retry_policy_snapshot']) + || PreparedLocalRetryPolicy::validateSnapshot($receipt['retry_policy_snapshot']) !== $graph['retry_policy_snapshot']) { + return null; + } + } catch (InvalidArgumentException) { + return null; + } + + return $claim; + } + + /** Original callback authority for a fresh recovery or its retained duplicate. */ + public static function recoveryAuthority(WorkflowTask $task, int $sequence, string $owner, int $epoch): ?array + { + if (! self::backendSupported()) { + return null; + } + $execution = ActivityExecution::query()->where('workflow_run_id', $task->workflow_run_id) + ->where('sequence', $sequence)->first(); + if (! $execution instanceof ActivityExecution || ! is_string($execution->current_attempt_id)) { + return null; + } + // A following attempt may already exist. A duplicate recovery still + // binds the attempt retained by this exact recovery identity. + $receipts = WorkflowHistoryEvent::query()->where('workflow_run_id', $task->workflow_run_id) + ->whereIn('event_type', [HistoryEventType::ActivityRetryScheduled, HistoryEventType::ActivityFailed, + HistoryEventType::ActivityTimedOut]) + ->where('payload->sequence', $sequence) + ->where('payload->local_recovery->workflow_task_id', $task->id) + ->where('payload->local_recovery->workflow_task_attempt', $epoch) + ->where('payload->local_recovery->lease_owner', $owner)->limit(2)->get(); + if ($receipts->count() > 1) { + return null; + } + $attemptId = $receipts->isEmpty() ? $execution->current_attempt_id : ($receipts->first()->payload['activity_attempt_id'] ?? null); + if (! is_string($attemptId)) { + return null; + } + $attempt = ActivityAttempt::query()->where('workflow_run_id', $task->workflow_run_id) + ->where('activity_execution_id', $execution->id)->find($attemptId); + $originalTask = $attempt instanceof ActivityAttempt + ? WorkflowTask::query()->where('namespace', $task->namespace)->where('workflow_run_id', $task->workflow_run_id)->find($attempt->workflow_task_id) + : null; + $originalEpoch = $originalTask?->payload[self::CLAIMS_KEY][$attemptId]['claim']['attempt'] ?? null; + if (! $attempt instanceof ActivityAttempt || ! $originalTask instanceof WorkflowTask || ! is_int($originalEpoch) + || ! is_string($attempt->lease_owner) || $attempt->lease_owner === '') { + return null; + } + $claim = self::originalClaim($originalTask, $attempt, $attempt->lease_owner, $originalEpoch); + + return $claim === null ? null : ['task' => $originalTask, 'attempt' => $attempt, 'claim' => $claim]; + } + + /** Read canonical Core authority. The engine rechecks it under its locks. */ + private static function graph(WorkflowTask $task, ActivityAttempt $attempt, string $owner, int $epoch): ?array + { + if ($epoch < 1 || $attempt->workflow_task_id !== $task->id || $attempt->workflow_run_id !== $task->workflow_run_id + || $attempt->lease_owner !== $owner || ! is_string($attempt->worker_attempt_id) || $attempt->worker_attempt_id === '') { + return null; + } + $run = WorkflowRun::query()->where('namespace', $task->namespace)->find($task->workflow_run_id); + $execution = ActivityExecution::query()->where('workflow_run_id', $task->workflow_run_id)->find($attempt->activity_execution_id); + if (! $run instanceof WorkflowRun || ! $execution instanceof ActivityExecution + || ($execution->activity_options['local_activity_contract_version'] ?? null) !== 2) { + return null; + } + // Do not let a first-row selection hide conflicting start authority. + $starts = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id) + ->where('event_type', HistoryEventType::ActivityStarted) + ->where('payload->activity_attempt_id', $attempt->id)->limit(2)->get(); + if ($starts->count() !== 1) { + return null; + } + $started = PortableLocalActivityPreparation::originalStart($run, $execution, $attempt, $owner, $epoch, 2); + if (! $started instanceof WorkflowHistoryEvent || $started->id !== $starts->first()->id + || ($started->payload['local_preparation']['version'] ?? null) !== 2 + || ($started->payload['local_preparation']['local_activity_contract_version'] ?? null) !== 2 + || ! is_string($started->payload['local_preparation']['descriptor_fingerprint'] ?? null) + || preg_match('/^[0-9a-f]{64}$/D', $started->payload['local_preparation']['descriptor_fingerprint']) !== 1 + || ! is_array($started->payload['activity']['retry_policy'] ?? null) + || ! is_array($execution->retry_policy)) { + return null; + } + try { + $snapshot = PreparedLocalRetryPolicy::validateSnapshot($execution->retry_policy); + if (PreparedLocalRetryPolicy::validateSnapshot($started->payload['activity']['retry_policy']) !== $snapshot) { + return null; + } + } catch (InvalidArgumentException) { + return null; + } + + return ['retry_policy_snapshot' => $snapshot, 'preparation_history_event_id' => $started->id, + 'descriptor_fingerprint' => $started->payload['local_preparation']['descriptor_fingerprint']]; + } + + private static function exactKeys(array $value, array $expected): bool + { + $keys = array_keys($value); + sort($keys); + sort($expected); + + return $keys === $expected; + } +} diff --git a/app/Support/RuntimePayloadCompletionLease.php b/app/Support/RuntimePayloadCompletionLease.php index c821fd5b4..2b8037960 100644 --- a/app/Support/RuntimePayloadCompletionLease.php +++ b/app/Support/RuntimePayloadCompletionLease.php @@ -7,6 +7,7 @@ use Workflow\V2\Contracts\ActivityTaskBridge; use Workflow\V2\Contracts\CancellationScopeTaskBridge; use Workflow\V2\Contracts\PreparedLocalActivityTaskBridge; +use Workflow\V2\Contracts\PreparedLocalActivityRetryTaskBridge; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\TaskType; use Workflow\V2\Models\ActivityAttempt; @@ -139,14 +140,21 @@ private function workflow(string $namespace, RuntimePayloadCompletionContext $co } if ($context->operation === 'local_activity_outcome') { $attempt = ActivityAttempt::query()->where('workflow_task_id', $task->id)->find($context->activityAttemptId); - if (! $attempt instanceof ActivityAttempt - || PreparedLocalActivityPolicy::originalClaim($task, $attempt, $context->leaseOwner, $context->attempt) !== $claim) { + $retry = $attempt instanceof ActivityAttempt && PreparedLocalRetryAdmission::retainedIntent($task, $attempt); + $originalClaim = $attempt instanceof ActivityAttempt + ? ($retry ? PreparedLocalRetryAdmission::originalClaim($task, $attempt, $context->leaseOwner, $context->attempt) + : PreparedLocalActivityPolicy::originalClaim($task, $attempt, $context->leaseOwner, $context->attempt)) + : null; + if (! $attempt instanceof ActivityAttempt || $originalClaim !== $claim) { throw self::rejected(); } - /** @var PreparedLocalActivityTaskBridge $bridge */ + /** @var PreparedLocalActivityTaskBridge|PreparedLocalActivityRetryTaskBridge $bridge */ $bridge = app(WorkflowTaskBridge::class); - $control = $bridge->controlLocalActivity($attempt->id, $context->leaseOwner, $context->attempt, false, - PreparedLocalActivityPolicy::MINIMUM_PROTOCOL_VERSION); + $control = $retry + ? $bridge->controlLocalActivityWithRetryPolicy($attempt->id, $context->leaseOwner, $context->attempt, false, + PreparedLocalActivityPolicy::MINIMUM_PROTOCOL_VERSION) + : $bridge->controlLocalActivity($attempt->id, $context->leaseOwner, $context->attempt, false, + PreparedLocalActivityPolicy::MINIMUM_PROTOCOL_VERSION); if (($control['active'] ?? null) !== true || ($control['activity_attempt_id'] ?? null) !== $attempt->id || ($control['workflow_task_id'] ?? null) !== $task->id diff --git a/docs/contracts/prepared-local-retry-v2.md b/docs/contracts/prepared-local-retry-v2.md index 89e17fe13..431a93ef2 100644 --- a/docs/contracts/prepared-local-retry-v2.md +++ b/docs/contracts/prepared-local-retry-v2.md @@ -157,3 +157,40 @@ canonical original mode/snapshot to the retained claim under the existing outer transaction. The current bound bridge does not expose that role. Native implementation, schema, installed/backend qualification and release remain separate work. + +## Guarded Server adapter + +The source adapter targets Core's optional interface and pure helpers at +`ce24207ef86856e64f97b939c9aeef9d1fb387c8`. The locked package and default +bridge remain unchanged. No retry capability is advertised by this adapter. +Policy-member or override presence selects only untrusted intent. A real bound +`PreparedLocalActivityRetryTaskBridge`, the original issued capability and +canonical contract/snapshot authority are required independently. + +A successful version-2 preparation receives a distinct private +`durable-workflow.server.prepared-local-authority/v2` receipt. Its exact keys +are `schema`, `claim`, `activity_execution_id`, `activity_attempt_id`, +`worker_attempt_id`, `local_activity_contract_version`, `retry_policy_snapshot`, +`preparation_history_event_id` and `descriptor_fingerprint`. The contract value +comes from canonical Core preparation. The complete normalized snapshot and +unique correlated Started event must agree on every read. A legacy receipt is +never rewritten or promoted. The hosting task's later current claim cannot +replace the original callback's issued claim. + +Recovery independently reads the original callback's retained authority before +entering Core. A duplicate binds the old attempt named by its exact recovery +identity even after a following attempt exists. Fresh recovery still needs the +recovering claim and existing registration fence. Post-write readback checks the +original graph within the same outer transaction, so a changed graph rolls the +operation back. Physical callback stop remains a separate authority. + +Completed external payload uploads retain their existing completion-lease, +namespace, storage and reference checks. The original receipt and supervisor +control dispatch use the same proven mode as the outcome route. This adds no +fetch-free completed replay guarantee. Failed and timed-out pure report checks +perform no payload resolution. + +Normal locked-package tests cover unavailable-role precedence and unchanged +legacy authority. The exact candidate writer/transaction/transport comparison +and all semantic cases remain required qualification work. PHP syntax or pure +intent controls alone do not qualify a prepared writer or enable a runtime role. diff --git a/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml b/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml index 022f00c17..5972f69d2 100644 --- a/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml +++ b/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: durable-workflow.v2.worker-protocol-api - version: "56" + version: "57" summary: Durable Workflow worker-plane HTTP+JSON API description: > Normative OpenAPI specification for the worker-plane HTTP+JSON API: @@ -2368,6 +2368,22 @@ components: type: object additionalProperties: false required: [type, activity_type, arguments] + allOf: + - if: + required: [retry_policy] + properties: + retry_policy: + type: object + anyOf: + - required: [contract_version] + - required: [snapshot_version] + - required: [mode] + - required: [maximum_interval_seconds] + then: + properties: + start_to_close_timeout: { type: [integer, "null"], minimum: 1, maximum: 2147483647 } + schedule_to_close_timeout: { type: [integer, "null"], minimum: 1, maximum: 2147483647 } + heartbeat_timeout: { type: [integer, "null"], minimum: 1, maximum: 2147483647 } properties: type: { type: string, const: record_local_activity } activity_type: { type: string, minLength: 1 } @@ -2405,10 +2421,38 @@ components: items: { type: object } retry_policy: type: [object, "null"] + description: Versioned members select only untrusted contract-2 intent. Admission requires the actual optional retry bridge and the original activity_retry_delay_override_v1 claim capability. Legacy policies retain their existing path. + allOf: + - if: + type: object + anyOf: + - required: [contract_version] + - required: [snapshot_version] + - required: [mode] + - required: [maximum_interval_seconds] + then: { $ref: "#/components/schemas/PreparedLocalRetryPolicyV2" } properties: max_attempts: { type: [integer, "null"], minimum: 1 } backoff_seconds: { type: array, items: { type: integer, minimum: 0 } } non_retryable_error_types: { type: array, items: { type: string } } + PreparedLocalRetryPolicyV2: + type: object + additionalProperties: false + required: [contract_version, mode, backoff_seconds] + description: Flattened prepared-per-attempt policy. This schema does not grant mode or callback admission. The complete retained ten-key snapshot must be version 2 and match original Started authority. New-route timeouts are null or integers 1..2147483647; schedule-to-start remains null. Backend timestamp checks remain independent of the portable delay bound. + properties: + contract_version: { type: integer, const: 2 } + mode: { type: string, enum: [explicit, generated] } + max_attempts: { type: integer, minimum: 1, maximum: 1024 } + backoff_seconds: + type: array + maxItems: 1024 + items: { type: integer, minimum: 0, maximum: 9007199254740991 } + maximum_interval_seconds: { type: integer, minimum: 0, maximum: 9007199254740991 } + non_retryable_error_types: + type: array + maxItems: 128 + items: { type: string, minLength: 1, maxLength: 255, x-durable-workflow-max-utf8-bytes: 255 } PreparedLocalRecoverRequest: allOf: - $ref: "#/components/schemas/PreparedLocalClaimRequest" @@ -2538,6 +2582,11 @@ components: message: { type: string } exception_type: { type: string, minLength: 1, maxLength: 255 } non_retryable: { type: boolean, default: false } + next_retry_delay_seconds: + type: integer + minimum: 0 + maximum: 9007199254740991 + description: Failed-only optional supplied delay. Absence differs from zero. Requires the installed retry role plus original claim capability, local contract 2 and retained snapshot 2; never upgrades a legacy report. Strict override failures are 422; other closed report failures are 409 invalid_local_activity_outcome. - type: object additionalProperties: false required: [outcome] diff --git a/tests/Feature/PreparedLocalRetryAdmissionTest.php b/tests/Feature/PreparedLocalRetryAdmissionTest.php new file mode 100644 index 000000000..8c35e5170 --- /dev/null +++ b/tests/Feature/PreparedLocalRetryAdmissionTest.php @@ -0,0 +1,139 @@ + '1.20']); + WorkflowNamespace::query()->create(['name' => 'default', 'description' => 'Test', + 'retention_days' => 30, 'status' => 'active']); + } + + public function test_policy_and_report_presence_are_only_untrusted_intent(): void + { + $this->assertFalse(PreparedLocalRetryAdmission::descriptorIntent(['retry_policy' => ['max_attempts' => 2, 'backoff_seconds' => [0]]])); + foreach (['contract_version', 'snapshot_version', 'mode', 'maximum_interval_seconds'] as $field) { + foreach ([null, false, '2', 2.0, 2] as $value) { + $this->assertTrue(PreparedLocalRetryAdmission::descriptorIntent(['retry_policy' => [$field => $value]])); + } + } + $this->assertFalse(PreparedLocalRetryAdmission::reportIntent(['outcome' => 'failed', 'message' => 'failed'])); + foreach ([null, false, '0', 0.0, -1, 0] as $value) { + $this->assertTrue(PreparedLocalRetryAdmission::reportIntent(['next_retry_delay_seconds' => $value])); + } + $this->assertFalse(PreparedLocalRetryAdmission::backendSupported()); + } + + public function test_original_capability_conjunction_keeps_literal_types(): void + { + $claim = ['protocol_version' => '1.20', 'capabilities' => [CooperativeCancellationPolicy::CAPABILITY, + PreparedLocalActivityPolicy::CAPABILITY, PreparedLocalRetryAdmission::CAPABILITY]]; + $this->assertTrue(PreparedLocalRetryAdmission::claimSupports($claim)); + foreach ($claim['capabilities'] as $capability) { + $this->assertFalse(PreparedLocalRetryAdmission::claimSupports([...$claim, + 'capabilities' => array_values(array_diff($claim['capabilities'], [$capability]))])); + } + foreach ([null, false, true, 1.20, 120, '1.19', '2.0', 'invalid'] as $version) { + $this->assertFalse(PreparedLocalRetryAdmission::claimSupports([...$claim, 'protocol_version' => $version])); + } + foreach ([null, true, PreparedLocalRetryAdmission::CAPABILITY] as $capabilities) { + $this->assertFalse(PreparedLocalRetryAdmission::claimSupports([...$claim, 'capabilities' => $capabilities])); + } + } + + public function test_versioned_wire_schema_preserves_legacy_and_failed_only_override(): void + { + $schema = OpenApiSchema::fromFile(base_path('resources/platform-protocol-specs/worker-protocol-api.openapi.yaml')); + foreach (['explicit', 'generated'] as $mode) { + $schema->assertReferenceMatches('#/components/schemas/PreparedLocalRetryPolicyV2', (object) [ + 'contract_version' => 2, 'mode' => $mode, 'backoff_seconds' => array_fill(0, 1024, 0), + ]); + } + foreach ([[], ['next_retry_delay_seconds' => 0], ['next_retry_delay_seconds' => 9007199254740991]] as $extra) { + $schema->assertReferenceMatches('#/components/schemas/PreparedLocalOutcomeRequest', (object) [ + 'lease_owner' => 'worker', 'workflow_task_attempt' => 1, + 'report' => (object) ['outcome' => 'failed', 'message' => 'failed', ...$extra], + ]); + } + foreach ([null, true, '0', 0.0, -1, 9007199254740992] as $value) { + try { + $schema->assertReferenceMatches('#/components/schemas/PreparedLocalOutcomeRequest', (object) [ + 'lease_owner' => 'worker', 'workflow_task_attempt' => 1, + 'report' => (object) ['outcome' => 'failed', 'message' => 'failed', 'next_retry_delay_seconds' => $value], + ]); + } catch (AssertionFailedError) { + continue; + } + $this->fail('A nonliteral or out-of-range override matched the wire schema.'); + } + foreach (['completed', 'timed_out'] as $outcome) { + try { + $schema->assertReferenceMatches('#/components/schemas/PreparedLocalOutcomeRequest', (object) [ + 'lease_owner' => 'worker', 'workflow_task_attempt' => 1, + 'report' => (object) ['outcome' => $outcome, ...($outcome === 'completed' ? ['result' => 'encoded'] : []), 'next_retry_delay_seconds' => 0], + ]); + } catch (AssertionFailedError) { + continue; + } + $this->fail('A nonfailed report accepted an override.'); + } + } + + public function test_a_base_prepared_bridge_does_not_acquire_the_optional_role(): void + { + $this->app->instance(WorkflowTaskBridge::class, \Mockery::mock(PreparedLocalActivityTaskBridge::class)); + $this->assertFalse(PreparedLocalRetryAdmission::backendSupported()); + } + + public function test_default_unavailable_role_precedes_override_and_durable_lookup(): void + { + DB::enableQueryLog(); + foreach ([null, true, '0', 0.0, -1, 0, 9007199254740992] as $delay) { + $this->withHeaders(['X-Namespace' => 'default', WorkerProtocol::HEADER => '1.20']) + ->postJson('/api/worker/workflow-tasks/missing/local-activities/absent/outcome', [ + 'lease_owner' => 'worker', 'workflow_task_attempt' => 1, + 'report' => ['outcome' => 'failed', 'message' => 'failed', 'next_retry_delay_seconds' => $delay], + ])->assertStatus(409)->assertJsonPath('reason', 'prepared_local_activity_not_supported'); + } + $lookups = array_filter(DB::getQueryLog(), static fn (array $row): bool => + preg_match('/\b(workflow_tasks|activity_attempts|activity_executions|workflow_history_events)\b/i', $row['query']) === 1); + DB::disableQueryLog(); + $this->assertSame([], array_values($lookups)); + } + + public function test_a_v1_receipt_is_preserved_and_cannot_supply_retry_authority(): void + { + $task = new WorkflowTask; + $task->forceFill(['id' => 'task', 'workflow_run_id' => 'run', 'namespace' => 'default', 'payload' => [ + '_server_prepared_local_activity_claims' => ['attempt' => [ + 'schema' => 'durable-workflow.server.prepared-local-authority/v1', 'claim' => ['attempt' => 1], + ]], + ]]); + $attempt = new ActivityAttempt; + $attempt->forceFill(['id' => 'attempt', 'workflow_task_id' => 'task', 'workflow_run_id' => 'run']); + $before = $task->getAttributes(); + $this->assertFalse(PreparedLocalRetryAdmission::retainedIntent($task, $attempt)); + $this->assertNull(PreparedLocalRetryAdmission::originalClaim($task, $attempt, 'worker', 1)); + $this->assertSame($before, $task->getAttributes()); + } +} From ef00c26f8ede161a8f1b034f808137d9b19f18d4 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 06:32:31 +0000 Subject: [PATCH 2/9] fix: retain prepared recovery authority atomically --- .github/workflows/phpunit-feature.yml | 2 +- .../Api/PreparedLocalActivityController.php | 57 ++- .../PreparedLocalActivityAdmissionRefused.php | 10 +- app/Support/PreparedLocalRetryAdmission.php | 169 ++++++++- docs/contracts/prepared-local-retry-v2.md | 26 +- .../PreparedLocalRetryAdmissionTest.php | 10 +- .../PreparedLocalRetryAdmissionSourceTest.php | 338 ++++++++++++++++++ 7 files changed, 593 insertions(+), 19 deletions(-) create mode 100644 tests/Source/PreparedLocalRetryAdmissionSourceTest.php diff --git a/.github/workflows/phpunit-feature.yml b/.github/workflows/phpunit-feature.yml index 78bbb34c0..e3083984e 100644 --- a/.github/workflows/phpunit-feature.yml +++ b/.github/workflows/phpunit-feature.yml @@ -582,7 +582,7 @@ jobs: composer install --no-interaction --no-progress --prefer-dist mv vendor/durable-workflow/workflow /tmp/locked-workflow-package cp -a prepared-workflow-source vendor/durable-workflow/workflow - php vendor/bin/phpunit tests/Feature/PreparedLocalActivityProtocolTest.php tests/Feature/ActivityAttemptStatusTest.php tests/Feature/HistoryRetentionTest.php tests/Unit/WorkerPollFenceTest.php tests/Source/CancellationCascadeDiagnosticsTest.php tests/Source/PreparedLocalCancellationPolicyTest.php tests/Source/CancellationScopeAdmissionTest.php tests/Source/ScopedActivityCancellationReceiptTest.php tests/Source/CancellationScopeDeliveryTest.php tests/Source/NexusTargetCancellationAdmissionTest.php tests/Feature/NexusTargetCancellationPolicyTest.php tests/Feature/WorkflowStreamsTest.php tests/Feature/WorkflowDebugTest.php --fail-on-skipped --log-junit "/evidence/$DW_PREPARED_DATABASE.xml" --colors=never + php vendor/bin/phpunit tests/Feature/PreparedLocalActivityProtocolTest.php tests/Feature/ActivityAttemptStatusTest.php tests/Feature/HistoryRetentionTest.php tests/Unit/WorkerPollFenceTest.php tests/Source/CancellationCascadeDiagnosticsTest.php tests/Source/PreparedLocalCancellationPolicyTest.php tests/Source/PreparedLocalRetryAdmissionSourceTest.php tests/Source/CancellationScopeAdmissionTest.php tests/Source/ScopedActivityCancellationReceiptTest.php tests/Source/CancellationScopeDeliveryTest.php tests/Source/NexusTargetCancellationAdmissionTest.php tests/Feature/NexusTargetCancellationPolicyTest.php tests/Feature/WorkflowStreamsTest.php tests/Feature/WorkflowDebugTest.php --fail-on-skipped --log-junit "/evidence/$DW_PREPARED_DATABASE.xml" --colors=never if [ "$DW_PREPARED_DATABASE" = sqlite ]; then php vendor/bin/phpunit tests/Feature/CooperativeCancellationProtocolTest.php tests/Feature/WorkflowWorkerProtocolTest.php tests/Feature/ActivityWorkerProtocolTest.php tests/Feature/SearchAttributeValueValidationTest.php tests/Feature/NamespaceDurableStateQuotaTest.php tests/Feature/RuntimeExternalPayloadTransportTest.php tests/Feature/RuntimePayloadCompletionUploadsTest.php tests/Unit/RuntimePayloadCompletionContextTest.php tests/Unit/WorkerProtocolOpenApiContractTest.php --log-junit /evidence/affected-regression.xml --colors=never fi diff --git a/app/Http/Controllers/Api/PreparedLocalActivityController.php b/app/Http/Controllers/Api/PreparedLocalActivityController.php index e58c2d1b5..ae91b1109 100644 --- a/app/Http/Controllers/Api/PreparedLocalActivityController.php +++ b/app/Http/Controllers/Api/PreparedLocalActivityController.php @@ -84,9 +84,13 @@ private function admit(Request $request, string $taskId, bool $recover): JsonRes && ($response = self::cancellationPolicyRefusal($request, $task, $validated['descriptor']['cancellation_policy']))) { return $response; } - $claim = PreparedLocalActivityPolicy::currentClaim($task); + $recoveryReplay = $retry && $recover + ? PreparedLocalRetryAdmission::retainedRecovery($task, (int) $validated['sequence'], $validated['lease_owner'], (int) $validated['workflow_task_attempt']) + : null; + $claim = $recoveryReplay['claim'] ?? PreparedLocalActivityPolicy::currentClaim($task); if ($claim === null) { - return self::refused($validated['lease_owner'], ['worker_claim_capability']); + return self::refused($validated['lease_owner'], ['worker_claim_capability'], + $retry ? PreparedLocalRetryAdmission::CAPABILITY : PreparedLocalActivityPolicy::CAPABILITY); } if ($retry && ! PreparedLocalRetryAdmission::claimSupports($claim)) { return self::refused($validated['lease_owner'], ['worker_claim_retry_capability'], PreparedLocalRetryAdmission::CAPABILITY); @@ -96,7 +100,10 @@ private function admit(Request $request, string $taskId, bool $recover): JsonRes return $response; } $validated['descriptor'] = $this->resolvePayload($validated['descriptor'], 'arguments', 'descriptor', $namespace); - $reply = $this->mutations->run(fn (): array|JsonResponse => DB::transaction(function () use ($request, $namespace, $taskId, $validated, $claim, $recover, $retry): array|JsonResponse { + $reply = $this->mutations->run(fn (): array|JsonResponse => DB::transaction(function () use ($request, $namespace, $taskId, $validated, $claim, $recover, $retry, $recoveryReplay): array|JsonResponse { + if ($recoveryReplay !== null) { + return $this->replayRetryRecovery($request, $namespace, $taskId, $validated, $recoveryReplay); + } $quotaSnapshot = $this->quota->snapshotForMutation($namespace, self::quotaResources()); // Registration precedes Native's attempt/execution/run/task // locks. Do not lock the hosting task before entering Native. @@ -114,6 +121,8 @@ private function admit(Request $request, string $taskId, bool $recover): JsonRes $currentTask = NamespaceWorkflowScope::task($namespace, $taskId); if (! PreparedLocalRetryAdmission::backendSupported() || ! $bridge instanceof PreparedLocalActivityRetryTaskBridge || ! $currentTask instanceof WorkflowTask || PreparedLocalActivityPolicy::currentClaim($currentTask) !== $claim + || ($claim['lease_owner'] ?? null) !== $validated['lease_owner'] + || ($claim['attempt'] ?? null) !== (int) $validated['workflow_task_attempt'] || ! PreparedLocalRetryAdmission::claimSupports($claim)) { return self::refused($validated['lease_owner'], ['worker_claim_retry_capability'], PreparedLocalRetryAdmission::CAPABILITY); } @@ -135,8 +144,11 @@ private function admit(Request $request, string $taskId, bool $recover): JsonRes $originalTask = $recoveryAuthority['task']->refresh(); $originalAttempt = $recoveryAuthority['attempt']->refresh(); $originalClaim = $recoveryAuthority['claim']; + $recoveringTask = NamespaceWorkflowScope::taskQuery($namespace)->lockForUpdate()->find($taskId); if (($reply['activity_attempt_id'] ?? null) !== $originalAttempt->id - || PreparedLocalRetryAdmission::originalClaim($originalTask, $originalAttempt, $originalClaim['lease_owner'], $originalClaim['attempt']) !== $originalClaim) { + || PreparedLocalRetryAdmission::originalClaim($originalTask, $originalAttempt, $originalClaim['lease_owner'], $originalClaim['attempt']) !== $originalClaim + || ! $recoveringTask instanceof WorkflowTask + || ! PreparedLocalRetryAdmission::rememberRecovery($recoveringTask, (int) $validated['sequence'], $claim, $reply)) { throw new PreparedLocalActivityAdmissionRefused; } } @@ -155,8 +167,13 @@ private function admit(Request $request, string $taskId, bool $recover): JsonRes return $reply; })); - } catch (PreparedLocalActivityAdmissionRefused) { - return self::refused($validated['lease_owner'], ['original_claim_persistence']); + } catch (PreparedLocalActivityAdmissionRefused $exception) { + if ($exception->reply !== null) { + return $this->reply($request, $exception->reply); + } + + return self::refused($validated['lease_owner'], ['original_claim_persistence'], + $retry ? PreparedLocalRetryAdmission::CAPABILITY : PreparedLocalActivityPolicy::CAPABILITY); } catch (\Throwable $exception) { if ($response = $this->payloadFailure($exception)) { return $response; @@ -171,6 +188,34 @@ private function admit(Request $request, string $taskId, bool $recover): JsonRes return $this->reply($request, $reply); } + /** The containing transaction rolls back every nonmatching bridge result. */ + private function replayRetryRecovery(Request $request, string $namespace, string $taskId, array $validated, array $original): array + { + $task = NamespaceWorkflowScope::task($namespace, $taskId); + $retained = $task instanceof WorkflowTask + ? PreparedLocalRetryAdmission::retainedRecovery($task, (int) $validated['sequence'], $validated['lease_owner'], (int) $validated['workflow_task_attempt']) + : null; + $bridge = app(WorkflowTaskBridge::class); + if ($retained === null || ! $bridge instanceof PreparedLocalActivityRetryTaskBridge + || $retained['event_id'] !== $original['event_id'] || $retained['claim'] !== $original['claim']) { + throw new PreparedLocalActivityAdmissionRefused; + } + $reply = $bridge->recoverLocalActivityWithRetryPolicy($taskId, $validated['lease_owner'], + (int) $validated['workflow_task_attempt'], (int) $validated['sequence'], $validated['descriptor'], WorkerProtocol::requestVersion($request)); + if (! PreparedLocalRetryAdmission::matchesRecoveryReply($retained, $reply, true)) { + // Preserve an existing engine refusal after rolling back. A fresh + // writer or any unexpected publication list gains no replay grant. + $refusal = ($reply['recovered'] ?? null) === false && ($reply['created_task_ids'] ?? null) === [] ? $reply : null; + throw new PreparedLocalActivityAdmissionRefused($refusal); + } + $current = PreparedLocalRetryAdmission::retainedRecovery($task->refresh(), (int) $validated['sequence'], $validated['lease_owner'], (int) $validated['workflow_task_attempt']); + if ($current === null || $current['event_id'] !== $retained['event_id'] || $current['claim'] !== $retained['claim']) { + throw new PreparedLocalActivityAdmissionRefused; + } + + return $reply; + } + private function cancellationScopeRefusal(Request $request, WorkflowTask $task, array $descriptor, int $sequence): ?JsonResponse { if (! array_key_exists('cancellation_scope_id', $descriptor)) { diff --git a/app/Support/PreparedLocalActivityAdmissionRefused.php b/app/Support/PreparedLocalActivityAdmissionRefused.php index a08d75e29..ef8efc1b7 100644 --- a/app/Support/PreparedLocalActivityAdmissionRefused.php +++ b/app/Support/PreparedLocalActivityAdmissionRefused.php @@ -2,5 +2,11 @@ namespace App\Support; -/** Roll back preparation if its original Server authority cannot be retained. */ -final class PreparedLocalActivityAdmissionRefused extends \RuntimeException {} +/** Roll back preparation or replay when original authority cannot be retained. */ +final class PreparedLocalActivityAdmissionRefused extends \RuntimeException +{ + public function __construct(public readonly ?array $reply = null) + { + parent::__construct('Original prepared authority could not be retained.'); + } +} diff --git a/app/Support/PreparedLocalRetryAdmission.php b/app/Support/PreparedLocalRetryAdmission.php index 499f8a37a..30db569c2 100644 --- a/app/Support/PreparedLocalRetryAdmission.php +++ b/app/Support/PreparedLocalRetryAdmission.php @@ -25,8 +25,16 @@ final class PreparedLocalRetryAdmission public const SCHEMA = 'durable-workflow.server.prepared-local-authority/v2'; + public const RECOVERY_SCHEMA = 'durable-workflow.server.prepared-local-recovery-authority/v2'; + private const CLAIMS_KEY = '_server_prepared_local_activity_claims'; + private const RECOVERIES_KEY = '_server_prepared_local_activity_recoveries'; + + private const RECOVERY_KEYS = ['schema', 'claim', 'activity_execution_id', 'activity_attempt_id', + 'local_activity_contract_version', 'retry_policy_snapshot', 'preparation_history_event_id', + 'descriptor_fingerprint', 'engine_event']; + private const RECEIPT_KEYS = ['schema', 'claim', 'activity_execution_id', 'activity_attempt_id', 'worker_attempt_id', 'local_activity_contract_version', 'retry_policy_snapshot', 'preparation_history_event_id', 'descriptor_fingerprint']; @@ -202,7 +210,166 @@ public static function recoveryAuthority(WorkflowTask $task, int $sequence, stri } $claim = self::originalClaim($originalTask, $attempt, $attempt->lease_owner, $originalEpoch); - return $claim === null ? null : ['task' => $originalTask, 'attempt' => $attempt, 'claim' => $claim]; + return $claim === null ? null : ['task' => $originalTask, 'attempt' => $attempt, 'claim' => $claim, + 'event' => $receipts->first()]; + } + + /** A retained recovery claim authorizes only Core's exact duplicate. */ + public static function retainedRecovery(WorkflowTask $task, int $sequence, string $owner, int $epoch): ?array + { + $authority = self::recoveryAuthority($task, $sequence, $owner, $epoch); + $event = $authority['event'] ?? null; + if (! $event instanceof WorkflowHistoryEvent) { + return null; + } + $receipt = $task->payload[self::RECOVERIES_KEY][$event->id] ?? null; + $claim = is_array($receipt) ? ($receipt['claim'] ?? null) : null; + if (! is_array($receipt) || ! self::exactKeys($receipt, self::RECOVERY_KEYS) + || ($receipt['schema'] ?? null) !== self::RECOVERY_SCHEMA + || ($receipt['local_activity_contract_version'] ?? null) !== 2 + || ! is_array($claim) || ! self::claimSupports($claim) + || ($claim['task_id'] ?? null) !== $task->id || ($claim['run_id'] ?? null) !== $task->workflow_run_id + || ($claim['namespace'] ?? null) !== $task->namespace + || ($claim['lease_owner'] ?? null) !== $owner || ($claim['worker_id'] ?? null) !== $owner + || ($claim['attempt'] ?? null) !== $epoch) { + return null; + } + $graph = self::graph($authority['task'], $authority['attempt'], $authority['claim']['lease_owner'], $authority['claim']['attempt']); + $engine = self::recoveryEvent($event, $task, $authority['attempt'], $claim); + if ($graph === null || $engine === null + || ($receipt['activity_execution_id'] ?? null) !== $authority['attempt']->activity_execution_id + || ($receipt['activity_attempt_id'] ?? null) !== $authority['attempt']->id + || ($receipt['preparation_history_event_id'] ?? null) !== $graph['preparation_history_event_id'] + || ($receipt['descriptor_fingerprint'] ?? null) !== $graph['descriptor_fingerprint'] + || ! self::sameRecoveryEvent($receipt['engine_event'], $engine)) { + return null; + } + try { + if (! is_array($receipt['retry_policy_snapshot']) + || PreparedLocalRetryPolicy::validateSnapshot($receipt['retry_policy_snapshot']) !== $graph['retry_policy_snapshot']) { + return null; + } + } catch (InvalidArgumentException) { + return null; + } + + return ['task' => $task, 'claim' => $claim, 'event_id' => $event->id, 'authority' => $authority]; + } + + /** No backfill: only the fresh engine recovery and its issued claim may create this receipt. */ + public static function rememberRecovery(WorkflowTask $task, int $sequence, array $claim, array $reply): bool + { + if (DB::transactionLevel() < 1 || ! self::backendSupported() || ! self::claimSupports($claim) + || ($reply['recovered'] ?? null) !== true || ($reply['duplicate'] ?? null) !== false + || ($task->payload['_server_workflow_claim'] ?? null) !== $claim) { + return false; + } + $authority = self::recoveryAuthority($task, $sequence, $claim['lease_owner'], $claim['attempt']); + $event = $authority['event'] ?? null; + if (! $event instanceof WorkflowHistoryEvent || ($reply['event_id'] ?? null) !== $event->id) { + return false; + } + $graph = self::graph($authority['task'], $authority['attempt'], $authority['claim']['lease_owner'], $authority['claim']['attempt']); + $engine = self::recoveryEvent($event, $task, $authority['attempt'], $claim); + $payload = is_array($task->payload) ? $task->payload : []; + $receipts = $payload[self::RECOVERIES_KEY] ?? []; + if ($graph === null || $engine === null || ! is_array($receipts) || array_key_exists($event->id, $receipts)) { + return false; + } + if (! self::matchesRecoveryReply(['task' => $task, 'claim' => $claim, 'event_id' => $event->id, + 'authority' => $authority], $reply, false)) { + return false; + } + $receipts[$event->id] = ['schema' => self::RECOVERY_SCHEMA, 'claim' => $claim, + 'activity_execution_id' => $authority['attempt']->activity_execution_id, + 'activity_attempt_id' => $authority['attempt']->id, 'local_activity_contract_version' => 2, + ...$graph, 'engine_event' => $engine]; + $payload[self::RECOVERIES_KEY] = $receipts; + $task->forceFill(['payload' => $payload])->save(); + + return true; + } + + /** The bridge's carrier must match independently retained event authority. */ + public static function matchesRecoveryReply(array $context, array $reply, bool $duplicate): bool + { + $event = $context['authority']['event']; + $attempt = $context['authority']['attempt']; + $engine = self::recoveryEvent($event, $context['task'], $attempt, $context['claim']); + if ($engine === null || ! self::exactKeys($reply, ['recovered', 'duplicate', 'reason', 'event_id', 'event_type', + 'activity_execution_id', 'activity_attempt_id', 'workflow_task_id', 'recorded_at', 'callback_stop_state', + 'claim_released', 'created_task_ids', 'retry_disposition'])) { + return false; + } + + return $reply['recovered'] === true && $reply['duplicate'] === $duplicate && $reply['reason'] === null + && $reply['event_id'] === $engine['event_id'] && $reply['event_type'] === $engine['event_type'] + && $reply['activity_execution_id'] === $attempt->activity_execution_id && $reply['activity_attempt_id'] === $attempt->id + && $reply['workflow_task_id'] === $context['task']->id && $reply['recorded_at'] === $engine['recorded_at'] + && $reply['callback_stop_state'] === 'unknown' + && $reply['claim_released'] === ($engine['event_type'] === HistoryEventType::ActivityRetryScheduled->value) + && $reply['created_task_ids'] === ($engine['retry_task_id'] === null ? [] : [$engine['retry_task_id']]) + && self::flatObject($reply['retry_disposition']) === $engine['retry_disposition']; + } + + private static function recoveryEvent(WorkflowHistoryEvent $event, WorkflowTask $task, ActivityAttempt $attempt, array $claim): ?array + { + $payload = $event->payload; + $recovery = self::flatObject($payload['local_recovery'] ?? null); + $decision = self::flatObject($payload['retry_disposition'] ?? null); + $retryTask = $payload['retry_task_id'] ?? null; + if ($recovery === null || $decision === null || $event->recorded_at === null + || ($decision['contract_version'] ?? null) !== 2 || ($decision['snapshot_version'] ?? null) !== 2 + || ($event->event_type === HistoryEventType::ActivityRetryScheduled && (! is_string($retryTask) || $retryTask === '')) + || ($event->event_type !== HistoryEventType::ActivityRetryScheduled && $retryTask !== null) + || $event->workflow_run_id !== $task->workflow_run_id || $event->workflow_task_id !== $task->id + || ($payload['activity_execution_id'] ?? null) !== $attempt->activity_execution_id + || ($payload['activity_attempt_id'] ?? null) !== $attempt->id + || ($payload['worker_attempt_id'] ?? null) !== $attempt->worker_attempt_id + || ($payload['task']['id'] ?? null) !== $task->id + || ($payload['task']['lease_owner'] ?? null) !== $claim['lease_owner'] + || ($payload['task']['attempt_count'] ?? null) !== $claim['attempt'] + || ($recovery['version'] ?? null) !== 1 || ($recovery['workflow_task_id'] ?? null) !== $task->id + || ($recovery['workflow_task_attempt'] ?? null) !== $claim['attempt'] + || ($recovery['lease_owner'] ?? null) !== $claim['lease_owner'] + || ($recovery['callback_stop_state'] ?? null) !== 'unknown') { + return null; + } + + return ['event_id' => $event->id, 'event_type' => $event->event_type->value, + 'event_sequence' => $event->sequence, 'recorded_at' => $event->recorded_at->toISOString(), + 'retry_task_id' => $retryTask, 'local_recovery' => $recovery, 'retry_disposition' => $decision]; + } + + private static function sameRecoveryEvent(mixed $recorded, array $current): bool + { + if (! is_array($recorded) || ! self::exactKeys($recorded, array_keys($current))) { + return false; + } + foreach (['event_id', 'event_type', 'event_sequence', 'recorded_at', 'retry_task_id'] as $field) { + if ($recorded[$field] !== $current[$field]) { + return false; + } + } + + return self::flatObject($recorded['local_recovery']) === $current['local_recovery'] + && self::flatObject($recorded['retry_disposition']) === $current['retry_disposition']; + } + + /** The closed engine decision and recovery identity contain only scalar fields. */ + private static function flatObject(mixed $value): ?array + { + if (! is_array($value) || $value === [] || array_is_list($value)) { + return null; + } + foreach ($value as $key => $item) { + if (! is_string($key) || ($item !== null && ! is_string($item) && ! is_int($item) && ! is_bool($item))) { + return null; + } + } + ksort($value); + + return $value; } /** Read canonical Core authority. The engine rechecks it under its locks. */ diff --git a/docs/contracts/prepared-local-retry-v2.md b/docs/contracts/prepared-local-retry-v2.md index 431a93ef2..f265ceaf8 100644 --- a/docs/contracts/prepared-local-retry-v2.md +++ b/docs/contracts/prepared-local-retry-v2.md @@ -177,12 +177,26 @@ unique correlated Started event must agree on every read. A legacy receipt is never rewritten or promoted. The hosting task's later current claim cannot replace the original callback's issued claim. -Recovery independently reads the original callback's retained authority before -entering Core. A duplicate binds the old attempt named by its exact recovery -identity even after a following attempt exists. Fresh recovery still needs the -recovering claim and existing registration fence. Post-write readback checks the -original graph within the same outer transaction, so a changed graph rolls the -operation back. Physical callback stop remains a separate authority. +A fresh recovery also retains a separate closed private +`durable-workflow.server.prepared-local-recovery-authority/v2` receipt on the +recovering task. Its keys are `schema`, `claim`, `activity_execution_id`, +`activity_attempt_id`, `local_activity_contract_version`, `retry_policy_snapshot`, +`preparation_history_event_id`, `descriptor_fingerprint` and `engine_event`. +The copied engine projection contains the event ID/type/sequence/insertion time, +retry-task identity, original recovery envelope and scalar retry decision. +The decision's retained failure sample and history insertion time are distinct +clocks. Their values are preserved independently rather than equated. + +Recovery reads the original callback authority and the recovering claim +independently. An exact duplicate binds the old attempt even after a following +attempt exists. A complete unchanged private recovery receipt permits Core's +matching committed duplicate before fresh registration, quota or clock work. +It grants no fresh writer authority. Missing receipts are never backfilled from +an existing engine event, and changed graph, event or output identities roll +back the transaction. Fresh recovery still requires the current recovering +claim and registration fence. Preparation and fresh recovery retain their +Server authority atomically with the Core writer. Physical callback stop remains +unknown. Completed external payload uploads retain their existing completion-lease, namespace, storage and reference checks. The original receipt and supervisor diff --git a/tests/Feature/PreparedLocalRetryAdmissionTest.php b/tests/Feature/PreparedLocalRetryAdmissionTest.php index 8c35e5170..20f1ad9ba 100644 --- a/tests/Feature/PreparedLocalRetryAdmissionTest.php +++ b/tests/Feature/PreparedLocalRetryAdmissionTest.php @@ -115,10 +115,14 @@ public function test_default_unavailable_role_precedes_override_and_durable_look 'report' => ['outcome' => 'failed', 'message' => 'failed', 'next_retry_delay_seconds' => $delay], ])->assertStatus(409)->assertJsonPath('reason', 'prepared_local_activity_not_supported'); } - $lookups = array_filter(DB::getQueryLog(), static fn (array $row): bool => - preg_match('/\b(workflow_tasks|activity_attempts|activity_executions|workflow_history_events)\b/i', $row['query']) === 1); + // Schema ownership checks precede controller role admission. Count + // actual durable-row access, rather than table names in catalog queries. + $isLookup = static fn (array $row): bool => + preg_match('/\b(?:from|join|update|into)\s+(?:["`]?[A-Za-z0-9_]+["`]?\.)?["`]?(?:workflow_tasks|activity_attempts|activity_executions|workflow_history_events)\b/i', $row['query']) === 1; + $this->assertSame([], array_values(array_filter(DB::getQueryLog(), $isLookup))); + WorkflowTask::query()->whereKey('counter-only')->first(); + $this->assertCount(1, array_filter(DB::getQueryLog(), $isLookup), 'The same counter must detect a real task-row read.'); DB::disableQueryLog(); - $this->assertSame([], array_values($lookups)); } public function test_a_v1_receipt_is_preserved_and_cannot_supply_retry_authority(): void diff --git a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php new file mode 100644 index 000000000..90788f1cd --- /dev/null +++ b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php @@ -0,0 +1,338 @@ +assertTrue(interface_exists(PreparedLocalActivityRetryTaskBridge::class), 'Select the exact maintained Core retry candidate.'); + Queue::fake(); + config(['server.worker_protocol.version' => '1.20', 'workflows.v2.workflow_task_lease_seconds' => 300, + 'workflows.v2.types.workflows' => ['tests.external-greeting-workflow' => ExternalGreetingWorkflow::class]]); + $this->travelTo(\Illuminate\Support\Carbon::parse('2026-11-06T00:00:00Z')); + WorkflowNamespace::query()->create(['name' => 'default', 'description' => 'Test', 'retention_days' => 30, 'status' => 'active']); + $this->bindRetryRole(); + $this->assertTrue(PreparedLocalRetryAdmission::backendSupported()); + } + + protected function tearDown(): void + { + $this->travelBack(); + parent::tearDown(); + } + + private function bindRetryRole(array $overrides = []): void + { + $base = app(WorkflowTaskBridge::class); + $interfaces = array_values(array_unique([...class_implements($base), PreparedLocalActivityRetryTaskBridge::class])); + $role = \Mockery::mock(implode(', ', $interfaces)); + $versioned = [ + 'prepareLocalActivityWithRetryPolicy' => [PortableLocalActivityPreparation::class, 'prepareWithRetryPolicy'], + 'recordLocalActivityRetryOutcome' => [app(LocalActivityExecutor::class), 'recordPortableRetryOutcome'], + 'controlLocalActivityWithRetryPolicy' => [PortableLocalActivityControl::class, 'pollWithRetryPolicy'], + 'heartbeatLocalActivityWithRetryPolicy' => [PortableLocalActivityControl::class, 'heartbeatWithRetryPolicy'], + 'recoverLocalActivityWithRetryPolicy' => [PortableLocalActivityPreparation::class, 'recoverWithRetryPolicy'], + ]; + $versioned = array_replace($versioned, $overrides); + $methods = []; + foreach ($interfaces as $interface) { + foreach ((new ReflectionClass($interface))->getMethods() as $method) { + $methods[$method->name] = true; + } + } + foreach (array_keys($methods) as $method) { + $role->shouldReceive($method)->andReturnUsing($versioned[$method] ?? [$base, $method]); + } + $this->app->instance(WorkflowTaskBridge::class, $role); + } + + public function test_real_preparation_retains_a_closed_original_v2_graph(): void + { + $task = $this->claim(); + $prepared = $this->local($task, 'prepare')->assertOk()->assertJsonPath('prepared', true)->json(); + $stored = WorkflowTask::query()->findOrFail($task['task_id']); + $attempt = ActivityAttempt::query()->findOrFail($prepared['activity_attempt_id']); + $receipt = $stored->payload['_server_prepared_local_activity_claims'][$attempt->id]; + $keys = array_keys($receipt); + sort($keys); + $expected = ['schema', 'claim', 'activity_execution_id', 'activity_attempt_id', 'worker_attempt_id', + 'local_activity_contract_version', 'retry_policy_snapshot', 'preparation_history_event_id', 'descriptor_fingerprint']; + sort($expected); + $this->assertSame($expected, $keys); + $this->assertSame(2, $receipt['local_activity_contract_version']); + $this->assertSame(2, $receipt['retry_policy_snapshot']['snapshot_version']); + $this->assertSame($receipt['claim'], PreparedLocalRetryAdmission::originalClaim($stored, $attempt, $task['lease_owner'], $task['workflow_task_attempt'])); + $before = $this->state(); + $this->local($task, 'prepare')->assertOk()->assertJsonPath('duplicate', true)->assertJsonPath('activity_attempt_id', $attempt->id); + $this->assertSame($before, $this->state()); + } + + public function test_supplied_override_and_other_report_errors_refuse_before_rows(): void + { + foreach ([null, true, '0', 0.0, -1, 9007199254740992] as $delay) { + DB::enableQueryLog(); + $this->outcome(['task_id' => 'missing', 'lease_owner' => 'worker', 'workflow_task_attempt' => 1], 'absent', + ['outcome' => 'failed', 'message' => 'failed', 'next_retry_delay_seconds' => $delay]) + ->assertStatus(422)->assertJsonPath('reason', 'invalid_activity_retry_delay'); + $this->assertNoAuthorityRows(); + } + DB::enableQueryLog(); + $this->outcome(['task_id' => 'missing', 'lease_owner' => 'worker', 'workflow_task_attempt' => 1], 'absent', + ['outcome' => 'failed', 'message' => [], 'next_retry_delay_seconds' => 0]) + ->assertStatus(409)->assertJsonPath('reason', 'invalid_local_activity_outcome'); + $this->assertNoAuthorityRows(); + $this->outcome(['task_id' => 'missing', 'lease_owner' => 'worker', 'workflow_task_attempt' => 1], 'absent', + ['outcome' => 'failed', 'message' => []])->assertStatus(404)->assertJsonPath('reason', 'activity_attempt_not_found'); + } + + public function test_outcome_duplicate_preserves_the_committed_decision_after_time_and_registration_change(): void + { + $task = $this->claim(); + $prepared = $this->local($task, 'prepare')->assertOk()->json(); + $this->travel(1)->seconds(); + $report = ['outcome' => 'failed', 'message' => 'again', 'next_retry_delay_seconds' => 2]; + $original = $this->outcome($task, $prepared['activity_attempt_id'], $report)->assertOk() + ->assertJsonPath('recorded', true)->assertJsonPath('duplicate', false)->json(); + $this->assertSame('scheduled', $original['retry_disposition']['decision']); + $before = $this->state(); + WorkerRegistration::query()->where('worker_id', 'retry-worker')->update(['status' => 'superseded']); + $this->travel(20)->seconds(); + $duplicate = $this->outcome($task, $prepared['activity_attempt_id'], $report)->assertOk()->assertJsonPath('duplicate', true)->json(); + $this->assertSame($original['event_id'], $duplicate['event_id']); + $this->assertSame($original['retry_disposition'], $duplicate['retry_disposition']); + $this->assertSame($before, $this->state()); + } + + public function test_recovery_retains_the_recovering_claim_and_replays_after_a_following_attempt(): void + { + config(['workflows.v2.workflow_task_lease_seconds' => 10]); + $descriptor = $this->descriptor(['start_to_close_timeout' => 100, 'schedule_to_close_timeout' => 300, + 'retry_policy' => ['contract_version' => 2, 'mode' => 'explicit', 'max_attempts' => 3, + 'backoff_seconds' => [0], 'maximum_interval_seconds' => 10]]); + $task = $this->claim(); + $this->local($task, 'prepare', $descriptor)->assertOk(); + $this->travel(11)->seconds(); + $recovering = $this->poll()->assertOk()->json('task'); + $original = $this->local($recovering, 'recover', $descriptor)->assertOk()->assertJsonPath('recovered', true) + ->assertJsonPath('duplicate', false)->assertJsonPath('callback_stop_state', 'unknown')->json(); + $this->assertSame('engine_interruption', $original['retry_disposition']['failure_source']); + $stored = WorkflowTask::query()->findOrFail($recovering['task_id']); + $receipt = $stored->payload['_server_prepared_local_activity_recoveries'][$original['event_id']]; + $this->assertSame(PreparedLocalRetryAdmission::RECOVERY_SCHEMA, $receipt['schema']); + $this->assertSame($recovering['workflow_task_attempt'], $receipt['claim']['attempt']); + $following = $this->poll()->assertOk()->json('task'); + $this->local($following, 'prepare', $descriptor, 'sdk-next')->assertOk()->assertJsonPath('attempt_number', 2); + WorkerRegistration::query()->where('worker_id', 'retry-worker')->update(['status' => 'superseded']); + $before = $this->state(); + $this->travel(20)->seconds(); + $duplicate = $this->local($recovering, 'recover', $descriptor)->assertOk()->assertJsonPath('duplicate', true)->json(); + $this->assertSame($original['event_id'], $duplicate['event_id']); + $this->assertSame($original['retry_disposition'], $duplicate['retry_disposition']); + $this->assertSame($before, $this->state()); + $stored = $stored->fresh(); + $payload = $stored->payload; + unset($payload['_server_prepared_local_activity_recoveries'][$original['event_id']]); + $stored->forceFill(['payload' => $payload])->save(); + $this->local($recovering, 'recover', $descriptor)->assertStatus(409)->assertJsonPath('reason', 'prepared_local_activity_not_supported'); + $this->assertSame($before, $this->state()); + $this->assertArrayNotHasKey($original['event_id'], $stored->fresh()->payload['_server_prepared_local_activity_recoveries']); + } + + public function test_corrupt_original_contract_or_snapshot_cannot_acquire_outcome_authority(): void + { + $task = $this->claim(); + $prepared = $this->local($task, 'prepare')->assertOk()->json(); + $execution = ActivityExecution::query()->findOrFail($prepared['activity_execution_id']); + $options = $execution->activity_options; + $options['local_activity_contract_version'] = '2'; + $execution->forceFill(['activity_options' => $options])->save(); + $before = $this->state(); + $this->outcome($task, $prepared['activity_attempt_id'], ['outcome' => 'failed', 'message' => 'failed']) + ->assertStatus(409)->assertJsonPath('reason', 'prepared_local_activity_not_supported'); + $this->assertSame($before, $this->state()); + } + + public function test_changed_original_receipts_refuse_without_repair(): void + { + $task = $this->claim(); + $prepared = $this->local($task, 'prepare')->assertOk()->json(); + $stored = WorkflowTask::query()->findOrFail($task['task_id']); + $payload = $stored->payload; + $attemptId = $prepared['activity_attempt_id']; + $receipt = $payload['_server_prepared_local_activity_claims'][$attemptId]; + $mutations = [ + [...$receipt, 'local_activity_contract_version' => '2'], + [...$receipt, 'activity_attempt_id' => 'foreign'], + [...$receipt, 'preparation_history_event_id' => 'foreign'], + [...$receipt, 'descriptor_fingerprint' => str_repeat('0', 64)], + [...$receipt, 'unexpected' => true], + [...$receipt, 'retry_policy_snapshot' => [...$receipt['retry_policy_snapshot'], 'max_attempts' => 4]], + [...$receipt, 'claim' => [...$receipt['claim'], 'capabilities' => [PreparedLocalActivityPolicy::CAPABILITY]]], + [...$receipt, 'claim' => [...$receipt['claim'], 'namespace' => 'foreign']], + ]; + foreach ($mutations as $mutation) { + $stored->forceFill(['payload' => [...$payload, '_server_prepared_local_activity_claims' => [$attemptId => $mutation]]])->save(); + $before = $this->state(); + $this->outcome($task, $attemptId, ['outcome' => 'failed', 'message' => 'failed']) + ->assertStatus(409)->assertJsonPath('reason', 'prepared_local_activity_not_supported'); + $this->assertSame($before, $this->state()); + } + $stored->forceFill(['payload' => $payload])->save(); + $this->outcome($task, $attemptId, ['outcome' => 'failed', 'message' => 'failed'])->assertOk()->assertJsonPath('recorded', true); + } + + public function test_failed_post_write_preparation_binding_rolls_back_the_real_writer(): void + { + $task = $this->claim(); + $this->bindRetryRole(['prepareLocalActivityWithRetryPolicy' => static function (...$arguments): array { + $reply = PortableLocalActivityPreparation::prepareWithRetryPolicy(...$arguments); + $execution = ActivityExecution::query()->findOrFail($reply['activity_execution_id']); + $execution->forceFill(['activity_options' => [...$execution->activity_options, 'local_activity_contract_version' => '2']])->save(); + + return $reply; + }]); + $before = $this->state(); + $this->local($task, 'prepare')->assertStatus(409)->assertJsonPath('reason', 'prepared_local_activity_not_supported'); + $this->assertSame($before, $this->state()); + } + + public function test_failed_post_write_recovery_binding_rolls_back_decision_and_following_task(): void + { + config(['workflows.v2.workflow_task_lease_seconds' => 10]); + $task = $this->claim(); + $this->local($task, 'prepare')->assertOk(); + $this->travel(11)->seconds(); + $recovering = $this->poll()->assertOk()->json('task'); + $this->bindRetryRole(['recoverLocalActivityWithRetryPolicy' => static function (...$arguments): array { + $reply = PortableLocalActivityPreparation::recoverWithRetryPolicy(...$arguments); + + return [...$reply, 'event_id' => 'foreign']; + }]); + $before = $this->state(); + $this->local($recovering, 'recover')->assertStatus(409)->assertJsonPath('reason', 'prepared_local_activity_not_supported'); + $this->assertSame($before, $this->state()); + } + + public function test_total_deadline_preserves_selection_and_engine_timeout_precedence(): void + { + foreach ([99, 100] as $seconds) { + $this->travelTo(\Illuminate\Support\Carbon::parse('2026-11-06T00:00:00Z')); + $task = $this->claim(); + $prepared = $this->local($task, 'prepare', $this->descriptor([ + 'start_to_close_timeout' => 200, 'schedule_to_close_timeout' => 100, + ]))->assertOk()->json(); + $before = ActivityAttempt::query()->count(); + $this->travel($seconds)->seconds(); + $reply = $this->outcome($task, $prepared['activity_attempt_id'], + ['outcome' => 'failed', 'message' => 'caller failure', 'next_retry_delay_seconds' => 1]) + ->assertOk()->assertJsonPath('recorded', true)->assertJsonPath('created_task_ids', [])->json(); + $this->assertSame('terminal', $reply['retry_disposition']['decision']); + $this->assertSame($seconds === 99 ? 'application' : 'engine_timeout', $reply['retry_disposition']['failure_source']); + if ($seconds === 99) { + $this->assertSame('retry_delay_exceeds_schedule_to_close', $reply['retry_disposition']['terminal_classification']); + } else { + $this->assertSame('schedule_to_close', $reply['retry_disposition']['timeout_kind']); + } + $this->assertSame($before, ActivityAttempt::query()->count()); + $this->assertSame(0, WorkflowHistoryEvent::query()->where('workflow_run_id', $task['run_id']) + ->where('event_type', HistoryEventType::ActivityRetryScheduled)->count()); + } + } + + private function assertNoAuthorityRows(): void + { + $queries = array_filter(DB::getQueryLog(), static fn (array $row): bool => + preg_match('/\b(?:from|join|update|into)\s+(?:["`]?[A-Za-z0-9_]+["`]?\.)?["`]?(?:workflow_tasks|activity_attempts|activity_executions|workflow_history_events|workflow_worker_registrations)\b/i', $row['query']) === 1); + DB::disableQueryLog(); + DB::flushQueryLog(); + $this->assertSame([], array_values($queries)); + } + + private function state(): array + { + $rows = []; + foreach (['workflow_history_events', 'activity_attempts', 'activity_executions', 'workflow_tasks'] as $table) { + $rows[$table] = DB::table($table)->orderBy('id')->get()->map(static fn ($row): array => (array) $row)->all(); + } + + return $rows; + } + + private function descriptor(array $overrides = []): array + { + return [...['type' => 'record_local_activity', 'activity_type' => 'opaque-local', + 'arguments' => Serializer::serializeWithCodec('avro', ['Ada']), 'payload_codec' => 'avro', + 'start_to_close_timeout' => 100, 'schedule_to_close_timeout' => 300, + 'retry_policy' => ['contract_version' => 2, 'mode' => 'explicit', 'max_attempts' => 3, + 'backoff_seconds' => [1], 'maximum_interval_seconds' => 10]], ...$overrides]; + } + + private function claim(): array + { + $this->withHeaders(['X-Namespace' => 'default', 'X-Durable-Workflow-Control-Plane-Version' => '2']) + ->postJson('/api/workflows', ['workflow_type' => 'tests.external-greeting-workflow', 'task_queue' => 'retry-v2', 'input' => ['Ada']])->assertCreated(); + $this->withHeaders($this->headers())->postJson('/api/worker/register', ['worker_id' => 'retry-worker', 'task_queue' => 'retry-v2', + 'runtime' => 'php', 'supported_workflow_types' => ['tests.external-greeting-workflow'], + 'capabilities' => [CooperativeCancellationPolicy::CAPABILITY, PreparedLocalActivityPolicy::CAPABILITY, PreparedLocalRetryAdmission::CAPABILITY], + 'capability_manifest' => $this->portableWorkerAffinityRefusalManifest(), 'max_concurrent_workflow_tasks' => 1, + 'process_metrics' => ['host' => 'test-worker', 'process_started_at' => now()->toISOString(), 'process_id' => 10]])->assertCreated(); + + return $this->poll()->assertOk()->json('task'); + } + + private function poll(): TestResponse + { + return $this->withHeaders($this->headers())->postJson('/api/worker/workflow-tasks/poll', + ['worker_id' => 'retry-worker', 'task_queue' => 'retry-v2', 'timeout_seconds' => 0]); + } + + private function local(array $task, string $operation, ?array $descriptor = null, string $workerAttempt = 'sdk-original'): TestResponse + { + return $this->withHeaders($this->headers())->postJson("/api/worker/workflow-tasks/{$task['task_id']}/local-activities/{$operation}", + ['lease_owner' => $task['lease_owner'], 'workflow_task_attempt' => $task['workflow_task_attempt'], 'sequence' => 1, + 'worker_attempt_id' => $workerAttempt, 'descriptor' => $descriptor ?? $this->descriptor()]); + } + + private function outcome(array $task, string $attempt, array $report): TestResponse + { + return $this->withHeaders($this->headers())->call('POST', "/api/worker/workflow-tasks/{$task['task_id']}/local-activities/{$attempt}/outcome", + [], [], [], ['CONTENT_TYPE' => 'application/json'], json_encode([ + 'lease_owner' => $task['lease_owner'], 'workflow_task_attempt' => $task['workflow_task_attempt'], 'report' => $report, + ], JSON_THROW_ON_ERROR | JSON_PRESERVE_ZERO_FRACTION)); + } + + private function headers(): array + { + return ['X-Namespace' => 'default', WorkerProtocol::HEADER => '1.20']; + } +} From 5bb1e23f0c87841077c7892812c602a452caa3a4 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 07:02:04 +0000 Subject: [PATCH 3/9] fix: bind recovery to its literal activity sequence --- app/Support/PreparedLocalRetryAdmission.php | 10 ++++++---- docs/contracts/prepared-local-retry-v2.md | 3 ++- tests/Source/PreparedLocalRetryAdmissionSourceTest.php | 10 +++++++++- 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/app/Support/PreparedLocalRetryAdmission.php b/app/Support/PreparedLocalRetryAdmission.php index 30db569c2..68fed5841 100644 --- a/app/Support/PreparedLocalRetryAdmission.php +++ b/app/Support/PreparedLocalRetryAdmission.php @@ -179,7 +179,7 @@ public static function recoveryAuthority(WorkflowTask $task, int $sequence, stri } $execution = ActivityExecution::query()->where('workflow_run_id', $task->workflow_run_id) ->where('sequence', $sequence)->first(); - if (! $execution instanceof ActivityExecution || ! is_string($execution->current_attempt_id)) { + if (! $execution instanceof ActivityExecution || $execution->sequence !== $sequence || ! is_string($execution->current_attempt_id)) { return null; } // A following attempt may already exist. A duplicate recovery still @@ -191,7 +191,8 @@ public static function recoveryAuthority(WorkflowTask $task, int $sequence, stri ->where('payload->local_recovery->workflow_task_id', $task->id) ->where('payload->local_recovery->workflow_task_attempt', $epoch) ->where('payload->local_recovery->lease_owner', $owner)->limit(2)->get(); - if ($receipts->count() > 1) { + if ($receipts->count() > 1 || (! $receipts->isEmpty() + && ($receipts->first()->payload['sequence'] ?? null) !== $sequence)) { return null; } $attemptId = $receipts->isEmpty() ? $execution->current_attempt_id : ($receipts->first()->payload['activity_attempt_id'] ?? null); @@ -319,6 +320,7 @@ private static function recoveryEvent(WorkflowHistoryEvent $event, WorkflowTask $decision = self::flatObject($payload['retry_disposition'] ?? null); $retryTask = $payload['retry_task_id'] ?? null; if ($recovery === null || $decision === null || $event->recorded_at === null + || ! is_int($payload['sequence'] ?? null) || $payload['sequence'] < 1 || ($decision['contract_version'] ?? null) !== 2 || ($decision['snapshot_version'] ?? null) !== 2 || ($event->event_type === HistoryEventType::ActivityRetryScheduled && (! is_string($retryTask) || $retryTask === '')) || ($event->event_type !== HistoryEventType::ActivityRetryScheduled && $retryTask !== null) @@ -337,7 +339,7 @@ private static function recoveryEvent(WorkflowHistoryEvent $event, WorkflowTask } return ['event_id' => $event->id, 'event_type' => $event->event_type->value, - 'event_sequence' => $event->sequence, 'recorded_at' => $event->recorded_at->toISOString(), + 'event_sequence' => $event->sequence, 'activity_sequence' => $payload['sequence'], 'recorded_at' => $event->recorded_at->toISOString(), 'retry_task_id' => $retryTask, 'local_recovery' => $recovery, 'retry_disposition' => $decision]; } @@ -346,7 +348,7 @@ private static function sameRecoveryEvent(mixed $recorded, array $current): bool if (! is_array($recorded) || ! self::exactKeys($recorded, array_keys($current))) { return false; } - foreach (['event_id', 'event_type', 'event_sequence', 'recorded_at', 'retry_task_id'] as $field) { + foreach (['event_id', 'event_type', 'event_sequence', 'activity_sequence', 'recorded_at', 'retry_task_id'] as $field) { if ($recorded[$field] !== $current[$field]) { return false; } diff --git a/docs/contracts/prepared-local-retry-v2.md b/docs/contracts/prepared-local-retry-v2.md index f265ceaf8..7e938148c 100644 --- a/docs/contracts/prepared-local-retry-v2.md +++ b/docs/contracts/prepared-local-retry-v2.md @@ -182,7 +182,8 @@ A fresh recovery also retains a separate closed private recovering task. Its keys are `schema`, `claim`, `activity_execution_id`, `activity_attempt_id`, `local_activity_contract_version`, `retry_policy_snapshot`, `preparation_history_event_id`, `descriptor_fingerprint` and `engine_event`. -The copied engine projection contains the event ID/type/sequence/insertion time, +The copied engine projection contains the event ID/type/history sequence, +strict integer activity sequence and insertion time, retry-task identity, original recovery envelope and scalar retry decision. The decision's retained failure sample and history insertion time are distinct clocks. Their values are preserved independently rather than equated. diff --git a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php index 90788f1cd..df815b929 100644 --- a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php +++ b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php @@ -159,11 +159,19 @@ public function test_recovery_retains_the_recovering_claim_and_replays_after_a_f $this->assertSame($original['event_id'], $duplicate['event_id']); $this->assertSame($original['retry_disposition'], $duplicate['retry_disposition']); $this->assertSame($before, $this->state()); + $event = WorkflowHistoryEvent::query()->findOrFail($original['event_id']); + $eventPayload = $event->payload; + $event->forceFill(['payload' => [...$eventPayload, 'sequence' => '1']])->save(); + $corrupt = $this->state(); + $this->local($recovering, 'recover', $descriptor)->assertStatus(409); + $this->assertSame($corrupt, $this->state()); + $event->forceFill(['payload' => $eventPayload])->save(); $stored = $stored->fresh(); $payload = $stored->payload; unset($payload['_server_prepared_local_activity_recoveries'][$original['event_id']]); $stored->forceFill(['payload' => $payload])->save(); - $this->local($recovering, 'recover', $descriptor)->assertStatus(409)->assertJsonPath('reason', 'prepared_local_activity_not_supported'); + $before = $this->state(); + $this->local($recovering, 'recover', $descriptor)->assertStatus(409); $this->assertSame($before, $this->state()); $this->assertArrayNotHasKey($original['event_id'], $stored->fresh()->payload['_server_prepared_local_activity_recoveries']); } From aa5b7ba1a78ee9f5af5b0917b07379ea9fb97a41 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 08:00:34 +0000 Subject: [PATCH 4/9] test: retain the real base bridge when replacing retry behavior --- tests/Source/PreparedLocalRetryAdmissionSourceTest.php | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php index df815b929..4e1f06c10 100644 --- a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php +++ b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php @@ -32,6 +32,8 @@ final class PreparedLocalRetryAdmissionSourceTest extends TestCase { use RefreshDatabase; + private WorkflowTaskBridge $baseBridge; + protected function setUp(): void { parent::setUp(); @@ -41,6 +43,8 @@ protected function setUp(): void 'workflows.v2.types.workflows' => ['tests.external-greeting-workflow' => ExternalGreetingWorkflow::class]]); $this->travelTo(\Illuminate\Support\Carbon::parse('2026-11-06T00:00:00Z')); WorkflowNamespace::query()->create(['name' => 'default', 'description' => 'Test', 'retention_days' => 30, 'status' => 'active']); + $this->baseBridge = app(WorkflowTaskBridge::class); + $this->assertNotInstanceOf(\Mockery\MockInterface::class, $this->baseBridge); $this->bindRetryRole(); $this->assertTrue(PreparedLocalRetryAdmission::backendSupported()); } @@ -53,7 +57,7 @@ protected function tearDown(): void private function bindRetryRole(array $overrides = []): void { - $base = app(WorkflowTaskBridge::class); + $base = $this->baseBridge; $interfaces = array_values(array_unique([...class_implements($base), PreparedLocalActivityRetryTaskBridge::class])); $role = \Mockery::mock(implode(', ', $interfaces)); $versioned = [ From 06ee03c6380f7b34e545cf28c704f4d7f79736cc Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 08:40:47 +0000 Subject: [PATCH 5/9] test: correct prepared retry source qualification fixtures --- tests/Feature/ActivityAttemptStatusTest.php | 30 +++++++++- .../Source/CancellationScopeAdmissionTest.php | 4 +- .../PreparedLocalRetryAdmissionSourceTest.php | 56 ++++++++++--------- 3 files changed, 61 insertions(+), 29 deletions(-) diff --git a/tests/Feature/ActivityAttemptStatusTest.php b/tests/Feature/ActivityAttemptStatusTest.php index 74d25b6e5..1d00d3da3 100644 --- a/tests/Feature/ActivityAttemptStatusTest.php +++ b/tests/Feature/ActivityAttemptStatusTest.php @@ -254,7 +254,7 @@ public function test_workflow_deadline_crossing_during_payload_resolution_cannot $this->withHeaders($this->headers())->postJson($this->path($task, 'complete'), $this->fence($task) + ['result' => null]) ->assertConflict()->assertJsonPath('recorded', false)->assertJsonPath('reason', $reason); - $this->assertSame($before, $this->snapshot($task)); + $this->assertRefusedOutcomeSnapshot($before, $task, 'completed', $reason); } #[DataProvider('expiredOutcomeProvider')] @@ -277,7 +277,7 @@ public function test_outcome_crossing_expiry_during_payload_resolution_cannot_re ->postJson($this->path($task, $action), $this->fence($task) + $payload) ->assertStatus(409)->assertJsonPath('recorded', false)->assertJsonPath('reason', 'lease_expired'); - $this->assertSame($before, $this->snapshot($task)); + $this->assertRefusedOutcomeSnapshot($before, $task, $action === 'complete' ? 'completed' : 'failed', 'lease_expired'); } public static function deadlineProvider(): array @@ -665,6 +665,32 @@ private function observe(array $task, array $overrides = [], string $namespace = return $this->withHeaders($this->headers($namespace))->postJson($this->path($task), array_replace($this->fence($task), $overrides)); } + private function assertRefusedOutcomeSnapshot(array $before, array $task, string $outcome, string $reason): void + { + $after = $this->snapshot($task); + // Maintained Core records a bounded rejection audit. The locked older + // package has no recorder and must retain its original history count. + $audited = class_exists(\Workflow\V2\Support\ActivityOutcomeRejection::class); + $this->assertSame($before[4] + ($audited ? 1 : 0), $after[4]); + unset($before[4], $after[4]); + $this->assertSame($before, $after); + if (! $audited) { + return; + } + $event = WorkflowHistoryEvent::query()->where('workflow_run_id', $task['run_id']) + ->where('event_type', 'ActivityOutcomeRejected')->sole(); + $this->assertSame($task['task_id'], $event->workflow_task_id); + $expected = ['activity_execution_id' => $task['activity_execution_id'], + 'activity_attempt_id' => $task['activity_attempt_id'], + 'attempt_number' => ActivityAttempt::query()->findOrFail($task['activity_attempt_id'])->attempt_number, + 'lease_owner' => $task['lease_owner'], 'outcome' => $outcome, 'reason' => $reason, + 'observed_at' => now()->toJSON()]; + $actual = $event->payload; + ksort($expected); + ksort($actual); + $this->assertSame($expected, $actual); + } + private function snapshot(array $task): array { return [ diff --git a/tests/Source/CancellationScopeAdmissionTest.php b/tests/Source/CancellationScopeAdmissionTest.php index 59c3d3445..0d65ddca8 100644 --- a/tests/Source/CancellationScopeAdmissionTest.php +++ b/tests/Source/CancellationScopeAdmissionTest.php @@ -22,6 +22,7 @@ use Workflow\V2\Contracts\CancellationScopeAdmission; use Workflow\V2\Contracts\CancellationScopeTaskBridge; use Workflow\V2\Contracts\PreparedLocalActivityTaskBridge; +use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Models\ActivityAttempt; use Workflow\V2\Models\ActivityExecution; @@ -494,10 +495,11 @@ public function test_admission_is_rechecked_under_the_run_lock_before_stream_eff { $task = $this->claim(); $current = app(WorkflowTaskBridge::class); - $bridge = \Mockery::mock(WorkflowTaskBridge::class.', '.CancellationScopeAdmission::class); + $bridge = \Mockery::mock(ProtocolAwareWorkflowTaskBridge::class.', '.CancellationScopeAdmission::class); $bridge->shouldReceive('status')->andReturnUsing(fn (string $id) => $current->status($id)); $bridge->shouldReceive('validateCancellationScopeMembership')->twice()->andReturn(null, 'operation_scope_not_recorded'); $bridge->shouldNotReceive('complete'); + $bridge->shouldNotReceive('completeForProtocol'); $this->app->instance(WorkflowTaskBridge::class, $bridge); $this->complete($task, 'root')->assertStatus(409)->assertJsonPath('reason', 'operation_scope_not_recorded'); $this->assertDatabaseCount('workflow_durable_stream_items', 0); diff --git a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php index 4e1f06c10..b4bcad08f 100644 --- a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php +++ b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php @@ -12,6 +12,7 @@ use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; use Illuminate\Testing\TestResponse; +use PHPUnit\Framework\Attributes\DataProvider; use ReflectionClass; use Tests\Fixtures\ExternalGreetingWorkflow; use Tests\TestCase; @@ -255,30 +256,34 @@ public function test_failed_post_write_recovery_binding_rolls_back_decision_and_ $this->assertSame($before, $this->state()); } - public function test_total_deadline_preserves_selection_and_engine_timeout_precedence(): void + public static function totalDeadlines(): array { - foreach ([99, 100] as $seconds) { - $this->travelTo(\Illuminate\Support\Carbon::parse('2026-11-06T00:00:00Z')); - $task = $this->claim(); - $prepared = $this->local($task, 'prepare', $this->descriptor([ - 'start_to_close_timeout' => 200, 'schedule_to_close_timeout' => 100, - ]))->assertOk()->json(); - $before = ActivityAttempt::query()->count(); - $this->travel($seconds)->seconds(); - $reply = $this->outcome($task, $prepared['activity_attempt_id'], - ['outcome' => 'failed', 'message' => 'caller failure', 'next_retry_delay_seconds' => 1]) - ->assertOk()->assertJsonPath('recorded', true)->assertJsonPath('created_task_ids', [])->json(); - $this->assertSame('terminal', $reply['retry_disposition']['decision']); - $this->assertSame($seconds === 99 ? 'application' : 'engine_timeout', $reply['retry_disposition']['failure_source']); - if ($seconds === 99) { - $this->assertSame('retry_delay_exceeds_schedule_to_close', $reply['retry_disposition']['terminal_classification']); - } else { - $this->assertSame('schedule_to_close', $reply['retry_disposition']['timeout_kind']); - } - $this->assertSame($before, ActivityAttempt::query()->count()); - $this->assertSame(0, WorkflowHistoryEvent::query()->where('workflow_run_id', $task['run_id']) - ->where('event_type', HistoryEventType::ActivityRetryScheduled)->count()); + return ['before original deadline' => [99], 'at original deadline' => [100]]; + } + + #[DataProvider('totalDeadlines')] + public function test_total_deadline_preserves_selection_and_engine_timeout_precedence(int $seconds): void + { + $task = $this->claim(); + $prepared = $this->local($task, 'prepare', $this->descriptor([ + 'start_to_close_timeout' => 200, 'schedule_to_close_timeout' => 100, + ]))->assertOk()->json(); + $before = ActivityAttempt::query()->count(); + $this->travel($seconds)->seconds(); + WorkerRegistration::query()->where('worker_id', 'retry-worker')->update(['last_heartbeat_at' => now()]); + $reply = $this->outcome($task, $prepared['activity_attempt_id'], + ['outcome' => 'failed', 'message' => 'caller failure', 'next_retry_delay_seconds' => 1]) + ->assertOk()->assertJsonPath('recorded', true)->assertJsonPath('created_task_ids', [])->json(); + $this->assertSame('terminal', $reply['retry_disposition']['decision']); + $this->assertSame($seconds === 99 ? 'application' : 'engine_timeout', $reply['retry_disposition']['failure_source']); + if ($seconds === 99) { + $this->assertSame('retry_delay_exceeds_schedule_to_close', $reply['retry_disposition']['terminal_classification']); + } else { + $this->assertSame('schedule_to_close', $reply['retry_disposition']['timeout_kind']); } + $this->assertSame($before, ActivityAttempt::query()->count()); + $this->assertSame(0, WorkflowHistoryEvent::query()->where('workflow_run_id', $task['run_id']) + ->where('event_type', HistoryEventType::ActivityRetryScheduled)->count()); } private function assertNoAuthorityRows(): void @@ -337,10 +342,9 @@ private function local(array $task, string $operation, ?array $descriptor = null private function outcome(array $task, string $attempt, array $report): TestResponse { - return $this->withHeaders($this->headers())->call('POST', "/api/worker/workflow-tasks/{$task['task_id']}/local-activities/{$attempt}/outcome", - [], [], [], ['CONTENT_TYPE' => 'application/json'], json_encode([ - 'lease_owner' => $task['lease_owner'], 'workflow_task_attempt' => $task['workflow_task_attempt'], 'report' => $report, - ], JSON_THROW_ON_ERROR | JSON_PRESERVE_ZERO_FRACTION)); + return $this->postJson("/api/worker/workflow-tasks/{$task['task_id']}/local-activities/{$attempt}/outcome", [ + 'lease_owner' => $task['lease_owner'], 'workflow_task_attempt' => $task['workflow_task_attempt'], 'report' => $report, + ], $this->headers(), JSON_THROW_ON_ERROR | JSON_PRESERVE_ZERO_FRACTION); } private function headers(): array From 868921c2a14c613559288c4d9d5a2b0d9bb6e561 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 09:12:48 +0000 Subject: [PATCH 6/9] test: prove deadline fixture claim prerequisites --- .../PreparedLocalRetryAdmissionSourceTest.php | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php index b4bcad08f..5626c6bea 100644 --- a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php +++ b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php @@ -7,6 +7,7 @@ use App\Support\CooperativeCancellationPolicy; use App\Support\PreparedLocalActivityPolicy; use App\Support\PreparedLocalRetryAdmission; +use App\Support\WorkerPollFence; use App\Support\WorkerProtocol; use Illuminate\Foundation\Testing\RefreshDatabase; use Illuminate\Support\Facades\DB; @@ -271,9 +272,15 @@ public function test_total_deadline_preserves_selection_and_engine_timeout_prece $before = ActivityAttempt::query()->count(); $this->travel($seconds)->seconds(); WorkerRegistration::query()->where('worker_id', 'retry-worker')->update(['last_heartbeat_at' => now()]); - $reply = $this->outcome($task, $prepared['activity_attempt_id'], - ['outcome' => 'failed', 'message' => 'caller failure', 'next_retry_delay_seconds' => 1]) - ->assertOk()->assertJsonPath('recorded', true)->assertJsonPath('created_task_ids', [])->json(); + $hosting = WorkflowTask::query()->findOrFail($task['task_id']); + $attempt = ActivityAttempt::query()->findOrFail($prepared['activity_attempt_id']); + $this->assertTrue(WorkerPollFence::isCurrent($hosting->payload['_server_workflow_claim']), 'The original registration must still be current.'); + $this->assertTrue($hosting->lease_expires_at->gt(now()), 'The hosting task lease must outlive the activity deadline sample.'); + $this->assertTrue($attempt->lease_expires_at->gt(now()), 'The callback lease must outlive the activity deadline sample.'); + $response = $this->outcome($task, $prepared['activity_attempt_id'], + ['outcome' => 'failed', 'message' => 'caller failure', 'next_retry_delay_seconds' => 1]); + $this->assertSame(200, $response->status(), $response->getContent()); + $reply = $response->assertJsonPath('recorded', true)->assertJsonPath('created_task_ids', [])->json(); $this->assertSame('terminal', $reply['retry_disposition']['decision']); $this->assertSame($seconds === 99 ? 'application' : 'engine_timeout', $reply['retry_disposition']['failure_source']); if ($seconds === 99) { From b24082cb542310b6ba0a4390119f4b6d085669d1 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 09:38:09 +0000 Subject: [PATCH 7/9] test: use a coherent prepared deadline fixture --- .../PreparedLocalRetryAdmissionSourceTest.php | 24 ++++++++++++++----- 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php index 5626c6bea..8af63c496 100644 --- a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php +++ b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php @@ -265,20 +265,22 @@ public static function totalDeadlines(): array #[DataProvider('totalDeadlines')] public function test_total_deadline_preserves_selection_and_engine_timeout_precedence(int $seconds): void { + config(['server.workers.stale_after_seconds' => 300]); $task = $this->claim(); - $prepared = $this->local($task, 'prepare', $this->descriptor([ - 'start_to_close_timeout' => 200, 'schedule_to_close_timeout' => 100, - ]))->assertOk()->json(); + $preparation = $this->local($task, 'prepare', $this->descriptor([ + 'start_to_close_timeout' => 100, 'schedule_to_close_timeout' => 100, + ])); + $this->assertSame(200, $preparation->status(), $preparation->getContent()); + $prepared = $preparation->json(); $before = ActivityAttempt::query()->count(); $this->travel($seconds)->seconds(); - WorkerRegistration::query()->where('worker_id', 'retry-worker')->update(['last_heartbeat_at' => now()]); $hosting = WorkflowTask::query()->findOrFail($task['task_id']); $attempt = ActivityAttempt::query()->findOrFail($prepared['activity_attempt_id']); $this->assertTrue(WorkerPollFence::isCurrent($hosting->payload['_server_workflow_claim']), 'The original registration must still be current.'); $this->assertTrue($hosting->lease_expires_at->gt(now()), 'The hosting task lease must outlive the activity deadline sample.'); $this->assertTrue($attempt->lease_expires_at->gt(now()), 'The callback lease must outlive the activity deadline sample.'); - $response = $this->outcome($task, $prepared['activity_attempt_id'], - ['outcome' => 'failed', 'message' => 'caller failure', 'next_retry_delay_seconds' => 1]); + $report = ['outcome' => 'failed', 'message' => 'caller failure', 'next_retry_delay_seconds' => 1]; + $response = $this->outcome($task, $prepared['activity_attempt_id'], $report); $this->assertSame(200, $response->status(), $response->getContent()); $reply = $response->assertJsonPath('recorded', true)->assertJsonPath('created_task_ids', [])->json(); $this->assertSame('terminal', $reply['retry_disposition']['decision']); @@ -291,6 +293,16 @@ public function test_total_deadline_preserves_selection_and_engine_timeout_prece $this->assertSame($before, ActivityAttempt::query()->count()); $this->assertSame(0, WorkflowHistoryEvent::query()->where('workflow_run_id', $task['run_id']) ->where('event_type', HistoryEventType::ActivityRetryScheduled)->count()); + if ($seconds === 99) { + $this->travel(20)->seconds(); + WorkerRegistration::query()->where('worker_id', 'retry-worker')->update(['status' => 'superseded']); + $before = $this->state(); + $duplicate = $this->outcome($task, $prepared['activity_attempt_id'], $report); + $this->assertSame(200, $duplicate->status(), $duplicate->getContent()); + $duplicate->assertJsonPath('duplicate', true)->assertJsonPath('event_id', $reply['event_id']); + $this->assertSame($reply['retry_disposition'], $duplicate->json('retry_disposition')); + $this->assertSame($before, $this->state()); + } } private function assertNoAuthorityRows(): void From 44062bb3a5c464f7ffeef49d00b6880a763b93b7 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 10:03:25 +0000 Subject: [PATCH 8/9] test: compare duplicate retry decisions as strict objects --- tests/Source/PreparedLocalRetryAdmissionSourceTest.php | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php index 8af63c496..4ec174c55 100644 --- a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php +++ b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php @@ -300,7 +300,11 @@ public function test_total_deadline_preserves_selection_and_engine_timeout_prece $duplicate = $this->outcome($task, $prepared['activity_attempt_id'], $report); $this->assertSame(200, $duplicate->status(), $duplicate->getContent()); $duplicate->assertJsonPath('duplicate', true)->assertJsonPath('event_id', $reply['event_id']); - $this->assertSame($reply['retry_disposition'], $duplicate->json('retry_disposition')); + $originalDecision = $reply['retry_disposition']; + $duplicateDecision = $duplicate->json('retry_disposition'); + ksort($originalDecision); + ksort($duplicateDecision); + $this->assertSame($originalDecision, $duplicateDecision); $this->assertSame($before, $this->state()); } } From d80a8db418a17ede74bbaf325f68e1c31253e86b Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 10:33:31 +0000 Subject: [PATCH 9/9] fix: refuse preparation receipt backfill on duplicate --- .../Api/PreparedLocalActivityController.php | 2 +- app/Support/PreparedLocalRetryAdmission.php | 6 +++++- docs/contracts/prepared-local-retry-v2.md | 5 +++++ .../PreparedLocalRetryAdmissionSourceTest.php | 15 +++++++++++++++ 4 files changed, 26 insertions(+), 2 deletions(-) diff --git a/app/Http/Controllers/Api/PreparedLocalActivityController.php b/app/Http/Controllers/Api/PreparedLocalActivityController.php index ae91b1109..4979c0446 100644 --- a/app/Http/Controllers/Api/PreparedLocalActivityController.php +++ b/app/Http/Controllers/Api/PreparedLocalActivityController.php @@ -157,7 +157,7 @@ private function admit(Request $request, string $taskId, bool $recover): JsonRes $attemptId = $reply['activity_attempt_id'] ?? null; $attempt = is_string($attemptId) ? ActivityAttempt::query()->find($attemptId) : null; if (! $task instanceof WorkflowTask || ! $attempt instanceof ActivityAttempt - || ! ($retry ? PreparedLocalRetryAdmission::remember($task, $attempt, $claim) + || ! ($retry ? PreparedLocalRetryAdmission::remember($task, $attempt, $claim, $reply) : PreparedLocalActivityPolicy::remember($task, $attempt, $claim))) { throw new PreparedLocalActivityAdmissionRefused; } diff --git a/app/Support/PreparedLocalRetryAdmission.php b/app/Support/PreparedLocalRetryAdmission.php index 68fed5841..065b32255 100644 --- a/app/Support/PreparedLocalRetryAdmission.php +++ b/app/Support/PreparedLocalRetryAdmission.php @@ -107,7 +107,7 @@ public static function reportRefusal(array $report): ?string } /** Persist only after the actual optional bridge prepared under the outer transaction. */ - public static function remember(WorkflowTask $task, ActivityAttempt $attempt, array $claim): bool + public static function remember(WorkflowTask $task, ActivityAttempt $attempt, array $claim, array $reply): bool { if (DB::transactionLevel() < 1 || ! self::backendSupported() || PreparedLocalActivityPolicy::currentClaim($task) !== $claim || ! self::claimSupports($claim)) { @@ -126,6 +126,10 @@ public static function remember(WorkflowTask $task, ActivityAttempt $attempt, ar // A v1 receipt, or a changed v2 graph, is never rewritten. return self::originalClaim($task, $attempt, $claim['lease_owner'], $claim['attempt']) === $claim; } + if (($reply['prepared'] ?? null) !== true || ($reply['duplicate'] ?? null) !== false) { + // A replay cannot repair missing host admission authority. + return false; + } $receipts[$attempt->id] = ['schema' => self::SCHEMA, 'claim' => $claim, 'activity_execution_id' => $attempt->activity_execution_id, 'activity_attempt_id' => $attempt->id, 'worker_attempt_id' => $attempt->worker_attempt_id, diff --git a/docs/contracts/prepared-local-retry-v2.md b/docs/contracts/prepared-local-retry-v2.md index 7e938148c..806eaef93 100644 --- a/docs/contracts/prepared-local-retry-v2.md +++ b/docs/contracts/prepared-local-retry-v2.md @@ -177,6 +177,11 @@ unique correlated Started event must agree on every read. A legacy receipt is never rewritten or promoted. The hosting task's later current claim cannot replace the original callback's issued claim. +A new private preparation receipt also requires the bound role's literal +`prepared: true` and `duplicate: false` result. A duplicate can only read an +already retained matching receipt. It cannot recreate missing Server authority +from an existing Core graph. + A fresh recovery also retains a separate closed private `durable-workflow.server.prepared-local-recovery-authority/v2` receipt on the recovering task. Its keys are `schema`, `claim`, `activity_execution_id`, diff --git a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php index 4ec174c55..1632c515f 100644 --- a/tests/Source/PreparedLocalRetryAdmissionSourceTest.php +++ b/tests/Source/PreparedLocalRetryAdmissionSourceTest.php @@ -121,6 +121,21 @@ public function test_supplied_override_and_other_report_errors_refuse_before_row ['outcome' => 'failed', 'message' => []])->assertStatus(404)->assertJsonPath('reason', 'activity_attempt_not_found'); } + public function test_preparation_duplicate_cannot_recreate_missing_original_authority(): void + { + $task = $this->claim(); + $prepared = $this->local($task, 'prepare')->assertOk()->json(); + $stored = WorkflowTask::query()->findOrFail($task['task_id']); + $payload = $stored->payload; + unset($payload['_server_prepared_local_activity_claims'][$prepared['activity_attempt_id']]); + $stored->forceFill(['payload' => $payload])->save(); + $before = $this->state(); + + $this->local($task, 'prepare')->assertStatus(409)->assertJsonPath('reason', 'prepared_local_activity_not_supported'); + $this->assertSame($before, $this->state()); + $this->assertArrayNotHasKey($prepared['activity_attempt_id'], $stored->fresh()->payload['_server_prepared_local_activity_claims']); + } + public function test_outcome_duplicate_preserves_the_committed_decision_after_time_and_registration_change(): void { $task = $this->claim();